From 69d2d5f76097f746c293336357868345ece8f87c Mon Sep 17 00:00:00 2001 From: Nadav Erell Date: Thu, 2 Jul 2026 02:20:19 +0300 Subject: [PATCH 01/11] Diagnose: persist investigations to SQLite so history survives restarts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Runs were RAM-only: a Radar restart lost every transcript, verdict, and the sessionId that makes "continue in your agent" work — even though the agent CLIs' own sessions survive on disk. Persist them to ~/.radar/ai-runs.db (modernc sqlite, 0700/0600, default-on, --ai-history / aiHistory config to disable). Design (settled via plan-loop, 2 codex cross-review cycles): - Two tables: runs (summary_json) + run_events keyed (run_id, seq). Writes go through an async single-writer goroutine, enqueued under the run mutex so per-run order matches seq; terminal events commit WITH their status in one transaction, so crash recovery can trust that a "running" row has no terminal marker. Loads drain the queue first (read-your-writes). - Hydration: run rows load eagerly at startup (seeds nextID past persisted ids; interrupted "running" rows flip to error + a restart marker; Cursor sessions are cleared — workspace-scoped resume can't survive the temp workdir). Event logs hydrate lazily on first Subscribe/turn/Stop. - History from another kube-context loads view-only: a lazy sweep (get/List) marks it stale with store-assigned terminal markers, and beginTurn refuses cross-context turns. - Shutdown marks in-flight runs stopped BEFORE cancelling so goroutines don't persist spurious "context canceled" errors, then drains and closes the store. - Retention: 100 runs / 30 days; eviction deletes rows. Store failure degrades loudly to memory-only (log + historyDegraded on the list response + panel notice). - UX: consent card now discloses local retention; Settings gains a confirmed "Clear history" action (POST /api/diagnose/history/clear — wipes finished runs, live ones survive and re-persist); stale copy updated. --- cmd/desktop/main.go | 2 + cmd/explorer/main.go | 4 + internal/ai/runs.go | 319 ++++++++++++++++- internal/ai/runs_test.go | 208 ++++++++++- internal/ai/store.go | 326 ++++++++++++++++++ internal/ai/store_test.go | 115 ++++++ internal/app/bootstrap.go | 14 + internal/config/config.go | 13 + internal/server/ai_diagnose.go | 26 +- internal/server/server.go | 15 +- web/src/api/diagnose.ts | 21 +- web/src/components/diagnose/AISettings.tsx | 74 +++- .../components/diagnose/DiagnoseContext.tsx | 8 +- .../components/diagnose/DiagnoseSurface.tsx | 2 + web/src/components/diagnose/Home.tsx | 17 +- .../components/diagnose/InvestigationView.tsx | 6 +- web/src/components/diagnose/parts.tsx | 3 +- .../components/settings/SettingsDialog.tsx | 1 + 18 files changed, 1147 insertions(+), 27 deletions(-) create mode 100644 internal/ai/store.go create mode 100644 internal/ai/store_test.go diff --git a/cmd/desktop/main.go b/cmd/desktop/main.go index be50e626e..85c88d349 100644 --- a/cmd/desktop/main.go +++ b/cmd/desktop/main.go @@ -117,6 +117,8 @@ func main() { PrometheusHeadersFromEnv: fileCfg.PrometheusHeadersFromEnv, Version: version, MCPEnabled: fileCfg.MCPEnabledOr(true), + AIHistory: fileCfg.AIHistoryOr(true), + AIHistoryDBPath: fileCfg.AIHistoryDBPath, } app.SetGlobals(cfg) diff --git a/cmd/explorer/main.go b/cmd/explorer/main.go index f4258527d..3f42d2b4a 100644 --- a/cmd/explorer/main.go +++ b/cmd/explorer/main.go @@ -69,6 +69,8 @@ func main() { timelineDBPath := flag.String("timeline-db", fileCfg.TimelineDBPath, "Path to timeline database file (default: ~/.radar/timeline.db)") timelineRetention := flag.Duration("timeline-retention", fileCfg.TimelineRetentionOr(7*24*time.Hour), "How long to retain timeline events when --timeline-storage=sqlite (e.g. 168h, 720h). 0 disables age-based cleanup.") timelineMaxSize := flag.String("timeline-max-size", fileCfg.TimelineMaxSizeOr("1Gi"), "Maximum SQLite timeline storage size before pruning oldest events (e.g. 800Mi, 8Gi). 0 disables size-based pruning.") + // AI history (Diagnose investigations) + aiHistory := flag.Bool("ai-history", fileCfg.AIHistoryOr(true), "Persist AI investigations (transcripts + verdicts) to ~/.radar/ai-runs.db so they survive restarts") // Traffic/metrics options prometheusURL := flag.String("prometheus-url", fileCfg.PrometheusURL, "Manual Prometheus/VictoriaMetrics URL (skips auto-discovery)") // --prometheus-header Key=Value, repeatable. Defaults populated from @@ -222,6 +224,8 @@ func main() { PrometheusHeaders: resolvedPrometheusHeaders, PrometheusHeadersFromEnv: promHeadersFromEnv.value(), MCPEnabled: mcpEnabled, + AIHistory: *aiHistory, + AIHistoryDBPath: fileCfg.AIHistoryDBPath, Version: version, AuthConfig: auth.Config{ Mode: *authMode, diff --git a/internal/ai/runs.go b/internal/ai/runs.go index c6268492d..fb6e6a9f4 100644 --- a/internal/ai/runs.go +++ b/internal/ai/runs.go @@ -7,6 +7,7 @@ import ( "log" "os" "path/filepath" + "strconv" "strings" "sync" "time" @@ -40,12 +41,17 @@ type RunManager struct { // its own temp workspace per turn, losing cross-turn resume but staying correct). workRoot string + // store persists runs + event logs across restarts (nil = memory-only, + // the historical behavior). Owned here: Shutdown closes it. + store RunStore + mu sync.Mutex runs map[string]*Run order []string // insertion order, for eviction nextID int - maxRetained int // total runs kept in memory (running + finished) - maxConcurrent int // concurrent IN-FLIGHT turns (= live agent processes) + maxRetained int // total runs kept in memory (running + finished) + maxConcurrent int // concurrent IN-FLIGHT turns (= live agent processes) + sweptCtx string // last kube-context the loaded-run staleness sweep ran for } // Run is one investigation: identity, status, the agent session to resume, and the @@ -65,16 +71,24 @@ type Run struct { Health *ResourceHealthSignal CreatedAt time.Time + // store mirrors RunManager.store (nil = memory-only) so the event hot path + // can persist without reaching back to the manager. + store RunStore + mu sync.Mutex status string // running | done | error | stopped | stale sessionID string preview string // last root cause, for the list updatedAt time.Time events []RunEvent - inFlight bool - subs map[int]chan RunEvent - nextSub int - cancel context.CancelFunc + // hydrated marks that r.events holds the run's full log. Runs created live + // are born hydrated; runs loaded from the store hydrate lazily on first + // read/mutation (ensureHydrated) so startup doesn't pay for old transcripts. + hydrated bool + inFlight bool + subs map[int]chan RunEvent + nextSub int + cancel context.CancelFunc } // RunEvent is a sequenced stream event. Seq drives SSE id: / Last-Event-ID replay. @@ -118,8 +132,12 @@ var ( ) const ( - defaultMaxConcurrent = 3 // running child processes - defaultMaxRetained = 20 // total runs kept in memory + defaultMaxConcurrent = 3 // running child processes + defaultMaxRetained = 100 // total runs kept (memory rows + store) + + // defaultHistoryAge is how long finished runs are kept in the store; older + // ones are dropped at startup. Count-based eviction still applies first. + defaultHistoryAge = 30 * 24 * time.Hour // defaultTurnTimeout bounds one agent turn's wall-clock time. Generous — a // deep multi-tool investigation runs minutes, not tens of minutes — while @@ -140,7 +158,9 @@ func turnTimeout() time.Duration { // NewRunManager builds a manager over a resolved Diagnoser. mcpPort/ctxLabel are // callbacks because the listener port and kube-context are only known at runtime. -func NewRunManager(d *Diagnoser, mcpPort func() int, ctxLabel func() string) *RunManager { +// store persists history across restarts (nil = memory-only); persisted runs are +// hydrated into the manager here. +func NewRunManager(d *Diagnoser, mcpPort func() int, ctxLabel func() string, store RunStore) *RunManager { ctx, cancel := context.WithCancel(context.Background()) // Best-effort: a failure here just means runs get no shared workdir (logged). root, err := os.MkdirTemp("", "radar-ai-") @@ -148,17 +168,93 @@ func NewRunManager(d *Diagnoser, mcpPort func() int, ctxLabel func() string) *Ru log.Printf("[ai] could not create AI scratch root: %v (Cursor resume will be degraded)", err) root = "" } - return &RunManager{ + m := &RunManager{ d: d, mcpPort: mcpPort, ctxLabel: ctxLabel, baseCtx: ctx, baseCancel: cancel, workRoot: root, + store: store, runs: map[string]*Run{}, maxRetained: defaultMaxRetained, maxConcurrent: defaultMaxConcurrent, } + m.loadPersisted() + return m +} + +// loadPersisted hydrates run ROWS from the store (event logs stay lazy — see +// ensureHydrated) and normalizes state that can't carry across a process: +// - a persisted "running" run was interrupted by the restart → error, with a +// terminal event appended so replay still ends in a terminal marker; +// - Cursor sessions are workspace-scoped and the workspace was a process- +// lifetime temp dir → drop the sessionID so follow-ups report "no session" +// instead of spawning an agent guaranteed to fail; +// - nextID reseeds past every persisted id so new runs can't collide. +func (m *RunManager) loadPersisted() { + if m.store == nil { + return + } + sums, err := m.store.LoadRuns() + if err != nil { + log.Printf("[ai] could not load run history: %v", err) + return + } + cutoff := nowUTC().Add(-defaultHistoryAge) + for _, s := range sums { + if s.ID == "" { + continue + } + if n := runIDNum(s.ID); n > m.nextID { + m.nextID = n + } + if s.UpdatedAt.Before(cutoff) { + m.store.DeleteRun(s.ID) // age-based retention + continue + } + if s.Agent == "cursor-agent" { + s.SessionID = "" + } + r := &Run{ + ID: s.ID, Kind: s.Kind, Namespace: s.Namespace, Name: s.Name, + Context: s.Context, Agent: s.Agent, Isolated: s.Isolated, + Model: s.Model, Effort: s.Effort, ManagedBy: s.ManagedBy, + Health: s.Health, CreatedAt: s.CreatedAt, + store: m.store, + status: s.Status, sessionID: s.SessionID, preview: s.Preview, + updatedAt: s.UpdatedAt, + subs: map[int]chan RunEvent{}, + } + if s.Status == "running" { + // Interrupted by the restart. Terminal statuses are written in the + // same transaction as their terminal event, so a "running" row means + // the log has no terminal marker yet — append one (store-assigned + // seq; the in-memory log stays lazy). + r.status = "error" + r.updatedAt = nowUTC() + sum := r.summaryLocked() + m.store.AppendEvent(r.ID, RunEvent{Event: StreamEvent{ + Type: "error", + Error: "Radar restarted while this investigation was running. Re-run Diagnose to analyze the current cluster.", + }}, &sum) + } + if r.status == "stale" { + r.subs = nil // stale runs were finalized — streams replay then close + } + m.runs[r.ID] = r + m.order = append(m.order, r.ID) + } + m.evictLocked() // the retention cap may have shrunk since the DB was written +} + +// runIDNum extracts N from "run-N" ids ( 0 for anything else). +func runIDNum(id string) int { + n, err := strconv.Atoi(strings.TrimPrefix(id, "run-")) + if err != nil { + return 0 + } + return n } // runWorkDir is the per-run scratch dir under the manager's private root — stable @@ -172,9 +268,11 @@ func (m *RunManager) runWorkDir(id string) string { } // Shutdown cancels every run (killing agent child processes) — called on server -// stop so local agents don't outlive radar. +// stop so local agents don't outlive radar. In-flight runs are marked stopped +// BEFORE their contexts are cancelled so the run goroutines' terminal-status +// guard keeps them from persisting a spurious "context canceled" error; the +// store then drains and closes, and anything appended after that is a no-op. func (m *RunManager) Shutdown() { - m.baseCancel() m.mu.Lock() runs := make([]*Run, 0, len(m.runs)) for _, r := range m.runs { @@ -183,12 +281,23 @@ func (m *RunManager) Shutdown() { m.mu.Unlock() for _, r := range runs { r.mu.Lock() + if r.inFlight { + r.status = "stopped" + r.updatedAt = nowUTC() + if r.store != nil { + r.store.SaveRun(r.summaryLocked()) + } + } c := r.cancel r.mu.Unlock() if c != nil { c() } } + m.baseCancel() + if m.store != nil { + m.store.Close() + } // Drop every run's scratch in one shot — the process is going away. if m.workRoot != "" { _ = os.RemoveAll(m.workRoot) @@ -269,13 +378,18 @@ func (m *RunManager) Start(kind, namespace, name, agent string, isolated bool, m ID: id, Kind: kind, Namespace: namespace, Name: name, Context: cur, Agent: agent, WorkDir: m.runWorkDir(id), Isolated: isolated, Model: model, Effort: effort, ManagedBy: managedBy, Health: health, CreatedAt: nowUTC(), + store: m.store, status: "running", inFlight: true, updatedAt: nowUTC(), - subs: map[int]chan RunEvent{}, + hydrated: true, // born live — its full log is in memory by construction + subs: map[int]chan RunEvent{}, } m.runs[r.ID] = r m.order = append(m.order, r.ID) m.evictLocked() m.mu.Unlock() + if m.store != nil { + m.store.SaveRun(r.Summary()) + } m.launchTurn(r, "", false, "", "") return r.Summary(), nil @@ -288,6 +402,9 @@ func (m *RunManager) AddTurn(id, question string, apply bool, fix string) error if r == nil { return ErrRunNotFound } + // A follow-up on a run loaded from history must extend the PERSISTED log — + // hydrate before beginTurn so the new turn's sequence numbers continue it. + r.ensureHydrated() session, err := m.beginTurn(r, true) if err != nil { return err @@ -364,6 +481,7 @@ func (m *RunManager) Stop(id string) error { if r == nil { return ErrRunNotFound } + r.ensureHydrated() // an in-flight run is always hydrated; harmless otherwise r.mu.Lock() if !r.inFlight { r.mu.Unlock() @@ -392,6 +510,17 @@ func (m *RunManager) OnContextSwitch() { for _, r := range runs { r.mu.Lock() c := r.cancel + inFlight := r.inFlight + hydrated := r.hydrated + r.mu.Unlock() + if !inFlight && !hydrated { + // Loaded, never-touched history: mark stale without paying to load + // its transcript (terminal markers get store-assigned seqs). + r.markStale() + r.removeWorkDir() + continue + } + r.mu.Lock() r.status = "stale" r.mu.Unlock() if c != nil { @@ -409,6 +538,7 @@ func (m *RunManager) Get(id string) *Run { return m.get(id) } func (m *RunManager) get(id string) *Run { m.mu.Lock() defer m.mu.Unlock() + m.sweepForeignLocked() return m.runs[id] } @@ -416,6 +546,7 @@ func (m *RunManager) get(id string) *Run { func (m *RunManager) List() []RunSummary { m.mu.Lock() defer m.mu.Unlock() + m.sweepForeignLocked() out := make([]RunSummary, 0, len(m.order)) for i := len(m.order) - 1; i >= 0; i-- { out = append(out, m.runs[m.order[i]].Summary()) @@ -423,6 +554,72 @@ func (m *RunManager) List() []RunSummary { return out } +// HistoryDegraded reports that run persistence stopped working — history will +// not survive a restart. Surfaced on the runs-list response so the UI can say so. +func (m *RunManager) HistoryDegraded() bool { + return m.store != nil && m.store.Degraded() +} + +// sweepForeignLocked marks runs loaded from a PREVIOUS process against a +// different kube-context as stale — the same treatment OnContextSwitch gives +// live runs when the context changes under them. Runs once per observed context +// label; the label callback resolves only after the cluster connects, which is +// why this can't happen at load time. Caller holds m.mu. +func (m *RunManager) sweepForeignLocked() { + cur := m.ctx() + if cur == "" || cur == m.sweptCtx { + return + } + m.sweptCtx = cur + for _, r := range m.runs { + if r.Context != cur { + r.markStale() + } + } +} + +// ClearHistory drops every terminal run from memory and the store. Live +// (running) runs survive and are re-persisted so their rows aren't orphaned by +// the wipe. +func (m *RunManager) ClearHistory() error { + m.mu.Lock() + kept := make([]string, 0, len(m.order)) + var keptRuns []*Run + for _, id := range m.order { + r := m.runs[id] + if r.snapshotStatus() == "running" { + kept = append(kept, id) + keptRuns = append(keptRuns, r) + continue + } + delete(m.runs, id) + r.finalize() + r.removeWorkDir() + } + m.order = kept + m.mu.Unlock() + + if m.store == nil { + return nil + } + if err := m.store.Clear(); err != nil { + return err + } + // Re-persist the survivors: their summary rows and any events already in + // memory (live runs are always hydrated, so this is the complete log). + for _, r := range keptRuns { + r.mu.Lock() + sum := r.summaryLocked() + events := append([]RunEvent(nil), r.events...) + r.mu.Unlock() + m.store.SaveRun(sum) + for _, e := range events { + m.store.AppendEvent(r.ID, e, nil) + } + } + return nil +} + // evictLocked drops the oldest finished run when over the retention cap. Running // runs are never evicted. Caller holds m.mu. func (m *RunManager) evictLocked() { @@ -443,6 +640,9 @@ func (m *RunManager) evictLocked() { m.order = append(m.order[:idx], m.order[idx+1:]...) victim.finalize() victim.removeWorkDir() // best-effort: drop the evicted run's scratch dir + if m.store != nil { + m.store.DeleteRun(id) + } } } @@ -458,6 +658,12 @@ func (r *Run) removeWorkDir() { func (r *Run) Summary() RunSummary { r.mu.Lock() defer r.mu.Unlock() + return r.summaryLocked() +} + +// summaryLocked builds the snapshot; the caller holds r.mu (or has exclusive +// access to a not-yet-shared run). +func (r *Run) summaryLocked() RunSummary { return RunSummary{ ID: r.ID, Kind: r.Kind, Namespace: r.Namespace, Name: r.Name, Context: r.Context, Agent: r.Agent, Isolated: r.Isolated, @@ -468,6 +674,74 @@ func (r *Run) Summary() RunSummary { } } +// ensureHydrated loads the run's event log from the store on first touch. Every +// path that reads or extends the log (Subscribe, follow-up turns, Stop) calls it +// first, so sequence numbers always continue from the persisted log. Idempotent +// and safe under concurrency: a racing second load just re-installs the same +// immutable prefix before either appends. +func (r *Run) ensureHydrated() { + if r.store == nil { + return + } + r.mu.Lock() + if r.hydrated { + r.mu.Unlock() + return + } + r.mu.Unlock() + events, err := r.store.LoadEvents(r.ID) // outside r.mu — a DB read may wait on the writer + r.mu.Lock() + if !r.hydrated { + if err != nil { + log.Printf("[ai] could not load transcript for %s: %v", r.ID, err) + } else { + r.events = events + } + r.hydrated = true + } + r.mu.Unlock() +} + +// markStale flips a non-running run to stale and finalizes its stream without +// requiring hydration: the terminal error + closed markers are appended in the +// STORE with store-assigned sequence numbers, and the in-memory log stays lazy — +// the next Subscribe hydrates and replays them. Used for runs loaded from a +// previous process whose kube-context no longer matches. +func (r *Run) markStale() { + r.mu.Lock() + if r.status == "stale" || r.inFlight { + r.mu.Unlock() + return + } + r.status = "stale" + r.updatedAt = nowUTC() + sum := r.summaryLocked() + hydrated := r.hydrated + for id, ch := range r.subs { + delete(r.subs, id) + close(ch) + } + r.subs = nil + r.mu.Unlock() + if r.store == nil { + return + } + if hydrated { + // The in-memory log is authoritative — persist with explicit seqs so + // memory and store stay aligned. + r.mu.Lock() + stale := RunEvent{Seq: len(r.events) + 1, Event: StreamEvent{Type: "error", Error: "Cluster context changed — this investigation was about a different cluster."}} + closed := RunEvent{Seq: len(r.events) + 2, Event: StreamEvent{Type: "closed"}} + r.events = append(r.events, stale, closed) + r.mu.Unlock() + r.store.AppendEvent(r.ID, stale, nil) + r.store.AppendEvent(r.ID, closed, &sum) + return + } + r.store.AppendEvent(r.ID, RunEvent{Event: StreamEvent{Type: "error", Error: "Cluster context changed — this investigation was about a different cluster."}}, nil) + r.store.AppendEvent(r.ID, RunEvent{Event: StreamEvent{Type: "closed"}}, &sum) +} + // matchesTarget reports whether r is the same investigation as a Start request — // same resource + cluster AND same agent/isolation mode. The mode is part of the // key so starting codex-isolated never silently focuses a live claude or my-setup @@ -488,6 +762,7 @@ func (r *Run) snapshotStatus() string { // The channel is closed only when the run is finalized (stale/evicted) — NOT when // a turn completes, so the same subscription sees later turns. func (r *Run) Subscribe(afterSeq int) (backlog []RunEvent, ch <-chan RunEvent, cancel func()) { + r.ensureHydrated() // a run loaded from history replays its persisted transcript r.mu.Lock() defer r.mu.Unlock() for _, e := range r.events { @@ -515,11 +790,23 @@ func (r *Run) Subscribe(afterSeq int) (backlog []RunEvent, ch <-chan RunEvent, c // append records an event and fans it out non-blockingly. A subscriber whose buffer // is full is dropped (it reconnects with Last-Event-ID to replay). +// Persistence rides along: the event is enqueued to the store under r.mu (enqueue +// never blocks), and TERMINAL events ("done"/"error") carry the run's summary so +// the status column and its terminal event commit in one transaction — crash +// recovery can then trust that a "running" row has no terminal marker. func (r *Run) append(ev StreamEvent) { r.mu.Lock() re := RunEvent{Seq: len(r.events) + 1, Event: ev} r.events = append(r.events, re) r.updatedAt = nowUTC() + if r.store != nil { + var sum *RunSummary + if ev.Type == "done" || ev.Type == "error" { + s := r.summaryLocked() + sum = &s + } + r.store.AppendEvent(r.ID, re, sum) + } for id, ch := range r.subs { select { case ch <- re: @@ -545,6 +832,12 @@ func (r *Run) finalize() { re := RunEvent{Seq: len(r.events) + 1, Event: StreamEvent{Type: "closed"}} r.events = append(r.events, re) r.updatedAt = nowUTC() + if r.store != nil && r.hydrated { + // Unhydrated finalize only happens on eviction, where the rows are + // deleted right after — persisting a wrong-seq sentinel would be noise. + sum := r.summaryLocked() + r.store.AppendEvent(r.ID, re, &sum) + } for id, ch := range r.subs { select { case ch <- re: diff --git a/internal/ai/runs_test.go b/internal/ai/runs_test.go index c3d8674cd..42d874dca 100644 --- a/internal/ai/runs_test.go +++ b/internal/ai/runs_test.go @@ -1,7 +1,10 @@ package ai import ( + "errors" + "fmt" "path/filepath" + "strings" "testing" ) @@ -28,8 +31,8 @@ func TestRunWorkDirUnderPrivateRoot(t *testing.T) { // backlog after its last-seen seq, then live events, then a close on terminal. func TestRunSubscribeReplay(t *testing.T) { r := &Run{subs: map[int]chan RunEvent{}} - r.append(StreamEvent{Type: "turn"}) // seq 1 - r.append(StreamEvent{Type: "phase"}) // seq 2 + r.append(StreamEvent{Type: "turn"}) // seq 1 + r.append(StreamEvent{Type: "phase"}) // seq 2 r.append(StreamEvent{Type: "thinking", Token: "x"}) // seq 3 backlog, ch, cancel := r.Subscribe(1) // everything after seq 1 @@ -169,3 +172,204 @@ func TestRunMatchesTarget(t *testing.T) { t.Error("different effort must NOT match") } } + +// persistedManager builds a manager over a store with no live diagnoser — good +// enough for persistence-path tests (nothing spawns an agent). +func persistedManager(t *testing.T, store RunStore, ctx string) *RunManager { + t.Helper() + m := NewRunManager(nil, func() int { return 0 }, func() string { return ctx }, store) + t.Cleanup(func() { + // Don't let Shutdown close the shared test store between phases. + m.baseCancel() + }) + return m +} + +// TestPersistenceRestartRoundtrip pins the core promise: a finished run's +// summary, transcript, and sessionId survive a "restart" (a second manager over +// the same store), replay parity included. +func TestPersistenceRestartRoundtrip(t *testing.T) { + st, _ := testStore(t) + + m1 := persistedManager(t, st, "ctx-a") + r := &Run{ + ID: "run-1", Kind: "Pod", Namespace: "ns", Name: "p", Context: "ctx-a", + Agent: "claude", store: st, status: "running", hydrated: true, + CreatedAt: nowUTC(), updatedAt: nowUTC(), subs: map[int]chan RunEvent{}, + } + m1.mu.Lock() + m1.runs[r.ID] = r + m1.order = append(m1.order, r.ID) + m1.nextID = 1 + m1.mu.Unlock() + st.SaveRun(r.Summary()) + + r.append(StreamEvent{Type: "turn"}) + r.append(StreamEvent{Type: "thinking", Token: "checking"}) + r.mu.Lock() + r.status = "done" + r.sessionID = "sess-42" + r.preview = "bad image" + r.mu.Unlock() + r.append(StreamEvent{Type: "done", Diag: &Diagnosis{RootCause: "bad image"}}) + st.(*sqliteRunStore).barrier() + + // "Restart": fresh manager, same store. + m2 := persistedManager(t, st, "ctx-a") + runs := m2.List() + if len(runs) != 1 || runs[0].Status != "done" || runs[0].SessionID != "sess-42" || runs[0].Preview != "bad image" { + t.Fatalf("restart lost state: %+v", runs) + } + // ID generation must not collide with the persisted run. + if m2.nextID != 1 { + t.Errorf("nextID = %d, want 1 (seeded from run-1)", m2.nextID) + } + // Replay parity: Subscribe hydrates the transcript from the store. + r2 := m2.Get("run-1") + backlog, _, cancel := r2.Subscribe(0) + defer cancel() + if len(backlog) != 3 || backlog[2].Event.Type != "done" || backlog[2].Event.Diag == nil { + t.Fatalf("replay after restart = %+v", backlog) + } +} + +// TestPersistenceInterruptedRun pins crash recovery: a run persisted as +// "running" loads as error with a terminal event appended, so replay still ends +// in a terminal marker and Start won't focus the dead run. +func TestPersistenceInterruptedRun(t *testing.T) { + st, _ := testStore(t) + st.SaveRun(RunSummary{ID: "run-3", Kind: "Pod", Name: "p", Context: "ctx-a", + Agent: "claude", Status: "running", CreatedAt: nowUTC(), UpdatedAt: nowUTC()}) + st.AppendEvent("run-3", RunEvent{Seq: 1, Event: StreamEvent{Type: "turn"}}, nil) + st.(*sqliteRunStore).barrier() + + m := persistedManager(t, st, "ctx-a") + runs := m.List() + if len(runs) != 1 || runs[0].Status != "error" { + t.Fatalf("interrupted run = %+v, want status error", runs) + } + r := m.Get("run-3") + backlog, _, cancel := r.Subscribe(0) + defer cancel() + last := backlog[len(backlog)-1] + if last.Event.Type != "error" || !strings.Contains(last.Event.Error, "restarted") { + t.Fatalf("replay must end in the restart marker, got %+v", last) + } +} + +// TestPersistenceCursorNotResumable pins the accepted Cursor degradation: its +// resume is workspace-scoped and the workspace died with the old process, so a +// loaded Cursor run must refuse follow-ups via ErrNoSession — never spawn an +// agent guaranteed to fail. +func TestPersistenceCursorNotResumable(t *testing.T) { + st, _ := testStore(t) + st.SaveRun(RunSummary{ID: "run-1", Kind: "Pod", Name: "p", Context: "ctx-a", + Agent: "cursor-agent", Status: "done", SessionID: "cursor-sess", + CreatedAt: nowUTC(), UpdatedAt: nowUTC()}) + st.(*sqliteRunStore).barrier() + + m := persistedManager(t, st, "ctx-a") + if err := m.AddTurn("run-1", "and?", false, ""); !errors.Is(err, ErrNoSession) { + t.Fatalf("cursor follow-up after restart = %v, want ErrNoSession", err) + } +} + +// TestPersistenceForeignContextSweep pins that history from another kube-context +// loads view-only: stale status, closed stream after replay, follow-ups refused. +func TestPersistenceForeignContextSweep(t *testing.T) { + st, _ := testStore(t) + st.SaveRun(RunSummary{ID: "run-1", Kind: "Pod", Name: "p", Context: "ctx-OLD", + Agent: "claude", Status: "done", SessionID: "s", CreatedAt: nowUTC(), UpdatedAt: nowUTC()}) + st.AppendEvent("run-1", RunEvent{Seq: 1, Event: StreamEvent{Type: "turn"}}, nil) + st.(*sqliteRunStore).barrier() + + m := persistedManager(t, st, "ctx-NEW") + runs := m.List() // sweep runs here (context label resolved) + if len(runs) != 1 || runs[0].Status != "stale" { + t.Fatalf("foreign-context run = %+v, want stale", runs) + } + if err := m.AddTurn("run-1", "and?", false, ""); !errors.Is(err, ErrStale) { + t.Fatalf("foreign-context follow-up = %v, want ErrStale", err) + } + st.(*sqliteRunStore).barrier() + // The persisted log gained terminal markers (store-assigned seqs), so a + // fresh subscribe replays and then CLOSES instead of hanging. + r := m.Get("run-1") + backlog, ch, cancel := r.Subscribe(0) + defer cancel() + last := backlog[len(backlog)-1] + if last.Event.Type != "closed" { + t.Fatalf("stale replay must end in closed, got %+v", backlog) + } + if _, ok := <-ch; ok { + t.Error("stale run's live channel must be closed") + } +} + +// TestPersistenceEvictionDeletesRows pins that count-based eviction removes the +// run from the store too — history and memory can't drift apart. +func TestPersistenceEvictionDeletesRows(t *testing.T) { + st, _ := testStore(t) + m := persistedManager(t, st, "ctx-a") + m.maxRetained = 2 + for i := 1; i <= 3; i++ { + id := fmt.Sprintf("run-%d", i) + r := &Run{ID: id, Kind: "Pod", Name: "p", Context: "ctx-a", store: st, + status: "done", hydrated: true, CreatedAt: nowUTC(), updatedAt: nowUTC(), + subs: map[int]chan RunEvent{}} + st.SaveRun(r.Summary()) + m.mu.Lock() + m.runs[id] = r + m.order = append(m.order, id) + m.evictLocked() + m.mu.Unlock() + } + st.(*sqliteRunStore).barrier() + runs, _ := st.LoadRuns() + if len(runs) != 2 { + t.Fatalf("store kept %d runs after eviction, want 2", len(runs)) + } + for _, r := range runs { + if r.ID == "run-1" { + t.Error("evicted run-1 still in store") + } + } +} + +// TestClearHistoryKeepsRunning pins that Clear wipes finished runs (memory + +// store) but a live investigation survives, fully re-persisted. +func TestClearHistoryKeepsRunning(t *testing.T) { + st, _ := testStore(t) + m := persistedManager(t, st, "ctx-a") + mk := func(id, status string) *Run { + r := &Run{ID: id, Kind: "Pod", Name: "p", Context: "ctx-a", store: st, + status: status, hydrated: true, CreatedAt: nowUTC(), updatedAt: nowUTC(), + subs: map[int]chan RunEvent{}} + st.SaveRun(r.Summary()) + m.mu.Lock() + m.runs[id] = r + m.order = append(m.order, id) + m.mu.Unlock() + return r + } + mk("run-1", "done") + live := mk("run-2", "running") + live.append(StreamEvent{Type: "turn"}) + + if err := m.ClearHistory(); err != nil { + t.Fatal(err) + } + st.(*sqliteRunStore).barrier() + runs := m.List() + if len(runs) != 1 || runs[0].ID != "run-2" { + t.Fatalf("memory after clear = %+v", runs) + } + stored, _ := st.LoadRuns() + if len(stored) != 1 || stored[0].ID != "run-2" { + t.Fatalf("store after clear = %+v", stored) + } + events, _ := st.LoadEvents("run-2") + if len(events) != 1 || events[0].Event.Type != "turn" { + t.Fatalf("live run's transcript not re-persisted: %+v", events) + } +} diff --git a/internal/ai/store.go b/internal/ai/store.go new file mode 100644 index 000000000..903e85b4e --- /dev/null +++ b/internal/ai/store.go @@ -0,0 +1,326 @@ +// SQLite persistence for AI investigation runs — the durable backing behind +// RunManager's in-memory state, so history, transcripts, and the agent-session +// hand-off survive a Radar restart. Single local user (the feature is gated to +// no-auth standalone radar), so one DB file with a single writer is plenty. +package ai + +import ( + "database/sql" + "encoding/json" + "fmt" + "log" + "os" + "path/filepath" + "sync" + "sync/atomic" + + _ "modernc.org/sqlite" +) + +// RunStore persists runs and their event logs. Write methods are asynchronous +// and must never block the caller — RunManager enqueues them while holding a +// run's mutex on the live event path. +type RunStore interface { + // SaveRun upserts a run's summary row. + SaveRun(s RunSummary) + // AppendEvent appends one event row. e.Seq > 0 inserts that exact sequence; + // e.Seq == 0 lets the store assign MAX(seq)+1 (used for terminal markers on + // runs whose log was never hydrated into memory). When summary is non-nil it + // is upserted in the SAME transaction — terminal events ride with their + // status so crash recovery can trust the status column. + AppendEvent(runID string, e RunEvent, summary *RunSummary) + // LoadRuns returns every persisted summary, oldest first. + LoadRuns() ([]RunSummary, error) + // LoadEvents returns a run's events ordered by seq. + LoadEvents(runID string) ([]RunEvent, error) + // DeleteRun removes a run and its events. + DeleteRun(id string) + // Clear synchronously removes every persisted run and event. + Clear() error + // Degraded reports that persistence has stopped working (disk error or a + // saturated write queue) — history will not survive a restart. + Degraded() bool + // Close drains pending writes and closes the DB. + Close() +} + +const storeSchema = ` +CREATE TABLE IF NOT EXISTS runs ( + id TEXT PRIMARY KEY, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + status TEXT NOT NULL, + summary_json TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS run_events ( + run_id TEXT NOT NULL, + seq INTEGER NOT NULL, + event_json TEXT NOT NULL, + PRIMARY KEY (run_id, seq) +); +` + +// sqliteRunStore implements RunStore over one SQLite file with a single writer +// goroutine. Writes are enqueued (non-blocking) and applied in order; reads use +// the same connection and may briefly wait on the writer (busy_timeout). +type sqliteRunStore struct { + db *sql.DB + ops chan func(db *sql.DB) error + done chan struct{} + // closeMu makes enqueue-vs-Close safe: senders hold RLock while checking the + // closed flag AND sending, Close holds Lock while flipping it and closing the + // channel — so a send can never race the close and panic. + closeMu sync.RWMutex + closed bool + degraded atomic.Bool +} + +// OpenRunStore opens (or creates) the run history DB. The file is created 0600 +// before SQLite touches it so the WAL/SHM siblings inherit private permissions — +// transcripts contain cluster data. +func OpenRunStore(dbPath string) (RunStore, error) { + dir := filepath.Dir(dbPath) + if dir != "" && dir != "." { + if err := os.MkdirAll(dir, 0o700); err != nil { + return nil, fmt.Errorf("ai history dir: %w", err) + } + } + f, err := os.OpenFile(dbPath, os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + return nil, fmt.Errorf("ai history file: %w", err) + } + _ = f.Close() + + db, err := sql.Open("sqlite", dbPath) + if err != nil { + return nil, fmt.Errorf("ai history open: %w", err) + } + // One writer at a time (SQLite's model); reads share the connection. + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) + for _, pragma := range []string{ + "PRAGMA journal_mode=WAL", + "PRAGMA busy_timeout=10000", + "PRAGMA synchronous=NORMAL", + "PRAGMA user_version=1", + } { + if _, err := db.Exec(pragma); err != nil { + _ = db.Close() + return nil, fmt.Errorf("ai history pragma: %w", err) + } + } + if _, err := db.Exec(storeSchema); err != nil { + _ = db.Close() + return nil, fmt.Errorf("ai history schema: %w", err) + } + + s := &sqliteRunStore{ + db: db, + ops: make(chan func(db *sql.DB) error, 4096), + done: make(chan struct{}), + } + go s.writer() + return s, nil +} + +func (s *sqliteRunStore) writer() { + defer close(s.done) + for op := range s.ops { + if op == nil { + continue + } + if err := op(s.db); err != nil && !s.degraded.Load() { + s.degraded.Store(true) + log.Printf("[ai] history write failed — investigations will NOT survive a restart: %v", err) + } + } +} + +// enqueue hands an op to the writer without ever blocking the caller (it runs +// under a run's mutex on the hot path). A full queue flips degraded and drops. +func (s *sqliteRunStore) enqueue(op func(db *sql.DB) error) { + s.closeMu.RLock() + defer s.closeMu.RUnlock() + if s.closed { + return + } + select { + case s.ops <- op: + default: + if !s.degraded.Load() { + s.degraded.Store(true) + log.Printf("[ai] history write queue full — dropping writes; investigations will NOT survive a restart") + } + } +} + +// enqueueWait hands an op to the writer and blocks until it ran (user-initiated +// operations like Clear, where dropping would be wrong). Returns false if the +// store is closed. +func (s *sqliteRunStore) enqueueWait(op func(db *sql.DB) error) bool { + done := make(chan struct{}) + s.closeMu.RLock() + if s.closed { + s.closeMu.RUnlock() + return false + } + s.ops <- func(db *sql.DB) error { + defer close(done) + return op(db) + } + s.closeMu.RUnlock() + <-done + return true +} + +// barrier waits until every op enqueued before it has been applied. +func (s *sqliteRunStore) barrier() { + s.enqueueWait(func(*sql.DB) error { return nil }) +} + +func upsertRun(db *sql.DB, s RunSummary) error { + b, err := json.Marshal(s) + if err != nil { + return err + } + _, err = db.Exec(`INSERT INTO runs (id, created_at, updated_at, status, summary_json) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET updated_at=excluded.updated_at, + status=excluded.status, summary_json=excluded.summary_json`, + s.ID, s.CreatedAt.UnixMilli(), s.UpdatedAt.UnixMilli(), s.Status, string(b)) + return err +} + +func (s *sqliteRunStore) SaveRun(sum RunSummary) { + s.enqueue(func(db *sql.DB) error { return upsertRun(db, sum) }) +} + +func (s *sqliteRunStore) AppendEvent(runID string, e RunEvent, summary *RunSummary) { + s.enqueue(func(db *sql.DB) error { + b, err := json.Marshal(e.Event) + if err != nil { + return err + } + tx, err := db.Begin() + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + if e.Seq > 0 { + _, err = tx.Exec(`INSERT OR REPLACE INTO run_events (run_id, seq, event_json) VALUES (?, ?, ?)`, + runID, e.Seq, string(b)) + } else { + // Store-assigned sequence: terminal markers appended to a run whose + // log was never loaded into memory this process. + _, err = tx.Exec(`INSERT INTO run_events (run_id, seq, event_json) + SELECT ?, COALESCE(MAX(seq), 0) + 1, ? FROM run_events WHERE run_id = ?`, + runID, string(b), runID) + } + if err != nil { + return err + } + if summary != nil { + b, err := json.Marshal(*summary) + if err != nil { + return err + } + if _, err := tx.Exec(`INSERT INTO runs (id, created_at, updated_at, status, summary_json) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET updated_at=excluded.updated_at, + status=excluded.status, summary_json=excluded.summary_json`, + summary.ID, summary.CreatedAt.UnixMilli(), summary.UpdatedAt.UnixMilli(), + summary.Status, string(b)); err != nil { + return err + } + } + return tx.Commit() + }) +} + +func (s *sqliteRunStore) LoadRuns() ([]RunSummary, error) { + s.barrier() // read-your-writes: drain queued ops so loads see them + rows, err := s.db.Query(`SELECT summary_json FROM runs ORDER BY created_at ASC, id ASC`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []RunSummary + for rows.Next() { + var raw string + if err := rows.Scan(&raw); err != nil { + return nil, err + } + var sum RunSummary + if err := json.Unmarshal([]byte(raw), &sum); err != nil { + continue // one corrupt row shouldn't lose the rest of history + } + out = append(out, sum) + } + return out, rows.Err() +} + +func (s *sqliteRunStore) LoadEvents(runID string) ([]RunEvent, error) { + s.barrier() // read-your-writes: a hydration right after markStale/startup must see those markers + rows, err := s.db.Query(`SELECT seq, event_json FROM run_events WHERE run_id = ? ORDER BY seq ASC`, runID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []RunEvent + for rows.Next() { + var seq int + var raw string + if err := rows.Scan(&seq, &raw); err != nil { + return nil, err + } + var ev StreamEvent + if err := json.Unmarshal([]byte(raw), &ev); err != nil { + continue + } + out = append(out, RunEvent{Seq: seq, Event: ev}) + } + return out, rows.Err() +} + +func (s *sqliteRunStore) DeleteRun(id string) { + s.enqueue(func(db *sql.DB) error { + if _, err := db.Exec(`DELETE FROM run_events WHERE run_id = ?`, id); err != nil { + return err + } + _, err := db.Exec(`DELETE FROM runs WHERE id = ?`, id) + return err + }) +} + +func (s *sqliteRunStore) Clear() error { + var out error + s.enqueueWait(func(db *sql.DB) error { + if _, err := db.Exec(`DELETE FROM run_events`); err != nil { + out = err + return err + } + _, err := db.Exec(`DELETE FROM runs`) + out = err + return err + }) + return out +} + +func (s *sqliteRunStore) Degraded() bool { return s.degraded.Load() } + +// Close drains pending writes and closes the DB. Late writers (agent goroutines +// finishing after shutdown began) become no-ops via the closed flag; the flag +// flip and the channel close happen under the write lock, so no sender can race +// them. +func (s *sqliteRunStore) Close() { + s.closeMu.Lock() + if s.closed { + s.closeMu.Unlock() + return + } + s.closed = true + close(s.ops) + s.closeMu.Unlock() + <-s.done + _ = s.db.Close() +} diff --git a/internal/ai/store_test.go b/internal/ai/store_test.go new file mode 100644 index 000000000..2ad0f99b9 --- /dev/null +++ b/internal/ai/store_test.go @@ -0,0 +1,115 @@ +package ai + +import ( + "os" + "path/filepath" + "testing" + "time" +) + +func testStore(t *testing.T) (RunStore, string) { + t.Helper() + dbPath := filepath.Join(t.TempDir(), "ai-runs.db") + st, err := OpenRunStore(dbPath) + if err != nil { + t.Fatal(err) + } + t.Cleanup(st.Close) + return st, dbPath +} + +func TestStoreRoundtrip(t *testing.T) { + st, _ := testStore(t) + sum := RunSummary{ + ID: "run-1", Kind: "Pod", Namespace: "ns", Name: "p", Context: "ctx-a", + Agent: "claude", Status: "running", SessionID: "sess-1", + CreatedAt: time.Now().UTC().Truncate(time.Millisecond), + UpdatedAt: time.Now().UTC().Truncate(time.Millisecond), + } + st.SaveRun(sum) + st.AppendEvent("run-1", RunEvent{Seq: 1, Event: StreamEvent{Type: "turn"}}, nil) + st.AppendEvent("run-1", RunEvent{Seq: 2, Event: StreamEvent{Type: "thinking", Token: "hmm"}}, nil) + // Terminal event rides with its summary in one transaction. + sum.Status = "done" + st.AppendEvent("run-1", RunEvent{Seq: 3, Event: StreamEvent{Type: "done"}}, &sum) + st.(*sqliteRunStore).barrier() + + runs, err := st.LoadRuns() + if err != nil { + t.Fatal(err) + } + if len(runs) != 1 || runs[0].ID != "run-1" || runs[0].Status != "done" || runs[0].SessionID != "sess-1" { + t.Fatalf("LoadRuns = %+v", runs) + } + events, err := st.LoadEvents("run-1") + if err != nil { + t.Fatal(err) + } + if len(events) != 3 || events[0].Seq != 1 || events[2].Event.Type != "done" { + t.Fatalf("LoadEvents = %+v", events) + } + if events[1].Event.Token != "hmm" { + t.Errorf("event payload lost: %+v", events[1]) + } +} + +func TestStoreAutoSeq(t *testing.T) { + st, _ := testStore(t) + st.AppendEvent("run-9", RunEvent{Seq: 1, Event: StreamEvent{Type: "turn"}}, nil) + // Seq 0 = store-assigned MAX+1: terminal markers on never-hydrated runs. + st.AppendEvent("run-9", RunEvent{Event: StreamEvent{Type: "error", Error: "stale"}}, nil) + st.AppendEvent("run-9", RunEvent{Event: StreamEvent{Type: "closed"}}, nil) + st.(*sqliteRunStore).barrier() + + events, err := st.LoadEvents("run-9") + if err != nil { + t.Fatal(err) + } + if len(events) != 3 || events[1].Seq != 2 || events[2].Seq != 3 { + t.Fatalf("auto-seq mis-assigned: %+v", events) + } + if events[2].Event.Type != "closed" { + t.Errorf("order lost: %+v", events) + } +} + +func TestStoreDeleteAndClear(t *testing.T) { + st, _ := testStore(t) + for _, id := range []string{"run-1", "run-2"} { + st.SaveRun(RunSummary{ID: id, Status: "done", CreatedAt: time.Now(), UpdatedAt: time.Now()}) + st.AppendEvent(id, RunEvent{Seq: 1, Event: StreamEvent{Type: "turn"}}, nil) + } + st.DeleteRun("run-1") + st.(*sqliteRunStore).barrier() + runs, _ := st.LoadRuns() + if len(runs) != 1 || runs[0].ID != "run-2" { + t.Fatalf("DeleteRun left %+v", runs) + } + if err := st.Clear(); err != nil { + t.Fatal(err) + } + runs, _ = st.LoadRuns() + events, _ := st.LoadEvents("run-2") + if len(runs) != 0 || len(events) != 0 { + t.Fatalf("Clear left runs=%d events=%d", len(runs), len(events)) + } +} + +func TestStoreFilePermissions(t *testing.T) { + _, dbPath := testStore(t) + info, err := os.Stat(dbPath) + if err != nil { + t.Fatal(err) + } + if perm := info.Mode().Perm(); perm != 0o600 { + t.Errorf("history DB is %v, want 0600 — transcripts hold cluster data", perm) + } +} + +func TestStoreOpenFailureIsClean(t *testing.T) { + // A path whose parent can't be created must error (caller degrades to + // memory-only), not panic or half-open. + if _, err := OpenRunStore(filepath.Join(string([]byte{0}), "nope.db")); err == nil { + t.Fatal("expected open error for impossible path") + } +} diff --git a/internal/app/bootstrap.go b/internal/app/bootstrap.go index d36be54e4..a6c9fa94f 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -55,6 +55,8 @@ type AppConfig struct { PrometheusHeadersFromEnv map[string]string Version string MCPEnabled bool + AIHistory bool // persist AI investigations across restarts + AIHistoryDBPath string // "" = ~/.radar/ai-runs.db AuthConfig auth.Config } @@ -263,6 +265,18 @@ func CreateServer(cfg AppConfig) *server.Server { AuthConfig: cfg.AuthConfig, } + // AI-history DB path: resolved here (like the timeline DB) so the server + // only sees a ready-to-open path. Only meaningful where the AI engine can + // actually enable (no-auth + MCP mounted) — the server checks that gate. + if cfg.AIHistory { + dbPath := cfg.AIHistoryDBPath + if dbPath == "" { + homeDir, _ := os.UserHomeDir() + dbPath = filepath.Join(homeDir, ".radar", "ai-runs.db") + } + serverCfg.AIHistoryDB = dbPath + } + if cfg.MCPEnabled { serverCfg.MCPHandler = mcppkg.NewHandler() serverCfg.MCPReadOnlyHandler = mcppkg.NewReadOnlyHandler() diff --git a/internal/config/config.go b/internal/config/config.go index f66d893c6..6198be3fe 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -36,6 +36,11 @@ type Config struct { // pods. Empty falls back to busybox:latest; set it to a reachable mirror for // air-gapped / private-registry clusters. DebugImage string `json:"debugImage,omitempty"` + // AIHistory persists AI investigations (transcripts + verdicts) to a local + // SQLite file so they survive restarts. nil = default (true), false = off. + AIHistory *bool `json:"aiHistory,omitempty"` + // AIHistoryDBPath overrides the history DB location (default ~/.radar/ai-runs.db). + AIHistoryDBPath string `json:"aiHistoryDbPath,omitempty"` } // mu serializes Load-mutate-Save cycles to prevent concurrent writes @@ -131,6 +136,14 @@ func (c Config) MCPEnabledOr(def bool) bool { return def } +// AIHistoryOr returns *c.AIHistory if non-nil, otherwise the provided default. +func (c Config) AIHistoryOr(def bool) bool { + if c.AIHistory != nil { + return *c.AIHistory + } + return def +} + // TimelineStorageOr returns c.TimelineStorage if non-empty, otherwise the provided default. func (c Config) TimelineStorageOr(def string) string { if c.TimelineStorage != "" { diff --git a/internal/server/ai_diagnose.go b/internal/server/ai_diagnose.go index 679f295b3..fa87f2a5d 100644 --- a/internal/server/ai_diagnose.go +++ b/internal/server/ai_diagnose.go @@ -203,12 +203,34 @@ func (s *Server) handleDiagnoseStart(w http.ResponseWriter, r *http.Request) { s.writeJSON(w, run) } -// handleDiagnoseList returns all in-memory runs (newest first). +// handleDiagnoseList returns all retained runs (newest first). historyDegraded +// warns that persistence stopped working, so the UI can say history won't +// survive a restart instead of letting the user believe otherwise. func (s *Server) handleDiagnoseList(w http.ResponseWriter, r *http.Request) { if !s.aiReady(w) { return } - s.writeJSON(w, map[string]any{"runs": s.aiRuns.List()}) + s.writeJSON(w, map[string]any{ + "runs": s.aiRuns.List(), + "historyDegraded": s.aiRuns.HistoryDegraded(), + }) +} + +// handleDiagnoseHistoryClear wipes the persisted investigation history (and +// drops finished runs from memory). Live runs survive. POST, same-origin only. +func (s *Server) handleDiagnoseHistoryClear(w http.ResponseWriter, r *http.Request) { + if !localOriginOK(r) { + s.writeError(w, http.StatusForbidden, "cross-origin request rejected") + return + } + if !s.aiReady(w) { + return + } + if err := s.aiRuns.ClearHistory(); err != nil { + s.writeError(w, http.StatusInternalServerError, "couldn't clear history: "+err.Error()) + return + } + s.writeJSON(w, map[string]any{"ok": true}) } // handleDiagnoseTurn adds a follow-up or apply turn to a run. POST {question?, diff --git a/internal/server/server.go b/internal/server/server.go index fb91a3f4f..03685d12b 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -122,6 +122,7 @@ type Config struct { DiagConfig *DiagConfig // Sanitized config for diagnostics endpoint EffectiveConfig *config.Config // Running startup config for GET /api/config AuthConfig auth.Config // Authentication configuration + AIHistoryDB string // AI run-history SQLite path ("" = memory-only runs) } // New creates a new server instance @@ -156,7 +157,18 @@ func New(cfg Config) *Server { if !s.authConfig.Enabled() && s.mcpHandler != nil { if d, err := ai.NewDetected(context.Background()); err == nil { s.aiDiagnoser = d - s.aiRuns = ai.NewRunManager(d, s.ActualPort, k8s.GetContextName) + // History store opens only when the engine actually enables, so a + // disabled feature never creates the DB. Open failure degrades to + // memory-only runs (the historical behavior), never blocks startup. + var store ai.RunStore + if cfg.AIHistoryDB != "" { + if st, err := ai.OpenRunStore(cfg.AIHistoryDB); err != nil { + log.Printf("[ai] run history disabled — could not open %s: %v", cfg.AIHistoryDB, err) + } else { + store = st + } + } + s.aiRuns = ai.NewRunManager(d, s.ActualPort, k8s.GetContextName, store) log.Printf("[ai] diagnose enabled (default agent: %s)", d.DefaultAgent()) } } @@ -330,6 +342,7 @@ func (s *Server) setupRoutes() { r.Get("/diagnose/runs", s.handleDiagnoseList) r.Post("/diagnose/runs/{id}/turns", s.handleDiagnoseTurn) r.Post("/diagnose/runs/{id}/stop", s.handleDiagnoseStop) + r.Post("/diagnose/history/clear", s.handleDiagnoseHistoryClear) r.Get("/diagnostics", s.handleDiagnostics) r.Get("/auth/me", s.handleAuthMe) r.Get("/version-check", s.handleVersionCheck) diff --git a/web/src/api/diagnose.ts b/web/src/api/diagnose.ts index 861b5be2f..bb5be1efd 100644 --- a/web/src/api/diagnose.ts +++ b/web/src/api/diagnose.ts @@ -148,16 +148,33 @@ export async function createRun( return res.json(); } +export interface RunsResponse { + runs: RunSummary[]; + // Persistence stopped working (disk error) — history will NOT survive a + // restart; the UI should say so instead of letting the user assume otherwise. + historyDegraded?: boolean; +} + // listRuns returns all server-side runs (newest first) — the source of truth for // the recent-investigations list. -export async function listRuns(signal?: AbortSignal): Promise { +export async function listRuns(signal?: AbortSignal): Promise { const res = await fetch(RUNS(), { credentials: getCredentialsMode(), signal, }); if (!res.ok) throw new DiagnoseError(res.status, await errorText(res)); const d = await res.json(); - return d.runs ?? []; + return { runs: d.runs ?? [], historyDegraded: !!d.historyDegraded }; +} + +// clearHistory wipes the persisted investigation history (finished runs); live +// investigations survive. +export async function clearHistory(): Promise { + const res = await fetch(`${getApiBase()}/diagnose/history/clear`, { + method: "POST", + credentials: getCredentialsMode(), + }); + if (!res.ok) throw new DiagnoseError(res.status, await errorText(res)); } // addTurn appends a follow-up (question) or an apply turn (apply + confirmed fix). diff --git a/web/src/components/diagnose/AISettings.tsx b/web/src/components/diagnose/AISettings.tsx index 58b4adb40..aeb5726f2 100644 --- a/web/src/components/diagnose/AISettings.tsx +++ b/web/src/components/diagnose/AISettings.tsx @@ -1,7 +1,9 @@ // The "AI Diagnosis" section of the Settings dialog. Controlled by the dialog: // it edits a STAGED draft and is committed on Save (like the rest of Settings), // not on every keystroke. Renders nothing when no supported agent CLI is installed. -import { type AgentInfo } from "../../api/diagnose"; +import { useState } from "react"; +import { Trash2 } from "lucide-react"; +import { clearHistory, type AgentInfo } from "../../api/diagnose"; import { AgentControls } from "./parts"; export interface AIDraft { @@ -11,16 +13,85 @@ export interface AIDraft { effort: string; } +// ClearHistoryRow is an immediate action (not part of the staged draft): a +// two-step confirm button that wipes finished investigations from the local +// history DB. Live investigations survive. +function ClearHistoryRow({ onCleared }: { onCleared: () => void }) { + const [confirming, setConfirming] = useState(false); + const [state, setState] = useState<"idle" | "busy" | "done" | "error">( + "idle", + ); + const run = () => { + setState("busy"); + clearHistory() + .then(() => { + setState("done"); + setConfirming(false); + onCleared(); + }) + .catch(() => setState("error")); + }; + return ( +
+

