feat: AI security and policy improvements for 5.0

- Add DOMPurify sanitization for AI chat markdown rendering (XSS fix)
- Configure DOMPurify to add target=_blank and rel=noopener to links
- Update system prompt to align with command approval policy
- Clarify safe vs destructive commands in prompt
- Improve patrol auto-fix mode guidance with safe operation list
- Add verification requirements for auto-fix actions
- Update observe-only mode to be clearer about read-only restrictions
This commit is contained in:
rcourtman 2025-12-12 17:38:55 +00:00
parent 2381631df0
commit 7f057432ed
17 changed files with 882 additions and 1838 deletions

View file

@ -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.

View file

@ -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

Binary file not shown.

View file

@ -27,6 +27,7 @@
}, },
"dependencies": { "dependencies": {
"@solidjs/router": "^0.10.10", "@solidjs/router": "^0.10.10",
"dompurify": "^3.3.1",
"lucide-solid": "^0.545.0", "lucide-solid": "^0.545.0",
"marked": "^17.0.1", "marked": "^17.0.1",
"solid-js": "^1.8.0" "solid-js": "^1.8.0"

View file

@ -1,5 +1,6 @@
import { Component, Show, createSignal, For, createEffect, createMemo, onMount, Switch, Match } from 'solid-js'; import { Component, Show, createSignal, For, createEffect, createMemo, onMount, Switch, Match } from 'solid-js';
import { marked } from 'marked'; import { marked } from 'marked';
import DOMPurify from 'dompurify';
import { AIAPI } from '@/api/ai'; import { AIAPI } from '@/api/ai';
import { notificationStore } from '@/stores/notifications'; import { notificationStore } from '@/stores/notifications';
import { logger } from '@/utils/logger'; import { logger } from '@/utils/logger';
@ -63,12 +64,41 @@ marked.setOptions({
gfm: true, // GitHub Flavored Markdown 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 => { const renderMarkdown = (content: string): string => {
try { 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 { } catch {
return content; // If parsing fails, escape HTML entities as fallback
return content.replace(/[&<>"']/g, (char) => {
const entities: Record<string, string> = { '&': '&amp;', '<': '&lt;', '>': '&gt;', '"': '&quot;', "'": '&#39;' };
return entities[char] || char;
});
} }
}; };

View file

@ -124,6 +124,7 @@ export interface AIExecuteRequest {
history?: AIConversationMessage[]; // Previous conversation messages history?: AIConversationMessage[]; // Previous conversation messages
finding_id?: string; // If fixing a patrol finding, the ID to resolve on success 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) model?: string; // Override model for this request (user selection in chat)
use_case?: 'chat' | 'patrol'; // Optional server-side routing/model selection
} }
// Tool execution info // Tool execution info
@ -140,6 +141,7 @@ export interface AIExecuteResponse {
input_tokens: number; input_tokens: number;
output_tokens: number; output_tokens: number;
tool_calls?: AIToolExecution[]; // Commands that were executed tool_calls?: AIToolExecution[]; // Commands that were executed
pending_approvals?: AIStreamApprovalNeededData[]; // Non-streaming approvals
} }
// Streaming event types // Streaming event types

View file

