diff --git a/.agent/docs/PULSE_AI_ARCHITECTURE.md b/.agent/docs/PULSE_AI_ARCHITECTURE.md deleted file mode 100644 index b7ec596..0000000 --- a/.agent/docs/PULSE_AI_ARCHITECTURE.md +++ /dev/null @@ -1,415 +0,0 @@ -# Pulse AI Architecture: Long-Term Vision - -## The Core Problem - -Pulse AI currently provides "AI that can talk to your infrastructure." But this is becoming commodity. Any user can: -1. Install Claude Code / Cursor / Windsurf -2. Give it SSH access to their Proxmox nodes -3. Ask "What's wrong with my infrastructure?" - -**We need to provide value that a stateless AI session cannot.** - ---- - -## The Fundamental Insight - -A stateless AI with SSH access can answer: **"What is the current state?"** - -Pulse, with its continuous monitoring, can answer: -- **"How has this changed over time?"** -- **"What does 'normal' look like for YOUR infrastructure?"** -- **"What's about to go wrong?"** -- **"Have we seen this pattern before?"** -- **"What did you do last time this happened?"** - -These require **persistent context** that accumulates over time. This is our moat. - ---- - -## Architecture Principles - -### 1. Context is King - -The AI is only as useful as the context we provide. We should think of Pulse as a **context accumulation engine** that happens to have an AI interface. - -Every piece of data Pulse collects should be available to the AI in a digestible form: -- Real-time metrics -- Historical trends -- User annotations -- Alert history -- Previous AI findings -- Configuration changes -- Remediation history - -### 2. Time-Aware Intelligence - -The AI should always know: -- What's happening **now** -- What happened **before** (trends, history) -- What will likely happen **next** (forecasts) -- What's **different** from normal (anomalies) - -### 3. Learning From Operations - -Every interaction with Pulse teaches it about the user's infrastructure: -- Dismissed findings → "This is expected behavior" -- User notes → "This VM runs the critical database" -- Alert patterns → "This resource is flaky on Tuesdays" -- Remediation actions → "Last time this happened, we restarted the service" - -### 4. Proactive, Not Just Reactive - -The goal isn't just to answer questions. It's to: -- Surface problems before users ask -- Predict capacity issues weeks in advance -- Notice patterns humans would miss -- Remember what humans would forget - ---- - -## Data Architecture - -### Layer 1: Real-Time State (Already Have) - -``` -StateSnapshot -├── Nodes[] -├── VMs[] -├── Containers[] -├── Storage[] -├── DockerHosts[] -├── PBSInstances[] -├── Hosts[] -└── PMGInstances[] -``` - -This is what we send to the AI today. Point-in-time. Commodity. - -### Layer 2: Historical Metrics (Partially Have) - -``` -MetricsHistory -├── NodeMetrics[nodeID] → {CPU[], Memory[], Disk[]} over time -├── GuestMetrics[guestID] → {CPU[], Memory[], Network[]} over time -└── StorageMetrics[storageID] → {Usage[], Used[], Total[]} over time -``` - -We collect this for the frontend trendlines, but **don't expose it to the AI**. - -### Layer 3: Computed Insights (Need to Build) - -``` -InsightsStore -├── Trends[resourceID] → {direction, rate_of_change, forecast} -├── Baselines[resourceID] → {normal_cpu_range, normal_memory_range, typical_patterns} -├── Anomalies[resourceID] → {current_deviations, severity} -├── Correlations[] → {resource_a, resource_b, relationship} -└── Predictions[] → {resource, metric, predicted_event, eta} -``` - -This is computed from historical data and provides **derived intelligence**. - -### Layer 4: Operational Memory (Partially Have) - -``` -OperationalMemory -├── Findings[findingID] → {status, user_response, resolution} -├── Knowledge[guestID] → {user_notes, learned_facts} -├── AlertHistory[] → {alert, duration, resolution, user_action} -├── RemediationLog[] → {problem, action_taken, outcome, timestamp} -└── ChangeLog[] → {resource, what_changed, when, detected_impact} -``` - -This captures **what happened and how it was handled**. - ---- - -## The AI Context Pipeline - -When the AI needs context (for chat, patrol, or alert analysis), we build it in layers: - -``` -┌─────────────────────────────────────────────────────────────┐ -│ CONTEXT ASSEMBLY │ -├─────────────────────────────────────────────────────────────┤ -│ │ -│ 1. CURRENT STATE (required) │ -│ - Real-time metrics for relevant resources │ -│ - Current alerts and their status │ -│ │ -│ 2. HISTORICAL CONTEXT (high value) │ -│ - Trends: "Memory has been growing 3%/day for 5 days" │ -│ - Baselines: "Normal CPU for this VM is 5-15%" │ -│ - Anomalies: "Current 45% is 3σ above normal" │ -│ │ -│ 3. OPERATIONAL CONTEXT (essential for continuity) │ -│ - Previous findings for this resource │ -│ - User notes: "This is the production database" │ -│ - Past remediations: "We increased RAM last month" │ -│ │ -│ 4. PREDICTIVE CONTEXT (proactive value) │ -│ - Forecasts: "At current rate, disk full in 12 days" │ -│ - Pattern alerts: "This usually fails after X" │ -│ - Correlations: "When A spikes, B usually follows" │ -│ │ -│ 5. USER CONTEXT (personalization) │ -│ - Infrastructure notes: "This is a homelab" │ -│ - Preferences: "I prefer conservative recommendations" │ -│ - Expertise level: "User is comfortable with CLI" │ -│ │ -└─────────────────────────────────────────────────────────────┘ -``` - ---- - -## Implementation Status - -### ✅ Phase 1: Historical Context Integration (COMPLETE) - -**Implemented in `internal/ai/context/` package:** - -- `builder.go` - Context builder with trend and prediction integration -- `formatter.go` - Format resources with metrics for AI consumption -- `trends.go` - Linear regression for trend direction and rate of change - -**Features:** -- Trend computation (growing/declining/stable/volatile) -- 24h and 7d trend summaries -- Rate of change calculations -- Integrated into patrol and chat via `buildEnrichedContext()` - -### ✅ Phase 2: Anomaly Detection (COMPLETE) - -**Implemented in `internal/ai/baseline/` package:** - -- `store.go` - Statistical baseline learning and anomaly detection - -**Features:** -- Rolling statistics per resource (mean, stddev, percentiles) -- Z-score based anomaly severity (low/medium/high/critical) -- Persists baselines to disk (`ai_baselines.json`) -- Background learning loop (hourly updates) -- 7-day learning window with minimum sample requirements - -### ✅ Phase 3: Operational Memory (COMPLETE) - -**Implemented in `internal/ai/memory/` package:** - -- `changes.go` - Change detection for infrastructure changes -- `remediation.go` - Remediation action logging - -**Change Detection tracks:** -- Resource creation/deletion -- Status changes (started, stopped) -- VM/container migrations between nodes -- CPU/memory configuration changes -- Backup completions - -**Remediation logging records:** -- Command executed and output -- Problem being addressed -- Linked finding ID (if any) -- Outcome (resolved/partial/failed/unknown) -- Automatic vs manual distinction - -### ✅ Phase 4: Remediation Integration (COMPLETE) - -**AI now learns from past fixes:** - -- Commands logged to remediation log after execution -- System prompts include "Past Successful Fixes for Similar Issues" -- System prompts include "Remediation History for This Resource" -- Keyword matching finds relevant past solutions - -**Example AI context now includes:** -```markdown -## Past Successful Fixes for Similar Issues -These actions worked for similar problems before: -- **High memory usage causing slo...**: `apt clean && apt autoremove` (resolved) - -## Remediation History for This Resource -- 2 hours ago: Memory at 95% → `systemctl restart nginx` (resolved) -- 1 day ago: Disk full warning → `journalctl --vacuum-time=1d` (resolved) -``` - ---- - -## Next Steps - -### ✅ Phase 5: Predictive Intelligence (COMPLETE) - -**Implemented in `internal/ai/patterns/` package:** - -- `detector.go` - Pattern detector for failure prediction - -**Features:** -1. **Capacity Forecasting** ✅ - - Extrapolate growth trends - - "Storage will be full in X days at current rate" - -2. **Failure Prediction** ✅ - - Track historical events (high memory, OOM, restarts, etc.) - - Detect recurring patterns with interval analysis - - Calculate confidence based on pattern consistency - - Predict next occurrence time - - Persists to `ai_patterns.json` - -3. **Alert History Integration** ✅ - - Callback system in `alerts.HistoryManager` - - Every alert is recorded as a historical event - - Pattern detector learns from production alerts - -**Example AI context now includes:** -```markdown -## ⏰ Failure Predictions -Based on historical patterns: -- high memory usage typically occurs every ~7 days (next expected in ~3 days) -- OOM events typically occurs every ~14 days (last: 12 days ago, overdue) -``` - -### ✅ Phase 6: Multi-Resource Correlation (COMPLETE) - -**Implemented in `internal/ai/correlation/` package:** - -- `detector.go` - Correlation detector for multi-resource relationships - -**Features:** -1. **Automatic Correlation Detection** ✅ - - Tracks events across resources - - Detects temporal relationships (when A happens, B follows) - - Calculates average delay between correlated events - - Confidence scoring based on occurrence count - -2. **Dependency Mapping** ✅ - - GetDependencies() - What resources depend on this one - - GetDependsOn() - What this resource depends on - - Inferred from observed event patterns - -3. **Cascade Analysis** ✅ - - PredictCascade() - Predict downstream effects - - "If storage goes critical, database VM may restart within 5 minutes" - -**Persistence:** `ai_correlations.json` - -**Example AI context now includes:** -```markdown -## 🔗 Resource Correlations -Observed relationships between resources: -- When local-zfs experiences disk_full, database often follows within 5 minutes -- When node-1 has high CPU, vm-100 experiences high memory within 3 minutes -``` - ---- - -## AI Prompt Structure - -With this architecture, a typical AI prompt would look like: - -```markdown -# Infrastructure Analysis Request - -## Target Resource -VM 'database' (ID: 102, Node: pve-main) - -## Current State -- Status: running -- CPU: 78% (normal: 15-30%) -- Memory: 92% (normal: 60-75%) -- Disk: 67% (stable) -- Uptime: 45 days - -## Historical Context (7 days) -- Memory: Growing +2.1%/day (was 77% 7 days ago) -- CPU: Elevated since 3 days ago (was 20%) -- Pattern: No daily cycles detected, continuous growth - -## Anomaly Score: HIGH -- Memory 2.8σ above baseline -- CPU 3.1σ above baseline -- Combined anomaly score: 87/100 - -## Operational History -- Last issue: 3 months ago, high memory (user added swap, resolved) -- User notes: "Production PostgreSQL, critical, no downtime allowed" -- Related resources: Depends on storage 'ceph-ssd', accessed by VMs 105, 107, 112 - -## Recent Changes -- 4 days ago: VM 105 ('app-server') was updated -- 3 days ago: This VM's CPU started increasing - -## Predictions -- At current rate, memory will hit 100% in ~4 days -- Similar pattern to last incident (high memory leading to OOM) - -## User Question -"Why is my database server slow?" -``` - -**This context is impossible to replicate with a stateless SSH session.** - ---- - -## Success Metrics - -How do we know Pulse AI is providing value? - -1. **Predictive Accuracy** - - Did our capacity forecasts come true? - - Did predicted failures occur? - -2. **Time to Resolution** - - How long from problem detection to resolution? - - Compare AI-assisted vs. manual - -3. **Proactive Catches** - - Problems found by patrol before user noticed - - Predictions that led to preventive action - -4. **User Engagement** - - Are users adding notes? (means they trust the system) - - Are they dismissing findings with reasons? (feedback loop) - - Repeat usage of chat feature - -5. **Context Utilization** - - Is the AI using historical context in responses? - - Are predictions being cited in findings? - ---- - -## Technical Considerations - -### Data Retention -- Short-term (24h): High-resolution metrics for immediate analysis -- Medium-term (7-30d): Hourly aggregates for trend analysis -- Long-term (90d+): Daily summaries for baseline/pattern learning - -### Performance -- Context building must be fast (<100ms) -- Precompute expensive analytics (trends, baselines) on schedule -- Cache formatted context, invalidate on significant changes - -### Storage -- Baselines and insights are small, store in SQLite or JSON -- Historical metrics can grow; implement rollup/aggregation -- Consider time-series database for scale (InfluxDB, TimescaleDB) - -### Privacy -- All data stays local (no cloud sync of infrastructure data) -- AI context is built locally, only prompts go to API -- User controls what context is included - ---- - -## Summary - -The path to differentiating Pulse AI: - -| Today | Tomorrow | -|-------|----------| -| "Here's your current state" | "Here's what's changed and why it matters" | -| "This metric is high" | "This is unusual for YOUR infrastructure" | -| "You should check X" | "Last time this happened, you did Y and it worked" | -| "Something might be wrong" | "X will fail in 5 days if this continues" | -| Stateless queries | Accumulated operational intelligence | - -**The AI becomes more valuable the longer Pulse runs.** This is the moat. diff --git a/.agent/docs/PULSE_AI_IMPLEMENTATION_PLAN.md b/.agent/docs/PULSE_AI_IMPLEMENTATION_PLAN.md deleted file mode 100644 index c91e3d7..0000000 --- a/.agent/docs/PULSE_AI_IMPLEMENTATION_PLAN.md +++ /dev/null @@ -1,709 +0,0 @@ -# Pulse AI Implementation Plan - -This document outlines the concrete implementation steps to realize the Pulse AI vision. - ---- - -## Current State Audit - -### What We Have - -| Component | Location | Status | -|-----------|----------|--------| -| Real-time state | `models.StateSnapshot` | ✅ Complete | -| Metrics collection | `monitoring.MetricsHistory` | ✅ Collecting, exposed to AI | -| Finding persistence | `ai.FindingsStore` | ✅ Works | -| Knowledge store | `ai/knowledge.Store` | ✅ Per-guest notes | -| Alert context | `ai.buildAlertContext()` | ✅ Current alerts only | -| User annotations | `buildUserAnnotationsContext()` | ✅ Basic | -| Base patrol | `patrol.go` | ✅ Heuristics + optional AI | -| **AI Context package** | `ai/context/` | ✅ **NEW - Phase 1** | -| **Trend computation** | `ai/context/trends.go` | ✅ **NEW - Linear regression** | -| **Context builder** | `ai/context/builder.go` | ✅ **NEW - Orchestration** | -| **Metrics adapter** | `ai/metrics_history_adapter.go` | ✅ **NEW - Wiring** | - -### What's Missing - -| Component | Impact | Priority | Status | -|-----------|--------|----------|--------| -| Historical context for AI | Core differentiator | P0 | ✅ Done | -| Trend computation | Predictive capability | P0 | ✅ Done | -| Baseline learning | Anomaly detection | P1 | 🔲 Next | -| Change detection | Root cause analysis | P1 | 🔲 Planned | -| Remediation logging | Operational memory | P2 | 🔲 Planned | -| Correlation engine | Advanced insights | P2 | 🔲 Future | -| Capacity forecasting | Proactive alerts | P1 | ⚡ Partial (storage predictions) | - ---- - -## Phase 1: Foundation - AI Context Package - -**Goal**: Create a clean abstraction for building AI context with historical data. - -### 1.1 New Package Structure - -``` -internal/ai/context/ -├── builder.go # Main context builder orchestrator -├── current.go # Current state formatting (refactor from patrol) -├── historical.go # Historical metrics integration -├── trends.go # Trend computation -├── insights.go # Combined insights (anomalies, predictions) -├── formatter.go # AI-friendly text formatting -└── types.go # Shared types -``` - -### 1.2 Core Types - -```go -// types.go - -// ResourceContext contains all context for a single resource -type ResourceContext struct { - ResourceID string - ResourceType string // "node", "vm", "container", "storage", "docker_host" - ResourceName string - - // Current state - Current CurrentState - - // Historical analysis - Trends map[string]Trend // metric -> trend - Baselines map[string]Baseline // metric -> baseline - Anomalies []Anomaly - - // Operational memory - PastFindings []FindingSummary - UserNotes []string - RecentChanges []Change - LastRemediation *RemediationRecord -} - -// Trend represents the direction and rate of change for a metric -type Trend struct { - Metric string - Direction TrendDirection // stable, growing, declining, volatile - RatePerHour float64 // rate of change per hour - RatePerDay float64 // rate of change per day - Current float64 - Average24h float64 - Average7d float64 - Min24h float64 - Max24h float64 - DataPoints int // how much history we have - Confidence float64 // 0-1, based on data quality -} - -type TrendDirection string -const ( - TrendStable TrendDirection = "stable" - TrendGrowing TrendDirection = "growing" - TrendDeclining TrendDirection = "declining" - TrendVolatile TrendDirection = "volatile" -) - -// Baseline represents learned "normal" for a metric -type Baseline struct { - Metric string - Mean float64 - StdDev float64 - P5 float64 // 5th percentile - P95 float64 // 95th percentile - SampleSize int - LearnedAt time.Time -} - -// Anomaly represents a detected deviation from normal -type Anomaly struct { - Metric string - Current float64 - Expected float64 // baseline mean - Deviation float64 // standard deviations from mean - Severity string // "low", "medium", "high", "critical" - Since time.Time - Description string -} - -// Prediction represents a forecasted event -type Prediction struct { - ResourceID string - Metric string - Event string // "capacity_full", "oom", "pattern_repeat" - ETA time.Time - Confidence float64 - Basis string // explanation of prediction -} -``` - -### 1.3 Context Builder - -```go -// builder.go - -type ContextBuilder struct { - stateProvider StateProvider - metricsHistory *monitoring.MetricsHistory - findingsStore *FindingsStore - knowledgeStore *knowledge.Store - baselineStore *BaselineStore - - // Configuration - includeTrends bool - includeBaselines bool - includeHistory bool - historicalWindow time.Duration -} - -// BuildForResource creates comprehensive context for a single resource -func (b *ContextBuilder) BuildForResource(resourceID string) (*ResourceContext, error) - -// BuildForInfrastructure creates summarized context for all infrastructure -func (b *ContextBuilder) BuildForInfrastructure() (*InfrastructureContext, error) - -// FormatForAI converts context to AI-consumable markdown -func (b *ContextBuilder) FormatForAI(ctx *ResourceContext) string - -// FormatInfrastructureForAI converts full infrastructure context -func (b *ContextBuilder) FormatInfrastructureForAI(ctx *InfrastructureContext) string -``` - -### 1.4 Trend Computation - -```go -// trends.go - -// ComputeTrend calculates trend from historical data points -func ComputeTrend(points []monitoring.MetricPoint, window time.Duration) Trend { - if len(points) < 2 { - return Trend{Confidence: 0} - } - - // Calculate basic statistics - avg, min, max, stddev := computeStats(points) - - // Linear regression for direction and rate - slope, r2 := linearRegression(points) - - // Classify direction - direction := classifyTrend(slope, stddev, avg) - - // Rate per hour/day - ratePerHour := slope * 3600 // slope is per second - ratePerDay := ratePerHour * 24 - - return Trend{ - Direction: direction, - RatePerHour: ratePerHour, - RatePerDay: ratePerDay, - Current: points[len(points)-1].Value, - Average24h: avg, - Min24h: min, - Max24h: max, - DataPoints: len(points), - Confidence: r2, - } -} - -func classifyTrend(slope, stddev, avg float64) TrendDirection { - // Normalize slope relative to value magnitude - if avg == 0 { - avg = 1 // avoid division by zero - } - normalizedSlope := (slope * 3600) / avg // hourly change as fraction of avg - - // Threshold based on volatility - threshold := 0.01 // 1% per hour is significant - - if stddev/avg > 0.2 { - return TrendVolatile - } - if normalizedSlope > threshold { - return TrendGrowing - } - if normalizedSlope < -threshold { - return TrendDeclining - } - return TrendStable -} -``` - -### 1.5 Integration with Existing Code - -```go -// In patrol.go, replace buildInfrastructureSummary: - -func (p *PatrolService) buildEnrichedContext(state models.StateSnapshot) string { - builder := context.NewBuilder( - p.stateProvider, - p.metricsHistory, - p.findings, - p.knowledgeStore, - p.baselineStore, - ) - - infraCtx, err := builder.BuildForInfrastructure() - if err != nil { - log.Warn().Err(err).Msg("Failed to build enriched context, falling back") - return p.buildBasicSummary(state) - } - - return builder.FormatInfrastructureForAI(infraCtx) -} -``` - ---- - -## Phase 2: Baseline Learning - -**Goal**: Learn what "normal" looks like for each resource so we can detect anomalies. - -### 2.1 Baseline Store - -```go -// internal/ai/baseline/store.go - -type Store struct { - mu sync.RWMutex - baselines map[string]*ResourceBaseline // resourceID -> baselines - - persistence Persistence - - // Configuration - learningWindow time.Duration // how far back to learn from (default: 7 days) - minSamples int // minimum samples needed (default: 100) - updateInterval time.Duration // how often to recompute (default: 1 hour) -} - -type ResourceBaseline struct { - ResourceID string - LastUpdated time.Time - - Metrics map[string]*MetricBaseline // metric name -> baseline -} - -type MetricBaseline struct { - Mean float64 - StdDev float64 - Percentiles map[int]float64 // 5, 25, 50, 75, 95 - SampleCount int - - // Time-of-day patterns (optional, phase 2+) - HourlyMeans [24]float64 -} - -// Learn computes baselines from historical data -func (s *Store) Learn(resourceID string, history *monitoring.MetricsHistory) error - -// GetBaseline returns the baseline for a resource/metric -func (s *Store) GetBaseline(resourceID, metric string) (*MetricBaseline, bool) - -// IsAnomaly checks if a value is anomalous given the baseline -func (s *Store) IsAnomaly(resourceID, metric string, value float64) (bool, float64) -``` - -### 2.2 Background Learning Loop - -```go -// Run as part of patrol service or separate goroutine - -func (s *Store) StartLearningLoop(ctx context.Context, interval time.Duration) { - ticker := time.NewTicker(interval) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - s.updateAllBaselines() - } - } -} - -func (s *Store) updateAllBaselines() { - // Get list of all resources with metrics - resources := s.metricsHistory.GetResourceIDs() - - for _, resourceID := range resources { - if err := s.Learn(resourceID, s.metricsHistory); err != nil { - log.Warn().Err(err).Str("resource", resourceID).Msg("Failed to update baseline") - } - } - - // Persist updated baselines - s.save() -} -``` - -### 2.3 Anomaly Detection - -```go -// internal/ai/anomaly/detector.go - -type Detector struct { - baselineStore *baseline.Store - - // Thresholds - warningThreshold float64 // default: 2.0 std devs - criticalThreshold float64 // default: 3.0 std devs -} - -type Detection struct { - ResourceID string - Metric string - CurrentValue float64 - ExpectedMean float64 - StdDev float64 - ZScore float64 - Severity AnomalySeverity - DetectedAt time.Time -} - -func (d *Detector) Check(resourceID, metric string, value float64) *Detection { - baseline, ok := d.baselineStore.GetBaseline(resourceID, metric) - if !ok || baseline.SampleCount < 50 { - return nil // not enough data yet - } - - zScore := (value - baseline.Mean) / baseline.StdDev - absZ := math.Abs(zScore) - - if absZ < d.warningThreshold { - return nil // within normal range - } - - severity := AnomalyWarning - if absZ >= d.criticalThreshold { - severity = AnomalyCritical - } - - return &Detection{ - ResourceID: resourceID, - Metric: metric, - CurrentValue: value, - ExpectedMean: baseline.Mean, - StdDev: baseline.StdDev, - ZScore: zScore, - Severity: severity, - DetectedAt: time.Now(), - } -} -``` - ---- - -## Phase 3: Operational Memory - -**Goal**: Remember what happened, what users said, and what worked. - -### 3.1 Change Detection - -```go -// internal/ai/memory/changes.go - -type ChangeDetector struct { - previousState map[string]ResourceSnapshot - mu sync.RWMutex - - changes []Change - maxChanges int - persistence Persistence -} - -type Change struct { - ID string - ResourceID string - ChangeType ChangeType - Before interface{} - After interface{} - DetectedAt time.Time - Description string -} - -type ChangeType string -const ( - ChangeCreated ChangeType = "created" - ChangeDeleted ChangeType = "deleted" - ChangeConfig ChangeType = "config" // RAM, CPU allocation changed - ChangeStatus ChangeType = "status" // started, stopped - ChangeMigrated ChangeType = "migrated" // moved to different node -) - -func (d *ChangeDetector) Detect(current models.StateSnapshot) []Change { - // Compare current state to previous - // Detect new resources, deleted resources, config changes - // Store changes and return new ones -} -``` - -### 3.2 Remediation Logging - -```go -// internal/ai/memory/remediation.go - -type RemediationLog struct { - mu sync.RWMutex - records []RemediationRecord - - persistence Persistence -} - -type RemediationRecord struct { - ID string - Timestamp time.Time - ResourceID string - FindingID string // linked AI finding if any - Problem string // what was wrong - Action string // what was done - Outcome Outcome // did it work? - Duration time.Duration // how long until resolved - Note string // optional user/AI note -} - -type Outcome string -const ( - OutcomeResolved Outcome = "resolved" - OutcomePartial Outcome = "partial" - OutcomeFailed Outcome = "failed" - OutcomeUnknown Outcome = "unknown" -) - -// Log records a remediation action -func (r *RemediationLog) Log(record RemediationRecord) error - -// GetForResource returns remediation history for a resource -func (r *RemediationLog) GetForResource(resourceID string, limit int) []RemediationRecord - -// GetSimilar finds similar past remediations -func (r *RemediationLog) GetSimilar(problem string, limit int) []RemediationRecord -``` - -### 3.3 Integration Points - -When the AI executes a command: -```go -func (s *Service) onToolComplete(toolID, command, output string, success bool) { - // Log the remediation attempt - s.remediationLog.Log(RemediationRecord{ - ID: uuid.New().String(), - Timestamp: time.Now(), - ResourceID: s.currentContext.TargetID, - FindingID: s.currentContext.FindingID, - Problem: s.currentContext.Problem, - Action: command, - Outcome: outcomeFromSuccess(success), - }) -} -``` - -When a finding is resolved: -```go -func (s *FindingsStore) Resolve(findingID string, auto bool) bool { - // Link to any remediation actions - // Record what was done -} -``` - ---- - -## Phase 4: Capacity Forecasting - -**Goal**: Predict when resources will run out. - -### 4.1 Forecaster - -```go -// internal/ai/forecast/capacity.go - -type CapacityForecaster struct { - metricsHistory *monitoring.MetricsHistory - minDataPoints int // minimum points needed for forecast -} - -type CapacityForecast struct { - ResourceID string - Metric string - CurrentUsage float64 - Limit float64 - - GrowthRate float64 // per day - ETA time.Time // when it hits limit - DaysLeft float64 - Confidence float64 // 0-1 - - // Projection points for visualization - Projection []ProjectionPoint -} - -func (f *CapacityForecaster) Forecast(resourceID, metric string, limit float64) (*CapacityForecast, error) { - points := f.metricsHistory.GetMetrics(resourceID, metric, 7*24*time.Hour) - if len(points) < f.minDataPoints { - return nil, ErrInsufficientData - } - - // Linear regression for growth rate - slope, r2 := linearRegression(points) - if slope <= 0 { - return nil, nil // not growing - } - - current := points[len(points)-1].Value - remaining := limit - current - hoursUntilFull := remaining / (slope * 3600) - - if hoursUntilFull <= 0 { - return nil, nil // already at limit - } - - eta := time.Now().Add(time.Duration(hoursUntilFull) * time.Hour) - - return &CapacityForecast{ - ResourceID: resourceID, - Metric: metric, - CurrentUsage: current, - Limit: limit, - GrowthRate: slope * 86400, // per day - ETA: eta, - DaysLeft: hoursUntilFull / 24, - Confidence: r2, - }, nil -} -``` - -### 4.2 Integration with Patrol - -```go -func (p *PatrolService) generateForecasts(state models.StateSnapshot) []Prediction { - var predictions []Prediction - - // Forecast storage capacity - for _, storage := range state.Storage { - if storage.Total == 0 { - continue - } - forecast, err := p.forecaster.Forecast(storage.ID, "used", float64(storage.Total)) - if err != nil || forecast == nil { - continue - } - - if forecast.DaysLeft < 30 && forecast.Confidence > 0.5 { - predictions = append(predictions, Prediction{ - ResourceID: storage.ID, - Metric: "storage_capacity", - Event: "capacity_full", - ETA: forecast.ETA, - Confidence: forecast.Confidence, - Basis: fmt.Sprintf("Growing %.1f GB/day", forecast.GrowthRate/1e9), - }) - } - } - - // Forecast VM memory (could predict OOM) - // Forecast backup storage growth - // etc. - - return predictions -} -``` - ---- - -## File System Layout (Final) - -``` -internal/ai/ -├── context/ -│ ├── builder.go # Main orchestrator -│ ├── current.go # Current state extraction -│ ├── historical.go # Historical data integration -│ ├── trends.go # Trend computation -│ ├── formatter.go # AI-friendly formatting -│ └── types.go # Shared types -├── baseline/ -│ ├── store.go # Baseline storage and learning -│ ├── persistence.go # Disk persistence -│ └── learning.go # Statistical learning -├── anomaly/ -│ ├── detector.go # Anomaly detection -│ └── types.go -├── forecast/ -│ ├── capacity.go # Capacity forecasting -│ └── patterns.go # Pattern-based prediction -├── memory/ -│ ├── changes.go # Change detection -│ ├── remediation.go # Remediation logging -│ └── persistence.go -├── knowledge/ # (existing) -│ ├── store.go -│ └── store_test.go -├── providers/ # (existing) -├── findings.go # (existing) -├── patrol.go # (existing, will use new context/) -├── service.go # (existing, will use new context/) -└── routing.go # (existing) -``` - ---- - -## Migration Strategy - -### Step 1: Add without changing - -Create new packages (`context/`, `baseline/`, etc.) that work alongside existing code. Don't break anything. - -### Step 2: Wire up to MetricsHistory - -Pass `*monitoring.MetricsHistory` to the AI service at startup. Required for historical context. - -### Step 3: Switch patrol to enriched context - -Replace `buildInfrastructureSummary` with `buildEnrichedContext` behind a feature flag. - -### Step 4: Add baseline learning - -Start computing baselines in background. Initially just store, don't act. - -### Step 5: Enable anomaly annotations - -Add anomaly context to AI prompts. Let AI mention anomalies in findings. - -### Step 6: Add forecasts - -Enable capacity forecasting. Create new finding types for predicted issues. - -### Step 7: Phase out old code - -Remove deprecated methods once new system is stable. - ---- - -## Testing Strategy - -1. **Unit tests** for trend computation, baseline learning, anomaly detection -2. **Integration tests** with mock metrics history -3. **Golden file tests** for AI context formatting (ensure consistent output) -4. **Baseline learning tests** with synthetic time-series data -5. **Forecast accuracy tests** with historical data validation - ---- - -## Success Criteria - -Phase 1 complete when: -- AI prompts include historical trends for all resources -- "24h trend" visible in patrol output - -Phase 2 complete when: -- Baselines computed automatically -- Anomalies flagged in AI context -- "X is unusual" appearing in findings - -Phase 3 complete when: -- Changes detected and logged -- Remediation history queryable -- "Last time this happened..." in AI responses - -Phase 4 complete when: -- Capacity forecasts generated -- "Full in X days" predictions accurate -- Predictive findings created before issues occur diff --git a/frontend-modern/package-lock.json b/frontend-modern/package-lock.json index 6ef332d..b13a360 100644 Binary files a/frontend-modern/package-lock.json and b/frontend-modern/package-lock.json differ diff --git a/frontend-modern/package.json b/frontend-modern/package.json index 9dcc925..9c94d57 100644 --- a/frontend-modern/package.json +++ b/frontend-modern/package.json @@ -27,6 +27,7 @@ }, "dependencies": { "@solidjs/router": "^0.10.10", + "dompurify": "^3.3.1", "lucide-solid": "^0.545.0", "marked": "^17.0.1", "solid-js": "^1.8.0" diff --git a/frontend-modern/src/components/AI/AIChat.tsx b/frontend-modern/src/components/AI/AIChat.tsx index 86537fd..0b0857d 100644 --- a/frontend-modern/src/components/AI/AIChat.tsx +++ b/frontend-modern/src/components/AI/AIChat.tsx @@ -1,5 +1,6 @@ import { Component, Show, createSignal, For, createEffect, createMemo, onMount, Switch, Match } from 'solid-js'; import { marked } from 'marked'; +import DOMPurify from 'dompurify'; import { AIAPI } from '@/api/ai'; import { notificationStore } from '@/stores/notifications'; import { logger } from '@/utils/logger'; @@ -63,12 +64,41 @@ marked.setOptions({ gfm: true, // GitHub Flavored Markdown }); -// Helper to render markdown safely +let domPurifyConfigured = false; +const configureDOMPurify = () => { + if (domPurifyConfigured) return; + domPurifyConfigured = true; + + DOMPurify.addHook('afterSanitizeAttributes', (node) => { + const element = node as Element | null; + if (!element || element.tagName !== 'A') return; + element.setAttribute('target', '_blank'); + element.setAttribute('rel', 'noopener noreferrer'); + }); +}; + +// Helper to render markdown safely with XSS protection +// LLM output should NEVER be trusted - always sanitize before rendering as HTML const renderMarkdown = (content: string): string => { try { - return marked.parse(content) as string; + configureDOMPurify(); + const rawHtml = marked.parse(content) as string; + // Sanitize to prevent XSS from malicious LLM output or injected content + return DOMPurify.sanitize(rawHtml, { + // Allow common formatting tags but block scripts, iframes, etc. + ALLOWED_TAGS: ['p', 'br', 'strong', 'em', 'b', 'i', 'u', 'code', 'pre', 'blockquote', + 'ul', 'ol', 'li', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'a', 'hr', 'table', + 'thead', 'tbody', 'tr', 'th', 'td', 'span', 'div'], + ALLOWED_ATTR: ['href', 'target', 'rel', 'class'], + // Force all links to open in new tab and prevent opener attacks + ADD_ATTR: ['target', 'rel'], + }); } catch { - return content; + // If parsing fails, escape HTML entities as fallback + return content.replace(/[&<>"']/g, (char) => { + const entities: Record = { '&': '&', '<': '<', '>': '>', '"': '"', "'": ''' }; + return entities[char] || char; + }); } }; diff --git a/frontend-modern/src/types/ai.ts b/frontend-modern/src/types/ai.ts index 22a2f63..be28a93 100644 --- a/frontend-modern/src/types/ai.ts +++ b/frontend-modern/src/types/ai.ts @@ -124,6 +124,7 @@ export interface AIExecuteRequest { history?: AIConversationMessage[]; // Previous conversation messages finding_id?: string; // If fixing a patrol finding, the ID to resolve on success model?: string; // Override model for this request (user selection in chat) + use_case?: 'chat' | 'patrol'; // Optional server-side routing/model selection } // Tool execution info @@ -140,6 +141,7 @@ export interface AIExecuteResponse { input_tokens: number; output_tokens: number; tool_calls?: AIToolExecution[]; // Commands that were executed + pending_approvals?: AIStreamApprovalNeededData[]; // Non-streaming approvals } // Streaming event types diff --git a/internal/agentexec/policy.go b/internal/agentexec/policy.go index 46429b7..5236b23 100644 --- a/internal/agentexec/policy.go +++ b/internal/agentexec/policy.go @@ -27,23 +27,23 @@ func DefaultPolicy() *CommandPolicy { p := &CommandPolicy{ AutoApprove: []string{ // System inspection - `^ps\s`, + `^ps(\s|$)`, `^top\s+-bn`, - `^df\s`, - `^free\s`, + `^df(\s|$)`, + `^free(\s|$)`, `^uptime$`, `^hostname$`, - `^uname\s`, + `^uname(\s|$)`, `^cat\s+/proc/`, `^cat\s+/etc/os-release`, - `^lsof\s`, - `^netstat\s`, - `^ss\s`, + `^lsof(\s|$)`, + `^netstat(\s|$)`, + `^ss(\s|$)`, `^ip\s+(addr|route|link)`, `^ifconfig`, `^w$`, `^who$`, - `^last\s`, + `^last(\s|$)`, // Log reading (read-only) `^cat\s+/var/log/`, @@ -82,10 +82,10 @@ func DefaultPolicy() *CommandPolicy { `^lsblk`, `^blkid`, `^fdisk\s+-l`, - `^du\s`, - `^ls\s`, - `^stat\s`, - `^file\s`, + `^du(\s|$)`, + `^ls(\s|$)`, + `^stat(\s|$)`, + `^file(\s|$)`, `^find\s+/.*-size`, // Find large files `^find\s+/.*-mtime`, // Find by modification time `^find\s+/.*-type`, // Find by type @@ -131,6 +131,10 @@ func DefaultPolicy() *CommandPolicy { // Proxmox control `^pct\s+(start|stop|shutdown|reboot|resize|set)\s`, `^qm\s+(start|stop|shutdown|reboot|reset|resize|set)\s`, + + // journalctl maintenance (modifies logs / persistent state) + `^journalctl\s+--vacuum`, + `^journalctl\s+--rotate`, }, Blocked: []string{ @@ -207,6 +211,14 @@ const ( // Evaluate checks a command against the policy func (p *CommandPolicy) Evaluate(command string) PolicyDecision { command = strings.TrimSpace(command) + // Normalize simple sudo prefix so policy applies consistently. + // For complex sudo invocations (sudo flags), keep the command as-is (conservative). + if strings.HasPrefix(command, "sudo ") { + parts := strings.Fields(command) + if len(parts) >= 2 && !strings.HasPrefix(parts[1], "-") { + command = strings.Join(parts[1:], " ") + } + } // Check blocked first (highest priority) for _, re := range p.blockedRe { diff --git a/internal/ai/correlation/detector.go b/internal/ai/correlation/detector.go index 219d680..f722b20 100644 --- a/internal/ai/correlation/detector.go +++ b/internal/ai/correlation/detector.go @@ -5,9 +5,11 @@ package correlation import ( "encoding/json" + "fmt" "os" "path/filepath" "sort" + "strings" "sync" "time" @@ -18,13 +20,13 @@ import ( type EventType string const ( - EventAlert EventType = "alert" // Alert triggered - EventRestart EventType = "restart" // Resource restarted - EventHighCPU EventType = "high_cpu" // CPU spike - EventHighMem EventType = "high_mem" // Memory spike - EventDiskFull EventType = "disk_full" // Disk space critical - EventOffline EventType = "offline" // Resource went offline - EventMigration EventType = "migration" // Resource migrated + EventAlert EventType = "alert" // Alert triggered + EventRestart EventType = "restart" // Resource restarted + EventHighCPU EventType = "high_cpu" // CPU spike + EventHighMem EventType = "high_mem" // Memory spike + EventDiskFull EventType = "disk_full" // Disk space critical + EventOffline EventType = "offline" // Resource went offline + EventMigration EventType = "migration" // Resource migrated ) // Event represents a tracked event for correlation analysis @@ -40,18 +42,18 @@ type Event struct { // Correlation represents a detected relationship between two resources type Correlation struct { - SourceID string `json:"source_id"` // Resource that triggers - SourceName string `json:"source_name"` - SourceType string `json:"source_type"` - TargetID string `json:"target_id"` // Resource that follows - TargetName string `json:"target_name"` - TargetType string `json:"target_type"` - EventPattern string `json:"event_pattern"` // e.g., "high_mem -> restart" - Occurrences int `json:"occurrences"` // Number of times observed - AvgDelay time.Duration `json:"avg_delay"` // Average time between events - Confidence float64 `json:"confidence"` // 0-1 confidence level - LastSeen time.Time `json:"last_seen"` - Description string `json:"description"` + SourceID string `json:"source_id"` // Resource that triggers + SourceName string `json:"source_name"` + SourceType string `json:"source_type"` + TargetID string `json:"target_id"` // Resource that follows + TargetName string `json:"target_name"` + TargetType string `json:"target_type"` + EventPattern string `json:"event_pattern"` // e.g., "high_mem -> restart" + Occurrences int `json:"occurrences"` // Number of times observed + AvgDelay time.Duration `json:"avg_delay"` // Average time between events + Confidence float64 `json:"confidence"` // 0-1 confidence level + LastSeen time.Time `json:"last_seen"` + Description string `json:"description"` } // Detector tracks events and detects correlations between resources @@ -59,13 +61,13 @@ type Detector struct { mu sync.RWMutex events []Event correlations map[string]*Correlation // key: sourceID:targetID:pattern - + // Configuration - maxEvents int + maxEvents int correlationWindow time.Duration // How long after source event to look for target - minOccurrences int // Minimum co-occurrences to form correlation - retentionWindow time.Duration // How long to keep events - + minOccurrences int // Minimum co-occurrences to form correlation + retentionWindow time.Duration // How long to keep events + // Persistence dataDir string } @@ -103,7 +105,7 @@ func NewDetector(cfg Config) *Detector { if cfg.RetentionWindow <= 0 { cfg.RetentionWindow = 30 * 24 * time.Hour } - + d := &Detector{ events: make([]Event, 0), correlations: make(map[string]*Correlation), @@ -113,7 +115,7 @@ func NewDetector(cfg Config) *Detector { retentionWindow: cfg.RetentionWindow, dataDir: cfg.DataDir, } - + // Load existing data if cfg.DataDir != "" { if err := d.loadFromDisk(); err != nil { @@ -123,7 +125,7 @@ func NewDetector(cfg Config) *Detector { Msg("Loaded correlation data from disk") } } - + return d } @@ -131,20 +133,20 @@ func NewDetector(cfg Config) *Detector { func (d *Detector) RecordEvent(event Event) { d.mu.Lock() defer d.mu.Unlock() - + if event.ID == "" { event.ID = generateEventID() } if event.Timestamp.IsZero() { event.Timestamp = time.Now() } - + d.events = append(d.events, event) d.trimEvents() - + // Check for correlations with recent events on OTHER resources d.detectCorrelations(event) - + // Persist asynchronously go func() { if err := d.saveToDisk(); err != nil { @@ -156,7 +158,7 @@ func (d *Detector) RecordEvent(event Event) { // detectCorrelations looks for patterns where this event follows a recent event on another resource func (d *Detector) detectCorrelations(newEvent Event) { cutoff := newEvent.Timestamp.Add(-d.correlationWindow) - + for _, oldEvent := range d.events { // Skip same resource if oldEvent.ResourceID == newEvent.ResourceID { @@ -166,12 +168,12 @@ func (d *Detector) detectCorrelations(newEvent Event) { if oldEvent.Timestamp.Before(cutoff) || oldEvent.Timestamp.After(newEvent.Timestamp) { continue } - + // Found a potential correlation: oldEvent -> newEvent key := correlationKey(oldEvent.ResourceID, newEvent.ResourceID, oldEvent.EventType, newEvent.EventType) pattern := string(oldEvent.EventType) + " -> " + string(newEvent.EventType) delay := newEvent.Timestamp.Sub(oldEvent.Timestamp) - + if existing, ok := d.correlations[key]; ok { // Update existing correlation existing.Occurrences++ @@ -222,9 +224,14 @@ func (d *Detector) formatCorrelationDescription(c *Correlation) string { if targetName == "" { targetName = c.TargetID } - + delayStr := formatDuration(c.AvgDelay) - return "When " + sourceName + " experiences " + string(c.EventPattern[:len(c.EventPattern)/2]) + + sourceEvent := c.EventPattern + if parts := strings.Split(c.EventPattern, " -> "); len(parts) == 2 { + sourceEvent = parts[0] + } + + return "When " + sourceName + " experiences " + sourceEvent + ", " + targetName + " often follows within " + delayStr } @@ -232,19 +239,19 @@ func (d *Detector) formatCorrelationDescription(c *Correlation) string { func (d *Detector) GetCorrelations() []*Correlation { d.mu.RLock() defer d.mu.RUnlock() - + var result []*Correlation for _, c := range d.correlations { if c.Occurrences >= d.minOccurrences && c.Confidence >= 0.3 { result = append(result, c) } } - + // Sort by confidence descending sort.Slice(result, func(i, j int) bool { return result[i].Confidence > result[j].Confidence }) - + return result } @@ -252,14 +259,14 @@ func (d *Detector) GetCorrelations() []*Correlation { func (d *Detector) GetCorrelationsForResource(resourceID string) []*Correlation { d.mu.RLock() defer d.mu.RUnlock() - + var result []*Correlation for _, c := range d.correlations { if (c.SourceID == resourceID || c.TargetID == resourceID) && c.Occurrences >= d.minOccurrences { result = append(result, c) } } - + return result } @@ -268,14 +275,14 @@ func (d *Detector) GetCorrelationsForResource(resourceID string) []*Correlation func (d *Detector) GetDependencies(resourceID string) []string { d.mu.RLock() defer d.mu.RUnlock() - + deps := make(map[string]bool) for _, c := range d.correlations { if c.SourceID == resourceID && c.Occurrences >= d.minOccurrences { deps[c.TargetID] = true } } - + result := make([]string, 0, len(deps)) for dep := range deps { result = append(result, dep) @@ -288,14 +295,14 @@ func (d *Detector) GetDependencies(resourceID string) []string { func (d *Detector) GetDependsOn(resourceID string) []string { d.mu.RLock() defer d.mu.RUnlock() - + deps := make(map[string]bool) for _, c := range d.correlations { if c.TargetID == resourceID && c.Occurrences >= d.minOccurrences { deps[c.SourceID] = true } } - + result := make([]string, 0, len(deps)) for dep := range deps { result = append(result, dep) @@ -307,13 +314,17 @@ func (d *Detector) GetDependsOn(resourceID string) []string { func (d *Detector) PredictCascade(resourceID string, eventType EventType) []CascadePrediction { d.mu.RLock() defer d.mu.RUnlock() - + var predictions []CascadePrediction - + for _, c := range d.correlations { if c.SourceID == resourceID && c.Occurrences >= d.minOccurrences { - // Check if the event pattern starts with the given event type - if len(c.EventPattern) > 0 && EventType(c.EventPattern[:len(string(eventType))]) == eventType { + // Check if the correlation's source event matches the given event type. + sourceEvent := c.EventPattern + if parts := strings.Split(c.EventPattern, " -> "); len(parts) == 2 { + sourceEvent = parts[0] + } + if EventType(sourceEvent) == eventType { predictions = append(predictions, CascadePrediction{ ResourceID: c.TargetID, ResourceName: c.TargetName, @@ -324,12 +335,12 @@ func (d *Detector) PredictCascade(resourceID string, eventType EventType) []Casc } } } - + // Sort by confidence sort.Slice(predictions, func(i, j int) bool { return predictions[i].Confidence > predictions[j].Confidence }) - + return predictions } @@ -350,15 +361,15 @@ func (d *Detector) FormatForContext(resourceID string) string { } else { correlations = d.GetCorrelations() } - + if len(correlations) == 0 { return "" } - + var result string result = "\n## 🔗 Resource Correlations\n" result += "Observed relationships between resources:\n" - + for i, c := range correlations { if i >= 10 { // Limit to 10 correlations result += "\n... and more\n" @@ -370,7 +381,7 @@ func (d *Detector) FormatForContext(resourceID string) string { result += "- " + c.EventPattern + " (" + formatConfidence(c.Confidence) + " confidence)\n" } } - + return result } @@ -380,7 +391,7 @@ func (d *Detector) trimEvents() { if len(d.events) > d.maxEvents { d.events = d.events[len(d.events)-d.maxEvents:] } - + // Remove events older than retention window cutoff := time.Now().Add(-d.retentionWindow) kept := make([]Event, 0, len(d.events)) @@ -397,7 +408,7 @@ func (d *Detector) saveToDisk() error { if d.dataDir == "" { return nil } - + d.mu.RLock() data := struct { Events []Event `json:"events"` @@ -407,18 +418,18 @@ func (d *Detector) saveToDisk() error { Correlations: d.correlations, } d.mu.RUnlock() - + jsonData, err := json.MarshalIndent(data, "", " ") if err != nil { return err } - + path := filepath.Join(d.dataDir, "ai_correlations.json") tmpPath := path + ".tmp" if err := os.WriteFile(tmpPath, jsonData, 0600); err != nil { return err } - + return os.Rename(tmpPath, path) } @@ -427,8 +438,14 @@ func (d *Detector) loadFromDisk() error { if d.dataDir == "" { return nil } - + path := filepath.Join(d.dataDir, "ai_correlations.json") + if st, err := os.Stat(path); err == nil { + const maxOnDiskBytes = 10 << 20 // 10 MiB safety cap + if st.Size() > maxOnDiskBytes { + return fmt.Errorf("correlation history file too large (%d bytes)", st.Size()) + } + } jsonData, err := os.ReadFile(path) if err != nil { if os.IsNotExist(err) { @@ -436,19 +453,30 @@ func (d *Detector) loadFromDisk() error { } return err } - + var data struct { Events []Event `json:"events"` Correlations map[string]*Correlation `json:"correlations"` } - + if err := json.Unmarshal(jsonData, &data); err != nil { return err } - + d.events = data.Events d.correlations = data.Correlations - + if d.correlations == nil { + d.correlations = make(map[string]*Correlation) + } + + d.trimEvents() + cutoff := time.Now().Add(-d.retentionWindow) + for k, v := range d.correlations { + if v == nil || v.LastSeen.Before(cutoff) { + delete(d.correlations, k) + } + } + return nil } diff --git a/internal/ai/memory/changes.go b/internal/ai/memory/changes.go index fa2f445..6cc1197 100644 --- a/internal/ai/memory/changes.go +++ b/internal/ai/memory/changes.go @@ -5,6 +5,7 @@ package memory import ( "encoding/json" + "fmt" "os" "path/filepath" "sort" @@ -18,26 +19,26 @@ import ( type ChangeType string const ( - ChangeCreated ChangeType = "created" // New resource appeared - ChangeDeleted ChangeType = "deleted" // Resource removed - ChangeConfig ChangeType = "config" // Configuration changed (RAM, CPU, etc) - ChangeStatus ChangeType = "status" // Status changed (started, stopped, paused) - ChangeMigrated ChangeType = "migrated" // Moved to different node - ChangeRestarted ChangeType = "restarted" // Resource was restarted - ChangeBackedUp ChangeType = "backed_up" // Backup completed + ChangeCreated ChangeType = "created" // New resource appeared + ChangeDeleted ChangeType = "deleted" // Resource removed + ChangeConfig ChangeType = "config" // Configuration changed (RAM, CPU, etc) + ChangeStatus ChangeType = "status" // Status changed (started, stopped, paused) + ChangeMigrated ChangeType = "migrated" // Moved to different node + ChangeRestarted ChangeType = "restarted" // Resource was restarted + ChangeBackedUp ChangeType = "backed_up" // Backup completed ) // Change represents a detected change to infrastructure type Change struct { - ID string `json:"id"` - ResourceID string `json:"resource_id"` - ResourceType string `json:"resource_type"` // vm, container, node, storage - ResourceName string `json:"resource_name"` - ChangeType ChangeType `json:"change_type"` - Before interface{} `json:"before,omitempty"` - After interface{} `json:"after,omitempty"` - DetectedAt time.Time `json:"detected_at"` - Description string `json:"description"` + ID string `json:"id"` + ResourceID string `json:"resource_id"` + ResourceType string `json:"resource_type"` // vm, container, node, storage + ResourceName string `json:"resource_name"` + ChangeType ChangeType `json:"change_type"` + Before interface{} `json:"before,omitempty"` + After interface{} `json:"after,omitempty"` + DetectedAt time.Time `json:"detected_at"` + Description string `json:"description"` } // ResourceSnapshot captures key attributes for change detection @@ -60,7 +61,7 @@ type ChangeDetector struct { previousState map[string]ResourceSnapshot // resourceID -> snapshot changes []Change maxChanges int - + // Persistence dataDir string } @@ -76,14 +77,14 @@ func NewChangeDetector(cfg ChangeDetectorConfig) *ChangeDetector { if cfg.MaxChanges <= 0 { cfg.MaxChanges = 1000 } - + d := &ChangeDetector{ previousState: make(map[string]ResourceSnapshot), changes: make([]Change, 0), maxChanges: cfg.MaxChanges, dataDir: cfg.DataDir, } - + // Load existing changes from disk if cfg.DataDir != "" { if err := d.loadFromDisk(); err != nil { @@ -92,7 +93,7 @@ func NewChangeDetector(cfg ChangeDetectorConfig) *ChangeDetector { log.Info().Int("count", len(d.changes)).Msg("Loaded change history from disk") } } - + return d } @@ -100,16 +101,16 @@ func NewChangeDetector(cfg ChangeDetectorConfig) *ChangeDetector { func (d *ChangeDetector) DetectChanges(currentSnapshots []ResourceSnapshot) []Change { d.mu.Lock() defer d.mu.Unlock() - + now := time.Now() var newChanges []Change - + // Track which resources we've seen in current snapshot currentIDs := make(map[string]bool) - + for _, current := range currentSnapshots { currentIDs[current.ID] = true - + prev, exists := d.previousState[current.ID] if !exists { // New resource @@ -129,11 +130,11 @@ func (d *ChangeDetector) DetectChanges(currentSnapshots []ResourceSnapshot) []Ch changes := d.detectResourceChanges(prev, current, now) newChanges = append(newChanges, changes...) } - + // Update previous state d.previousState[current.ID] = current } - + // Check for deleted resources for id, prev := range d.previousState { if !currentIDs[id] { @@ -151,12 +152,12 @@ func (d *ChangeDetector) DetectChanges(currentSnapshots []ResourceSnapshot) []Ch delete(d.previousState, id) } } - + // Store new changes if len(newChanges) > 0 { d.changes = append(d.changes, newChanges...) d.trimChanges() - + // Persist asynchronously go func() { if err := d.saveToDisk(); err != nil { @@ -164,14 +165,14 @@ func (d *ChangeDetector) DetectChanges(currentSnapshots []ResourceSnapshot) []Ch } }() } - + return newChanges } // detectResourceChanges checks for changes between two snapshots of the same resource func (d *ChangeDetector) detectResourceChanges(prev, current ResourceSnapshot, now time.Time) []Change { var changes []Change - + // Status change if prev.Status != current.Status { change := Change{ @@ -187,7 +188,7 @@ func (d *ChangeDetector) detectResourceChanges(prev, current ResourceSnapshot, n } changes = append(changes, change) } - + // Node change (migration) if prev.Node != "" && current.Node != "" && prev.Node != current.Node { change := Change{ @@ -203,7 +204,7 @@ func (d *ChangeDetector) detectResourceChanges(prev, current ResourceSnapshot, n } changes = append(changes, change) } - + // CPU change if prev.CPUCores > 0 && current.CPUCores > 0 && prev.CPUCores != current.CPUCores { change := Change{ @@ -219,7 +220,7 @@ func (d *ChangeDetector) detectResourceChanges(prev, current ResourceSnapshot, n } changes = append(changes, change) } - + // Memory change (significant change > 5%) if prev.MemoryBytes > 0 && current.MemoryBytes > 0 { pctChange := float64(current.MemoryBytes-prev.MemoryBytes) / float64(prev.MemoryBytes) @@ -238,10 +239,10 @@ func (d *ChangeDetector) detectResourceChanges(prev, current ResourceSnapshot, n changes = append(changes, change) } } - + // Backup completed - if !prev.LastBackup.IsZero() && !current.LastBackup.IsZero() && - current.LastBackup.After(prev.LastBackup) { + if !prev.LastBackup.IsZero() && !current.LastBackup.IsZero() && + current.LastBackup.After(prev.LastBackup) { change := Change{ ID: generateChangeID(), ResourceID: current.ID, @@ -255,7 +256,7 @@ func (d *ChangeDetector) detectResourceChanges(prev, current ResourceSnapshot, n } changes = append(changes, change) } - + return changes } @@ -263,7 +264,7 @@ func (d *ChangeDetector) detectResourceChanges(prev, current ResourceSnapshot, n func (d *ChangeDetector) GetChangesForResource(resourceID string, limit int) []Change { d.mu.RLock() defer d.mu.RUnlock() - + var result []Change // Iterate in reverse to get most recent first for i := len(d.changes) - 1; i >= 0 && len(result) < limit; i-- { @@ -278,7 +279,7 @@ func (d *ChangeDetector) GetChangesForResource(resourceID string, limit int) []C func (d *ChangeDetector) GetRecentChanges(limit int, since time.Time) []Change { d.mu.RLock() defer d.mu.RUnlock() - + var result []Change for i := len(d.changes) - 1; i >= 0 && len(result) < limit; i-- { if d.changes[i].DetectedAt.After(since) { @@ -294,7 +295,7 @@ func (d *ChangeDetector) GetChangesSummary(since time.Time, maxChanges int) stri if len(changes) == 0 { return "" } - + var result string for _, c := range changes { ago := time.Since(c.DetectedAt) @@ -316,23 +317,23 @@ func (d *ChangeDetector) saveToDisk() error { if d.dataDir == "" { return nil } - + d.mu.RLock() changes := make([]Change, len(d.changes)) copy(changes, d.changes) d.mu.RUnlock() - + path := filepath.Join(d.dataDir, "ai_changes.json") data, err := json.MarshalIndent(changes, "", " ") if err != nil { return err } - + tmpPath := path + ".tmp" if err := os.WriteFile(tmpPath, data, 0600); err != nil { return err } - + return os.Rename(tmpPath, path) } @@ -341,8 +342,14 @@ func (d *ChangeDetector) loadFromDisk() error { if d.dataDir == "" { return nil } - + path := filepath.Join(d.dataDir, "ai_changes.json") + if st, err := os.Stat(path); err == nil { + const maxOnDiskBytes = 10 << 20 // 10 MiB safety cap + if st.Size() > maxOnDiskBytes { + return fmt.Errorf("change history file too large (%d bytes)", st.Size()) + } + } data, err := os.ReadFile(path) if err != nil { if os.IsNotExist(err) { @@ -350,18 +357,19 @@ func (d *ChangeDetector) loadFromDisk() error { } return err } - + var changes []Change if err := json.Unmarshal(data, &changes); err != nil { return err } - + // Sort by time sort.Slice(changes, func(i, j int) bool { return changes[i].DetectedAt.Before(changes[j].DetectedAt) }) - + d.changes = changes + d.trimChanges() return nil } @@ -371,7 +379,7 @@ var changeCounter int64 func generateChangeID() string { changeCounter++ - return time.Now().Format("20060102150405") + "-" + string(rune('0'+changeCounter%10)) + return time.Now().Format("20060102150405") + "-" + intToString(int(changeCounter%1000)) } func formatDuration(d time.Duration) string { @@ -391,7 +399,7 @@ func formatUnit(n int, unit string) string { if n == 1 { return "1 " + unit } - return string(rune('0'+n/10)) + string(rune('0'+n%10)) + " " + unit + "s" + return intToString(n) + " " + unit + "s" } func formatBytes(bytes int64) string { @@ -408,7 +416,7 @@ func formatBytes(bytes int64) string { case bytes >= KB: return formatFloat(float64(bytes)/KB) + " KB" default: - return string(rune('0'+bytes/10)) + string(rune('0'+bytes%10)) + " B" + return intToString(int(bytes)) + " B" } } diff --git a/internal/ai/memory/remediation.go b/internal/ai/memory/remediation.go index 6a89b88..ebf9ecc 100644 --- a/internal/ai/memory/remediation.go +++ b/internal/ai/memory/remediation.go @@ -2,6 +2,7 @@ package memory import ( "encoding/json" + "fmt" "os" "path/filepath" "sort" @@ -308,6 +309,12 @@ func (r *RemediationLog) loadFromDisk() error { } path := filepath.Join(r.dataDir, "ai_remediations.json") + if st, err := os.Stat(path); err == nil { + const maxOnDiskBytes = 10 << 20 // 10 MiB safety cap + if st.Size() > maxOnDiskBytes { + return fmt.Errorf("remediation history file too large (%d bytes)", st.Size()) + } + } data, err := os.ReadFile(path) if err != nil { if os.IsNotExist(err) { @@ -327,6 +334,7 @@ func (r *RemediationLog) loadFromDisk() error { }) r.records = records + r.trimRecords() return nil } @@ -351,7 +359,7 @@ func extractKeywords(text string) []string { // In production, this could use NLP or embeddings var keywords []string var current string - + for _, c := range text { if c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' { current += string(c) @@ -365,7 +373,7 @@ func extractKeywords(text string) []string { if len(current) > 3 { keywords = append(keywords, current) } - + return keywords } @@ -374,7 +382,7 @@ func countMatches(a, b []string) int { for _, s := range b { bSet[s] = true } - + count := 0 for _, s := range a { if bSet[s] { diff --git a/internal/ai/patrol.go b/internal/ai/patrol.go index 23fa0f2..309c6aa 100644 --- a/internal/ai/patrol.go +++ b/internal/ai/patrol.go @@ -1659,12 +1659,28 @@ If everything looks healthy, you can say so briefly without any FINDING blocks.` if autoFix { return basePrompt + ` -AUTO-FIX MODE ENABLED: You may use the run_command tool to attempt automatic remediation of issues you find. Use caution and only fix issues where you are confident the fix is safe. Always log what you're doing.` +AUTO-FIX MODE ENABLED: You may use the run_command tool to attempt automatic remediation of issues you find. + +Safe operations you can perform autonomously: +- Restart services (systemctl restart) +- Clear caches and temp files +- Rotate/compress logs +- Trigger garbage collection + +Operations requiring extra caution: +- Deleting files (prefer moving to /tmp first) +- Installing packages +- Modifying configurations + +Always: +1. Run a verification command after any fix to confirm success +2. Log what action was taken and the outcome +3. Stop and report if the fix doesn't resolve the issue` } return basePrompt + ` -OBSERVE ONLY MODE: You are in observation mode. Gather data using read-only commands (like checking status, memory usage, disk space) to investigate issues, but DO NOT attempt to fix or modify anything. Present your findings for the user to review and action.` +OBSERVE ONLY MODE: You are in observation mode. You may use read-only commands to gather diagnostic information (checking status, memory usage, disk space, logs, etc.) but DO NOT modify anything. Present your findings with clear recommendations for the user to review and action manually.` } // buildInfrastructureSummary creates a text summary of infrastructure state for the AI diff --git a/internal/ai/patterns/detector.go b/internal/ai/patterns/detector.go index 11726d0..ec5027b 100644 --- a/internal/ai/patterns/detector.go +++ b/internal/ai/patterns/detector.go @@ -4,6 +4,7 @@ package patterns import ( "encoding/json" + "fmt" "math" "os" "path/filepath" @@ -18,37 +19,37 @@ import ( type EventType string const ( - EventHighMemory EventType = "high_memory" // Memory exceeded threshold - EventHighCPU EventType = "high_cpu" // CPU exceeded threshold - EventDiskFull EventType = "disk_full" // Disk space critical - EventOOM EventType = "oom" // Out of memory kill - EventRestart EventType = "restart" // Resource restarted - EventUnresponsive EventType = "unresponsive" // Resource became unresponsive - EventBackupFailed EventType = "backup_failed" // Backup job failed + EventHighMemory EventType = "high_memory" // Memory exceeded threshold + EventHighCPU EventType = "high_cpu" // CPU exceeded threshold + EventDiskFull EventType = "disk_full" // Disk space critical + EventOOM EventType = "oom" // Out of memory kill + EventRestart EventType = "restart" // Resource restarted + EventUnresponsive EventType = "unresponsive" // Resource became unresponsive + EventBackupFailed EventType = "backup_failed" // Backup job failed ) // HistoricalEvent represents a recorded event type HistoricalEvent struct { - ID string `json:"id"` - ResourceID string `json:"resource_id"` - EventType EventType `json:"event_type"` - Timestamp time.Time `json:"timestamp"` - Description string `json:"description,omitempty"` - Resolved bool `json:"resolved"` - ResolvedAt time.Time `json:"resolved_at,omitempty"` + ID string `json:"id"` + ResourceID string `json:"resource_id"` + EventType EventType `json:"event_type"` + Timestamp time.Time `json:"timestamp"` + Description string `json:"description,omitempty"` + Resolved bool `json:"resolved"` + ResolvedAt time.Time `json:"resolved_at,omitempty"` Duration time.Duration `json:"duration,omitempty"` // How long it lasted } // Pattern represents a detected recurring pattern type Pattern struct { - ResourceID string `json:"resource_id"` - EventType EventType `json:"event_type"` - Occurrences int `json:"occurrences"` // Number of times event occurred + ResourceID string `json:"resource_id"` + EventType EventType `json:"event_type"` + Occurrences int `json:"occurrences"` // Number of times event occurred AverageInterval time.Duration `json:"average_interval"` // Average time between occurrences - StdDevInterval time.Duration `json:"stddev_interval"` // Standard deviation - LastOccurrence time.Time `json:"last_occurrence"` - NextPredicted time.Time `json:"next_predicted"` // When we expect it to happen again - Confidence float64 `json:"confidence"` // 0-1, based on consistency + StdDevInterval time.Duration `json:"stddev_interval"` // Standard deviation + LastOccurrence time.Time `json:"last_occurrence"` + NextPredicted time.Time `json:"next_predicted"` // When we expect it to happen again + Confidence float64 `json:"confidence"` // 0-1, based on consistency AverageDuration time.Duration `json:"average_duration,omitempty"` // How long events typically last } @@ -68,13 +69,13 @@ type Detector struct { mu sync.RWMutex events []HistoricalEvent patterns map[string]*Pattern // resourceID:eventType -> pattern - + // Configuration maxEvents int minOccurrences int // Minimum occurrences to form a pattern patternWindow time.Duration // How far back to look for patterns predictionLimit time.Duration // How far ahead to predict - + // Persistence dataDir string } @@ -112,7 +113,7 @@ func NewDetector(cfg DetectorConfig) *Detector { if cfg.PredictionLimit <= 0 { cfg.PredictionLimit = 30 * 24 * time.Hour } - + d := &Detector{ events: make([]HistoricalEvent, 0), patterns: make(map[string]*Pattern), @@ -122,7 +123,7 @@ func NewDetector(cfg DetectorConfig) *Detector { predictionLimit: cfg.PredictionLimit, dataDir: cfg.DataDir, } - + // Load existing data if cfg.DataDir != "" { if err := d.loadFromDisk(); err != nil { @@ -132,7 +133,7 @@ func NewDetector(cfg DetectorConfig) *Detector { Msg("Loaded pattern history from disk") } } - + return d } @@ -140,21 +141,26 @@ func NewDetector(cfg DetectorConfig) *Detector { func (d *Detector) RecordEvent(event HistoricalEvent) { d.mu.Lock() defer d.mu.Unlock() - + if event.ID == "" { event.ID = generateEventID() } if event.Timestamp.IsZero() { event.Timestamp = time.Now() } - + d.events = append(d.events, event) d.trimEvents() - + // Recompute pattern for this resource/event type key := patternKey(event.ResourceID, event.EventType) - d.patterns[key] = d.computePattern(event.ResourceID, event.EventType) - + pattern := d.computePattern(event.ResourceID, event.EventType) + if pattern == nil { + delete(d.patterns, key) + } else { + d.patterns[key] = pattern + } + // Persist asynchronously go func() { if err := d.saveToDisk(); err != nil { @@ -169,7 +175,7 @@ func (d *Detector) RecordFromAlert(resourceID string, alertType string, timestam if eventType == "" { return // Not a trackable event type } - + d.RecordEvent(HistoricalEvent{ ResourceID: resourceID, EventType: eventType, @@ -182,23 +188,26 @@ func (d *Detector) RecordFromAlert(resourceID string, alertType string, timestam func (d *Detector) GetPredictions() []FailurePrediction { d.mu.RLock() defer d.mu.RUnlock() - + var predictions []FailurePrediction now := time.Now() - + for _, pattern := range d.patterns { + if pattern == nil { + continue + } // Only predict if pattern has sufficient confidence if pattern.Confidence < 0.3 || pattern.Occurrences < d.minOccurrences { continue } - + // Check if prediction is within our limit if pattern.NextPredicted.Before(now) || pattern.NextPredicted.After(now.Add(d.predictionLimit)) { continue } - + daysUntil := pattern.NextPredicted.Sub(now).Hours() / 24 - + predictions = append(predictions, FailurePrediction{ ResourceID: pattern.ResourceID, EventType: pattern.EventType, @@ -209,12 +218,12 @@ func (d *Detector) GetPredictions() []FailurePrediction { Pattern: pattern, }) } - + // Sort by days until (soonest first) sort.Slice(predictions, func(i, j int) bool { return predictions[i].DaysUntil < predictions[j].DaysUntil }) - + return predictions } @@ -234,9 +243,12 @@ func (d *Detector) GetPredictionsForResource(resourceID string) []FailurePredict func (d *Detector) GetPatterns() map[string]*Pattern { d.mu.RLock() defer d.mu.RUnlock() - + result := make(map[string]*Pattern) for k, v := range d.patterns { + if v == nil { + continue + } result[k] = v } return result @@ -245,7 +257,7 @@ func (d *Detector) GetPatterns() map[string]*Pattern { // computePattern analyzes events to find patterns for a resource/event type func (d *Detector) computePattern(resourceID string, eventType EventType) *Pattern { cutoff := time.Now().Add(-d.patternWindow) - + // Get all events for this resource/type within the window var events []HistoricalEvent for _, e := range d.events { @@ -253,74 +265,83 @@ func (d *Detector) computePattern(resourceID string, eventType EventType) *Patte events = append(events, e) } } - + if len(events) < d.minOccurrences { return nil } - + // Sort by timestamp sort.Slice(events, func(i, j int) bool { return events[i].Timestamp.Before(events[j].Timestamp) }) - + // Calculate intervals between events var intervals []time.Duration var durations []time.Duration - + for i := 1; i < len(events); i++ { interval := events[i].Timestamp.Sub(events[i-1].Timestamp) intervals = append(intervals, interval) - + if events[i-1].Duration > 0 { durations = append(durations, events[i-1].Duration) } } - + if len(intervals) == 0 { return nil } - + // Calculate average and stddev of intervals avgInterval := averageDuration(intervals) stddevInterval := stddevDuration(intervals, avgInterval) - + // Calculate confidence based on consistency // If stddev is low relative to mean, pattern is more reliable consistency := 1.0 if avgInterval > 0 { cv := float64(stddevInterval) / float64(avgInterval) // Coefficient of variation - consistency = 1.0 - math.Min(cv, 1.0) // Higher consistency = lower CV + consistency = 1.0 - math.Min(cv, 1.0) // Higher consistency = lower CV } - + // Adjust confidence based on number of occurrences occurrenceBonus := math.Min(float64(len(events))/10.0, 0.3) confidence := consistency*0.7 + occurrenceBonus - + // Predict next occurrence lastEvent := events[len(events)-1] nextPredicted := lastEvent.Timestamp.Add(avgInterval) - + // Calculate average duration if available var avgDuration time.Duration if len(durations) > 0 { avgDuration = averageDuration(durations) } - + return &Pattern{ - ResourceID: resourceID, - EventType: eventType, - Occurrences: len(events), - AverageInterval: avgInterval, - StdDevInterval: stddevInterval, - LastOccurrence: lastEvent.Timestamp, - NextPredicted: nextPredicted, - Confidence: confidence, - AverageDuration: avgDuration, + ResourceID: resourceID, + EventType: eventType, + Occurrences: len(events), + AverageInterval: avgInterval, + StdDevInterval: stddevInterval, + LastOccurrence: lastEvent.Timestamp, + NextPredicted: nextPredicted, + Confidence: confidence, + AverageDuration: avgDuration, } } // trimEvents removes old events beyond maxEvents func (d *Detector) trimEvents() { + cutoff := time.Now().Add(-d.patternWindow) + kept := d.events[:0] + for _, e := range d.events { + if e.Timestamp.After(cutoff) { + kept = append(kept, e) + } + } + d.events = kept + if len(d.events) > d.maxEvents { d.events = d.events[len(d.events)-d.maxEvents:] } @@ -331,7 +352,7 @@ func (d *Detector) saveToDisk() error { if d.dataDir == "" { return nil } - + d.mu.RLock() data := struct { Events []HistoricalEvent `json:"events"` @@ -341,18 +362,18 @@ func (d *Detector) saveToDisk() error { Patterns: d.patterns, } d.mu.RUnlock() - + jsonData, err := json.MarshalIndent(data, "", " ") if err != nil { return err } - + path := filepath.Join(d.dataDir, "ai_patterns.json") tmpPath := path + ".tmp" if err := os.WriteFile(tmpPath, jsonData, 0600); err != nil { return err } - + return os.Rename(tmpPath, path) } @@ -361,8 +382,14 @@ func (d *Detector) loadFromDisk() error { if d.dataDir == "" { return nil } - + path := filepath.Join(d.dataDir, "ai_patterns.json") + if st, err := os.Stat(path); err == nil { + const maxOnDiskBytes = 10 << 20 // 10 MiB safety cap + if st.Size() > maxOnDiskBytes { + return fmt.Errorf("pattern history file too large (%d bytes)", st.Size()) + } + } jsonData, err := os.ReadFile(path) if err != nil { if os.IsNotExist(err) { @@ -370,19 +397,37 @@ func (d *Detector) loadFromDisk() error { } return err } - + var data struct { Events []HistoricalEvent `json:"events"` Patterns map[string]*Pattern `json:"patterns"` } - + if err := json.Unmarshal(jsonData, &data); err != nil { return err } - + d.events = data.Events - d.patterns = data.Patterns - + d.patterns = make(map[string]*Pattern, len(data.Patterns)) + for k, v := range data.Patterns { + if v == nil { + continue + } + d.patterns[k] = v + } + + d.trimEvents() + cutoff := time.Now().Add(-d.patternWindow) + for k, v := range d.patterns { + if v == nil { + delete(d.patterns, k) + continue + } + if v.Occurrences < d.minOccurrences || v.LastOccurrence.Before(cutoff) { + delete(d.patterns, k) + } + } + return nil } @@ -394,15 +439,15 @@ func (d *Detector) FormatForContext(resourceID string) string { } else { predictions = d.GetPredictions() } - + if len(predictions) == 0 { return "" } - + var result string result = "\n## ⏰ Failure Predictions\n" result += "Based on historical patterns:\n" - + for _, p := range predictions { if len(result) > 2000 { // Limit context size result += "\n... and more\n" @@ -410,7 +455,7 @@ func (d *Detector) FormatForContext(resourceID string) string { } result += "- " + p.Basis + "\n" } - + return result } @@ -488,7 +533,7 @@ func formatPatternBasis(p *Pattern) string { daysInterval := p.AverageInterval.Hours() / 24 daysSinceLast := time.Since(p.LastOccurrence).Hours() / 24 daysUntilNext := p.NextPredicted.Sub(time.Now()).Hours() / 24 - + eventName := string(p.EventType) switch p.EventType { case EventHighMemory: @@ -506,13 +551,13 @@ func formatPatternBasis(p *Pattern) string { case EventBackupFailed: eventName = "backup failures" } - + if daysUntilNext < 0 { - return eventName + " typically occurs every ~" + formatDays(daysInterval) + + return eventName + " typically occurs every ~" + formatDays(daysInterval) + " (last: " + formatDays(daysSinceLast) + " ago, overdue)" } - - return eventName + " typically occurs every ~" + formatDays(daysInterval) + + + return eventName + " typically occurs every ~" + formatDays(daysInterval) + " (next expected in ~" + formatDays(daysUntilNext) + ")" } diff --git a/internal/ai/service.go b/internal/ai/service.go index 7a9388c..fed10a8 100644 --- a/internal/ai/service.go +++ b/internal/ai/service.go @@ -2,12 +2,12 @@ package ai import ( "context" - "encoding/base64" "encoding/json" "fmt" "io" + "net" "net/http" - "path/filepath" + "net/url" "regexp" "strconv" "strings" @@ -48,6 +48,23 @@ type Service struct { // Alert-triggered analysis - token-efficient real-time AI insights alertTriggeredAnalyzer *AlertTriggeredAnalyzer + + limits executionLimits + + modelsCache modelsCache +} + +type executionLimits struct { + chatSlots chan struct{} + patrolSlots chan struct{} +} + +type modelsCache struct { + mu sync.RWMutex + at time.Time + key string + models []providers.ModelInfo + ttl time.Duration } // NewService creates a new AI service @@ -72,9 +89,67 @@ func NewService(persistence *config.ConfigPersistence, agentServer *agentexec.Se policy: agentexec.DefaultPolicy(), knowledgeStore: knowledgeStore, costStore: costStore, + limits: executionLimits{ + chatSlots: make(chan struct{}, 4), + patrolSlots: make(chan struct{}, 1), + }, + modelsCache: modelsCache{ + ttl: 5 * time.Minute, + }, } } +func (s *Service) acquireExecutionSlot(ctx context.Context, useCase string) (func(), error) { + normalized := strings.TrimSpace(strings.ToLower(useCase)) + if normalized == "" { + normalized = "chat" + } + + var slots chan struct{} + if normalized == "patrol" { + slots = s.limits.patrolSlots + } else { + slots = s.limits.chatSlots + } + + select { + case slots <- struct{}{}: + return func() { <-slots }, nil + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(5 * time.Second): + return nil, fmt.Errorf("AI is busy - too many concurrent requests") + } +} + +func (s *Service) enforceBudget(useCase string) error { + s.mu.RLock() + cfg := s.cfg + store := s.costStore + s.mu.RUnlock() + + if cfg == nil || cfg.CostBudgetUSD30d <= 0 || store == nil { + return nil + } + + summary := store.GetSummary(30) + if !summary.Totals.PricingKnown { + // We can't reliably enforce without pricing. Keep tracking and allow. + return nil + } + + if summary.Totals.EstimatedUSD >= cfg.CostBudgetUSD30d { + normalized := strings.TrimSpace(strings.ToLower(useCase)) + if normalized == "" { + normalized = "chat" + } + return fmt.Errorf("AI cost budget exceeded (%.2f/%.2f USD over %d days) - disable AI or raise budget to continue", + summary.Totals.EstimatedUSD, cfg.CostBudgetUSD30d, summary.EffectiveDays) + } + + return nil +} + // SetStateProvider sets the state provider for infrastructure context func (s *Service) SetStateProvider(sp StateProvider) { s.mu.Lock() @@ -502,6 +577,66 @@ func formatPolicyBlockedToolResult(command, reason string) string { return "POLICY_BLOCKED: " + string(b) } +func parseApprovalNeededMarker(content string) (ApprovalNeededData, bool) { + const prefix = "APPROVAL_REQUIRED:" + if !strings.HasPrefix(content, prefix) { + return ApprovalNeededData{}, false + } + trimmed := strings.TrimSpace(strings.TrimPrefix(content, prefix)) + if trimmed == "" { + return ApprovalNeededData{}, false + } + + var payload struct { + Type string `json:"type"` + Command string `json:"command"` + ToolID string `json:"tool_id"` + Reason string `json:"reason"` + } + if err := json.Unmarshal([]byte(trimmed), &payload); err != nil { + return ApprovalNeededData{}, false + } + if payload.Type != "approval_required" || payload.Command == "" { + return ApprovalNeededData{}, false + } + + return ApprovalNeededData{ + Command: payload.Command, + ToolID: payload.ToolID, + }, true +} + +func approvalNeededFromToolCall(req ExecuteRequest, tc providers.ToolCall, result string) (ApprovalNeededData, bool) { + if !strings.HasPrefix(result, "APPROVAL_REQUIRED:") { + return ApprovalNeededData{}, false + } + if tc.Name != "run_command" { + return ApprovalNeededData{}, false + } + + cmd, _ := tc.Input["command"].(string) + runOnHost, _ := tc.Input["run_on_host"].(bool) + targetHost, _ := tc.Input["target_host"].(string) + + if targetHost == "" { + if node, ok := req.Context["node"].(string); ok && node != "" { + targetHost = node + } else if node, ok := req.Context["hostname"].(string); ok && node != "" { + targetHost = node + } else if node, ok := req.Context["host_name"].(string); ok && node != "" { + targetHost = node + } + } + + return ApprovalNeededData{ + Command: cmd, + ToolID: tc.ID, + ToolName: tc.Name, + RunOnHost: runOnHost, + TargetHost: targetHost, + }, true +} + // LoadConfig loads the AI configuration and initializes the provider func (s *Service) LoadConfig() error { s.mu.Lock() @@ -671,232 +806,6 @@ func (s *Service) IsAutonomous() bool { return s.cfg != nil && s.cfg.AutonomousMode } -// isDangerousCommand checks if a command is too dangerous to auto-execute -// These commands ALWAYS require approval, even in autonomous mode -func isDangerousCommand(cmd string) bool { - cmd = strings.TrimSpace(strings.ToLower(cmd)) - parts := strings.Fields(cmd) - if len(parts) == 0 { - return false - } - baseCmd := parts[0] - if baseCmd == "sudo" && len(parts) > 1 { - baseCmd = parts[1] - } - - // Commands that are too dangerous to ever auto-execute - dangerousCommands := map[string]bool{ - // Deletion commands - "rm": true, - "rmdir": true, - "unlink": true, - "shred": true, - // Disk/filesystem destructive operations - "dd": true, - "mkfs": true, - "fdisk": true, - "parted": true, - "wipefs": true, - "sgdisk": true, - "gdisk": true, - "zpool": true, // Allow reads but not modifications - "zfs": true, // Allow reads but not modifications - "lvremove": true, - "vgremove": true, - "pvremove": true, - // System state changes - "reboot": true, - "shutdown": true, - "poweroff": true, - "halt": true, - "init": true, - "systemctl": true, // could stop critical services - "service": true, - // User/permission changes - "chmod": true, - "chown": true, - "useradd": true, - "userdel": true, - "passwd": true, - // Package management - "apt": true, - "apt-get": true, - "dpkg": true, - "yum": true, - "dnf": true, - "pacman": true, - "pip": true, - "npm": true, - // Proxmox destructive - "vzdump": true, - "vzrestore": true, - "pveam": true, - // Network changes - "iptables": true, - "nft": true, - "firewall-cmd": true, - } - - if dangerousCommands[baseCmd] { - // Special case: allow read-only apt/apt-get operations - if baseCmd == "apt" || baseCmd == "apt-get" { - // First, check if it's a dry-run/simulate command (safe even for upgrade/install) - for _, part := range parts { - if part == "--dry-run" || part == "-s" || part == "--simulate" || part == "--just-print" { - return false // Dry-run is always safe - } - } - // Check for inherently read-only operations - safeAptOps := []string{"update", "list", "show", "search", "policy", "madison", "depends", "rdepends", "changelog"} - for _, safeOp := range safeAptOps { - if len(parts) > 1 && parts[1] == safeOp { - return false // Safe read-only operation - } - // Also handle sudo apt - if len(parts) > 2 && parts[0] == "sudo" && parts[2] == safeOp { - return false - } - } - } - // Special case: allow read-only systemctl operations - if baseCmd == "systemctl" { - safeSystemctlOps := []string{"status", "show", "list-units", "list-unit-files", "is-active", "is-enabled", "is-failed", "cat"} - for _, safeOp := range safeSystemctlOps { - if len(parts) > 1 && parts[1] == safeOp { - return false - } - if len(parts) > 2 && parts[0] == "sudo" && parts[2] == safeOp { - return false - } - } - } - // Special case: allow read-only dpkg operations - if baseCmd == "dpkg" { - safeDpkgOps := []string{"-l", "--list", "-L", "--listfiles", "-s", "--status", "-S", "--search", "-p", "--print-avail", "--get-selections"} - for _, safeOp := range safeDpkgOps { - if len(parts) > 1 && parts[1] == safeOp { - return false - } - if len(parts) > 2 && parts[0] == "sudo" && parts[2] == safeOp { - return false - } - } - } - return true - } - - // Detect dangerous patterns in the full command - dangerousPatterns := []string{ - "rm -rf", "rm -fr", "rm -r", - "> /dev/", "| tee /", - "mkfs.", "dd if=", "dd of=", - ":(){ :|:& };:", // fork bomb - "chmod -R 777", "chmod 777", - "drop database", "drop table", - "truncate ", - } - for _, pattern := range dangerousPatterns { - if strings.Contains(cmd, pattern) { - return true - } - } - - return false -} - -// isReadOnlyCommand checks if a command is read-only (doesn't modify state) -// Read-only commands can be executed without approval even in non-autonomous mode -func isReadOnlyCommand(cmd string) bool { - cmd = strings.TrimSpace(cmd) - // Get the base command (first word, ignoring sudo) - parts := strings.Fields(cmd) - if len(parts) == 0 { - return false - } - baseCmd := parts[0] - if baseCmd == "sudo" && len(parts) > 1 { - baseCmd = parts[1] - } - - // List of read-only commands that are safe to auto-execute - readOnlyCommands := map[string]bool{ - // File/disk inspection - "ls": true, "ll": true, "dir": true, - "cat": true, "head": true, "tail": true, "less": true, "more": true, - "df": true, "du": true, "stat": true, "file": true, - "find": true, "locate": true, "which": true, "whereis": true, - "wc": true, "diff": true, "cmp": true, - // Process inspection - "ps": true, "top": true, "htop": true, "pgrep": true, - "lsof": true, "fuser": true, - // System info - "uname": true, "hostname": true, "uptime": true, "whoami": true, "id": true, - "free": true, "vmstat": true, "iostat": true, "sar": true, - "lscpu": true, "lsmem": true, "lsblk": true, "lspci": true, "lsusb": true, - "dmesg": true, "journalctl": true, - // Network inspection - "ip": true, "ifconfig": true, "netstat": true, "ss": true, - "ping": true, "traceroute": true, "tracepath": true, "mtr": true, - "dig": true, "nslookup": true, "host": true, - "curl": true, "wget": true, // typically read-only when not writing files - // Docker inspection - "docker": true, // we'll check subcommands below - // Package info (not install/remove) - "dpkg": true, "rpm": true, "apt-cache": true, - // Text processing (read-only) - "grep": true, "egrep": true, "fgrep": true, "rg": true, - "awk": true, "sed": true, // can be used read-only - "sort": true, "uniq": true, "cut": true, "tr": true, - "jq": true, "yq": true, - // Proxmox inspection - "pct": true, "qm": true, "pvesh": true, "pvecm": true, - "zpool": true, "zfs": true, - } - - if !readOnlyCommands[baseCmd] { - return false - } - - // Special handling for commands with dangerous subcommands - cmdLower := strings.ToLower(cmd) - - // Docker: only allow inspection subcommands - if baseCmd == "docker" { - dangerousDockerCmds := []string{ - "docker rm", "docker rmi", "docker kill", "docker stop", "docker start", - "docker restart", "docker prune", "docker pull", "docker push", - "docker exec", "docker run", "docker build", "docker compose", - "docker volume rm", "docker network rm", "docker system prune", - "docker image prune", "docker container prune", "docker builder prune", - } - for _, dangerous := range dangerousDockerCmds { - if strings.Contains(cmdLower, dangerous) { - return false - } - } - return true - } - - // Proxmox: only allow list/status subcommands - if baseCmd == "pct" || baseCmd == "qm" { - safeSubcmds := []string{"list", "status", "config", "pending", "snapshot"} - for _, safe := range safeSubcmds { - if strings.Contains(cmdLower, " "+safe) { - return true - } - } - // If no safe subcommand found, assume dangerous - return len(parts) == 1 // bare "pct" or "qm" is safe (shows help) - } - - // journalctl with --vacuum is not read-only - if baseCmd == "journalctl" && strings.Contains(cmdLower, "vacuum") { - return false - } - - return true -} - // ConversationMessage represents a message in conversation history type ConversationMessage struct { Role string `json:"role"` // "user" or "assistant" @@ -918,11 +827,12 @@ type ExecuteRequest struct { // ExecuteResponse represents the AI's response type ExecuteResponse struct { - Content string `json:"content"` - Model string `json:"model"` - InputTokens int `json:"input_tokens"` - OutputTokens int `json:"output_tokens"` - ToolCalls []ToolExecution `json:"tool_calls,omitempty"` // Commands that were executed + Content string `json:"content"` + Model string `json:"model"` + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` + ToolCalls []ToolExecution `json:"tool_calls,omitempty"` // Commands that were executed + PendingApprovals []ApprovalNeededData `json:"pending_approvals,omitempty"` // Commands that require approval (non-streaming) } // ToolExecution represents a tool that was executed during the AI conversation @@ -988,7 +898,7 @@ type ToolEndData struct { type ApprovalNeededData struct { Command string `json:"command"` ToolID string `json:"tool_id"` // ID to reference when approving - ToolName string `json:"tool_name"` // "run_command", "read_file", etc. + ToolName string `json:"tool_name"` // "run_command" RunOnHost bool `json:"run_on_host"` TargetHost string `json:"target_host,omitempty"` // Explicit host to route to } @@ -996,6 +906,16 @@ type ApprovalNeededData struct { // Execute sends a prompt to the AI and returns the response // If tools are available and the AI requests them, it executes them in a loop func (s *Service) Execute(ctx context.Context, req ExecuteRequest) (*ExecuteResponse, error) { + release, err := s.acquireExecutionSlot(ctx, req.UseCase) + if err != nil { + return nil, err + } + defer release() + + if err := s.enforceBudget(req.UseCase); err != nil { + return nil, err + } + s.mu.RLock() defaultProvider := s.provider agentServer := s.agentServer @@ -1078,6 +998,10 @@ Always execute the commands rather than telling the user how to do it.` // Agentic loop - keep going while AI requests tools maxIterations := 10 // Safety limit for i := 0; i < maxIterations; i++ { + if err := s.enforceBudget(req.UseCase); err != nil { + return nil, err + } + resp, err := provider.Chat(ctx, providers.ChatRequest{ Messages: messages, Model: s.getModelForRequest(req), @@ -1127,10 +1051,16 @@ Always execute the commands rather than telling the user how to do it.` }) // Execute each tool call and add results + var pendingApprovals []ApprovalNeededData for _, tc := range resp.ToolCalls { result, execution := s.executeTool(ctx, req, tc) toolExecutions = append(toolExecutions, execution) + if approval, ok := approvalNeededFromToolCall(req, tc, result); ok { + pendingApprovals = append(pendingApprovals, approval) + continue + } + // Add tool result to messages messages = append(messages, providers.Message{ Role: "user", @@ -1140,6 +1070,20 @@ Always execute the commands rather than telling the user how to do it.` IsError: !execution.Success, }, }) + + } + + // Stop the agentic loop when approvals are required. + // The caller can execute approvals via /api/ai/run-command and continue. + if len(pendingApprovals) > 0 { + return &ExecuteResponse{ + Content: finalContent, + Model: model, + InputTokens: totalInputTokens, + OutputTokens: totalOutputTokens, + ToolCalls: toolExecutions, + PendingApprovals: pendingApprovals, + }, nil } } @@ -1155,6 +1099,16 @@ Always execute the commands rather than telling the user how to do it.` // ExecuteStream sends a prompt to the AI and streams events via callback // This allows the UI to show real-time progress during tool execution func (s *Service) ExecuteStream(ctx context.Context, req ExecuteRequest, callback StreamCallback) (*ExecuteResponse, error) { + release, err := s.acquireExecutionSlot(ctx, req.UseCase) + if err != nil { + return nil, err + } + defer release() + + if err := s.enforceBudget(req.UseCase); err != nil { + return nil, err + } + s.mu.RLock() defaultProvider := s.provider agentServer := s.agentServer @@ -1267,6 +1221,11 @@ Always execute the commands rather than telling the user how to do it.` callback(StreamEvent{Type: "processing", Data: fmt.Sprintf("Analyzing results (iteration %d)...", iteration)}) } + if err := s.enforceBudget(req.UseCase); err != nil { + callback(StreamEvent{Type: "error", Data: err.Error()}) + return nil, err + } + resp, err := provider.Chat(ctx, providers.ChatRequest{ Messages: messages, Model: s.getModelForRequest(req), @@ -1371,24 +1330,49 @@ Always execute the commands rather than telling the user how to do it.` } isAuto := s.IsAutonomous() - isReadOnly := isReadOnlyCommand(cmd) - isDangerous := isDangerousCommand(cmd) + policyDecision := s.policy.Evaluate(cmd) log.Debug(). Bool("autonomous", isAuto). - Bool("read_only", isReadOnly). - Bool("dangerous", isDangerous). + Str("policy_decision", string(policyDecision)). Str("command", cmd). Str("target_host", targetHost). - Msg("Checking command approval") + Msg("Checking command policy/approval") - // In autonomous mode, NO commands need approval - full trust - // In non-autonomous mode: - // - Dangerous commands always need approval - // - Non-read-only commands need approval - if !isAuto && (isDangerous || !isReadOnly) { + // Always block commands blocked by policy (even in autonomous mode). + if policyDecision == agentexec.PolicyBlock { + result := formatPolicyBlockedToolResult(cmd, "This command is blocked by security policy") + execution := ToolExecution{ + Name: tc.Name, + Input: toolInput, + Output: result, + Success: false, + } + toolExecutions = append(toolExecutions, execution) + + callback(StreamEvent{ + Type: "tool_start", + Data: ToolStartData{Name: tc.Name, Input: toolInput}, + }) + callback(StreamEvent{ + Type: "tool_end", + Data: ToolEndData{Name: tc.Name, Input: toolInput, Output: result, Success: false}, + }) + + messages = append(messages, providers.Message{ + Role: "user", + ToolResult: &providers.ToolResult{ + ToolUseID: tc.ID, + Content: result, + IsError: true, + }, + }) + continue + } + + // If policy requires approval and we're not in autonomous mode, request approval. + if !isAuto && policyDecision == agentexec.PolicyRequireApproval { needsApproval = true anyNeedsApproval = true - // Send approval needed event callback(StreamEvent{ Type: "approval_needed", Data: ApprovalNeededData{ @@ -1433,7 +1417,7 @@ Always execute the commands rather than telling the user how to do it.` Type: "tool_end", Data: ToolEndData{Name: tc.Name, Input: toolInput, Output: result, Success: execution.Success}, }) - + // Log to remediation log for operational memory // Only log run_command since that's the main remediation action if tc.Name == "run_command" { @@ -1504,9 +1488,6 @@ func (s *Service) getToolInputDisplay(tc providers.ToolCall) string { return fmt.Sprintf("[host] %s", cmd) } return cmd - case "read_file": - path, _ := tc.Input["path"].(string) - return path case "fetch_url": url, _ := tc.Input["url"].(string) return url @@ -1525,16 +1506,16 @@ func (s *Service) logRemediation(req ExecuteRequest, command, output string, suc s.mu.RLock() patrol := s.patrolService s.mu.RUnlock() - + if patrol == nil { return } - + remLog := patrol.GetRemediationLog() if remLog == nil { return } - + // Determine outcome outcome := OutcomeUnknown if success { @@ -1542,13 +1523,13 @@ func (s *Service) logRemediation(req ExecuteRequest, command, output string, suc } else { outcome = OutcomeFailed } - + // Extract problem from the original prompt (first 200 chars as summary) problem := req.Prompt if len(problem) > 200 { problem = problem[:200] + "..." } - + // Get resource name from context if available resourceName := "" if req.Context != nil { @@ -1556,14 +1537,14 @@ func (s *Service) logRemediation(req ExecuteRequest, command, output string, suc resourceName = name } } - + // Truncate output for storage truncatedOutput := output const maxOutputLen = 1000 if len(truncatedOutput) > maxOutputLen { truncatedOutput = truncatedOutput[:maxOutputLen] + "..." } - + remLog.Log(RemediationRecord{ ResourceID: req.TargetID, ResourceType: req.TargetType, @@ -1575,7 +1556,7 @@ func (s *Service) logRemediation(req ExecuteRequest, command, output string, suc Outcome: outcome, Automatic: req.UseCase == "patrol", // Patrol runs are automatic }) - + log.Debug(). Str("resource_id", req.TargetID). Str("command", command). @@ -1620,46 +1601,6 @@ func (s *Service) getTools() []providers.Tool { "required": []string{"command"}, }, }, - { - Name: "read_file", - Description: "Read the contents of a file on the target. Use this to examine configuration files or logs.", - InputSchema: map[string]interface{}{ - "type": "object", - "properties": map[string]interface{}{ - "path": map[string]interface{}{ - "type": "string", - "description": "Absolute path to the file (e.g., '/etc/nginx/nginx.conf')", - }, - }, - "required": []string{"path"}, - }, - }, - { - Name: "write_file", - Description: "Write content to a file on the target. Use this to create or modify configuration files, scripts, or other text files. Creates parent directories if needed.", - InputSchema: map[string]interface{}{ - "type": "object", - "properties": map[string]interface{}{ - "path": map[string]interface{}{ - "type": "string", - "description": "Absolute path to the file (e.g., '/etc/myapp/config.yaml')", - }, - "content": map[string]interface{}{ - "type": "string", - "description": "The content to write to the file", - }, - "mode": map[string]interface{}{ - "type": "string", - "description": "Optional file permissions in octal (e.g., '0644' for rw-r--r--, '0755' for executable). Defaults to '0644'.", - }, - "append": map[string]interface{}{ - "type": "boolean", - "description": "If true, append to the file instead of overwriting. Defaults to false.", - }, - }, - "required": []string{"path", "content"}, - }, - }, { Name: "fetch_url", Description: "Fetch content from a URL. Use this to check if web services are responding, read API endpoints, or fetch documentation. Works with local network URLs and public sites.", @@ -1753,18 +1694,18 @@ func (s *Service) executeTool(ctx context.Context, req ExecuteRequest, tc provid return execution.Output, execution } - // Check security policy (skip if autonomous mode is enabled) - if !s.IsAutonomous() { - decision := s.policy.Evaluate(command) - if decision == agentexec.PolicyBlock { - execution.Output = formatPolicyBlockedToolResult(command, "This command is blocked by security policy") - return execution.Output, execution - } - if decision == agentexec.PolicyRequireApproval { - execution.Output = formatApprovalNeededToolResult(command, tc.ID, "Security policy requires approval") - execution.Success = true // Not an error, just needs approval - return execution.Output, execution - } + // Enforce security policy. + // - Blocked commands are ALWAYS blocked (even in autonomous mode). + // - Approval-required commands only require approval when not in autonomous mode. + decision := s.policy.Evaluate(command) + if decision == agentexec.PolicyBlock { + execution.Output = formatPolicyBlockedToolResult(command, "This command is blocked by security policy") + return execution.Output, execution + } + if decision == agentexec.PolicyRequireApproval && !s.IsAutonomous() { + execution.Output = formatApprovalNeededToolResult(command, tc.ID, "Security policy requires approval") + execution.Success = true // Not an error, just needs approval + return execution.Output, execution } // Build execution request with proper targeting @@ -1821,122 +1762,6 @@ func (s *Service) executeTool(ctx context.Context, req ExecuteRequest, tc provid execution.Success = true return result, execution - case "read_file": - path, _ := tc.Input["path"].(string) - execution.Input = path - - if path == "" { - execution.Output = "Error: path is required" - return execution.Output, execution - } - - // Use cat command to read file (simple approach) - command := fmt.Sprintf("cat %q 2>&1 | head -c 65536", path) - result, err := s.executeOnAgent(ctx, req, command) - if err != nil { - execution.Output = fmt.Sprintf("Error reading file: %s", err) - return execution.Output, execution - } - - execution.Output = result - execution.Success = true - return result, execution - - case "write_file": - path, _ := tc.Input["path"].(string) - content, _ := tc.Input["content"].(string) - mode, _ := tc.Input["mode"].(string) - appendMode, _ := tc.Input["append"].(bool) - execution.Input = path - - if path == "" { - execution.Output = "Error: path is required" - return execution.Output, execution - } - if content == "" { - execution.Output = "Error: content is required" - return execution.Output, execution - } - - // Size limit: 1MB max to prevent filling disk - const maxFileSize = 1024 * 1024 // 1MB - if len(content) > maxFileSize { - execution.Output = fmt.Sprintf("Error: content too large (%d bytes). Maximum allowed is %d bytes (1MB)", len(content), maxFileSize) - return execution.Output, execution - } - - // Path blocklist: prevent writes to critical system files - blockedPaths := []string{ - "/etc/passwd", "/etc/shadow", "/etc/group", "/etc/gshadow", - "/etc/sudoers", "/etc/ssh/sshd_config", - "/boot/", "/lib/", "/lib64/", "/usr/lib/", - "/bin/", "/sbin/", "/usr/bin/", "/usr/sbin/", - "/proc/", "/sys/", "/dev/", - } - cleanPath := filepath.Clean(path) - for _, blocked := range blockedPaths { - if cleanPath == blocked || strings.HasPrefix(cleanPath, blocked) { - execution.Output = fmt.Sprintf("Error: writing to %s is blocked for safety. This is a critical system path.", path) - return execution.Output, execution - } - } - - // Default mode if not specified - if mode == "" { - mode = "0644" - } - - // Build the write command using base64 to safely handle any content - // This avoids issues with special characters, quotes, newlines, etc. - encoded := base64.StdEncoding.EncodeToString([]byte(content)) - - var command string - if appendMode { - // Append mode: decode and append to file (no backup needed for append) - command = fmt.Sprintf("echo %q | base64 -d >> %q && echo 'Content appended to %s (%d bytes)'", encoded, path, path, len(content)) - } else { - // Overwrite mode with safety features: - // 1. Create parent directory if needed - // 2. Backup existing file if it exists (atomic - only if backup succeeds) - // 3. Write to temp file first - // 4. Atomic move temp file to target - // 5. Set permissions - dir := filepath.Dir(path) - tempFile := path + ".pulse-tmp" - backupFile := path + ".bak" - - // Build a safe multi-step command: - // - mkdir -p for parent dir - // - if file exists, copy to .bak - // - write content to temp file - // - mv temp file to target (atomic) - // - chmod to set permissions - command = fmt.Sprintf( - "mkdir -p %q && "+ - "([ -f %q ] && cp %q %q 2>/dev/null || true) && "+ - "echo %q | base64 -d > %q && "+ - "mv %q %q && "+ - "chmod %s %q && "+ - "echo 'Written %d bytes to %s (backup: %s.bak if existed)'", - dir, - path, path, backupFile, - encoded, tempFile, - tempFile, path, - mode, path, - len(content), path, path, - ) - } - - result, err := s.executeOnAgent(ctx, req, command) - if err != nil { - execution.Output = fmt.Sprintf("Error writing file: %s", err) - return execution.Output, execution - } - - execution.Output = result - execution.Success = true - return result, execution - case "fetch_url": urlStr, _ := tc.Input["url"].(string) execution.Input = urlStr @@ -2076,13 +1901,27 @@ func (s *Service) DeleteGuestNote(guestID, noteID string) error { // fetchURL fetches content from a URL with size limits and timeout func (s *Service) fetchURL(ctx context.Context, urlStr string) (string, error) { + parsedURL, err := parseAndValidateFetchURL(ctx, urlStr) + if err != nil { + return "", err + } + // Create HTTP client with timeout client := &http.Client{ Timeout: 30 * time.Second, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + if len(via) >= 3 { + return fmt.Errorf("too many redirects") + } + if _, err := parseAndValidateFetchURL(ctx, req.URL.String()); err != nil { + return err + } + return nil + }, } // Create request with context - req, err := http.NewRequestWithContext(ctx, "GET", urlStr, nil) + req, err := http.NewRequestWithContext(ctx, "GET", parsedURL.String(), nil) if err != nil { return "", fmt.Errorf("invalid URL: %w", err) } @@ -2118,6 +1957,78 @@ func (s *Service) fetchURL(ctx context.Context, urlStr string) (string, error) { return result, nil } +func parseAndValidateFetchURL(ctx context.Context, urlStr string) (*url.URL, error) { + clean := strings.TrimSpace(urlStr) + if clean == "" { + return nil, fmt.Errorf("url is required") + } + + parsed, err := url.ParseRequestURI(clean) + if err != nil { + return nil, fmt.Errorf("invalid URL: %w", err) + } + if parsed.Scheme != "http" && parsed.Scheme != "https" { + return nil, fmt.Errorf("only http/https URLs are allowed") + } + if parsed.User != nil { + return nil, fmt.Errorf("URLs with embedded credentials are not allowed") + } + if parsed.Fragment != "" { + return nil, fmt.Errorf("URL fragments are not allowed") + } + + host := parsed.Hostname() + if host == "" { + return nil, fmt.Errorf("URL must include a host") + } + + if isBlockedFetchHost(host) { + return nil, fmt.Errorf("URL host is blocked") + } + + if ip := net.ParseIP(host); ip != nil { + if isBlockedFetchIP(ip) { + return nil, fmt.Errorf("URL IP is blocked") + } + return parsed, nil + } + + addrs, err := net.DefaultResolver.LookupIPAddr(ctx, host) + if err != nil { + // DNS failures are surfaced directly to the caller. + return nil, fmt.Errorf("failed to resolve host: %w", err) + } + for _, addr := range addrs { + if isBlockedFetchIP(addr.IP) { + return nil, fmt.Errorf("URL host resolves to a blocked address") + } + } + + return parsed, nil +} + +func isBlockedFetchHost(host string) bool { + h := strings.TrimSpace(strings.ToLower(host)) + if h == "localhost" || h == "localhost." { + return true + } + return false +} + +func isBlockedFetchIP(ip net.IP) bool { + if ip == nil { + return true + } + if ip.IsLoopback() || ip.IsUnspecified() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() { + return true + } + // Block multicast and other non-unicast targets. + if !ip.IsGlobalUnicast() && !ip.IsPrivate() { + return true + } + return false +} + // sanitizeError cleans up error messages to remove internal networking details // that are not helpful to users or AI models (IP addresses, port numbers, etc.) func sanitizeError(err error) error { @@ -2342,17 +2253,31 @@ func (s *Service) buildSystemPrompt(req ExecuteRequest) string { - Only show the conclusion, not every command run - If something fails, briefly state what and why -## When Performing Actions -- Just do it. Don't ask "Would you like me to..." -- Report the result, not the intent -- If destructive (delete data, stop services), execute but clearly state what was done +## Command Approval System +Pulse has a command approval policy that protects against accidental damage: +- **Safe commands** (read-only: ls, df, cat, ps, etc.) execute immediately +- **Potentially destructive commands** (rm, service restart, apt install, etc.) require approval unless autonomous mode is enabled +- **Blocked commands** are never executed +- When approval is required, the user will see the command and can approve or reject it + +When you need to run a potentially destructive command: +1. Execute it normally - Pulse will handle the approval flow +2. If a command is blocked or needs approval, you'll get feedback +3. Continue with other work while waiting for approval, or explain what the command will do + +Do NOT say things like: +- "I need your permission to..." +- "Would you like me to run...?" +- "This is destructive, are you sure?" + +Instead, just execute the command. Pulse's approval system will handle user confirmation for risky operations. ## Response Format BAD: "I'll check that for you. Let me run some commands..." GOOD: State findings directly. -BAD: "Would you like me to..." -GOOD: Do it, then report the result. +BAD: "Would you like me to run rm -rf...?" +GOOD: Execute the command. User will approve/reject via Pulse. BAD: Tables, headers, bullet-heavy summaries GOOD: Plain prose, 2-4 sentences. @@ -2361,7 +2286,7 @@ GOOD: Plain prose, 2-4 sentences. When the user asks you to DO something (install, fix, update, configure), ACT IMMEDIATELY: - Don't extensively investigate before acting. Run 1-2 diagnostic commands max, then DO the thing. - Don't explain what you're about to do - just do it. -- Don't ask for confirmation. The user asked you to do it, so do it. +- If the command needs approval, Pulse will queue it. You don't need to ask separately. - If the first approach fails, try the next most obvious approach. Don't stop to report. - Complete the task END TO END. If asked to "install and run X", you're not done until X is running. @@ -2369,7 +2294,7 @@ INVESTIGATION ANTI-PATTERNS TO AVOID: - Running 10+ diagnostic commands before taking action - Explaining each step before doing it - Stopping to report partial progress -- Asking "would you like me to proceed?" after the user already asked you to do it +- Asking "would you like me to proceed?" (Pulse handles approval automatically) - Checking version, checking config, checking service, checking ports, checking this, checking that... JUST ACT. GOOD PATTERN for "install X and make sure it's running": @@ -2383,6 +2308,7 @@ BAD PATTERN: 5. Check another config... 6. Try to enable something... 7. Check if it worked... 8. Read a script... 9. Check yet another file... [user: "do it"] 10. Still investigating... [user: "DO IT"] + ## Using Context Data Pulse provides real metrics in "Current Metrics and State". Use this data directly - don't ask users to check things you already know. @@ -2503,7 +2429,7 @@ This is a 3-command job. Don't over-investigate.` } else if req.TargetID != "" { prompt += fmt.Sprintf("\n\n## Current Focus\nYou are analyzing %s '%s'", req.TargetType, req.TargetID) } - + // Add past remediation history for this resource prompt += s.buildRemediationContext(req.TargetID, req.Prompt) } @@ -2721,14 +2647,29 @@ func (s *Service) TestConnection(ctx context.Context) error { // ListModels fetches available models from ALL configured AI providers // Returns a unified list with models prefixed by provider name func (s *Service) ListModels(ctx context.Context) ([]providers.ModelInfo, error) { + models, _, err := s.ListModelsWithCache(ctx) + return models, err +} + +func (s *Service) ListModelsWithCache(ctx context.Context) ([]providers.ModelInfo, bool, error) { cfg, err := s.persistence.LoadAIConfig() if err != nil { - return nil, fmt.Errorf("failed to load AI config: %w", err) + return nil, false, fmt.Errorf("failed to load AI config: %w", err) } if cfg == nil { - return nil, fmt.Errorf("AI not configured") + return nil, false, fmt.Errorf("AI not configured") } + cacheKey := buildModelsCacheKey(cfg) + s.modelsCache.mu.RLock() + if s.modelsCache.key == cacheKey && s.modelsCache.ttl > 0 && time.Since(s.modelsCache.at) < s.modelsCache.ttl && len(s.modelsCache.models) > 0 { + out := make([]providers.ModelInfo, len(s.modelsCache.models)) + copy(out, s.modelsCache.models) + s.modelsCache.mu.RUnlock() + return out, true, nil + } + s.modelsCache.mu.RUnlock() + var allModels []providers.ModelInfo // Query each configured provider @@ -2764,7 +2705,33 @@ func (s *Service) ListModels(ctx context.Context) ([]providers.ModelInfo, error) } } - return allModels, nil + s.modelsCache.mu.Lock() + s.modelsCache.key = cacheKey + s.modelsCache.at = time.Now() + s.modelsCache.models = make([]providers.ModelInfo, len(allModels)) + copy(s.modelsCache.models, allModels) + s.modelsCache.mu.Unlock() + + return allModels, false, nil +} + +func buildModelsCacheKey(cfg *config.AIConfig) string { + if cfg == nil { + return "nil" + } + + var b strings.Builder + b.WriteString("providers=") + b.WriteString(strings.Join(cfg.GetConfiguredProviders(), ",")) + b.WriteString("|auth=") + b.WriteString(string(cfg.AuthMethod)) + + b.WriteString("|openai_base=") + b.WriteString(cfg.OpenAIBaseURL) + b.WriteString("|ollama_base=") + b.WriteString(cfg.OllamaBaseURL) + + return b.String() } // providerDisplayName returns a user-friendly name for a provider @@ -2793,18 +2760,18 @@ func (s *Service) buildRemediationContext(resourceID, currentProblem string) str s.mu.RLock() patrol := s.patrolService s.mu.RUnlock() - + if patrol == nil { return "" } - + remLog := patrol.GetRemediationLog() if remLog == nil { return "" } - + var context string - + // Get similar past remediations based on the current problem if currentProblem != "" { successful := remLog.GetSuccessfulRemediations(currentProblem, 3) @@ -2812,14 +2779,14 @@ func (s *Service) buildRemediationContext(resourceID, currentProblem string) str context += "\n\n## Past Successful Fixes for Similar Issues\n" context += "These actions worked for similar problems before:\n" for _, rec := range successful { - context += fmt.Sprintf("- **%s**: `%s` (%s)\n", - truncateString(rec.Problem, 60), + context += fmt.Sprintf("- **%s**: `%s` (%s)\n", + truncateString(rec.Problem, 60), truncateString(rec.Action, 80), rec.Outcome) } } } - + // Get history for this specific resource if resourceID != "" { history := remLog.GetForResource(resourceID, 5) @@ -2836,7 +2803,7 @@ func (s *Service) buildRemediationContext(resourceID, currentProblem string) str } } } - + return context } diff --git a/internal/api/ai_handlers.go b/internal/api/ai_handlers.go index 611856c..aa8d85d 100644 --- a/internal/api/ai_handlers.go +++ b/internal/api/ai_handlers.go @@ -12,6 +12,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" @@ -142,13 +143,13 @@ type AISettingsResponse struct { Provider string `json:"provider"` // DEPRECATED: legacy single provider APIKeySet bool `json:"api_key_set"` // DEPRECATED: true if legacy API key is configured Model string `json:"model"` - ChatModel string `json:"chat_model,omitempty"` // Model for interactive chat (empty = use default) - PatrolModel string `json:"patrol_model,omitempty"` // Model for patrol (empty = use default) + ChatModel string `json:"chat_model,omitempty"` // Model for interactive chat (empty = use default) + PatrolModel string `json:"patrol_model,omitempty"` // Model for patrol (empty = use default) AutoFixModel string `json:"auto_fix_model,omitempty"` // Model for auto-fix (empty = use patrol model) - BaseURL string `json:"base_url,omitempty"` // DEPRECATED: legacy base URL - Configured bool `json:"configured"` // true if AI is ready to use - AutonomousMode bool `json:"autonomous_mode"` // true if AI can execute without approval - CustomContext string `json:"custom_context"` // user-provided infrastructure context + BaseURL string `json:"base_url,omitempty"` // DEPRECATED: legacy base URL + Configured bool `json:"configured"` // true if AI is ready to use + AutonomousMode bool `json:"autonomous_mode"` // true if AI can execute without approval + CustomContext string `json:"custom_context"` // user-provided infrastructure context // OAuth fields for Claude Pro/Max subscription authentication AuthMethod string `json:"auth_method"` // "api_key" or "oauth" OAuthConnected bool `json:"oauth_connected"` // true if OAuth tokens are configured @@ -176,10 +177,10 @@ type AISettingsUpdateRequest struct { Provider *string `json:"provider,omitempty"` // DEPRECATED: use model selection instead APIKey *string `json:"api_key,omitempty"` // DEPRECATED: use per-provider keys Model *string `json:"model,omitempty"` - ChatModel *string `json:"chat_model,omitempty"` // Model for interactive chat - PatrolModel *string `json:"patrol_model,omitempty"` // Model for background patrol + ChatModel *string `json:"chat_model,omitempty"` // Model for interactive chat + PatrolModel *string `json:"patrol_model,omitempty"` // Model for background patrol AutoFixModel *string `json:"auto_fix_model,omitempty"` // Model for auto-fix remediation - BaseURL *string `json:"base_url,omitempty"` // DEPRECATED: use per-provider URLs + BaseURL *string `json:"base_url,omitempty"` // DEPRECATED: use per-provider URLs AutonomousMode *bool `json:"autonomous_mode,omitempty"` CustomContext *string `json:"custom_context,omitempty"` // user-provided infrastructure context AuthMethod *string `json:"auth_method,omitempty"` // "api_key" or "oauth" @@ -650,7 +651,7 @@ func (h *AISettingsHandler) HandleListModels(w http.ResponseWriter, r *http.Requ Cached bool `json:"cached"` } - models, err := h.aiService.ListModels(ctx) + models, cached, err := h.aiService.ListModelsWithCache(ctx) if err != nil { // Return error but don't fail the request - frontend can show a fallback resp := Response{ @@ -675,6 +676,7 @@ func (h *AISettingsHandler) HandleListModels(w http.ResponseWriter, r *http.Requ resp := Response{ Models: responseModels, + Cached: cached, } if err := utils.WriteJSONResponse(w, resp); err != nil { @@ -695,15 +697,19 @@ type AIExecuteRequest struct { TargetID string `json:"target_id,omitempty"` Context map[string]interface{} `json:"context,omitempty"` // Current metrics, state, etc. History []AIConversationMessage `json:"history,omitempty"` // Previous conversation messages + FindingID string `json:"finding_id,omitempty"` + Model string `json:"model,omitempty"` + UseCase string `json:"use_case,omitempty"` // "chat" or "patrol" } // AIExecuteResponse is the response from POST /api/ai/execute type AIExecuteResponse struct { - Content string `json:"content"` - Model string `json:"model"` - InputTokens int `json:"input_tokens"` - OutputTokens int `json:"output_tokens"` - ToolCalls []ai.ToolExecution `json:"tool_calls,omitempty"` // Commands that were executed + Content string `json:"content"` + Model string `json:"model"` + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` + ToolCalls []ai.ToolExecution `json:"tool_calls,omitempty"` // Commands that were executed + PendingApprovals []ai.ApprovalNeededData `json:"pending_approvals,omitempty"` // Commands that require approval (non-streaming) } // HandleExecute executes an AI prompt (POST /api/ai/execute) @@ -741,11 +747,29 @@ func (h *AISettingsHandler) HandleExecute(w http.ResponseWriter, r *http.Request ctx, cancel := context.WithTimeout(r.Context(), 120*time.Second) defer cancel() + // Convert history from API type to service type + var history []ai.ConversationMessage + for _, msg := range req.History { + history = append(history, ai.ConversationMessage{ + Role: msg.Role, + Content: msg.Content, + }) + } + + useCase := strings.TrimSpace(req.UseCase) + if useCase == "" { + useCase = "chat" + } + resp, err := h.aiService.Execute(ctx, ai.ExecuteRequest{ Prompt: req.Prompt, TargetType: req.TargetType, TargetID: req.TargetID, Context: req.Context, + History: history, + FindingID: req.FindingID, + Model: req.Model, + UseCase: useCase, }) if err != nil { log.Error().Err(err).Msg("AI execution failed") @@ -754,11 +778,12 @@ func (h *AISettingsHandler) HandleExecute(w http.ResponseWriter, r *http.Request } response := AIExecuteResponse{ - Content: resp.Content, - Model: resp.Model, - InputTokens: resp.InputTokens, - OutputTokens: resp.OutputTokens, - ToolCalls: resp.ToolCalls, + Content: resp.Content, + Model: resp.Model, + InputTokens: resp.InputTokens, + OutputTokens: resp.OutputTokens, + ToolCalls: resp.ToolCalls, + PendingApprovals: resp.PendingApprovals, } if err := utils.WriteJSONResponse(w, response); err != nil { @@ -815,7 +840,7 @@ func (h *AISettingsHandler) HandleExecuteStream(w http.ResponseWriter, r *http.R } log.Info(). - Str("prompt", req.Prompt). + Int("prompt_len", len(req.Prompt)). Str("target_type", req.TargetType). Str("target_id", req.TargetID). Msg("AI streaming request started") @@ -859,7 +884,7 @@ func (h *AISettingsHandler) HandleExecuteStream(w http.ResponseWriter, r *http.R // NOTE: We don't check r.Context().Done() because Vite proxy may close // the request context prematurely. We detect real disconnection via write failures. heartbeatDone := make(chan struct{}) - var clientDisconnected bool + var clientDisconnected atomic.Bool go func() { ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() @@ -872,7 +897,7 @@ func (h *AISettingsHandler) HandleExecuteStream(w http.ResponseWriter, r *http.R _, err := w.Write([]byte(": heartbeat\n\n")) if err != nil { log.Debug().Err(err).Msg("Heartbeat write failed, stopping heartbeat (AI continues)") - clientDisconnected = true + clientDisconnected.Store(true) // Don't cancel the AI request - let it complete with its own timeout // The SSE connection may have issues but the AI work can still finish return @@ -888,14 +913,14 @@ func (h *AISettingsHandler) HandleExecuteStream(w http.ResponseWriter, r *http.R // Helper to safely write SSE events, tracking if client disconnected safeWrite := func(data []byte) bool { - if clientDisconnected { + if clientDisconnected.Load() { return false } _ = rc.SetWriteDeadline(time.Now().Add(10 * time.Second)) _, err := w.Write(data) if err != nil { log.Debug().Err(err).Msg("Failed to write SSE event (client may have disconnected)") - clientDisconnected = true + clientDisconnected.Store(true) return false } flusher.Flush() @@ -934,9 +959,14 @@ func (h *AISettingsHandler) HandleExecuteStream(w http.ResponseWriter, r *http.R }) } + useCase := strings.TrimSpace(req.UseCase) + if useCase == "" { + useCase = "chat" + } + // Ensure we always send a final 'done' event defer func() { - if !clientDisconnected { + if !clientDisconnected.Load() { doneEvent := ai.StreamEvent{Type: "done"} data, _ := json.Marshal(doneEvent) safeWrite([]byte("data: " + string(data) + "\n\n")) @@ -951,6 +981,9 @@ func (h *AISettingsHandler) HandleExecuteStream(w http.ResponseWriter, r *http.R TargetID: req.TargetID, Context: req.Context, History: history, + FindingID: req.FindingID, + Model: req.Model, + UseCase: useCase, }, callback) if err != nil { @@ -1018,7 +1051,7 @@ func (h *AISettingsHandler) HandleRunCommand(w http.ResponseWriter, r *http.Requ http.Error(w, "Invalid request body", http.StatusBadRequest) return } - log.Debug().Str("body", string(bodyBytes)).Msg("run-command request body") + log.Debug().Int("body_len", len(bodyBytes)).Msg("run-command request received") var req AIRunCommandRequest if err := json.Unmarshal(bodyBytes, &req); err != nil { @@ -1454,7 +1487,7 @@ func (h *AISettingsHandler) HandleInvestigateAlert(w http.ResponseWriter, r *htt // Heartbeat routine heartbeatDone := make(chan struct{}) - var clientDisconnected bool + var clientDisconnected atomic.Bool go func() { ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() @@ -1464,7 +1497,7 @@ func (h *AISettingsHandler) HandleInvestigateAlert(w http.ResponseWriter, r *htt _ = rc.SetWriteDeadline(time.Now().Add(10 * time.Second)) _, err := w.Write([]byte(": heartbeat\n\n")) if err != nil { - clientDisconnected = true + clientDisconnected.Store(true) return } flusher.Flush() @@ -1476,13 +1509,13 @@ func (h *AISettingsHandler) HandleInvestigateAlert(w http.ResponseWriter, r *htt defer close(heartbeatDone) safeWrite := func(data []byte) bool { - if clientDisconnected { + if clientDisconnected.Load() { return false } _ = rc.SetWriteDeadline(time.Now().Add(10 * time.Second)) _, err := w.Write(data) if err != nil { - clientDisconnected = true + clientDisconnected.Store(true) return false } flusher.Flush() @@ -1518,7 +1551,7 @@ func (h *AISettingsHandler) HandleInvestigateAlert(w http.ResponseWriter, r *htt // Execute with streaming defer func() { - if !clientDisconnected { + if !clientDisconnected.Load() { doneEvent := ai.StreamEvent{Type: "done"} data, _ := json.Marshal(doneEvent) safeWrite([]byte("data: " + string(data) + "\n\n")) diff --git a/internal/monitoring/monitor_docker_test.go b/internal/monitoring/monitor_docker_test.go index d343279..5c12d44 100644 --- a/internal/monitoring/monitor_docker_test.go +++ b/internal/monitoring/monitor_docker_test.go @@ -19,6 +19,7 @@ func newTestMonitor(t *testing.T) *Monitor { alertManager: alerts.NewManager(), removedDockerHosts: make(map[string]time.Time), rateTracker: NewRateTracker(), + metricsHistory: NewMetricsHistory(1000, 24*time.Hour), dockerTokenBindings: make(map[string]string), dockerMetadataStore: config.NewDockerMetadataStore(t.TempDir()), } diff --git a/internal/sensors/parser.go b/internal/sensors/parser.go index 19444da..4f68353 100644 --- a/internal/sensors/parser.go +++ b/internal/sensors/parser.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "math" + "sort" "strings" "github.com/rs/zerolog/log" @@ -37,6 +38,7 @@ func Parse(jsonStr string) (*TemperatureData, error) { } foundCPUChip := false + nvmeTempsByChip := make(map[string]float64) // Parse each sensor chip for chipName, chipData := range sensorsData { @@ -54,8 +56,10 @@ func Parse(jsonStr string) (*TemperatureData, error) { } // Handle NVMe temperature sensors - if strings.Contains(chipName, "nvme") { - parseNVMeTemps(chipName, chipMap, data) + if strings.Contains(chipLower, "nvme") { + if tempVal, ok := extractNVMeCompositeTemp(chipMap); ok { + nvmeTempsByChip[chipName] = tempVal + } } // Handle GPU temperature sensors @@ -75,6 +79,23 @@ func Parse(jsonStr string) (*TemperatureData, error) { data.CPUPackage = data.CPUMax } + if len(nvmeTempsByChip) > 0 { + chips := make([]string, 0, len(nvmeTempsByChip)) + for chip := range nvmeTempsByChip { + chips = append(chips, chip) + } + sort.Strings(chips) + for i, chip := range chips { + normalizedName := fmt.Sprintf("nvme%d", i) + data.NVMe[normalizedName] = nvmeTempsByChip[chip] + log.Debug(). + Str("chip", chip). + Str("normalizedName", normalizedName). + Float64("temp", nvmeTempsByChip[chip]). + Msg("Found NVMe temperature") + } + } + data.Available = foundCPUChip || len(data.NVMe) > 0 || len(data.GPU) > 0 log.Debug(). @@ -194,7 +215,7 @@ func parseCPUTemps(chipMap map[string]interface{}, data *TemperatureData) { } } -func parseNVMeTemps(chipName string, chipMap map[string]interface{}, data *TemperatureData) { +func extractNVMeCompositeTemp(chipMap map[string]interface{}) (float64, bool) { for sensorName, sensorData := range chipMap { sensorMap, ok := sensorData.(map[string]interface{}) if !ok { @@ -203,20 +224,12 @@ func parseNVMeTemps(chipName string, chipMap map[string]interface{}, data *Tempe // Look for Composite temperature (main NVMe temp) if strings.Contains(sensorName, "Composite") { - if tempVal := extractTempInput(sensorMap); !math.IsNaN(tempVal) { - // Normalize chip name to nvme0, nvme1 format - // Input format is like "nvme-pci-0200" or "nvme-pci-0300" - // We extract the device index based on how many NVMe devices we've seen - normalizedName := fmt.Sprintf("nvme%d", len(data.NVMe)) - data.NVMe[normalizedName] = tempVal - log.Debug(). - Str("chip", chipName). - Str("normalizedName", normalizedName). - Float64("temp", tempVal). - Msg("Found NVMe temperature") + if tempVal := extractTempInput(sensorMap); !math.IsNaN(tempVal) && tempVal > 0 { + return tempVal, true } } } + return 0, false } func parseGPUTemps(chipName string, chipMap map[string]interface{}, data *TemperatureData) { diff --git a/internal/sensors/parser_test.go b/internal/sensors/parser_test.go index d4ee80d..ac8949f 100644 --- a/internal/sensors/parser_test.go +++ b/internal/sensors/parser_test.go @@ -146,12 +146,12 @@ func TestParse_NVMe(t *testing.T) { t.Errorf("len(NVMe) = %d, want 2", len(data.NVMe)) } - if data.NVMe["nvme-pci-0100"] != 38.85 { - t.Errorf("nvme-pci-0100 = %v, want 38.85", data.NVMe["nvme-pci-0100"]) + if data.NVMe["nvme0"] != 38.85 { + t.Errorf("nvme0 = %v, want 38.85", data.NVMe["nvme0"]) } - if data.NVMe["nvme-pci-0200"] != 42.0 { - t.Errorf("nvme-pci-0200 = %v, want 42.0", data.NVMe["nvme-pci-0200"]) + if data.NVMe["nvme1"] != 42.0 { + t.Errorf("nvme1 = %v, want 42.0", data.NVMe["nvme1"]) } } @@ -223,6 +223,10 @@ func TestParse_Combined(t *testing.T) { t.Errorf("len(NVMe) = %d, want 1", len(data.NVMe)) } + if data.NVMe["nvme0"] != 40.0 { + t.Errorf("nvme0 = %v, want 40.0", data.NVMe["nvme0"]) + } + if len(data.GPU) != 1 { t.Errorf("len(GPU) = %d, want 1", len(data.GPU)) }