+ Investigation transcripts are kept on this machine ( + ~/.radar) so history survives + restarts. + {state === "done" && ( + + History cleared. + + )} + {state === "error" && ( + + Couldn't clear history. + + )} +

+ {confirming ? ( +
+ + +
+ ) : ( + + )} +
+ ); +} + export function AISettingsSection({ available, agents, draft, onChange, + onHistoryCleared, }: { available: boolean; agents: AgentInfo[]; draft: AIDraft; onChange: (patch: Partial) => void; + onHistoryCleared: () => void; }) { if (!available || agents.length === 0) return null; return ( @@ -44,6 +115,7 @@ export function AISettingsSection({ effort={draft.effort} onSetEffort={(v) => onChange({ effort: v })} /> + ); } diff --git a/web/src/components/diagnose/DiagnoseContext.tsx b/web/src/components/diagnose/DiagnoseContext.tsx index 99100ee59..9a8d2de65 100644 --- a/web/src/components/diagnose/DiagnoseContext.tsx +++ b/web/src/components/diagnose/DiagnoseContext.tsx @@ -45,6 +45,7 @@ interface DiagnoseCtx { view: DiagnoseView; activeRunId: string | null; runs: RunSummary[]; + historyDegraded: boolean; // persistence broke — history won't survive a restart needsConsent: boolean; // a start is pending the one-time consent startError: string | null; openInvestigation: (t: Target) => void; @@ -179,6 +180,7 @@ export function DiagnoseProvider({ children }: { children: ReactNode }) { const [view, setView] = useState("home"); const [activeRunId, setActiveRunId] = useState(null); const [runs, setRuns] = useState([]); + const [historyDegraded, setHistoryDegraded] = useState(false); const [pendingTarget, setPendingTarget] = useState(null); const [startError, setStartError] = useState(null); const [width, setWidth] = useState(() => { @@ -265,7 +267,10 @@ export function DiagnoseProvider({ children }: { children: ReactNode }) { const refreshRuns = useCallback(() => { if (!available) return; listRuns() - .then(setRuns) + .then((r) => { + setRuns(r.runs); + setHistoryDegraded(!!r.historyDegraded); + }) .catch(() => {}); }, [available]); @@ -388,6 +393,7 @@ export function DiagnoseProvider({ children }: { children: ReactNode }) { view, activeRunId, runs, + historyDegraded, // pendingTarget is set ONLY when the current agent's consent is missing, and // cleared on approve/cancel — so its presence is exactly "consent needed now". needsConsent: !!pendingTarget, diff --git a/web/src/components/diagnose/DiagnoseSurface.tsx b/web/src/components/diagnose/DiagnoseSurface.tsx index 587c3f6d0..d5854a648 100644 --- a/web/src/components/diagnose/DiagnoseSurface.tsx +++ b/web/src/components/diagnose/DiagnoseSurface.tsx @@ -333,6 +333,7 @@ export function DiagnoseSurface({ topInset = 0 }: { topInset?: number }) { runs={d.runs} selectedId={d.activeRunId} onSelect={d.openRun} + historyDegraded={d.historyDegraded} /> )} @@ -345,6 +346,7 @@ export function DiagnoseSurface({ topInset = 0 }: { topInset?: number }) { agentLabel={d.agentLabel} runs={d.runs} onSelect={d.openRun} + historyDegraded={d.historyDegraded} /> ) : ( diff --git a/web/src/components/diagnose/Home.tsx b/web/src/components/diagnose/Home.tsx index 597748fe6..c4684754f 100644 --- a/web/src/components/diagnose/Home.tsx +++ b/web/src/components/diagnose/Home.tsx @@ -59,16 +59,29 @@ export function RecentList({ runs, onSelect, selectedId, + historyDegraded = false, }: { agentLabel: string; runs: RunSummary[]; onSelect: (id: string) => void; selectedId?: string | null; + historyDegraded?: boolean; }) { const now = Date.now(); + // Persistence broke (disk error) — without this the user reasonably assumes + // their history survives a restart, and it won't. + const degradedNote = historyDegraded ? ( +
+ History isn't being saved right now (disk error) — investigations + won't survive a restart. +
+ ) : null; + if (runs.length === 0) { return ( +
+ {degradedNote}
@@ -80,14 +93,16 @@ export function RecentList({ action to investigate it with {agentLabel} —{" "} Diagnose{" "} a problem, or just ask about it. Investigations run in the background - and stay here until you restart Radar. + and are kept in your history here.

+
); } return (
+ {degradedNote}
Investigations
diff --git a/web/src/components/diagnose/InvestigationView.tsx b/web/src/components/diagnose/InvestigationView.tsx index e8659a67c..c52531cb0 100644 --- a/web/src/components/diagnose/InvestigationView.tsx +++ b/web/src/components/diagnose/InvestigationView.tsx @@ -476,9 +476,9 @@ export function InvestigationView({
- This investigation is no longer available — investigations are - kept in memory and clear when Radar restarts or after enough - newer ones. Re-run Diagnose to analyze the current cluster. + This investigation is no longer available — history keeps + the most recent investigations, and this one has been + cleared. Re-run Diagnose to analyze the current cluster.
)} + {turns.map((t, i) => { const isLast = i === turns.length - 1; const canApply = i === lastRemediationIdx && !stale; diff --git a/web/src/components/diagnose/parts.tsx b/web/src/components/diagnose/parts.tsx index 002bbf84f..1e6f60d51 100644 --- a/web/src/components/diagnose/parts.tsx +++ b/web/src/components/diagnose/parts.tsx @@ -26,7 +26,9 @@ import { type Diagnosis, type DiagnoseStep, type AgentInfo, + type RunSummary, } from "../../api/diagnose"; +import { StatusDot } from "@skyhook-io/k8s-ui"; import { Markdown } from "../ui/Markdown"; // Segmented two-or-more-way selector — shared shape for the agent picker and the @@ -474,6 +476,72 @@ export function TurnView({ ); } +// RunContextCard opens every investigation with what RADAR already knows — the +// health frame the server captured at run start. It renders instantly (no agent +// round-trip), so the agent's boot time reads as "context, then deepening" +// instead of dead air — and it anchors the verdict against Radar's own signal. +export function RunContextCard({ run }: { run: RunSummary }) { + const h = run.health; + const issueCount = h?.issueCount ?? 0; + const lines: ReactNode[] = []; + if (issueCount > 0) { + lines.push( + + + {issueCount} active issue{issueCount === 1 ? "" : "s"} + {h?.topReason ? ( + <> + {" — "} + {h.topReason} + + ) : null} + , + ); + } else if (h?.health === "healthy") { + lines.push( + + + Reported healthy — 0 active issues + , + ); + } else if (h) { + lines.push( + + + 0 active issues{h.health ? ` — status ${h.health}` : ""} + , + ); + } + if ((h?.auditCount ?? 0) > 0) { + lines.push( + + {h!.auditCount} configuration finding{h!.auditCount === 1 ? "" : "s"} from + cluster audit{h?.topFinding ? ` — ${h.topFinding}` : ""} + , + ); + } + if (run.managedBy) { + lines.push( + + Managed by {run.managedBy} + , + ); + } + if (lines.length === 0) return null; + return ( +
+
+ Radar's read at start +
+
+ {lines.map((l, i) => ( +
{l}
+ ))} +
+
+ ); +} + // The first-run consent + trust card. Its copy is the OSS BYO-local trust story // ("your own agent, on your machine, nothing to Radar"). // TODO(cloud): this copy must become embedder-overridable for Radar Cloud, where From 57702a736e47b4a9807b80fe8145d96883f4de24 Mon Sep 17 00:00:00 2001 From: Nadav Erell Date: Thu, 2 Jul 2026 11:58:12 +0300 Subject: [PATCH 07/11] Diagnose: concrete health rows in the context card + CLI thinking spinner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The "Radar's read at start" card rendered aggregates ("2 active issues — Unschedulable") — vague, when the issues engine had the full sentences all along. The health frame now carries the top actual rows (3 issues + 2 audit findings, engine messages capped at 220 chars), so: - the context card shows "Unschedulable — 2 node(s) insufficient cpu (0/2 nodes available)" with per-row severity dots and +N-more lines; - the first-turn prompt frames the agent with the same specifics; - `radar diagnose` prints them under its header before the agent boots. The CLI also gains a thinking indicator: on real TTYs, >1s of silence shows a live "⠙ · thinking… 12s" line (erased before any real output, suppressed mid-reasoning-line and under NO_COLOR/pipes) — the model's long quiet gaps read as work, not a hang. Renderer writes are now mutex-serialized against the spinner goroutine. --- internal/ai/diagnoser.go | 33 +++++-- internal/diagnosecli/diagnosecli.go | 16 ++++ internal/diagnosecli/render.go | 119 ++++++++++++++++++++++++-- internal/server/ai_diagnose.go | 34 +++++++- internal/server/ai_handlers.go | 29 +++++-- web/src/api/diagnose.ts | 8 ++ web/src/components/diagnose/parts.tsx | 90 ++++++++++++------- 7 files changed, 271 insertions(+), 58 deletions(-) diff --git a/internal/ai/diagnoser.go b/internal/ai/diagnoser.go index b90228879..1881a53d2 100644 --- a/internal/ai/diagnoser.go +++ b/internal/ai/diagnoser.go @@ -84,14 +84,26 @@ type Request struct { // ResourceHealthSignal is the compact server-side health frame captured when a // run starts. Issue fields describe live operational findings; audit fields are // static posture findings and should not be treated as proof of an outage. +// Issues/AuditFindings carry the top actual rows (capped) — aggregates alone +// read as vague in the UI and starve the prompt of detail Radar already has. type ResourceHealthSignal struct { - Health string `json:"health,omitempty"` - IssueCount int `json:"issueCount,omitempty"` - HighestSeverity string `json:"highestSeverity,omitempty"` - TopReason string `json:"topReason,omitempty"` - AuditCount int `json:"auditCount,omitempty"` - AuditSeverity string `json:"auditSeverity,omitempty"` - TopFinding string `json:"topFinding,omitempty"` + Health string `json:"health,omitempty"` + IssueCount int `json:"issueCount,omitempty"` + HighestSeverity string `json:"highestSeverity,omitempty"` + TopReason string `json:"topReason,omitempty"` + Issues []HealthLine `json:"issues,omitempty"` + AuditCount int `json:"auditCount,omitempty"` + AuditSeverity string `json:"auditSeverity,omitempty"` + TopFinding string `json:"topFinding,omitempty"` + AuditFindings []HealthLine `json:"auditFindings,omitempty"` +} + +// HealthLine is one concrete issue/finding row: what Radar's issue engine or +// audit suite actually said, not just a count. +type HealthLine struct { + Severity string `json:"severity,omitempty"` + Reason string `json:"reason,omitempty"` // issue Reason / audit CheckID + Message string `json:"message,omitempty"` // human detail, capped } // Diagnosis is the engine's final result. @@ -473,6 +485,13 @@ func healthFrame(target string, health *ResourceHealthSignal) string { } } b.WriteString(".") + for _, line := range health.Issues { + fmt.Fprintf(&b, " [%s] %s", line.Severity, line.Reason) + if line.Message != "" { + fmt.Fprintf(&b, ": %s", line.Message) + } + b.WriteString(".") + } case health.Health == "healthy": fmt.Fprintf(&b, "Radar currently reports %s as healthy and flags 0 active issues.", target) case health.Health != "": diff --git a/internal/diagnosecli/diagnosecli.go b/internal/diagnosecli/diagnosecli.go index 86827a704..55622a937 100644 --- a/internal/diagnosecli/diagnosecli.go +++ b/internal/diagnosecli/diagnosecli.go @@ -279,6 +279,20 @@ type runSummary struct { Name string `json:"name"` Agent string `json:"agent"` SessionID string `json:"sessionId"` + ManagedBy string `json:"managedBy"` + Health *struct { + IssueCount int `json:"issueCount"` + AuditCount int `json:"auditCount"` + Issues []struct { + Severity string `json:"severity"` + Reason string `json:"reason"` + Message string `json:"message"` + } `json:"issues"` + AuditFindings []struct { + Reason string `json:"reason"` + Message string `json:"message"` + } `json:"auditFindings"` + } `json:"health"` } func startRun(base, kind, namespace, name, agent string) (runSummary, error) { @@ -343,6 +357,8 @@ func streamRun(base, id string, out *renderer) (json.RawMessage, bool) { return nil, false } + out.startSpinner() + defer out.stopSpinner() sc := bufio.NewScanner(resp.Body) sc.Buffer(make([]byte, 0, 64*1024), 4<<20) steps := map[string]stepInfo{} diff --git a/internal/diagnosecli/render.go b/internal/diagnosecli/render.go index 1e1f02e84..ecbb30bc1 100644 --- a/internal/diagnosecli/render.go +++ b/internal/diagnosecli/render.go @@ -5,16 +5,27 @@ import ( "os" "regexp" "strings" + "sync" + "time" "golang.org/x/term" ) // renderer writes the live transcript + verdict to the terminal. In --json mode // everything human goes to stderr so stdout stays a clean JSON document. +// A single mutex serializes event writes with the spinner goroutine: the model +// goes quiet for long stretches (its own thinking + slow tools), and without a +// live indicator a silent terminal reads as a hang. type renderer struct { - w *os.File - color bool - inThinking bool + w *os.File + color bool + + mu sync.Mutex + inThinking bool + spinnerOn bool // spinner line currently drawn (must be erased before real output) + lastEvent time.Time // last real output, for the quiet-gap threshold + stopSpin chan struct{} + spinStopped bool } func newRenderer(jsonMode bool) *renderer { @@ -22,7 +33,63 @@ func newRenderer(jsonMode bool) *renderer { if jsonMode { w = os.Stderr } - return &renderer{w: w, color: term.IsTerminal(int(w.Fd())) && os.Getenv("NO_COLOR") == ""} + return &renderer{ + w: w, + color: term.IsTerminal(int(w.Fd())) && os.Getenv("NO_COLOR") == "", + lastEvent: time.Now(), + stopSpin: make(chan struct{}), + } +} + +var spinnerFrames = []string{"⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"} + +// startSpinner shows "⠙ thinking… 12s" after a second of silence — only on a +// real terminal (it repaints its own line), and never mid-thinking-line. +func (r *renderer) startSpinner() { + if !r.color { + return + } + go func() { + t := time.NewTicker(120 * time.Millisecond) + defer t.Stop() + frame := 0 + for { + select { + case <-r.stopSpin: + return + case <-t.C: + } + r.mu.Lock() + quiet := time.Since(r.lastEvent) + if quiet > time.Second && !r.inThinking { + fmt.Fprintf(r.w, " %s%s %s thinking… %ds%s", + cDim, spinnerFrames[frame%len(spinnerFrames)], cAmber+"·"+cReset+cDim, int(quiet.Seconds()), cReset) + r.spinnerOn = true + frame++ + } + r.mu.Unlock() + } + }() +} + +func (r *renderer) stopSpinner() { + r.mu.Lock() + defer r.mu.Unlock() + if r.spinStopped { + return + } + r.spinStopped = true + close(r.stopSpin) + r.clearSpinnerLocked() +} + +// clearSpinnerLocked erases the spinner line before real output. Caller holds r.mu. +func (r *renderer) clearSpinnerLocked() { + if r.spinnerOn { + fmt.Fprint(r.w, " ") + r.spinnerOn = false + } + r.lastEvent = time.Now() } const ( @@ -49,7 +116,28 @@ func (r *renderer) header(run runSummary, base string) { } target += run.Name fmt.Fprintf(r.w, "%s %s\n", r.c(cBold, "◉ Investigating"), target) - fmt.Fprintf(r.w, "%s\n\n", r.c(cDim, fmt.Sprintf("run %s · via %s · watch: %s/?ai-run=%s", run.ID, agentDisplay(run.Agent), base, run.ID))) + fmt.Fprintf(r.w, "%s\n", r.c(cDim, fmt.Sprintf("run %s · via %s · watch: %s/?ai-run=%s", run.ID, agentDisplay(run.Agent), base, run.ID))) + // Radar's read at start — the concrete issue rows the server captured, shown + // before the agent produces anything (its boot is the longest silent gap). + if h := run.Health; h != nil { + for _, line := range h.Issues { + sev := r.c(cRed, "●") + if line.Severity != "critical" { + sev = r.c(cAmber, "●") + } + fmt.Fprintf(r.w, "%s %s — %s\n", sev, r.c(cBold, line.Reason), line.Message) + } + if extra := h.IssueCount - len(h.Issues); extra > 0 { + fmt.Fprintf(r.w, "%s\n", r.c(cDim, fmt.Sprintf(" +%d more active issues", extra))) + } + for _, f := range h.AuditFindings { + fmt.Fprintf(r.w, "%s\n", r.c(cDim, fmt.Sprintf(" audit: %s — %s", f.Reason, f.Message))) + } + } + if run.ManagedBy != "" { + fmt.Fprintf(r.w, "%s\n", r.c(cDim, " managed by "+run.ManagedBy)) + } + fmt.Fprintln(r.w) } func agentDisplay(name string) string { @@ -69,6 +157,9 @@ func (r *renderer) thinking(token string) { if token == "" { return } + r.mu.Lock() + defer r.mu.Unlock() + r.clearSpinnerLocked() if r.color { fmt.Fprint(r.w, cDim+token+cReset) } else { @@ -77,7 +168,8 @@ func (r *renderer) thinking(token string) { r.inThinking = !strings.HasSuffix(token, "\n") } -func (r *renderer) breakThinking() { +// breakThinkingLocked ends a partial reasoning line. Caller holds r.mu. +func (r *renderer) breakThinkingLocked() { if r.inThinking { fmt.Fprintln(r.w) r.inThinking = false @@ -86,7 +178,10 @@ func (r *renderer) breakThinking() { // step prints one completed tool call: " ✓ get resource {"name":"web"} · 233ms". func (r *renderer) step(s stepInfo) { - r.breakThinking() + r.mu.Lock() + defer r.mu.Unlock() + r.clearSpinnerLocked() + r.breakThinkingLocked() line := " " + r.c(cGreen, "✓") + " " + prettyTool(s.Tool) if s.Summary != "" { line += " " + r.c(cDim, compact(s.Summary, 80)) @@ -110,12 +205,18 @@ func compact(s string, max int) string { } func (r *renderer) errorLine(msg string) { - r.breakThinking() + r.mu.Lock() + defer r.mu.Unlock() + r.clearSpinnerLocked() + r.breakThinkingLocked() fmt.Fprintf(r.w, "\n%s %s\n", r.c(cRed, "✗"), msg) } func (r *renderer) verdict(d diagnosis) { - r.breakThinking() + r.mu.Lock() + defer r.mu.Unlock() + r.clearSpinnerLocked() + r.breakThinkingLocked() fmt.Fprintln(r.w) switch { case d.Healthy && d.RootCause == "": diff --git a/internal/server/ai_diagnose.go b/internal/server/ai_diagnose.go index fa87f2a5d..ea98b0877 100644 --- a/internal/server/ai_diagnose.go +++ b/internal/server/ai_diagnose.go @@ -49,8 +49,8 @@ func (s *Server) detectDiagnoseHealth(ctx context.Context, kind, namespace, name if canonicalKind == "" { canonicalKind = kind } - issueSum := computeIssueSummaryForResource(cache, gvk.Group, canonicalKind, namespace, name) - auditSum := computeAuditSummaryForResource(cache, gvk.Group, canonicalKind, namespace, name) + issueSum, issueRows := computeIssueSummaryAndRows(cache, gvk.Group, canonicalKind, namespace, name) + auditSum, auditRows := computeAuditSummaryAndRows(cache, gvk.Group, canonicalKind, namespace, name) var issueCount int signal := &ai.ResourceHealthSignal{} @@ -59,6 +59,13 @@ func (s *Server) detectDiagnoseHealth(ctx context.Context, kind, namespace, name signal.IssueCount = issueSum.Count signal.HighestSeverity = issueSum.HighestSeverity signal.TopReason = issueSum.TopReason + for _, row := range issueRows[:min(len(issueRows), maxHealthLines)] { + signal.Issues = append(signal.Issues, ai.HealthLine{ + Severity: string(row.Severity), + Reason: row.Reason, + Message: capHealthMessage(row.Message), + }) + } } if summary := resourcecontext.BuildSummary(obj, resourcecontext.SummaryOptions{IssueCount: issueCount}); summary != nil { signal.Health = summary.Health @@ -67,10 +74,33 @@ func (s *Server) detectDiagnoseHealth(ctx context.Context, kind, namespace, name signal.AuditCount = auditSum.Count signal.AuditSeverity = auditSum.HighestSeverity signal.TopFinding = auditSum.TopFinding + for _, row := range auditRows[:min(len(auditRows), maxHealthAuditLines)] { + signal.AuditFindings = append(signal.AuditFindings, ai.HealthLine{ + Severity: normalizeAuditSeverity(row.Severity), + Reason: row.CheckID, + Message: capHealthMessage(row.Message), + }) + } } return signal } +const ( + // maxHealthLines / maxHealthAuditLines cap the rows carried on the health + // frame: enough to make the context card + prompt concrete, small enough to + // stay a frame rather than a report (the agent reads the full state itself). + maxHealthLines = 3 + maxHealthAuditLines = 2 +) + +func capHealthMessage(msg string) string { + const max = 220 + if len(msg) <= max { + return msg + } + return msg[:max-1] + "…" +} + func managedByFromMeta(obj *unstructured.Unstructured) string { labels, ann := obj.GetLabels(), obj.GetAnnotations() has := func(m map[string]string, k string) bool { _, ok := m[k]; return ok } diff --git a/internal/server/ai_handlers.go b/internal/server/ai_handlers.go index 158745e36..7ab0969fc 100644 --- a/internal/server/ai_handlers.go +++ b/internal/server/ai_handlers.go @@ -501,12 +501,20 @@ func (s *Server) topologyForContext(namespace string) (*topology.Topology, topol // // Returns nil when no issues match — Build then omits the IssueSummary field. func computeIssueSummaryForResource(cache *k8s.ResourceCache, group, kind, namespace, name string) *resourcecontext.IssueSummary { + sum, _ := computeIssueSummaryAndRows(cache, group, kind, namespace, name) + return sum +} + +// computeIssueSummaryAndRows additionally returns the matched rows sorted by +// (severity desc, reason asc) — the diagnose health frame shows the actual +// lines, not just the rollup. +func computeIssueSummaryAndRows(cache *k8s.ResourceCache, group, kind, namespace, name string) (*resourcecontext.IssueSummary, []issues.Issue) { if cache == nil { - return nil + return nil, nil } provider := issues.NewCacheProvider() if provider == nil { - return nil + return nil, nil } var namespaces []string if namespace != "" { @@ -518,7 +526,7 @@ func computeIssueSummaryForResource(cache *k8s.ResourceCache, group, kind, names // both (a Deployment matched no Kind=Pod evidence rows → empty summary). matched := issues.RelatedIssues(provider, namespaces, group, kind, namespace, name) if len(matched) == 0 { - return nil + return nil, nil } bySource := make(map[string]int, len(matched)) for _, row := range matched { @@ -542,7 +550,7 @@ func computeIssueSummaryForResource(cache *k8s.ResourceCache, group, kind, names HighestSeverity: string(topSeverity), TopReason: topReason, BySource: bySource, - } + }, matched } // computeAuditSummaryForResource looks up audit findings for the subject @@ -558,8 +566,13 @@ func computeIssueSummaryForResource(cache *k8s.ResourceCache, group, kind, names // influence the choice — agents pinning regression tests on // resourceContext output rely on stable field values across runs. func computeAuditSummaryForResource(cache *k8s.ResourceCache, group, kind, namespace, name string) *resourcecontext.AuditSummary { + sum, _ := computeAuditSummaryAndRows(cache, group, kind, namespace, name) + return sum +} + +func computeAuditSummaryAndRows(cache *k8s.ResourceCache, group, kind, namespace, name string) (*resourcecontext.AuditSummary, []bpaudit.Finding) { if cache == nil || kind == "" { - return nil + return nil, nil } // Match computeIssueSummaryForResource's guard: passing []string{""} to // RunFromCache would filter to literally namespace="" resources instead @@ -572,12 +585,12 @@ func computeAuditSummaryForResource(cache *k8s.ResourceCache, group, kind, names } results := audit.RunFromCache(cache, namespaces, nil) if results == nil || len(results.Findings) == 0 { - return nil + return nil, nil } idx := bpaudit.IndexByResource(results.Findings) match := idx[bpaudit.ResourceKey(group, kind, namespace, name)] if len(match) == 0 { - return nil + return nil, nil } // Sort by (severity desc, CheckID asc) so TopFinding is deterministic @@ -594,7 +607,7 @@ func computeAuditSummaryForResource(cache *k8s.ResourceCache, group, kind, names Count: len(match), HighestSeverity: normalizeAuditSeverity(match[0].Severity), TopFinding: topFinding, - } + }, match } // normalizeAuditSeverity maps the audit suite's emission vocabulary diff --git a/web/src/api/diagnose.ts b/web/src/api/diagnose.ts index bb5be1efd..b3ab03495 100644 --- a/web/src/api/diagnose.ts +++ b/web/src/api/diagnose.ts @@ -41,14 +41,22 @@ export interface Diagnosis { sessionId?: string; } +export interface HealthLine { + severity?: string; + reason?: string; // issue Reason / audit CheckID + message?: string; +} + export interface ResourceHealthSignal { health?: string; issueCount?: number; highestSeverity?: string; topReason?: string; + issues?: HealthLine[]; auditCount?: number; auditSeverity?: string; topFinding?: string; + auditFindings?: HealthLine[]; } export interface DiagnoseStreamEvent { diff --git a/web/src/components/diagnose/parts.tsx b/web/src/components/diagnose/parts.tsx index 1e6f60d51..553779f39 100644 --- a/web/src/components/diagnose/parts.tsx +++ b/web/src/components/diagnose/parts.tsx @@ -480,51 +480,81 @@ export function TurnView({ // health frame the server captured at run start. It renders instantly (no agent // round-trip), so the agent's boot time reads as "context, then deepening" // instead of dead air — and it anchors the verdict against Radar's own signal. +function healthLineTone(severity?: string): "unhealthy" | "degraded" | "alert" { + if (severity === "critical") return "unhealthy"; + if (severity === "warning") return "degraded"; + return "alert"; +} + export function RunContextCard({ run }: { run: RunSummary }) { const h = run.health; const issueCount = h?.issueCount ?? 0; + const issues = h?.issues ?? []; + const findings = h?.auditFindings ?? []; const lines: ReactNode[] = []; - if (issueCount > 0) { - lines.push( - - - {issueCount} active issue{issueCount === 1 ? "" : "s"} - {h?.topReason ? ( - <> - {" — "} - {h.topReason} - - ) : null} - , - ); + if (issues.length > 0) { + // The actual issue rows Radar's engine flagged — the reason bolded, the + // engine's own detail sentence after it. Concrete beats a count. + for (const [i, line] of issues.entries()) { + lines.push( +
+ + + + {line.reason} + + {line.message ? <> — {line.message} : null} + +
, + ); + } + if (issueCount > issues.length) { + lines.push( +
+ +{issueCount - issues.length} more active issue + {issueCount - issues.length === 1 ? "" : "s"} +
, + ); + } } else if (h?.health === "healthy") { lines.push( - - +
+ Reported healthy — 0 active issues - , +
, ); } else if (h) { lines.push( - - - 0 active issues{h.health ? ` — status ${h.health}` : ""} - , +
+ 0 active issues + {h.health ? ` — status ${h.health}` : ""} +
, + ); + } + for (const [i, f] of findings.entries()) { + lines.push( +
+ Audit: {f.reason} + {f.message ? <> — {f.message} : null} +
, ); } - if ((h?.auditCount ?? 0) > 0) { + if ((h?.auditCount ?? 0) > findings.length && findings.length > 0) { lines.push( - - {h!.auditCount} configuration finding{h!.auditCount === 1 ? "" : "s"} from - cluster audit{h?.topFinding ? ` — ${h.topFinding}` : ""} - , +
+ +{h!.auditCount! - findings.length} more audit finding + {h!.auditCount! - findings.length === 1 ? "" : "s"} +
, ); } if (run.managedBy) { lines.push( - +
Managed by {run.managedBy} - , +
, ); } if (lines.length === 0) return null; @@ -533,11 +563,7 @@ export function RunContextCard({ run }: { run: RunSummary }) {
Radar's read at start
-
- {lines.map((l, i) => ( -
{l}
- ))} -
+
{lines}
); } From 07b952abd66604f154e6fefb80da6eef086bc57f Mon Sep 17 00:00:00 2001 From: Nadav Erell Date: Thu, 2 Jul 2026 12:15:31 +0300 Subject: [PATCH 08/11] radar diagnose: activity-verb spinner + terminal output polish MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wait line now speaks the web panel's vocabulary instead of a flat "thinking" — while a tool runs the spinner narrates it ("reading logs… 4s", "tracing dependencies…", from the step-running events), "starting investigation…" covers the agent boot, and "thinking…" only the genuine model gaps. Spinner glyph turned accent-amber. Output pass, element by element: - tool args render as terse k=v pairs (kind/namespace/name first) instead of raw JSON braces and quotes; - target in the header is bold-cyan; the meta line drops the "run run-4" doubling; - confidence band is color-coded (green/amber/red), not just dim text; - non-recommended remediation numbers dim so the ★ pick carries the row; "★ recommended —" phrasing; - rewrote render.go with the ANSI escapes as escape TEXT — a previous edit had embedded raw CR/ESC control bytes into the source string literals (compiled, but invisible-character source rot). --- internal/diagnosecli/diagnosecli.go | 4 +- internal/diagnosecli/render.go | 149 ++++++++++++++++++++++------ 2 files changed, 124 insertions(+), 29 deletions(-) diff --git a/internal/diagnosecli/diagnosecli.go b/internal/diagnosecli/diagnosecli.go index 55622a937..eeee806b0 100644 --- a/internal/diagnosecli/diagnosecli.go +++ b/internal/diagnosecli/diagnosecli.go @@ -389,7 +389,9 @@ func streamRun(base, id string, out *renderer) (json.RawMessage, bool) { cur.Ms = ev.Step.Ms cur.Status = ev.Step.Status steps[ev.Step.ID] = cur - if ev.Step.Status == "done" { + if ev.Step.Status == "running" { + out.toolStarted(cur.Tool) + } else if ev.Step.Status == "done" { out.step(cur) } case "done": diff --git a/internal/diagnosecli/render.go b/internal/diagnosecli/render.go index ecbb30bc1..77be7875e 100644 --- a/internal/diagnosecli/render.go +++ b/internal/diagnosecli/render.go @@ -1,9 +1,11 @@ package diagnosecli import ( + "encoding/json" "fmt" "os" "regexp" + "sort" "strings" "sync" "time" @@ -11,6 +13,19 @@ import ( "golang.org/x/term" ) +const ( + cReset = "\x1b[0m" + cDim = "\x1b[2m" + cBold = "\x1b[1m" + cGreen = "\x1b[32m" + cRed = "\x1b[31m" + cAmber = "\x1b[33m" + cCyan = "\x1b[36m" + + // clearLine returns the cursor to column 0 and erases the spinner line. + clearLine = "\r\x1b[K" +) + // renderer writes the live transcript + verdict to the terminal. In --json mode // everything human goes to stderr so stdout stays a clean JSON document. // A single mutex serializes event writes with the spinner goroutine: the model @@ -24,6 +39,8 @@ type renderer struct { inThinking bool spinnerOn bool // spinner line currently drawn (must be erased before real output) lastEvent time.Time // last real output, for the quiet-gap threshold + activeTool string // tool currently running (the spinner speaks its activity verb) + sawAnything bool // false until the agent's first output ("starting investigation…") stopSpin chan struct{} spinStopped bool } @@ -43,8 +60,9 @@ func newRenderer(jsonMode bool) *renderer { var spinnerFrames = []string{"⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"} -// startSpinner shows "⠙ thinking… 12s" after a second of silence — only on a -// real terminal (it repaints its own line), and never mid-thinking-line. +// startSpinner shows a live activity line ("⠙ reading logs… 12s") after a +// second of silence — only on a real terminal (it repaints its own line), and +// never mid-thinking-line. func (r *renderer) startSpinner() { if !r.color { return @@ -62,8 +80,9 @@ func (r *renderer) startSpinner() { r.mu.Lock() quiet := time.Since(r.lastEvent) if quiet > time.Second && !r.inThinking { - fmt.Fprintf(r.w, " %s%s %s thinking… %ds%s", - cDim, spinnerFrames[frame%len(spinnerFrames)], cAmber+"·"+cReset+cDim, int(quiet.Seconds()), cReset) + fmt.Fprintf(r.w, "%s%s%s %s %ds%s", + clearLine, cAmber+spinnerFrames[frame%len(spinnerFrames)]+cReset, + cDim, r.activityVerbLocked(), int(quiet.Seconds()), cReset) r.spinnerOn = true frame++ } @@ -86,27 +105,49 @@ func (r *renderer) stopSpinner() { // clearSpinnerLocked erases the spinner line before real output. Caller holds r.mu. func (r *renderer) clearSpinnerLocked() { if r.spinnerOn { - fmt.Fprint(r.w, " ") + fmt.Fprint(r.w, clearLine) r.spinnerOn = false } r.lastEvent = time.Now() + r.sawAnything = true } -const ( - cReset = "\x1b[0m" - cDim = "\x1b[2m" - cBold = "\x1b[1m" - cGreen = "\x1b[32m" - cRed = "\x1b[31m" - cAmber = "\x1b[33m" - cCyan = "\x1b[36m" -) +// activityVerbLocked mirrors the web panel's live status vocabulary: the wait +// names what the agent is actually doing, not a generic "thinking". Caller +// holds r.mu. +func (r *renderer) activityVerbLocked() string { + if t := strings.ToLower(r.activeTool); t != "" { + switch { + case strings.Contains(t, "log"): + return "reading logs…" + case strings.Contains(t, "event"): + return "checking recent events…" + case strings.Contains(t, "prometheus") || strings.Contains(t, "metric") || strings.Contains(t, "top"): + return "checking metrics…" + case strings.Contains(t, "topology") || strings.Contains(t, "neighborhood") || strings.Contains(t, "graph"): + return "tracing dependencies…" + case strings.Contains(t, "list") || strings.Contains(t, "search"): + return "scanning related resources…" + case strings.Contains(t, "diagnose"): + return "running diagnostics…" + case strings.Contains(t, "resource") || strings.Contains(t, "describe"): + return "inspecting the resource…" + } + return prettyTool(r.activeTool) + "…" + } + if !r.sawAnything { + return "starting investigation…" + } + return "thinking…" +} -func (r *renderer) c(code, s string) string { - if !r.color { - return s +// toolStarted records the running tool so the spinner narrates it. +func (r *renderer) toolStarted(tool string) { + r.mu.Lock() + if tool != "" { + r.activeTool = tool } - return code + s + cReset + r.mu.Unlock() } func (r *renderer) header(run runSummary, base string) { @@ -115,8 +156,8 @@ func (r *renderer) header(run runSummary, base string) { target += run.Namespace + "/" } target += run.Name - fmt.Fprintf(r.w, "%s %s\n", r.c(cBold, "◉ Investigating"), target) - fmt.Fprintf(r.w, "%s\n", r.c(cDim, fmt.Sprintf("run %s · via %s · watch: %s/?ai-run=%s", run.ID, agentDisplay(run.Agent), base, run.ID))) + fmt.Fprintf(r.w, "%s %s\n", r.c(cBold, "◉ Investigating"), r.c(cBold, r.c(cCyan, target))) + fmt.Fprintf(r.w, "%s\n", r.c(cDim, fmt.Sprintf("%s · via %s · watch: %s/?ai-run=%s", run.ID, agentDisplay(run.Agent), base, run.ID))) // Radar's read at start — the concrete issue rows the server captured, shown // before the agent produces anything (its boot is the longest silent gap). if h := run.Health; h != nil { @@ -152,6 +193,13 @@ func agentDisplay(name string) string { return name } +func (r *renderer) c(code, s string) string { + if !r.color { + return s + } + return code + s + cReset +} + // thinking streams the agent's interleaved reasoning, dimmed. func (r *renderer) thinking(token string) { if token == "" { @@ -176,18 +224,21 @@ func (r *renderer) breakThinkingLocked() { } } -// step prints one completed tool call: " ✓ get resource {"name":"web"} · 233ms". +// step prints one completed tool call: " ✓ get resource kind=node name=… 44ms". func (r *renderer) step(s stepInfo) { r.mu.Lock() defer r.mu.Unlock() r.clearSpinnerLocked() r.breakThinkingLocked() + if s.Tool == r.activeTool { + r.activeTool = "" + } line := " " + r.c(cGreen, "✓") + " " + prettyTool(s.Tool) - if s.Summary != "" { - line += " " + r.c(cDim, compact(s.Summary, 80)) + if args := prettyArgs(s.Summary); args != "" { + line += " " + r.c(cDim, args) } if s.Ms != nil { - line += r.c(cDim, fmt.Sprintf(" · %dms", *s.Ms)) + line += r.c(cDim, fmt.Sprintf(" %dms", *s.Ms)) } fmt.Fprintln(r.w, line) } @@ -196,6 +247,40 @@ func prettyTool(tool string) string { return strings.ReplaceAll(tool, "_", " ") } +// prettyArgs renders a tool's JSON input as terse k=v pairs — raw braces and +// quotes read as noise at a glance. Identity keys lead (kind, namespace, name), +// the rest follow sorted; anything non-JSON falls back to a compacted string. +func prettyArgs(raw string) string { + raw = strings.TrimSpace(raw) + if raw == "" { + return "" + } + var m map[string]any + if err := json.Unmarshal([]byte(raw), &m); err != nil || len(m) == 0 { + return compact(raw, 80) + } + lead := []string{"kind", "namespace", "name"} + parts := make([]string, 0, len(m)) + seen := map[string]bool{} + for _, k := range lead { + if v, ok := m[k]; ok { + parts = append(parts, fmt.Sprintf("%s=%v", k, v)) + seen[k] = true + } + } + rest := make([]string, 0, len(m)) + for k := range m { + if !seen[k] { + rest = append(rest, k) + } + } + sort.Strings(rest) + for _, k := range rest { + parts = append(parts, fmt.Sprintf("%s=%v", k, m[k])) + } + return compact(strings.Join(parts, " "), 90) +} + func compact(s string, max int) string { s = strings.Join(strings.Fields(s), " ") if len(s) > max { @@ -232,21 +317,29 @@ func (r *renderer) verdict(d diagnosis) { case d.RootCause != "": conf := "" if d.Confidence != nil { - conf = r.c(cDim, " · confidence "+confidenceLabel(*d.Confidence)) + label := confidenceLabel(*d.Confidence) + col := cGreen + switch label { + case "medium": + col = cAmber + case "low": + col = cRed + } + conf = r.c(cDim, " · confidence ") + r.c(col, label) } fmt.Fprintf(r.w, "%s%s\n", r.c(cAmber, r.c(cBold, "▲ Root cause")), conf) fmt.Fprintln(r.w, r.md(d.RootCause)) if len(d.Remediation) > 0 { fmt.Fprintf(r.w, "\n%s\n", r.c(cBold, "Remediation")) for i, step := range d.Remediation { - marker := fmt.Sprintf(" %d.", i+1) + marker := fmt.Sprintf(" %s", r.c(cDim, fmt.Sprintf("%d.", i+1))) if d.RecommendedIndex != nil && *d.RecommendedIndex == i+1 { - marker = r.c(cGreen, " ★"+fmt.Sprintf("%d.", i+1)) + marker = " " + r.c(cGreen, fmt.Sprintf("★%d.", i+1)) } fmt.Fprintf(r.w, "%s %s\n", marker, r.md(step)) } if d.RecommendedIndex != nil && d.RecommendedReason != "" { - fmt.Fprintf(r.w, " %s\n", r.c(cDim, "★ recommended: "+d.RecommendedReason)) + fmt.Fprintf(r.w, " %s\n", r.c(cDim, "★ recommended — "+d.RecommendedReason)) } } default: From 170c7c7f88bebda0bf8d06cd8643bf6688721cdf Mon Sep 17 00:00:00 2001 From: Nadav Erell Date: Thu, 2 Jul 2026 14:20:42 +0300 Subject: [PATCH 09/11] radar diagnose: name the older-Radar/stale-port-file failure instead of raw 404 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Multiple radar instances share ~/.radar/mcp-port (last writer wins), so discovery can land on a stale pre-Diagnose instance — which answers 404 on /api/agents. Explain that and point at --server rather than printing the bare status. --- internal/diagnosecli/diagnosecli.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/internal/diagnosecli/diagnosecli.go b/internal/diagnosecli/diagnosecli.go index eeee806b0..d987be210 100644 --- a/internal/diagnosecli/diagnosecli.go +++ b/internal/diagnosecli/diagnosecli.go @@ -137,7 +137,14 @@ Flags: } agents, err := fetchAgents(base) if err != nil { - fmt.Fprintf(os.Stderr, "found Radar at %s but couldn't query it: %v\n", base, err) + // A 404 here means SOMETHING answered but has no /api/agents — almost + // always an older Radar (or a stale ~/.radar/mcp-port pointing at one + // when several instances ran). Say that, not just "404". + if strings.Contains(err.Error(), "404") { + fmt.Fprintf(os.Stderr, "the Radar at %s doesn't support AI diagnosis — it's likely an older version (or a stale ~/.radar/mcp-port from another instance). Upgrade/restart it, or pass --server for the right instance.\n", base) + } else { + fmt.Fprintf(os.Stderr, "found Radar at %s but couldn't query it: %v\n", base, err) + } return 1 } if !agents.Enabled { From dc6e510d9643b3e7b650b6940c365beaa8f3bb41 Mon Sep 17 00:00:00 2001 From: Nadav Erell Date: Thu, 2 Jul 2026 14:35:34 +0300 Subject: [PATCH 10/11] =?UTF-8?q?radar=20diagnose:=20standalone=20mode=20?= =?UTF-8?q?=E2=80=94=20run=20cold=20without=20a=20running=20Radar?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --standalone (and automatic fallback when discovery finds nothing listening) boots a temporary in-process Radar for the one investigation: headless, random port, timeline in memory, --kubeconfig to pick a cluster. It skips the ~/.radar/mcp-port claim entirely (new app.DisableMCPPortFile guards both write AND the shutdown-time removal, so an ephemeral run can't clobber or delete a real instance's discovery slot) — but SHARES ai-runs.db, so a cold run's transcript, verdict, and session land in the same history the UI shows (run ids continue the persisted sequence). Boot UX: consent prompt BEFORE the cluster connect, a single "starting a temporary Radar — connecting to cluster… 12s" spinner during it, and all of Radar's boot + request logging (stdlib AND chi's own request logger, which binds os.Stdout at init) captured to a 64KB tail buffer that only surfaces on failure. Readiness = the diagnose endpoints stopping 503 (requireConnected), with a 2-minute ceiling; an explicit --server that isn't answering errors instead of falling back. Verified cold end-to-end against EKS with a stubbed agent: fallback note → connect → clean transcript → verdict → exit 0, no port file written. --- internal/app/bootstrap.go | 12 ++- internal/diagnosecli/diagnosecli.go | 77 ++++++++++++-- internal/diagnosecli/standalone.go | 156 ++++++++++++++++++++++++++++ 3 files changed, 232 insertions(+), 13 deletions(-) create mode 100644 internal/diagnosecli/standalone.go diff --git a/internal/app/bootstrap.go b/internal/app/bootstrap.go index a6c9fa94f..2a08a7289 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -422,11 +422,19 @@ func InitializeCluster() { }() } +// mcpPortFileDisabled suppresses port-file writes AND removals — an ephemeral +// instance (radar diagnose --standalone) must never clobber or delete the slot +// a real long-running Radar owns. +var mcpPortFileDisabled bool + +// DisableMCPPortFile makes Write/RemoveMCPPortFile no-ops for this process. +func DisableMCPPortFile() { mcpPortFileDisabled = true } + // WriteMCPPortFile writes the actual server port to ~/.radar/mcp-port so MCP // clients can discover the running instance without hardcoding a port. func WriteMCPPortFile(port int) { path := mcpPortFilePath() - if path == "" { + if path == "" || mcpPortFileDisabled { return } if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { @@ -443,7 +451,7 @@ func WriteMCPPortFile(port int) { // RemoveMCPPortFile removes the port discovery file on shutdown. func RemoveMCPPortFile() { path := mcpPortFilePath() - if path == "" { + if path == "" || mcpPortFileDisabled { return } if err := os.Remove(path); err != nil && !os.IsNotExist(err) { diff --git a/internal/diagnosecli/diagnosecli.go b/internal/diagnosecli/diagnosecli.go index d987be210..002988dbb 100644 --- a/internal/diagnosecli/diagnosecli.go +++ b/internal/diagnosecli/diagnosecli.go @@ -7,6 +7,7 @@ package diagnosecli import ( "bufio" + "context" "encoding/json" "flag" "fmt" @@ -19,6 +20,8 @@ import ( "time" "golang.org/x/term" + + "github.com/skyhook-io/radar/internal/ai" ) // kindAliases maps kubectl-style short/plural names to the canonical Kind. @@ -55,12 +58,14 @@ func normalizeKind(k string) string { } type options struct { - namespace string - agent string - server string - jsonOut bool - open bool - yes bool + namespace string + agent string + server string + kubeconfig string + standalone bool + jsonOut bool + open bool + yes bool } func newFlagSet() (*flag.FlagSet, *options) { @@ -73,6 +78,8 @@ func newFlagSet() (*flag.FlagSet, *options) { fs.BoolVar(&o.jsonOut, "json", false, "Print the final verdict as JSON on stdout (progress goes to stderr)") fs.BoolVar(&o.open, "open", false, "Also open the investigation in the Radar UI") fs.BoolVar(&o.yes, "yes", false, "Skip the first-run consent prompt") + fs.BoolVar(&o.standalone, "standalone", false, "Run against a temporary in-process Radar instead of a running instance (slower: connects to the cluster first)") + fs.StringVar(&o.kubeconfig, "kubeconfig", "", "Kubeconfig for --standalone (default: ~/.kube/config)") return fs, o } @@ -130,18 +137,52 @@ Flags: out := newRenderer(o.jsonOut) - base, err := resolveServer(o.server) - if err != nil { - fmt.Fprintln(os.Stderr, err) - return 1 + // Resolve the engine: an explicit --server, else the running instance from + // ~/.radar/mcp-port, else fall back to a temporary in-process Radar (what + // --standalone forces). Cold start pays a cluster connect up front — the + // running-instance path stays the fast default. + standalone := o.standalone + var base string + if !standalone { + var err error + base, err = resolveServer(o.server) + if err != nil || !probeListening(base) { + if o.server != "" { + // An explicit --server that isn't answering is an error, not a + // cue to boot something else. + fmt.Fprintf(os.Stderr, "nothing is listening at %s\n", o.server) + return 1 + } + fmt.Fprintln(os.Stderr, "no running Radar found — starting a temporary one for this investigation (use --server to target a running instance)") + standalone = true + } + } + + if standalone { + // Consent BEFORE the boot: nobody wants to answer a prompt after + // watching a cluster connect for 30 seconds. + if !o.yes && !consentGiven() { + if !promptConsent(localAgentLabel(o.agent)) { + fmt.Fprintln(os.Stderr, "aborted") + return 1 + } + } + b, shutdown, err := bootEphemeral(o.kubeconfig, false) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + defer shutdown() + base = b } + agents, err := fetchAgents(base) if err != nil { // A 404 here means SOMETHING answered but has no /api/agents — almost // always an older Radar (or a stale ~/.radar/mcp-port pointing at one // when several instances ran). Say that, not just "404". if strings.Contains(err.Error(), "404") { - fmt.Fprintf(os.Stderr, "the Radar at %s doesn't support AI diagnosis — it's likely an older version (or a stale ~/.radar/mcp-port from another instance). Upgrade/restart it, or pass --server for the right instance.\n", base) + fmt.Fprintf(os.Stderr, "the Radar at %s doesn't support AI diagnosis — it's likely an older version (or a stale ~/.radar/mcp-port from another instance). Upgrade/restart it, pass --server for the right instance, or use --standalone.\n", base) } else { fmt.Fprintf(os.Stderr, "found Radar at %s but couldn't query it: %v\n", base, err) } @@ -184,6 +225,20 @@ Flags: return 0 } +// localAgentLabel names the agent for the standalone consent prompt by probing +// PATH directly — there's no server to ask yet. +func localAgentLabel(pick string) string { + for _, info := range ai.DetectAgents(context.Background(), false) { + if !info.Supported { + continue + } + if pick == "" || info.Name == pick { + return info.Label + } + } + return "your agent CLI" +} + // --- server discovery ------------------------------------------------------- func resolveServer(explicit string) (string, error) { diff --git a/internal/diagnosecli/standalone.go b/internal/diagnosecli/standalone.go new file mode 100644 index 000000000..ce360be0c --- /dev/null +++ b/internal/diagnosecli/standalone.go @@ -0,0 +1,156 @@ +package diagnosecli + +import ( + "fmt" + "log" + "net" + "net/http" + "os" + "strings" + "sync" + "time" + + chimiddleware "github.com/go-chi/chi/v5/middleware" + "golang.org/x/term" + + "github.com/skyhook-io/radar/internal/app" +) + +// bootEphemeral starts a temporary in-process Radar for one investigation: +// headless, random port, no ~/.radar/mcp-port claim (a real instance may own +// it), timeline in memory — but the SHARED ai-runs history, so a cold run's +// transcript still shows up in the UI later. Radar's boot logging is captured +// to a tail buffer and surfaced only on failure; the terminal shows a single +// connecting spinner instead. +func bootEphemeral(kubeconfig string, quiet bool) (base string, shutdown func(), err error) { + tail := &tailBuffer{limit: 64 << 10} + log.SetOutput(tail) // for the whole process — request logs would drown the transcript + // chi's request logger holds its OWN logger bound to os.Stdout at package + // init — redirect it too, or every API call the CLI makes prints a line + // into the transcript. Must happen before CreateServer builds the router. + chimiddleware.DefaultLogger = chimiddleware.RequestLogger(&chimiddleware.DefaultLogFormatter{ + Logger: log.New(tail, "", log.LstdFlags), NoColor: true, + }) + + app.DisableMCPPortFile() + cfg := app.AppConfig{ + Kubeconfig: kubeconfig, + Port: 0, // random free port + NoBrowser: true, + HistoryLimit: 1000, // in-memory timeline floor; this instance lives minutes + MCPEnabled: true, + AIHistory: true, + TimelineStorage: "memory", + } + app.SetGlobals(cfg) + if err := app.InitializeK8s(cfg); err != nil { + return "", nil, fmt.Errorf("kubeconfig: %w", err) + } + app.RegisterCallbacks(cfg, app.BuildTimelineStoreConfig(cfg)) + srv := app.CreateServer(cfg) + + ready := make(chan struct{}) + go func() { + if err := srv.StartWithReady(ready); err != nil && !strings.Contains(err.Error(), "closed") { + log.Printf("ephemeral server error: %v", err) + } + }() + select { + case <-ready: + case <-time.After(15 * time.Second): + return "", nil, fmt.Errorf("temporary Radar didn't start listening\n%s", tail.String()) + } + base = fmt.Sprintf("http://localhost:%d", srv.ActualPort()) + + stopSpin := make(chan struct{}) + if !quiet { + go bootSpinner(stopSpin) + } + app.InitializeCluster() + + // Connected = the diagnose endpoints stop returning 503 (requireConnected). + deadline := time.Now().Add(2 * time.Minute) + for { + resp, err := http.Get(base + "/api/diagnose/runs") + if err == nil { + code := resp.StatusCode + resp.Body.Close() + if code == http.StatusOK { + break + } + if code == http.StatusNotImplemented { + close(stopSpin) + return "", nil, fmt.Errorf("no supported agent CLI found — install Claude Code, Codex, or Cursor") + } + } + if time.Now().After(deadline) { + close(stopSpin) + return "", nil, fmt.Errorf("couldn't connect to the cluster\n--- radar log tail ---\n%s", tail.String()) + } + time.Sleep(500 * time.Millisecond) + } + close(stopSpin) + + return base, func() { app.Shutdown(srv) }, nil +} + +// bootSpinner is the pre-run wait line ("starting a temporary Radar — +// connecting to cluster… 12s") on stderr; the renderer's own spinner takes +// over once the investigation streams. +func bootSpinner(stop <-chan struct{}) { + if os.Getenv("NO_COLOR") != "" || !term.IsTerminal(int(os.Stderr.Fd())) { + return + } + start := time.Now() + t := time.NewTicker(120 * time.Millisecond) + defer t.Stop() + frame := 0 + for { + select { + case <-stop: + fmt.Fprint(os.Stderr, clearLine) + return + case <-t.C: + fmt.Fprintf(os.Stderr, "%s%s%s starting a temporary Radar — connecting to cluster… %ds%s", + clearLine, cAmber+spinnerFrames[frame%len(spinnerFrames)]+cReset, cDim, + int(time.Since(start).Seconds()), cReset) + frame++ + } + } +} + +// tailBuffer keeps the last `limit` bytes written — enough boot log to debug a +// failed cluster connect without ever holding the whole request log. +type tailBuffer struct { + mu sync.Mutex + buf []byte + limit int +} + +func (t *tailBuffer) Write(p []byte) (int, error) { + t.mu.Lock() + defer t.mu.Unlock() + t.buf = append(t.buf, p...) + if len(t.buf) > t.limit { + t.buf = t.buf[len(t.buf)-t.limit:] + } + return len(p), nil +} + +func (t *tailBuffer) String() string { + t.mu.Lock() + defer t.mu.Unlock() + return string(t.buf) +} + +// probeListening reports whether anything answers on the discovered base — +// used to distinguish "stale port file" from "no Radar at all". +func probeListening(base string) bool { + u := strings.TrimPrefix(strings.TrimPrefix(base, "http://"), "https://") + conn, err := net.DialTimeout("tcp", u, time.Second) + if err != nil { + return false + } + _ = conn.Close() + return true +} From b412dc475ea8d812a8b2b4236590d84b915fa727 Mon Sep 17 00:00:00 2001 From: Nadav Erell Date: Thu, 2 Jul 2026 15:27:09 +0300 Subject: [PATCH 11/11] radar diagnose: keep klog off the standalone transcript MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit client-go logs (apiserver deprecation warnings etc.) go through klog, which writes directly to stderr and bypasses the stdlib-log redirect — in standalone mode they stomped the streaming transcript. Route klog into the same tail buffer, mirroring main()'s server-path setup (the subcommand exits before reaching it). --- internal/diagnosecli/standalone.go | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/internal/diagnosecli/standalone.go b/internal/diagnosecli/standalone.go index ce360be0c..b74a5bf47 100644 --- a/internal/diagnosecli/standalone.go +++ b/internal/diagnosecli/standalone.go @@ -1,6 +1,7 @@ package diagnosecli import ( + "flag" "fmt" "log" "net" @@ -12,6 +13,7 @@ import ( chimiddleware "github.com/go-chi/chi/v5/middleware" "golang.org/x/term" + "k8s.io/klog/v2" "github.com/skyhook-io/radar/internal/app" ) @@ -25,6 +27,16 @@ import ( func bootEphemeral(kubeconfig string, quiet bool) (base string, shutdown func(), err error) { tail := &tailBuffer{limit: 64 << 10} log.SetOutput(tail) // for the whole process — request logs would drown the transcript + // client-go logs through klog, which writes DIRECTLY to stderr by default + // (bypassing stdlib log) — e.g. apiserver deprecation warnings fired by the + // agent's list calls would stomp the transcript mid-spinner. The server + // path tames this in main(); the subcommand exits before reaching it, so + // repeat it here, pointed at the tail buffer. + klog.InitFlags(nil) + _ = flag.Set("v", "0") + _ = flag.Set("logtostderr", "false") + _ = flag.Set("alsologtostderr", "false") + klog.SetOutput(tail) // chi's request logger holds its OWN logger bound to os.Stdout at package // init — redirect it too, or every API call the CLI makes prints a line // into the transcript. Must happen before CreateServer builds the router.