@ -27,23 +27,23 @@ func DefaultPolicy() *CommandPolicy {
p := &CommandPolicy{ p := &CommandPolicy{
AutoApprove: []string{ AutoApprove: []string{
// System inspection // System inspection
`^ps\s`, `^ps(\s|$)`,
`^top\s+-bn`, `^top\s+-bn`,
`^df\s`, `^df(\s|$)`,
`^free\s`, `^free(\s|$)`,
`^uptime$`, `^uptime$`,
`^hostname$`, `^hostname$`,
`^uname\s`, `^uname(\s|$)`,
`^cat\s+/proc/`, `^cat\s+/proc/`,
`^cat\s+/etc/os-release`, `^cat\s+/etc/os-release`,
`^lsof\s`, `^lsof(\s|$)`,
`^netstat\s`, `^netstat(\s|$)`,
`^ss\s`, `^ss(\s|$)`,
`^ip\s+(addr|route|link)`, `^ip\s+(addr|route|link)`,
`^ifconfig`, `^ifconfig`,
`^w$`, `^w$`,
`^who$`, `^who$`,
`^last\s`, `^last(\s|$)`,
// Log reading (read-only) // Log reading (read-only)
`^cat\s+/var/log/`, `^cat\s+/var/log/`,
@ -82,10 +82,10 @@ func DefaultPolicy() *CommandPolicy {
`^lsblk`, `^lsblk`,
`^blkid`, `^blkid`,
`^fdisk\s+-l`, `^fdisk\s+-l`,
`^du\s`, `^du(\s|$)`,
`^ls\s`, `^ls(\s|$)`,
`^stat\s`, `^stat(\s|$)`,
`^file\s`, `^file(\s|$)`,
`^find\s+/.*-size`, // Find large files `^find\s+/.*-size`, // Find large files
`^find\s+/.*-mtime`, // Find by modification time `^find\s+/.*-mtime`, // Find by modification time
`^find\s+/.*-type`, // Find by type `^find\s+/.*-type`, // Find by type
@ -131,6 +131,10 @@ func DefaultPolicy() *CommandPolicy {
// Proxmox control // Proxmox control
`^pct\s+(start|stop|shutdown|reboot|resize|set)\s`, `^pct\s+(start|stop|shutdown|reboot|resize|set)\s`,
`^qm\s+(start|stop|shutdown|reboot|reset|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{ Blocked: []string{
@ -207,6 +211,14 @@ const (
// Evaluate checks a command against the policy // Evaluate checks a command against the policy
func (p *CommandPolicy) Evaluate(command string) PolicyDecision { func (p *CommandPolicy) Evaluate(command string) PolicyDecision {
command = strings.TrimSpace(command) 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) // Check blocked first (highest priority)
for _, re := range p.blockedRe { for _, re := range p.blockedRe {

View file

@ -5,9 +5,11 @@ package correlation
import ( import (
"encoding/json" "encoding/json"
"fmt"
"os" "os"
"path/filepath" "path/filepath"
"sort" "sort"
"strings"
"sync" "sync"
"time" "time"
@ -224,7 +226,12 @@ func (d *Detector) formatCorrelationDescription(c *Correlation) string {
} }
delayStr := formatDuration(c.AvgDelay) 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 ", " + targetName + " often follows within " + delayStr
} }
@ -312,8 +319,12 @@ func (d *Detector) PredictCascade(resourceID string, eventType EventType) []Casc
for _, c := range d.correlations { for _, c := range d.correlations {
if c.SourceID == resourceID && c.Occurrences >= d.minOccurrences { if c.SourceID == resourceID && c.Occurrences >= d.minOccurrences {
// Check if the event pattern starts with the given event type // Check if the correlation's source event matches the given event type.
if len(c.EventPattern) > 0 && EventType(c.EventPattern[:len(string(eventType))]) == eventType { sourceEvent := c.EventPattern
if parts := strings.Split(c.EventPattern, " -> "); len(parts) == 2 {
sourceEvent = parts[0]
}
if EventType(sourceEvent) == eventType {
predictions = append(predictions, CascadePrediction{ predictions = append(predictions, CascadePrediction{
ResourceID: c.TargetID, ResourceID: c.TargetID,
ResourceName: c.TargetName, ResourceName: c.TargetName,
@ -429,6 +440,12 @@ func (d *Detector) loadFromDisk() error {
} }
path := filepath.Join(d.dataDir, "ai_correlations.json") 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) jsonData, err := os.ReadFile(path)
if err != nil { if err != nil {
if os.IsNotExist(err) { if os.IsNotExist(err) {
@ -448,6 +465,17 @@ func (d *Detector) loadFromDisk() error {
d.events = data.Events d.events = data.Events
d.correlations = data.Correlations 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 return nil
} }

View file

@ -5,6 +5,7 @@ package memory
import ( import (
"encoding/json" "encoding/json"
"fmt"
"os" "os"
"path/filepath" "path/filepath"
"sort" "sort"
@ -343,6 +344,12 @@ func (d *ChangeDetector) loadFromDisk() error {
} }
path := filepath.Join(d.dataDir, "ai_changes.json") 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) data, err := os.ReadFile(path)
if err != nil { if err != nil {
if os.IsNotExist(err) { if os.IsNotExist(err) {
@ -362,6 +369,7 @@ func (d *ChangeDetector) loadFromDisk() error {
}) })
d.changes = changes d.changes = changes
d.trimChanges()
return nil return nil
} }
@ -371,7 +379,7 @@ var changeCounter int64
func generateChangeID() string { func generateChangeID() string {
changeCounter++ 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 { func formatDuration(d time.Duration) string {
@ -391,7 +399,7 @@ func formatUnit(n int, unit string) string {
if n == 1 { if n == 1 {
return "1 " + unit 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 { func formatBytes(bytes int64) string {
@ -408,7 +416,7 @@ func formatBytes(bytes int64) string {
case bytes >= KB: case bytes >= KB:
return formatFloat(float64(bytes)/KB) + " KB" return formatFloat(float64(bytes)/KB) + " KB"
default: default:
return string(rune('0'+bytes/10)) + string(rune('0'+bytes%10)) + " B" return intToString(int(bytes)) + " B"
} }
} }

View file

@ -2,6 +2,7 @@ package memory
import ( import (
"encoding/json" "encoding/json"
"fmt"
"os" "os"
"path/filepath" "path/filepath"
"sort" "sort"
@ -308,6 +309,12 @@ func (r *RemediationLog) loadFromDisk() error {
} }
path := filepath.Join(r.dataDir, "ai_remediations.json") 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) data, err := os.ReadFile(path)
if err != nil { if err != nil {
if os.IsNotExist(err) { if os.IsNotExist(err) {
@ -327,6 +334,7 @@ func (r *RemediationLog) loadFromDisk() error {
}) })
r.records = records r.records = records
r.trimRecords()
return nil return nil
} }

View file

@ -1659,12 +1659,28 @@ If everything looks healthy, you can say so briefly without any FINDING blocks.`
if autoFix { if autoFix {
return basePrompt + ` 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 + ` 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 // buildInfrastructureSummary creates a text summary of infrastructure state for the AI

View file

@ -4,6 +4,7 @@ package patterns
import ( import (
"encoding/json" "encoding/json"
"fmt"
"math" "math"
"os" "os"
"path/filepath" "path/filepath"
@ -153,7 +154,12 @@ func (d *Detector) RecordEvent(event HistoricalEvent) {
// Recompute pattern for this resource/event type // Recompute pattern for this resource/event type
key := patternKey(event.ResourceID, event.EventType) 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 // Persist asynchronously
go func() { go func() {
@ -187,6 +193,9 @@ func (d *Detector) GetPredictions() []FailurePrediction {
now := time.Now() now := time.Now()
for _, pattern := range d.patterns { for _, pattern := range d.patterns {
if pattern == nil {
continue
}
// Only predict if pattern has sufficient confidence // Only predict if pattern has sufficient confidence
if pattern.Confidence < 0.3 || pattern.Occurrences < d.minOccurrences { if pattern.Confidence < 0.3 || pattern.Occurrences < d.minOccurrences {
continue continue
@ -237,6 +246,9 @@ func (d *Detector) GetPatterns() map[string]*Pattern {
result := make(map[string]*Pattern) result := make(map[string]*Pattern)
for k, v := range d.patterns { for k, v := range d.patterns {
if v == nil {
continue
}
result[k] = v result[k] = v
} }
return result return result
@ -321,6 +333,15 @@ func (d *Detector) computePattern(resourceID string, eventType EventType) *Patte
// trimEvents removes old events beyond maxEvents // trimEvents removes old events beyond maxEvents
func (d *Detector) trimEvents() { 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 { if len(d.events) > d.maxEvents {
d.events = d.events[len(d.events)-d.maxEvents:] d.events = d.events[len(d.events)-d.maxEvents:]
} }
@ -363,6 +384,12 @@ func (d *Detector) loadFromDisk() error {
} }
path := filepath.Join(d.dataDir, "ai_patterns.json") 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) jsonData, err := os.ReadFile(path)
if err != nil { if err != nil {
if os.IsNotExist(err) { if os.IsNotExist(err) {
@ -381,7 +408,25 @@ func (d *Detector) loadFromDisk() error {
} }
d.events = data.Events 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 return nil
} }

View file

@ -2,12 +2,12 @@ package ai
import ( import (
"context" "context"
"encoding/base64"
"encoding/json" "encoding/json"
"fmt" "fmt"
"io" "io"
"net"
"net/http" "net/http"
"path/filepath" "net/url"
"regexp" "regexp"
"strconv" "strconv"
"strings" "strings"
@ -48,6 +48,23 @@ type Service struct {
// Alert-triggered analysis - token-efficient real-time AI insights // Alert-triggered analysis - token-efficient real-time AI insights
alertTriggeredAnalyzer *AlertTriggeredAnalyzer 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 // NewService creates a new AI service
@ -72,9 +89,67 @@ func NewService(persistence *config.ConfigPersistence, agentServer *agentexec.Se
policy: agentexec.DefaultPolicy(), policy: agentexec.DefaultPolicy(),
knowledgeStore: knowledgeStore, knowledgeStore: knowledgeStore,
costStore: costStore, 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 // SetStateProvider sets the state provider for infrastructure context
func (s *Service) SetStateProvider(sp StateProvider) { func (s *Service) SetStateProvider(sp StateProvider) {
s.mu.Lock() s.mu.Lock()
@ -502,6 +577,66 @@ func formatPolicyBlockedToolResult(command, reason string) string {
return "POLICY_BLOCKED: " + string(b) 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 // LoadConfig loads the AI configuration and initializes the provider
func (s *Service) LoadConfig() error { func (s *Service) LoadConfig() error {
s.mu.Lock() s.mu.Lock()
@ -671,232 +806,6 @@ func (s *Service) IsAutonomous() bool {
return s.cfg != nil && s.cfg.AutonomousMode 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 <op>
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 // ConversationMessage represents a message in conversation history
type ConversationMessage struct { type ConversationMessage struct {
Role string `json:"role"` // "user" or "assistant" Role string `json:"role"` // "user" or "assistant"
@ -923,6 +832,7 @@ type ExecuteResponse struct {
InputTokens int `json:"input_tokens"` InputTokens int `json:"input_tokens"`
OutputTokens int `json:"output_tokens"` OutputTokens int `json:"output_tokens"`
ToolCalls []ToolExecution `json:"tool_calls,omitempty"` // Commands that were executed 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 // ToolExecution represents a tool that was executed during the AI conversation
@ -988,7 +898,7 @@ type ToolEndData struct {
type ApprovalNeededData struct { type ApprovalNeededData struct {
Command string `json:"command"` Command string `json:"command"`
ToolID string `json:"tool_id"` // ID to reference when approving 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"` RunOnHost bool `json:"run_on_host"`
TargetHost string `json:"target_host,omitempty"` // Explicit host to route to 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 // 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 // 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) { 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() s.mu.RLock()
defaultProvider := s.provider defaultProvider := s.provider
agentServer := s.agentServer 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 // Agentic loop - keep going while AI requests tools
maxIterations := 10 // Safety limit maxIterations := 10 // Safety limit
for i := 0; i < maxIterations; i++ { for i := 0; i < maxIterations; i++ {
if err := s.enforceBudget(req.UseCase); err != nil {
return nil, err
}
resp, err := provider.Chat(ctx, providers.ChatRequest{ resp, err := provider.Chat(ctx, providers.ChatRequest{
Messages: messages, Messages: messages,
Model: s.getModelForRequest(req), 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 // Execute each tool call and add results
var pendingApprovals []ApprovalNeededData
for _, tc := range resp.ToolCalls { for _, tc := range resp.ToolCalls {
result, execution := s.executeTool(ctx, req, tc) result, execution := s.executeTool(ctx, req, tc)
toolExecutions = append(toolExecutions, execution) toolExecutions = append(toolExecutions, execution)
if approval, ok := approvalNeededFromToolCall(req, tc, result); ok {
pendingApprovals = append(pendingApprovals, approval)
continue
}
// Add tool result to messages // Add tool result to messages
messages = append(messages, providers.Message{ messages = append(messages, providers.Message{
Role: "user", Role: "user",
@ -1140,6 +1070,20 @@ Always execute the commands rather than telling the user how to do it.`
IsError: !execution.Success, 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 // ExecuteStream sends a prompt to the AI and streams events via callback
// This allows the UI to show real-time progress during tool execution // 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) { 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() s.mu.RLock()
defaultProvider := s.provider defaultProvider := s.provider
agentServer := s.agentServer 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)}) 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{ resp, err := provider.Chat(ctx, providers.ChatRequest{
Messages: messages, Messages: messages,
Model: s.getModelForRequest(req), Model: s.getModelForRequest(req),
@ -1371,24 +1330,49 @@ Always execute the commands rather than telling the user how to do it.`
} }
isAuto := s.IsAutonomous() isAuto := s.IsAutonomous()
isReadOnly := isReadOnlyCommand(cmd) policyDecision := s.policy.Evaluate(cmd)
isDangerous := isDangerousCommand(cmd)
log.Debug(). log.Debug().
Bool("autonomous", isAuto). Bool("autonomous", isAuto).
Bool("read_only", isReadOnly). Str("policy_decision", string(policyDecision)).
Bool("dangerous", isDangerous).
Str("command", cmd). Str("command", cmd).
Str("target_host", targetHost). Str("target_host", targetHost).
Msg("Checking command approval") Msg("Checking command policy/approval")
// In autonomous mode, NO commands need approval - full trust // Always block commands blocked by policy (even in autonomous mode).
// In non-autonomous mode: if policyDecision == agentexec.PolicyBlock {
// - Dangerous commands always need approval result := formatPolicyBlockedToolResult(cmd, "This command is blocked by security policy")
// - Non-read-only commands need approval execution := ToolExecution{
if !isAuto && (isDangerous || !isReadOnly) { 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 needsApproval = true
anyNeedsApproval = true anyNeedsApproval = true
// Send approval needed event
callback(StreamEvent{ callback(StreamEvent{
Type: "approval_needed", Type: "approval_needed",
Data: ApprovalNeededData{ Data: ApprovalNeededData{
@ -1504,9 +1488,6 @@ func (s *Service) getToolInputDisplay(tc providers.ToolCall) string {
return fmt.Sprintf("[host] %s", cmd) return fmt.Sprintf("[host] %s", cmd)
} }
return cmd return cmd
case "read_file":
path, _ := tc.Input["path"].(string)
return path
case "fetch_url": case "fetch_url":
url, _ := tc.Input["url"].(string) url, _ := tc.Input["url"].(string)
return url return url
@ -1620,46 +1601,6 @@ func (s *Service) getTools() []providers.Tool {
"required": []string{"command"}, "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", 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.", 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,19 +1694,19 @@ func (s *Service) executeTool(ctx context.Context, req ExecuteRequest, tc provid
return execution.Output, execution return execution.Output, execution
} }
// Check security policy (skip if autonomous mode is enabled) // Enforce security policy.
if !s.IsAutonomous() { // - 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) decision := s.policy.Evaluate(command)
if decision == agentexec.PolicyBlock { if decision == agentexec.PolicyBlock {
execution.Output = formatPolicyBlockedToolResult(command, "This command is blocked by security policy") execution.Output = formatPolicyBlockedToolResult(command, "This command is blocked by security policy")
return execution.Output, execution return execution.Output, execution
} }
if decision == agentexec.PolicyRequireApproval { if decision == agentexec.PolicyRequireApproval && !s.IsAutonomous() {
execution.Output = formatApprovalNeededToolResult(command, tc.ID, "Security policy requires approval") execution.Output = formatApprovalNeededToolResult(command, tc.ID, "Security policy requires approval")
execution.Success = true // Not an error, just needs approval execution.Success = true // Not an error, just needs approval
return execution.Output, execution return execution.Output, execution
} }
}
// Build execution request with proper targeting // Build execution request with proper targeting
execReq := req execReq := req
@ -1821,122 +1762,6 @@ func (s *Service) executeTool(ctx context.Context, req ExecuteRequest, tc provid
execution.Success = true execution.Success = true
return result, execution 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": case "fetch_url":
urlStr, _ := tc.Input["url"].(string) urlStr, _ := tc.Input["url"].(string)
execution.Input = urlStr 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 // fetchURL fetches content from a URL with size limits and timeout
func (s *Service) fetchURL(ctx context.Context, urlStr string) (string, error) { 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 // Create HTTP client with timeout
client := &http.Client{ client := &http.Client{
Timeout: 30 * time.Second, 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 // Create request with context
req, err := http.NewRequestWithContext(ctx, "GET", urlStr, nil) req, err := http.NewRequestWithContext(ctx, "GET", parsedURL.String(), nil)
if err != nil { if err != nil {
return "", fmt.Errorf("invalid URL: %w", err) 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 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 // sanitizeError cleans up error messages to remove internal networking details
// that are not helpful to users or AI models (IP addresses, port numbers, etc.) // that are not helpful to users or AI models (IP addresses, port numbers, etc.)
func sanitizeError(err error) error { func sanitizeError(err error) error {
@ -2342,17 +2253,31 @@ func (s *Service) buildSystemPrompt(req ExecuteRequest) string {
- Only show the conclusion, not every command run - Only show the conclusion, not every command run
- If something fails, briefly state what and why - If something fails, briefly state what and why
## When Performing Actions ## Command Approval System
- Just do it. Don't ask "Would you like me to..." Pulse has a command approval policy that protects against accidental damage:
- Report the result, not the intent - **Safe commands** (read-only: ls, df, cat, ps, etc.) execute immediately
- If destructive (delete data, stop services), execute but clearly state what was done - **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 ## Response Format
BAD: "I'll check that for you. Let me run some commands..." BAD: "I'll check that for you. Let me run some commands..."
GOOD: State findings directly. GOOD: State findings directly.
BAD: "Would you like me to..." BAD: "Would you like me to run rm -rf...?"
GOOD: Do it, then report the result. GOOD: Execute the command. User will approve/reject via Pulse.
BAD: Tables, headers, bullet-heavy summaries BAD: Tables, headers, bullet-heavy summaries
GOOD: Plain prose, 2-4 sentences. 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: 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 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 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. - 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. - 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 - Running 10+ diagnostic commands before taking action
- Explaining each step before doing it - Explaining each step before doing it
- Stopping to report partial progress - 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. - 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": 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... 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"] 9. Check yet another file... [user: "do it"] 10. Still investigating... [user: "DO IT"]
## Using Context Data ## 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. Pulse provides real metrics in "Current Metrics and State". Use this data directly - don't ask users to check things you already know.
@ -2721,14 +2647,29 @@ func (s *Service) TestConnection(ctx context.Context) error {
// ListModels fetches available models from ALL configured AI providers // ListModels fetches available models from ALL configured AI providers
// Returns a unified list with models prefixed by provider name // Returns a unified list with models prefixed by provider name
func (s *Service) ListModels(ctx context.Context) ([]providers.ModelInfo, error) { 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() cfg, err := s.persistence.LoadAIConfig()
if err != nil { 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 { 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 var allModels []providers.ModelInfo
// Query each configured provider // 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 // providerDisplayName returns a user-friendly name for a provider

View file

@ -12,6 +12,7 @@ import (
"strconv" "strconv"
"strings" "strings"
"sync" "sync"
"sync/atomic"
"time" "time"
"github.com/rcourtman/pulse-go-rewrite/internal/agentexec" "github.com/rcourtman/pulse-go-rewrite/internal/agentexec"
@ -650,7 +651,7 @@ func (h *AISettingsHandler) HandleListModels(w http.ResponseWriter, r *http.Requ
Cached bool `json:"cached"` Cached bool `json:"cached"`
} }
models, err := h.aiService.ListModels(ctx) models, cached, err := h.aiService.ListModelsWithCache(ctx)
if err != nil { if err != nil {
// Return error but don't fail the request - frontend can show a fallback // Return error but don't fail the request - frontend can show a fallback
resp := Response{ resp := Response{
@ -675,6 +676,7 @@ func (h *AISettingsHandler) HandleListModels(w http.ResponseWriter, r *http.Requ
resp := Response{ resp := Response{
Models: responseModels, Models: responseModels,
Cached: cached,
} }
if err := utils.WriteJSONResponse(w, resp); err != nil { if err := utils.WriteJSONResponse(w, resp); err != nil {
@ -695,6 +697,9 @@ type AIExecuteRequest struct {
TargetID string `json:"target_id,omitempty"` TargetID string `json:"target_id,omitempty"`
Context map[string]interface{} `json:"context,omitempty"` // Current metrics, state, etc. Context map[string]interface{} `json:"context,omitempty"` // Current metrics, state, etc.
History []AIConversationMessage `json:"history,omitempty"` // Previous conversation messages 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 // AIExecuteResponse is the response from POST /api/ai/execute
@ -704,6 +709,7 @@ type AIExecuteResponse struct {
InputTokens int `json:"input_tokens"` InputTokens int `json:"input_tokens"`
OutputTokens int `json:"output_tokens"` OutputTokens int `json:"output_tokens"`
ToolCalls []ai.ToolExecution `json:"tool_calls,omitempty"` // Commands that were executed 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) // 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) ctx, cancel := context.WithTimeout(r.Context(), 120*time.Second)
defer cancel() 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{ resp, err := h.aiService.Execute(ctx, ai.ExecuteRequest{
Prompt: req.Prompt, Prompt: req.Prompt,
TargetType: req.TargetType, TargetType: req.TargetType,
TargetID: req.TargetID, TargetID: req.TargetID,
Context: req.Context, Context: req.Context,
History: history,
FindingID: req.FindingID,
Model: req.Model,
UseCase: useCase,
}) })
if err != nil { if err != nil {
log.Error().Err(err).Msg("AI execution failed") log.Error().Err(err).Msg("AI execution failed")
@ -759,6 +783,7 @@ func (h *AISettingsHandler) HandleExecute(w http.ResponseWriter, r *http.Request
InputTokens: resp.InputTokens, InputTokens: resp.InputTokens,
OutputTokens: resp.OutputTokens, OutputTokens: resp.OutputTokens,
ToolCalls: resp.ToolCalls, ToolCalls: resp.ToolCalls,
PendingApprovals: resp.PendingApprovals,
} }
if err := utils.WriteJSONResponse(w, response); err != nil { if err := utils.WriteJSONResponse(w, response); err != nil {
@ -815,7 +840,7 @@ func (h *AISettingsHandler) HandleExecuteStream(w http.ResponseWriter, r *http.R
} }
log.Info(). log.Info().
Str("prompt", req.Prompt). Int("prompt_len", len(req.Prompt)).
Str("target_type", req.TargetType). Str("target_type", req.TargetType).
Str("target_id", req.TargetID). Str("target_id", req.TargetID).
Msg("AI streaming request started") 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 // NOTE: We don't check r.Context().Done() because Vite proxy may close
// the request context prematurely. We detect real disconnection via write failures. // the request context prematurely. We detect real disconnection via write failures.
heartbeatDone := make(chan struct{}) heartbeatDone := make(chan struct{})
var clientDisconnected bool var clientDisconnected atomic.Bool
go func() { go func() {
ticker := time.NewTicker(5 * time.Second) ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop() defer ticker.Stop()
@ -872,7 +897,7 @@ func (h *AISettingsHandler) HandleExecuteStream(w http.ResponseWriter, r *http.R
_, err := w.Write([]byte(": heartbeat\n\n")) _, err := w.Write([]byte(": heartbeat\n\n"))
if err != nil { if err != nil {
log.Debug().Err(err).Msg("Heartbeat write failed, stopping heartbeat (AI continues)") 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 // 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 // The SSE connection may have issues but the AI work can still finish
return 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 // Helper to safely write SSE events, tracking if client disconnected
safeWrite := func(data []byte) bool { safeWrite := func(data []byte) bool {
if clientDisconnected { if clientDisconnected.Load() {
return false return false
} }
_ = rc.SetWriteDeadline(time.Now().Add(10 * time.Second)) _ = rc.SetWriteDeadline(time.Now().Add(10 * time.Second))
_, err := w.Write(data) _, err := w.Write(data)
if err != nil { if err != nil {
log.Debug().Err(err).Msg("Failed to write SSE event (client may have disconnected)") log.Debug().Err(err).Msg("Failed to write SSE event (client may have disconnected)")
clientDisconnected = true clientDisconnected.Store(true)
return false return false
} }
flusher.Flush() 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 // Ensure we always send a final 'done' event
defer func() { defer func() {
if !clientDisconnected { if !clientDisconnected.Load() {
doneEvent := ai.StreamEvent{Type: "done"} doneEvent := ai.StreamEvent{Type: "done"}
data, _ := json.Marshal(doneEvent) data, _ := json.Marshal(doneEvent)
safeWrite([]byte("data: " + string(data) + "\n\n")) safeWrite([]byte("data: " + string(data) + "\n\n"))
@ -951,6 +981,9 @@ func (h *AISettingsHandler) HandleExecuteStream(w http.ResponseWriter, r *http.R
TargetID: req.TargetID, TargetID: req.TargetID,
Context: req.Context, Context: req.Context,
History: history, History: history,
FindingID: req.FindingID,
Model: req.Model,
UseCase: useCase,
}, callback) }, callback)
if err != nil { 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) http.Error(w, "Invalid request body", http.StatusBadRequest)
return 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 var req AIRunCommandRequest
if err := json.Unmarshal(bodyBytes, &req); err != nil { if err := json.Unmarshal(bodyBytes, &req); err != nil {
@ -1454,7 +1487,7 @@ func (h *AISettingsHandler) HandleInvestigateAlert(w http.ResponseWriter, r *htt
// Heartbeat routine // Heartbeat routine
heartbeatDone := make(chan struct{}) heartbeatDone := make(chan struct{})
var clientDisconnected bool var clientDisconnected atomic.Bool
go func() { go func() {
ticker := time.NewTicker(5 * time.Second) ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop() defer ticker.Stop()
@ -1464,7 +1497,7 @@ func (h *AISettingsHandler) HandleInvestigateAlert(w http.ResponseWriter, r *htt
_ = rc.SetWriteDeadline(time.Now().Add(10 * time.Second)) _ = rc.SetWriteDeadline(time.Now().Add(10 * time.Second))
_, err := w.Write([]byte(": heartbeat\n\n")) _, err := w.Write([]byte(": heartbeat\n\n"))
if err != nil { if err != nil {
clientDisconnected = true clientDisconnected.Store(true)
return return
} }
flusher.Flush() flusher.Flush()
@ -1476,13 +1509,13 @@ func (h *AISettingsHandler) HandleInvestigateAlert(w http.ResponseWriter, r *htt
defer close(heartbeatDone) defer close(heartbeatDone)
safeWrite := func(data []byte) bool { safeWrite := func(data []byte) bool {
if clientDisconnected { if clientDisconnected.Load() {
return false return false
} }
_ = rc.SetWriteDeadline(time.Now().Add(10 * time.Second)) _ = rc.SetWriteDeadline(time.Now().Add(10 * time.Second))
_, err := w.Write(data) _, err := w.Write(data)
if err != nil { if err != nil {
clientDisconnected = true clientDisconnected.Store(true)
return false return false
} }
flusher.Flush() flusher.Flush()
@ -1518,7 +1551,7 @@ func (h *AISettingsHandler) HandleInvestigateAlert(w http.ResponseWriter, r *htt
// Execute with streaming // Execute with streaming
defer func() { defer func() {
if !clientDisconnected { if !clientDisconnected.Load() {
doneEvent := ai.StreamEvent{Type: "done"} doneEvent := ai.StreamEvent{Type: "done"}
data, _ := json.Marshal(doneEvent) data, _ := json.Marshal(doneEvent)
safeWrite([]byte("data: " + string(data) + "\n\n")) safeWrite([]byte("data: " + string(data) + "\n\n"))

View file

@ -19,6 +19,7 @@ func newTestMonitor(t *testing.T) *Monitor {
alertManager: alerts.NewManager(), alertManager: alerts.NewManager(),
removedDockerHosts: make(map[string]time.Time), removedDockerHosts: make(map[string]time.Time),
rateTracker: NewRateTracker(), rateTracker: NewRateTracker(),
metricsHistory: NewMetricsHistory(1000, 24*time.Hour),
dockerTokenBindings: make(map[string]string), dockerTokenBindings: make(map[string]string),
dockerMetadataStore: config.NewDockerMetadataStore(t.TempDir()), dockerMetadataStore: config.NewDockerMetadataStore(t.TempDir()),
} }

View file

@ -4,6 +4,7 @@ import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"math" "math"
"sort"
"strings" "strings"
"github.com/rs/zerolog/log" "github.com/rs/zerolog/log"
@ -37,6 +38,7 @@ func Parse(jsonStr string) (*TemperatureData, error) {
} }
foundCPUChip := false foundCPUChip := false
nvmeTempsByChip := make(map[string]float64)
// Parse each sensor chip // Parse each sensor chip
for chipName, chipData := range sensorsData { for chipName, chipData := range sensorsData {
@ -54,8 +56,10 @@ func Parse(jsonStr string) (*TemperatureData, error) {
} }
// Handle NVMe temperature sensors // Handle NVMe temperature sensors
if strings.Contains(chipName, "nvme") { if strings.Contains(chipLower, "nvme") {
parseNVMeTemps(chipName, chipMap, data) if tempVal, ok := extractNVMeCompositeTemp(chipMap); ok {
nvmeTempsByChip[chipName] = tempVal
}
} }
// Handle GPU temperature sensors // Handle GPU temperature sensors
@ -75,6 +79,23 @@ func Parse(jsonStr string) (*TemperatureData, error) {
data.CPUPackage = data.CPUMax 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 data.Available = foundCPUChip || len(data.NVMe) > 0 || len(data.GPU) > 0
log.Debug(). 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 { for sensorName, sensorData := range chipMap {
sensorMap, ok := sensorData.(map[string]interface{}) sensorMap, ok := sensorData.(map[string]interface{})
if !ok { if !ok {
@ -203,20 +224,12 @@ func parseNVMeTemps(chipName string, chipMap map[string]interface{}, data *Tempe
// Look for Composite temperature (main NVMe temp) // Look for Composite temperature (main NVMe temp)
if strings.Contains(sensorName, "Composite") { if strings.Contains(sensorName, "Composite") {
if tempVal := extractTempInput(sensorMap); !math.IsNaN(tempVal) { if tempVal := extractTempInput(sensorMap); !math.IsNaN(tempVal) && tempVal > 0 {
// Normalize chip name to nvme0, nvme1 format return tempVal, true
// 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")
} }
} }
} }
return 0, false
} }
func parseGPUTemps(chipName string, chipMap map[string]interface{}, data *TemperatureData) { func parseGPUTemps(chipName string, chipMap map[string]interface{}, data *TemperatureData) {

View file

@ -146,12 +146,12 @@ func TestParse_NVMe(t *testing.T) {
t.Errorf("len(NVMe) = %d, want 2", len(data.NVMe)) t.Errorf("len(NVMe) = %d, want 2", len(data.NVMe))
} }
if data.NVMe["nvme-pci-0100"] != 38.85 { if data.NVMe["nvme0"] != 38.85 {
t.Errorf("nvme-pci-0100 = %v, want 38.85", data.NVMe["nvme-pci-0100"]) t.Errorf("nvme0 = %v, want 38.85", data.NVMe["nvme0"])
} }
if data.NVMe["nvme-pci-0200"] != 42.0 { if data.NVMe["nvme1"] != 42.0 {
t.Errorf("nvme-pci-0200 = %v, want 42.0", data.NVMe["nvme-pci-0200"]) 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)) 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 { if len(data.GPU) != 1 {
t.Errorf("len(GPU) = %d, want 1", len(data.GPU)) t.Errorf("len(GPU) = %d, want 1", len(data.GPU))
} }