package web import ( "bufio" "encoding/json" "os" "strings" "sync" "time" ) // LogEntry represents a single parsed log line. type LogEntry struct { Timestamp time.Time `json:"timestamp"` Level string `json:"level"` // "DEBUG", "INFO", "WARN", "ERROR" Message string `json:"message"` Source string `json:"source"` // "file.go:123" if Lshortfile enabled } // LogBuffer is a thread-safe ring buffer that captures log output. // It implements io.Writer so it can be used with log.New(io.MultiWriter(...)). type LogBuffer struct { mu sync.RWMutex entries []LogEntry size int pos int full bool } // NewLogBuffer creates a ring buffer that keeps the last `size` log entries. func NewLogBuffer(size int) *LogBuffer { return &LogBuffer{ entries: make([]LogEntry, size), size: size, } } // Write implements io.Writer. It parses Go's standard log output format. // Handles two formats: // - With Lshortfile: "2026/02/21 18:33:35 file.go:123: [LEVEL] message" // - Without: "2026/02/21 18:33:35 [LEVEL] message" func (lb *LogBuffer) Write(p []byte) (n int, err error) { line := strings.TrimRight(string(p), "\n\r") if line == "" { return len(p), nil } entry := parseLine(line) // fix-6 (CAMPAIGN-3): TRACE lines (periodic-job routine success — the every-cycle scheduler + // refreshStatusLocked noise) are DROPPED from the ring so the finite capacity holds SIGNAL, not // "nothing happened, again". They still reach stdout if the configured level allows. Failures and // state changes are never TRACE, so this never loses them. if levelPriority(entry.Level) < levelPriority("DEBUG") { return len(p), nil } lb.mu.Lock() lb.entries[lb.pos] = entry lb.pos = (lb.pos + 1) % lb.size if lb.pos == 0 && !lb.full { lb.full = true } lb.mu.Unlock() return len(p), nil } // Entries returns log entries filtered by minimum level, limited by count, // and optionally filtered to entries after a given timestamp. // Returns the matching entries and the total count in the buffer. func (lb *LogBuffer) Entries(minLevel string, limit int, after time.Time) ([]LogEntry, int) { lb.mu.RLock() defer lb.mu.RUnlock() // Collect all entries in chronological order total := lb.size if !lb.full { total = lb.pos } if limit <= 0 || limit > 1000 { limit = 200 } levelOrder := levelPriority(minLevel) var result []LogEntry start := 0 if lb.full { start = lb.pos } for i := 0; i < total; i++ { idx := (start + i) % lb.size e := lb.entries[idx] // Filter by level if levelPriority(e.Level) < levelOrder { continue } // Filter by timestamp if !after.IsZero() && !e.Timestamp.After(after) { continue } result = append(result, e) } // Apply limit (keep the most recent entries) if len(result) > limit { result = result[len(result)-limit:] } return result, total } // Lines renders the held entries as plain text lines (chronological), dropping // from the HEAD (oldest) to honor maxBytes so the newest lines survive — the // report-channel controller_log_tail budget (maxBytes ≤ 0 → no byte cap). // Format mirrors the agent ring's: " [] ". func (lb *LogBuffer) Lines(maxBytes int) []string { lb.mu.RLock() total := lb.size start := 0 if !lb.full { total = lb.pos } else { start = lb.pos } entries := make([]LogEntry, 0, total) for i := 0; i < total; i++ { entries = append(entries, lb.entries[(start+i)%lb.size]) } lb.mu.RUnlock() lines := make([]string, len(entries)) for i, e := range entries { src := "" if e.Source != "" { src = e.Source + ": " } lines[i] = e.Timestamp.Format(time.RFC3339) + " [" + e.Level + "] " + src + e.Message } if maxBytes <= 0 { return lines } budget := 0 keepFrom := len(lines) for i := len(lines) - 1; i >= 0; i-- { budget += len(lines[i]) + 1 // +1 for the newline it represents if budget > maxBytes { break } keepFrom = i } return lines[keepFrom:] } // snapshot returns the held entries in chronological order (oldest→newest). func (lb *LogBuffer) snapshot() []LogEntry { lb.mu.RLock() defer lb.mu.RUnlock() total := lb.size start := 0 if !lb.full { total = lb.pos } else { start = lb.pos } out := make([]LogEntry, 0, total) for i := 0; i < total; i++ { out = append(out, lb.entries[(start+i)%lb.size]) } return out } // SpillTo persists the ring to path as JSON-lines (one entry per line), newest-last, via an atomic // tmp+rename so a crash mid-write never corrupts the on-disk ring (fix-6, CAMPAIGN-3). The path MUST // be on the SSD/state dir — NEVER a NAS/HDD path (a ring that dies with the NAS is worse than none). func (lb *LogBuffer) SpillTo(path string) error { entries := lb.snapshot() tmp := path + ".tmp" f, err := os.Create(tmp) if err != nil { return err } w := bufio.NewWriter(f) enc := json.NewEncoder(w) for _, e := range entries { if err := enc.Encode(e); err != nil { f.Close() os.Remove(tmp) return err } } if err := w.Flush(); err != nil { f.Close() os.Remove(tmp) return err } if err := f.Sync(); err != nil { f.Close() os.Remove(tmp) return err } if err := f.Close(); err != nil { os.Remove(tmp) return err } return os.Rename(tmp, path) } // LoadFrom seeds the ring from a spill file written by SpillTo, keeping the newest `size` entries so // the pre-restart window is present in the viewer immediately (fix-6). Corruption-safe: a truncated // or partially-written line simply fails to parse and is SKIPPED — the ring loads what is valid and // NEVER panics or errors fatally (a missing file is a clean no-op). Call once at startup, before the // logger writes anything. func (lb *LogBuffer) LoadFrom(path string) { f, err := os.Open(path) if err != nil { return // no prior spill (fresh box / first boot) — clean no-op } defer f.Close() var loaded []LogEntry sc := bufio.NewScanner(f) sc.Buffer(make([]byte, 0, 64*1024), 1024*1024) // tolerate long lines for sc.Scan() { var e LogEntry if err := json.Unmarshal(sc.Bytes(), &e); err != nil { continue // skip a corrupt/truncated line — never fatal } loaded = append(loaded, e) } if len(loaded) == 0 { return } if len(loaded) > lb.size { loaded = loaded[len(loaded)-lb.size:] // keep the newest `size` } lb.mu.Lock() defer lb.mu.Unlock() for i, e := range loaded { lb.entries[i] = e } if len(loaded) == lb.size { lb.pos = 0 lb.full = true } else { lb.pos = len(loaded) lb.full = false } } // parseLine parses a single log line into a LogEntry. func parseLine(line string) LogEntry { entry := LogEntry{ Level: "INFO", Message: line, } // Try to parse timestamp: "2006/01/02 15:04:05" // Use Local timezone because Go's log.LstdFlags outputs in local time. if len(line) >= 19 { if t, err := time.ParseInLocation("2006/01/02 15:04:05", line[:19], time.Local); err == nil { entry.Timestamp = t rest := line[19:] if len(rest) > 0 && rest[0] == ' ' { rest = rest[1:] } // Check for source file (Lshortfile): "file.go:123: [LEVEL] ..." if colonIdx := strings.Index(rest, ": "); colonIdx > 0 && colonIdx < 40 { candidate := rest[:colonIdx] // Source file pattern: contains ".go:" or ".go" before the colon if strings.Contains(candidate, ".go:") || strings.HasSuffix(candidate, ".go") { entry.Source = candidate rest = rest[colonIdx+2:] } } // Extract level tag: [DEBUG], [INFO], [WARN], [ERROR], [SYNC], [SCHED], etc. entry.Level, entry.Message = extractLevel(rest) } } if entry.Timestamp.IsZero() { entry.Timestamp = time.Now() } return entry } // extractLevel finds and removes a [LEVEL] tag from the beginning of a string. func extractLevel(s string) (string, string) { s = strings.TrimSpace(s) if len(s) < 3 || s[0] != '[' { return "INFO", s } end := strings.Index(s, "]") if end < 0 || end > 20 { return "INFO", s } tag := s[1:end] msg := strings.TrimSpace(s[end+1:]) switch tag { case "TRACE": return "TRACE", msg case "DEBUG": return "DEBUG", msg case "INFO": return "INFO", msg case "WARN": return "WARN", msg case "ERROR": return "ERROR", msg case "FATAL": return "ERROR", msg default: // Tags like [SYNC], [SCHED], [STORAGE] etc. — treat as INFO, keep tag in message return "INFO", s } } // levelPriority returns numeric priority for log levels. func levelPriority(level string) int { switch strings.ToUpper(level) { case "TRACE": return -1 // below DEBUG — dropped from the ring (fix-6 periodic-noise policy) case "DEBUG": return 0 case "INFO": return 1 case "WARN": return 2 case "ERROR": return 3 default: return 0 } }