From 47628c10b06dca6cf5c98d08f9cf744a078d28b0 Mon Sep 17 00:00:00 2001 From: Abhijit Date: Thu, 8 Oct 2026 17:22:21 +0530 Subject: [PATCH 1/7] Signal history and incident records MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cometcli now remembers: - every node.triage run is saved to the node's signal history (~/.cometcli/history//, 30 days); triage opens with what changed since the last check (node.peers 8→1, val.jailed false→true) - node.history: first→last, min/max, trend line and change times per signal over a window ('since when?', 'is it getting worse?') - mon.record: a sampler that records every interval between triages - incident.record saves a finished incident (root cause, case, evidence, actions, outcome) as JSON + a Markdown postmortem, with the timeline and tx hashes rebuilt from the audit log; incident.list / incident.show - triage names the node's past incidents and flags recurring cases - /incident and the node prompt use them; both are core tools Co-Authored-By: Claude Opus 5.5 --- README.md | 13 +- docs/COMMANDS.md | 25 ++ internal/agent/builtin_cmds.go | 5 +- internal/agent/prompt.go | 7 +- internal/agent/toolsearch.go | 1 + internal/cli/agent.go | 2 +- internal/history/history.go | 269 ++++++++++++++++++++ internal/history/history_test.go | 62 +++++ internal/incidents/incidents.go | 227 +++++++++++++++++ internal/incidents/incidents_test.go | 65 +++++ internal/tools/triage/collect.go | 7 + internal/tools/triage/memory.go | 365 +++++++++++++++++++++++++++ internal/tools/triage/tool.go | 21 ++ internal/tools/triage/triage_test.go | 35 +++ 14 files changed, 1097 insertions(+), 7 deletions(-) create mode 100644 internal/history/history.go create mode 100644 internal/history/history_test.go create mode 100644 internal/incidents/incidents.go create mode 100644 internal/incidents/incidents_test.go create mode 100644 internal/tools/triage/memory.go diff --git a/README.md b/README.md index ea2b9d7..b5c36f0 100644 --- a/README.md +++ b/README.md @@ -179,6 +179,12 @@ Each case lists how to confirm it, its causes, fix steps tagged `[read]` / **`/incident [what you see]`** works the top case to resolution: confirm → fix the root cause → wait (`wait.until`, without spending model calls) → verify → report. When the agent finds a cause no case covered, it offers to record it with `kb.add`. + +cometcli remembers. Every triage is saved to the node's history, so it can tell you what +changed since the last check and answer "since when?" (`node history --since 7d`). +Every finished incident is saved as a postmortem with its root cause, actions, +transaction hashes and a timeline from the audit log. Triage points out when the same +problem comes back. Add your own cases in `~/.cometcli/kb/` — format in [docs/EXTENDING.md](docs/EXTENDING.md#knowledge-base-cases). @@ -222,17 +228,18 @@ Tools are grouped by domain. Required args also bind positionally — | Domain | Highlights | |---|---| -| `node` | `triage`, `status`, `peers`, `health`, `consensus`, `logs`, `config --action lint`, `service status\|restart`, `version-check` | +| `node` | `triage`, `history`, `status`, `peers`, `health`, `consensus`, `logs`, `config --action lint`, `service status\|restart`, `version-check` | | `val` | `status`, `signing`, `jail-check`, `consensus`, `rewards`, `votes` · on-chain: `unjail`, `withdraw`, `vote`, `edit`, `create` | | `chain` | `validators` (`--status jailed`), `params`, `gov`, `upgrade-plan`, `pool`, `balance` | | `gov` | `propose` (text / upgrade / messages), `deposit` | +| `incident` | `list`, `show`, `record` — postmortems of past incidents | | `kb` | `search`, `show`, `add` — the knowledge base of failure cases | | `wait` | `until` — synced, height, unjailable, signing, in-consensus, tx-committed, proposal-status, or any triage signal | | `evm` | `chainid`, `parity` (comet↔JSON-RPC drift), `gasprice`, `txpool` | | `tx` | `send`, `delegate`, `undelegate`, `redelegate`, `get` — every tx tool takes `--gas-price`, `--gas-limit`, `--fee-denom`, `--seq` | | `keys` | `add` (`--recover`/`--privkey-hex`), `list`, `show`, `rm`, `convert` (bech32↔0x) | | `sec` | `exposure` (listening ports), `perms` (key-file permissions), `doublesign` (priv_validator_state check) | -| `mon` | `snapshot`, `watch` (live TUI), `alerts` (rules → stdout/Slack/Discord/Telegram) | +| `mon` | `snapshot`, `watch` (live TUI), `record` (signal history), `alerts` (rules → stdout/Slack/Discord/Telegram) | | `upgrade` | `check`, `prepare` (download, verify, stage for cosmovisor), `watch` | | `snap` | `list`, `prune`, `statesync` | | `runbook` | `list`, `show`, `run` — builtins + your own YAML | @@ -309,6 +316,8 @@ credentials # API keys saved by `config set-key` (0600) settings.json # permission rules, hooks, MCP servers (also per project: .cometcli/) COMET.md # memory the agent always reads (also per project) kb/ # your own knowledge-base cases +history/ # triage signal history per profile (30 days) +incidents/ # incident records and postmortems per profile keys/ # ops keyring (file backend) sessions/ # saved conversations (cometcli -c / -r) runbooks/ # your custom YAML playbooks diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index d64fb3e..128c954 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -116,6 +116,31 @@ format in [EXTENDING.md](EXTENDING.md#knowledge-base-cases). cometcli wait until --condition signal --value "host.disk_used_pct < 85 && proc.running == true" ``` +### History and incident records + +Every `node.triage` run is saved to the node's signal history +(`~/.cometcli/history//`, kept 30 days), and triage starts by saying what +changed since the previous run (`node.peers 8→1, val.jailed false→true`). For trends: + +```bash +cometcli node history # key health signals, last 24h +cometcli node history --signals val.,node.peers --since 7d +cometcli mon record --interval 5m # sample regularly, not only when someone triages +``` + +Each answer gives first → last, min/max, a trend line and when the value changed. + +When an incident is finished, the agent saves it with `incident.record`: title, root +cause, matching case, evidence, actions and outcome. The timeline (prompts, tool calls, +approvals) and transaction hashes are filled in from the audit log. Records live in +`~/.cometcli/incidents//` as JSON plus a Markdown postmortem. Triage +mentions a node's past incidents and flags a case that keeps coming back. + +```bash +cometcli incident list # last 30 days (--since 90d) +cometcli incident show --id 20261007-0533-val-jailed-downtime +``` + ### Incident tools Tools built for running an incident to resolution — usable directly or by the agent: diff --git a/internal/agent/builtin_cmds.go b/internal/agent/builtin_cmds.go index db8ad2b..758039f 100644 --- a/internal/agent/builtin_cmds.go +++ b/internal/agent/builtin_cmds.go @@ -10,13 +10,14 @@ var builtinCommands = map[string]*settings.Command{ Description: "Work an incident to resolution: triage, match a known case, fix the root cause, wait, verify, report", Body: `Work this incident on the node to resolution. Keep the operator's checklist current with todo_write; don't stop between steps unless something needs their decision. -1. node.triage (since=2h if the problem started earlier). Note which sources were unavailable. +1. node.triage (since=2h if the problem started earlier). Note which sources were unavailable, what changed since the last check, and past incidents on this node — a repeat means the last fix missed something (incident.show it). For "since when?", use node.history. 2. Take the top matched case: kb.show it, run its confirm steps. If confirmation fails, move to the next case or kb.search the symptoms/log lines. No case at all → investigate from the signals and logs (node.logs, bash read-only). 3. If several cases match, fix the one that CAUSES the others first (disk full → crash → behind → jailed: disk first). 4. Apply the fix steps by CALLING the tools. [change] and [tx] steps are approved at the tool's own prompt: state in one line what you're doing and why, then make the call in the same step — never stop to ask for approval in text. Never do anything a case lists under NEVER. 5. Wait for the effect with wait.until (synced, signing, in-consensus, or condition=signal value=""). If progress stalls, re-triage instead of waiting longer. 6. Verify with the case's verify steps, then node.triage again: the case must no longer match. -7. Report: root cause with evidence, actions taken (tx hashes), final state. If you found a cause no case described, propose a kb.add case (with a test fixture) for the operator to approve. +7. Record it with incident.record (title, root cause, case id, evidence, actions, outcome) — the timeline and tx hashes are filled in for you. +8. Report: root cause with evidence, actions taken (tx hashes), final state. If you found a cause no case described, propose a kb.add case (with a test fixture) for the operator to approve. What the operator sees: $ARGUMENTS`, }, diff --git a/internal/agent/prompt.go b/internal/agent/prompt.go index d5959cd..0ee7b4d 100644 --- a/internal/agent/prompt.go +++ b/internal/agent/prompt.go @@ -42,8 +42,11 @@ Incidents and "is something wrong?" questions — work the method: height, or signal="") instead of guessing; if progress stalls, re-triage. 5. Verify with the case's verify steps, then re-run node.triage. -6. Report cause + evidence, actions (tx hashes), final state. If no case - fit and you found the cause, offer to record it with kb.add. +6. Save it with incident.record, then report cause + evidence, actions + (tx hashes), final state. If no case fit and you found the cause, + offer to record it with kb.add. +Triage shows what changed since the last check and this node's past +incidents; node.history answers "since when?" and "is it getting worse?". `) b.WriteString(strings.Replace(generalToolsText, "General tools:", "General tools (they run on the node's host — local or over SSH):", 1)) diff --git a/internal/agent/toolsearch.go b/internal/agent/toolsearch.go index d0a112e..45deaf3 100644 --- a/internal/agent/toolsearch.go +++ b/internal/agent/toolsearch.go @@ -22,6 +22,7 @@ var coreTools = map[string]bool{ "val.status": true, "val.signing": true, "val.jail-check": true, "val.consensus": true, "wait.until": true, "chain.validators": true, "node.triage": true, "kb.show": true, "kb.search": true, + "node.history": true, "incident.record": true, } var toolSearchDef = ToolDef{ diff --git a/internal/cli/agent.go b/internal/cli/agent.go index 4a09cda..3d5f239 100644 --- a/internal/cli/agent.go +++ b/internal/cli/agent.go @@ -13,7 +13,7 @@ func agentFlags(cmd *cobra.Command) { cmd.Flags().StringSlice("autopilot", nil, "tiers that skip the confirm prompt (only local-change; on-chain never)") cmd.Flags().Bool("no-stream", false, "disable token streaming") cmd.Flags().Int("budget", 0, "max tool calls per turn (0 = unlimited)") - cmd.Flags().Int("max-iter", 0, "max model iterations per turn (default 50)") + cmd.Flags().Int("max-iter", 0, "max model steps per request (default 200, or agent.max_turns)") sessionFlags(cmd) } diff --git a/internal/history/history.go b/internal/history/history.go new file mode 100644 index 0000000..458b14d --- /dev/null +++ b/internal/history/history.go @@ -0,0 +1,269 @@ +// Package history keeps a per-profile record of triage signals over time +// (~/.cometcli/history//YYYY-MM-DD.jsonl), so questions like +// "when did peers start dropping?" can be answered from data instead of +// a single snapshot. Records older than Retention are pruned. +package history + +import ( + "bufio" + "encoding/json" + "fmt" + "math" + "os" + "path/filepath" + "sort" + "strings" + "time" + + "github.com/abhijitkrm/cometcli/internal/config" +) + +// Retention is how long records are kept. +var Retention = 30 * 24 * time.Hour + +// Record is one sample of signals. +type Record struct { + TS time.Time `json:"ts"` + Signals map[string]any `json:"s"` +} + +func dir(profile string) (string, error) { + if profile == "" || strings.ContainsAny(profile, `/\`) || strings.HasPrefix(profile, ".") { + return "", fmt.Errorf("history: bad profile name %q", profile) + } + return config.Path("history", profile) +} + +// keep reports whether a signal is worth storing: numbers, booleans and +// short strings (states, versions) — not addresses, log lines or paths. +func keep(k string, v any) bool { + switch x := v.(type) { + case float64, bool: + return true + case string: + return len(x) <= 32 && !strings.Contains(k, "addr") && !strings.HasSuffix(k, ".file") && + k != "logs.last_error" && k != "val.valoper" + } + return false +} + +// Append stores one sample for profile. +func Append(profile string, ts time.Time, signals map[string]any) error { + d, err := dir(profile) + if err != nil { + return err + } + if err := os.MkdirAll(d, 0o700); err != nil { + return err + } + rec := Record{TS: ts.UTC(), Signals: map[string]any{}} + for k, v := range signals { + if keep(k, v) { + rec.Signals[k] = v + } + } + p := filepath.Join(d, ts.UTC().Format("2006-01-02")+".jsonl") + _, statErr := os.Stat(p) + f, err := os.OpenFile(p, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600) + if err != nil { + return err + } + defer f.Close() + if err := json.NewEncoder(f).Encode(rec); err != nil { + return err + } + if os.IsNotExist(statErr) { + prune(d, ts) + } + return nil +} + +// prune drops day files older than Retention (on the first write of a day). +func prune(d string, now time.Time) { + files, _ := filepath.Glob(filepath.Join(d, "*.jsonl")) + cut := now.Add(-Retention).UTC().Format("2006-01-02") + for _, f := range files { + if strings.TrimSuffix(filepath.Base(f), ".jsonl") < cut { + _ = os.Remove(f) + } + } +} + +// Load returns profile's records since t, oldest first. +func Load(profile string, since time.Time) ([]Record, error) { + d, err := dir(profile) + if err != nil { + return nil, err + } + files, _ := filepath.Glob(filepath.Join(d, "*.jsonl")) + sort.Strings(files) + first := since.UTC().Format("2006-01-02") + var out []Record + for _, f := range files { + if strings.TrimSuffix(filepath.Base(f), ".jsonl") < first { + continue + } + fh, err := os.Open(f) + if err != nil { + continue + } + sc := bufio.NewScanner(fh) + sc.Buffer(make([]byte, 64<<10), 1<<20) + for sc.Scan() { + var r Record + if json.Unmarshal(sc.Bytes(), &r) == nil && !r.TS.Before(since) { + out = append(out, r) + } + } + fh.Close() + } + sort.Slice(out, func(i, j int) bool { return out[i].TS.Before(out[j].TS) }) + return out, nil +} + +// Last is the most recent record before t (nil if none within Retention). +func Last(profile string, before time.Time) *Record { + recs, err := Load(profile, before.Add(-Retention)) + if err != nil { + return nil + } + for i := len(recs) - 1; i >= 0; i-- { + if recs[i].TS.Before(before) { + return &recs[i] + } + } + return nil +} + +// Change is a moment a signal changed value. +type Change struct { + TS time.Time + From, To any +} + +// Summary describes one signal across records. +type Summary struct { + Name string + Samples int + First, Last any + Min, Max float64 // numeric signals only + Numeric bool + Changes []Change // value changes (bool/string), or big jumps (numeric) + Spark string // numeric trend, oldest → newest +} + +// Summarize describes name across recs. +func Summarize(recs []Record, name string) Summary { + s := Summary{Name: name, Min: math.Inf(1), Max: math.Inf(-1)} + var prev any + var nums []float64 + for _, r := range recs { + v, ok := r.Signals[name] + if !ok { + continue + } + s.Samples++ + if s.Samples == 1 { + s.First = v + } + s.Last = v + if f, ok := v.(float64); ok { + s.Numeric = true + nums = append(nums, f) + s.Min, s.Max = math.Min(s.Min, f), math.Max(s.Max, f) + if p, ok := prev.(float64); ok && bigJump(p, f) { + s.Changes = append(s.Changes, Change{r.TS, p, f}) + } + } else if prev != nil && fmt.Sprint(prev) != fmt.Sprint(v) { + s.Changes = append(s.Changes, Change{r.TS, prev, v}) + } + prev = v + } + if s.Numeric { + s.Spark = spark(nums, 24) + } + return s +} + +// bigJump is a change worth calling out: ≥ 20% and ≥ 1 in absolute terms. +func bigJump(a, b float64) bool { + d := math.Abs(b - a) + return d >= 1 && d >= 0.2*math.Max(math.Abs(a), math.Abs(b)) +} + +// spark draws values (averaged into at most n buckets) as a sparkline. +func spark(vals []float64, n int) string { + if len(vals) == 0 { + return "" + } + if len(vals) > n { + b := make([]float64, n) + for i := range b { + lo, hi := i*len(vals)/n, (i+1)*len(vals)/n + sum := 0.0 + for _, v := range vals[lo:hi] { + sum += v + } + b[i] = sum / float64(hi-lo) + } + vals = b + } + lo, hi := vals[0], vals[0] + for _, v := range vals { + lo, hi = math.Min(lo, v), math.Max(hi, v) + } + bars := []rune("▁▂▃▄▅▆▇█") + var out []rune + for _, v := range vals { + i := 0 + if hi > lo { + i = int((v - lo) / (hi - lo) * float64(len(bars)-1)) + } + out = append(out, bars[i]) + } + return string(out) +} + +// noisy signals change every sample by design; deltas skip them. +func noisy(k string) bool { + for _, s := range []string{"height", "block_age", "uptime", "window_min", "logs.lines", "disk_free", "pvs.ahead_by", "load_per_cpu", "mem_used_pct_of_limit"} { + if strings.Contains(k, s) { + return true + } + } + return false +} + +// Deltas lists what changed meaningfully between two samples, sorted: +// any boolean/string change, numeric jumps of ≥ 20%. +func Deltas(prev, cur map[string]any) []string { + var out []string + for k, v := range cur { + if noisy(k) || !keep(k, v) { + continue + } + p, ok := prev[k] + if !ok { + continue + } + switch x := v.(type) { + case float64: + if pf, ok := p.(float64); ok && bigJump(pf, x) { + out = append(out, fmt.Sprintf("%s %s→%s", k, num(pf), num(x))) + } + default: + if fmt.Sprint(p) != fmt.Sprint(v) { + out = append(out, fmt.Sprintf("%s %v→%v", k, p, v)) + } + } + } + sort.Strings(out) + return out +} + +func num(f float64) string { + if f == math.Trunc(f) { + return fmt.Sprintf("%d", int64(f)) + } + return fmt.Sprintf("%.1f", f) +} diff --git a/internal/history/history_test.go b/internal/history/history_test.go new file mode 100644 index 0000000..95f4a84 --- /dev/null +++ b/internal/history/history_test.go @@ -0,0 +1,62 @@ +package history + +import ( + "strings" + "testing" + "time" +) + +func TestAppendLoadSummarizeDeltas(t *testing.T) { + t.Setenv("COMETCLI_HOME", t.TempDir()) + base := time.Now().Add(-3 * time.Hour) + peers := []float64{8, 8, 7, 1, 0, 0} + for i, p := range peers { + jailed := i >= 4 + if err := Append("val01", base.Add(time.Duration(i)*30*time.Minute), map[string]any{ + "node.peers": p, "val.jailed": jailed, "node.height": float64(1000 + i*100), + "val.cons_addr_hex": "ABCDEF", "logs.last_error": "a long log line that must not be stored", + }); err != nil { + t.Fatal(err) + } + } + recs, err := Load("val01", base.Add(-time.Minute)) + if err != nil || len(recs) != len(peers) { + t.Fatalf("loaded %d records: %v", len(recs), err) + } + if _, ok := recs[0].Signals["logs.last_error"]; ok { + t.Fatal("log line stored") + } + s := Summarize(recs, "node.peers") + if s.First != 8.0 || s.Last != 0.0 || s.Min != 0 || s.Max != 8 || len([]rune(s.Spark)) != len(peers) { + t.Fatalf("summary %+v", s) + } + if len(s.Changes) == 0 || s.Changes[0].From != 7.0 || s.Changes[0].To != 1.0 { + t.Fatalf("big drop not marked: %+v", s.Changes) + } + j := Summarize(recs, "val.jailed") + if len(j.Changes) != 1 || j.Changes[0].To != true { + t.Fatalf("jail change: %+v", j.Changes) + } + d := Deltas(recs[2].Signals, recs[4].Signals) + got := strings.Join(d, "; ") + if !strings.Contains(got, "node.peers 7→0") || !strings.Contains(got, "val.jailed false→true") || strings.Contains(got, "height") { + t.Fatalf("deltas: %s", got) + } + if last := Last("val01", time.Now()); last == nil || last.Signals["node.peers"] != 0.0 { + t.Fatalf("last: %+v", last) + } + if err := Append("../escape", time.Now(), nil); err == nil { + t.Fatal("path traversal accepted") + } +} + +func TestPrune(t *testing.T) { + t.Setenv("COMETCLI_HOME", t.TempDir()) + old := time.Now().Add(-40 * 24 * time.Hour) + _ = Append("v", old, map[string]any{"x": 1.0}) + _ = Append("v", time.Now(), map[string]any{"x": 2.0}) // first write of a new day prunes + recs, _ := Load("v", old.Add(-time.Hour)) + if len(recs) != 1 { + t.Fatalf("old day not pruned: %d records", len(recs)) + } +} diff --git a/internal/incidents/incidents.go b/internal/incidents/incidents.go new file mode 100644 index 0000000..7f721da --- /dev/null +++ b/internal/incidents/incidents.go @@ -0,0 +1,227 @@ +// Package incidents keeps a record of every incident cometcli worked on, +// per profile (~/.cometcli/incidents//.{json,md}): what was +// wrong, the root cause, what was done (with tx hashes), the outcome, and +// a timeline rebuilt from the audit log. Triage reads it back, so a repeat +// problem is recognized as one. +package incidents + +import ( + "bufio" + "encoding/json" + "fmt" + "os" + "path/filepath" + "regexp" + "sort" + "strings" + "time" + + "github.com/abhijitkrm/cometcli/internal/audit" + "github.com/abhijitkrm/cometcli/internal/config" +) + +// Incident is one worked incident. +type Incident struct { + ID string `json:"id"` + Profile string `json:"profile"` + Title string `json:"title"` + Case string `json:"case,omitempty"` // knowledge-base case id + Started time.Time `json:"started"` + Resolved time.Time `json:"resolved,omitempty"` + RootCause string `json:"root_cause"` + Evidence []string `json:"evidence,omitempty"` + Actions []string `json:"actions,omitempty"` + TxHashes []string `json:"tx_hashes,omitempty"` + Outcome string `json:"outcome"` + Timeline []string `json:"timeline,omitempty"` + Session string `json:"session,omitempty"` +} + +func dir(profile string) (string, error) { + if profile == "" || strings.ContainsAny(profile, `/\`) || strings.HasPrefix(profile, ".") { + return "", fmt.Errorf("incidents: bad profile name %q", profile) + } + return config.Path("incidents", profile) +} + +var slugRe = regexp.MustCompile(`[^a-z0-9]+`) + +// Save writes inc (assigning an ID if empty) and returns the Markdown path. +func Save(inc *Incident) (string, error) { + d, err := dir(inc.Profile) + if err != nil { + return "", err + } + if err := os.MkdirAll(d, 0o700); err != nil { + return "", err + } + if inc.ID == "" { + slug := inc.Case + if slug == "" { + slug = strings.Trim(slugRe.ReplaceAllString(strings.ToLower(inc.Title), "-"), "-") + } + if len(slug) > 40 { + slug = slug[:40] + } + inc.ID = inc.Started.UTC().Format("20060102-1504") + "-" + slug + } + raw, err := json.MarshalIndent(inc, "", " ") + if err != nil { + return "", err + } + if err := os.WriteFile(filepath.Join(d, inc.ID+".json"), raw, 0o600); err != nil { + return "", err + } + md := filepath.Join(d, inc.ID+".md") + return md, os.WriteFile(md, []byte(Markdown(inc)), 0o600) +} + +// Markdown renders an incident as a short postmortem. +func Markdown(inc *Incident) string { + var b strings.Builder + fmt.Fprintf(&b, "# %s\n\n", inc.Title) + fmt.Fprintf(&b, "- **Node:** %s\n", inc.Profile) + if inc.Case != "" { + fmt.Fprintf(&b, "- **Known case:** `%s`\n", inc.Case) + } + fmt.Fprintf(&b, "- **Started:** %s\n", inc.Started.UTC().Format("2006-01-02 15:04 UTC")) + if !inc.Resolved.IsZero() { + fmt.Fprintf(&b, "- **Resolved:** %s (%s)\n", inc.Resolved.UTC().Format("2006-01-02 15:04 UTC"), + inc.Resolved.Sub(inc.Started).Round(time.Minute)) + } + fmt.Fprintf(&b, "- **Outcome:** %s\n\n", inc.Outcome) + fmt.Fprintf(&b, "## Root cause\n\n%s\n", inc.RootCause) + list := func(title string, items []string, code bool) { + if len(items) == 0 { + return + } + fmt.Fprintf(&b, "\n## %s\n\n", title) + for _, it := range items { + if code { + fmt.Fprintf(&b, "- `%s`\n", it) + } else { + fmt.Fprintf(&b, "- %s\n", it) + } + } + } + list("Evidence", inc.Evidence, false) + list("Actions", inc.Actions, false) + list("Transactions", inc.TxHashes, true) + list("Timeline (UTC)", inc.Timeline, false) + return b.String() +} + +// List returns profile's incidents since t, newest first. +func List(profile string, since time.Time) ([]Incident, error) { + d, err := dir(profile) + if err != nil { + return nil, err + } + files, _ := filepath.Glob(filepath.Join(d, "*.json")) + var out []Incident + for _, f := range files { + raw, err := os.ReadFile(f) + if err != nil { + continue + } + var inc Incident + if json.Unmarshal(raw, &inc) == nil && !inc.Started.Before(since) { + out = append(out, inc) + } + } + sort.Slice(out, func(i, j int) bool { return out[i].Started.After(out[j].Started) }) + return out, nil +} + +// Get loads one incident by id. +func Get(profile, id string) (*Incident, error) { + d, err := dir(profile) + if err != nil { + return nil, err + } + if strings.ContainsAny(id, `/\`) || strings.HasPrefix(id, ".") { + return nil, fmt.Errorf("bad incident id %q", id) + } + raw, err := os.ReadFile(filepath.Join(d, id+".json")) + if err != nil { + return nil, fmt.Errorf("no incident %s for %s", id, profile) + } + var inc Incident + return &inc, json.Unmarshal(raw, &inc) +} + +// FromAudit fills the timeline and tx hashes from the audit log: the +// session's prompts, tool calls, approvals and broadcasts between since +// and until (all of profile's events when session is ""). +func FromAudit(profile, session string, since, until time.Time) (timeline, hashes []string) { + d, err := config.Path("audit") + if err != nil { + return nil, nil + } + for day := since.UTC().Truncate(24 * time.Hour); !day.After(until); day = day.Add(24 * time.Hour) { + f, err := os.Open(filepath.Join(d, day.Format("2006-01-02")+".jsonl")) + if err != nil { + continue + } + sc := bufio.NewScanner(f) + sc.Buffer(make([]byte, 64<<10), 4<<20) + for sc.Scan() { + var e audit.Event + if json.Unmarshal(sc.Bytes(), &e) != nil || e.TS.Before(since) || e.TS.After(until) { + continue + } + if e.Profile != profile || (session != "" && e.Session != session) { + continue + } + if line := describe(e); line != "" { + timeline = append(timeline, e.TS.UTC().Format("15:04:05")+" "+line) + } + if e.Kind == audit.KindTx && e.Detail["stage"] == "broadcast" { + if h, _ := e.Detail["hash"].(string); h != "" { + hashes = append(hashes, fmt.Sprintf("%s (code %v)", h, e.Detail["code"])) + } + } + } + f.Close() + } + if len(timeline) > 60 { // keep the start and the end + timeline = append(append(timeline[:30:30], fmt.Sprintf("… %d events …", len(timeline)-50)), timeline[len(timeline)-20:]...) + } + return timeline, hashes +} + +func describe(e audit.Event) string { + d := e.Detail + str := func(k string) string { s, _ := d[k].(string); return clip(s, 120) } + switch e.Kind { + case audit.KindPrompt: + return "operator: " + str("text") + case audit.KindTool: + line := "tool " + str("name") + if err := str("error"); err != "" { + line += " — error: " + err + } + return line + case audit.KindShell: + return "shell: " + str("cmd") + case audit.KindApproval: + verdict := "declined" + if g, _ := d["granted"].(bool); g { + verdict = "approved" + } + return verdict + ": " + strings.SplitN(str("prompt"), "\n", 2)[0] + case audit.KindTx: + if d["stage"] == "broadcast" { + return fmt.Sprintf("tx broadcast %v (code %v)", d["hash"], d["code"]) + } + } + return "" +} + +func clip(s string, n int) string { + s = strings.Join(strings.Fields(s), " ") + if len(s) > n { + return s[:n-1] + "…" + } + return s +} diff --git a/internal/incidents/incidents_test.go b/internal/incidents/incidents_test.go new file mode 100644 index 0000000..96d806b --- /dev/null +++ b/internal/incidents/incidents_test.go @@ -0,0 +1,65 @@ +package incidents + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/abhijitkrm/cometcli/internal/audit" +) + +func TestSaveListGetAndAuditTimeline(t *testing.T) { + home := t.TempDir() + t.Setenv("COMETCLI_HOME", home) + now := time.Now().UTC() + // an audit trail: one session for val01, noise from another session and profile + ad := filepath.Join(home, "audit") + _ = os.MkdirAll(ad, 0o700) + var lines []string + add := func(e audit.Event) { b, _ := json.Marshal(e); lines = append(lines, string(b)) } + add(audit.Event{TS: now.Add(-20 * time.Minute), Kind: audit.KindPrompt, Profile: "val01", Session: "s1", Detail: map[string]any{"text": "is anyone jailed?"}}) + add(audit.Event{TS: now.Add(-19 * time.Minute), Kind: audit.KindTool, Profile: "val01", Session: "s1", Detail: map[string]any{"name": "node.triage"}}) + add(audit.Event{TS: now.Add(-10 * time.Minute), Kind: audit.KindApproval, Profile: "val01", Session: "s1", Detail: map[string]any{"prompt": "broadcast transaction\n{…}", "granted": true}}) + add(audit.Event{TS: now.Add(-9 * time.Minute), Kind: audit.KindTx, Profile: "val01", Session: "s1", Detail: map[string]any{"stage": "broadcast", "hash": "ABC123", "code": 0}}) + add(audit.Event{TS: now.Add(-9 * time.Minute), Kind: audit.KindTool, Profile: "val02", Session: "s1", Detail: map[string]any{"name": "other profile"}}) + add(audit.Event{TS: now.Add(-8 * time.Minute), Kind: audit.KindTool, Profile: "val01", Session: "s2", Detail: map[string]any{"name": "other session"}}) + _ = os.WriteFile(filepath.Join(ad, now.Format("2006-01-02")+".jsonl"), []byte(strings.Join(lines, "\n")+"\n"), 0o600) + + tl, hashes := FromAudit("val01", "s1", now.Add(-time.Hour), now) + joined := strings.Join(tl, "\n") + if len(tl) != 4 || !strings.Contains(joined, "operator: is anyone jailed?") || !strings.Contains(joined, "approved: broadcast transaction") || + strings.Contains(joined, "other") { + t.Fatalf("timeline:\n%s", joined) + } + if len(hashes) != 1 || !strings.HasPrefix(hashes[0], "ABC123") { + t.Fatalf("hashes %v", hashes) + } + + inc := &Incident{Profile: "val01", Title: "Jailed for downtime", Case: "val-jailed-downtime", + Started: now.Add(-20 * time.Minute), Resolved: now, RootCause: "container stopped", Outcome: "resolved", + Actions: []string{"restarted", "unjailed"}, Timeline: tl, TxHashes: hashes} + md, err := Save(inc) + if err != nil { + t.Fatal(err) + } + raw, _ := os.ReadFile(md) + for _, want := range []string{"# Jailed for downtime", "`val-jailed-downtime`", "## Root cause", "container stopped", "`ABC123 (code 0)`", "## Timeline"} { + if !strings.Contains(string(raw), want) { + t.Errorf("postmortem lacks %q:\n%s", want, raw) + } + } + list, _ := List("val01", now.Add(-24*time.Hour)) + if len(list) != 1 || list[0].ID != inc.ID { + t.Fatalf("list %+v", list) + } + got, err := Get("val01", inc.ID) + if err != nil || got.RootCause != "container stopped" { + t.Fatalf("get %+v %v", got, err) + } + if _, err := Get("val01", "../x"); err == nil { + t.Fatal("path traversal accepted") + } +} diff --git a/internal/tools/triage/collect.go b/internal/tools/triage/collect.go index 9ddfff7..114496f 100644 --- a/internal/tools/triage/collect.go +++ b/internal/tools/triage/collect.go @@ -22,6 +22,7 @@ import ( "github.com/abhijitkrm/cometcli/internal/client/host" "github.com/abhijitkrm/cometcli/internal/config" + "github.com/abhijitkrm/cometcli/internal/incidents" "github.com/abhijitkrm/cometcli/internal/kb" "github.com/abhijitkrm/cometcli/internal/keys" "github.com/abhijitkrm/cometcli/internal/logscan" @@ -41,6 +42,12 @@ type Report struct { Samples map[string][]string Since time.Duration // log window + // Deltas are meaningful changes since the previous recorded sweep. + Deltas []string + DeltaSince time.Time + // Past are this node's recorded incidents (30 days, newest first). + Past []incidents.Incident + jailLines []string } diff --git a/internal/tools/triage/memory.go b/internal/tools/triage/memory.go new file mode 100644 index 0000000..4f00196 --- /dev/null +++ b/internal/tools/triage/memory.go @@ -0,0 +1,365 @@ +package triage + +import ( + "fmt" + "strings" + "time" + + "github.com/abhijitkrm/cometcli/internal/history" + "github.com/abhijitkrm/cometcli/internal/incidents" + "github.com/abhijitkrm/cometcli/internal/toolkit" +) + +// remember stores this sweep in the signal history and fills the report +// with what changed since the previous one and the node's past incidents. +func remember(c *toolkit.Context, r *Report) { + if c.Profile == nil || c.Profile.Name == "" { + return + } + name := c.Profile.Name + now := time.Now() + if prev := history.Last(name, now); prev != nil { + r.Deltas = history.Deltas(prev.Signals, r.Signals) + r.DeltaSince = prev.TS + } + _ = history.Append(name, now, r.Signals) + r.Past, _ = incidents.List(name, now.Add(-30*24*time.Hour)) +} + +// pastLine summarizes earlier incidents on this node for the triage text. +func pastLine(past []incidents.Incident, hits map[string]bool) string { + if len(past) == 0 { + return "" + } + last := past[0] + line := fmt.Sprintf("past incidents on this node (30d): %d — last %s: %s", len(past), + last.Started.UTC().Format("2006-01-02"), clip(last.Title, 80)) + if last.RootCause != "" { + line += " (cause: " + clip(last.RootCause, 80) + ")" + } + var repeats []string + seen := map[string]int{} + for _, p := range past { + if p.Case != "" && hits[p.Case] { + seen[p.Case]++ + } + } + for id, n := range seen { + repeats = append(repeats, fmt.Sprintf("%s ×%d", id, n)) + } + if len(repeats) > 0 { + line += "\nrecurring: " + strings.Join(repeats, ", ") + " — check what the last fix missed (incident.show)" + } + return line +} + +// --- node.history --------------------------------------------------------------- + +// History is node.history: how signals moved over time. +type History struct{} + +func (History) Name() string { return "node.history" } +func (History) Desc() string { + return "How node signals changed over time, from recorded triage sweeps: first→last, min/max, when it changed, a trend line. " + + "Answers 'since when?' and 'is it getting worse?' (e.g. signals=val.missed,node.peers since=24h). Records come from every " + + "node.triage run and from mon.record." +} +func (History) Schema() map[string]any { + return toolkit.ObjSchema(map[string]any{ + "signals": toolkit.Str("comma-separated signal names or prefixes (val., host.disk_used_pct) — default: the key health signals"), + "since": toolkit.Str("window, e.g. 6h, 24h, 7d (default 24h)"), + }) +} +func (History) Tier() toolkit.Tier { return toolkit.TierObserve } + +var defaultHistory = []string{"node.peers", "node.catching_up", "node.reachable", "val.jailed", "val.missed", + "val.signed_recent_pct", "host.disk_used_pct", "host.mem_used_pct", "proc.restarts", "proc.running"} + +func (History) Run(c *toolkit.Context, a toolkit.Args) (*toolkit.Result, error) { + if c.Profile == nil { + return nil, fmt.Errorf("no node profile — use_node first") + } + window, err := parseWindow(a.String("since", "24h")) + if err != nil { + return nil, err + } + recs, err := history.Load(c.Profile.Name, time.Now().Add(-window)) + if err != nil { + return nil, err + } + if len(recs) == 0 { + return &toolkit.Result{Text: fmt.Sprintf("no history for %s in the last %s — run node.triage (each run is recorded) or `cometcli mon record` for regular samples", c.Profile.Name, window)}, nil + } + names := defaultHistory + if s := strings.TrimSpace(a.String("signals", "")); s != "" { + names = expand(strings.Split(s, ","), recs) + } + var b strings.Builder + fmt.Fprintf(&b, "%s: %d samples, %s → %s\n", c.Profile.Name, len(recs), + recs[0].TS.Local().Format("Jan 2 15:04"), recs[len(recs)-1].TS.Local().Format("Jan 2 15:04")) + data := map[string]any{} + for _, n := range names { + s := history.Summarize(recs, n) + if s.Samples == 0 { + continue + } + data[n] = map[string]any{"first": s.First, "last": s.Last, "changes": len(s.Changes)} + if s.Numeric { + fmt.Fprintf(&b, "%s: %v → %v (min %v, max %v) %s", n, fmtVal(s.First), fmtVal(s.Last), fmtVal(s.Min), fmtVal(s.Max), s.Spark) + } else { + fmt.Fprintf(&b, "%s: %v → %v", n, fmtVal(s.First), fmtVal(s.Last)) + } + if k := len(s.Changes); k > 0 { + ch := s.Changes + if k > 4 { + ch = ch[k-4:] + } + var parts []string + for _, x := range ch { + parts = append(parts, fmt.Sprintf("%s %v→%v", x.TS.Local().Format("Jan 2 15:04"), fmtVal(x.From), fmtVal(x.To))) + } + fmt.Fprintf(&b, " · changed %d× (last: %s)", k, strings.Join(parts, "; ")) + } + b.WriteString("\n") + } + return &toolkit.Result{Text: strings.TrimRight(b.String(), "\n"), Data: data}, nil +} + +// expand turns prefixes ("val.") into the signal names present in recs. +func expand(want []string, recs []history.Record) []string { + have := map[string]bool{} + for _, r := range recs { + for k := range r.Signals { + have[k] = true + } + } + var out []string + seen := map[string]bool{} + for _, w := range want { + w = strings.TrimSpace(w) + if w == "" { + continue + } + if have[w] && !seen[w] { + out, seen[w] = append(out, w), true + continue + } + for _, k := range sortedKeys(toAny(have)) { + if strings.HasPrefix(k, w) && !seen[k] { + out, seen[k] = append(out, k), true + } + } + } + return out +} + +func toAny(m map[string]bool) map[string]any { + out := make(map[string]any, len(m)) + for k, v := range m { + out[k] = v + } + return out +} + +func parseWindow(s string) (time.Duration, error) { + s = strings.TrimSpace(s) + if strings.HasSuffix(s, "d") { + var n int + if _, err := fmt.Sscanf(s, "%dd", &n); err == nil && n > 0 { + return time.Duration(n) * 24 * time.Hour, nil + } + } + d, err := time.ParseDuration(s) + if err != nil || d <= 0 { + return 0, fmt.Errorf("since %q: want a window like 6h, 24h or 7d", s) + } + return d, nil +} + +// --- mon.record ----------------------------------------------------------------- + +// Record is mon.record: a sampler that keeps the signal history fresh. +type Record struct{} + +func (Record) Name() string { return "mon.record" } +func (Record) Desc() string { + return "Record a triage sample every interval into the signal history (for node.history trends); runs until stopped" +} +func (Record) Schema() map[string]any { + return toolkit.ObjSchema(map[string]any{ + "interval": toolkit.Str("time between samples, e.g. 5m (default 5m, minimum 30s)"), + }) +} +func (Record) Tier() toolkit.Tier { return toolkit.TierObserve } +func (Record) LongRunning() bool { return true } + +func (Record) Run(c *toolkit.Context, a toolkit.Args) (*toolkit.Result, error) { + if c.Profile == nil { + return nil, fmt.Errorf("no node profile") + } + iv, err := time.ParseDuration(a.String("interval", "5m")) + if err != nil || iv < 30*time.Second { + return nil, fmt.Errorf("interval: want a duration ≥ 30s, e.g. 5m") + } + n := 0 + for { + r := Collect(c, iv) + if err := history.Append(c.Profile.Name, time.Now(), r.Signals); err != nil { + return nil, err + } + n++ + if c.Out != nil { + fmt.Fprintf(c.Out, "%s recorded sample %d for %s (%d signals)\n", time.Now().Format("15:04:05"), n, c.Profile.Name, len(r.Signals)) + } + select { + case <-c.Done(): + return &toolkit.Result{Text: fmt.Sprintf("recorded %d samples", n)}, nil + case <-time.After(iv): + } + } +} + +// --- incidents ------------------------------------------------------------------ + +// RecordIncident is incident.record. +type RecordIncident struct{} + +func (RecordIncident) Name() string { return "incident.record" } +func (RecordIncident) Desc() string { + return "Save a finished incident to this node's record (a JSON entry and a Markdown postmortem): title, root cause, case id, " + + "evidence, actions, outcome. The timeline and tx hashes are filled in from the audit log. Call it at the end of every incident." +} +func (RecordIncident) Schema() map[string]any { + return toolkit.ObjSchema(map[string]any{ + "title": toolkit.Str("one line: what went wrong"), + "root_cause": toolkit.Str("the cause, with the decisive evidence"), + "case": toolkit.Str("knowledge-base case id, if one matched"), + "evidence": toolkit.Str("key evidence, one item per line"), + "actions": toolkit.Str("what was done, one item per line"), + "outcome": toolkit.Str("resolved | mitigated | unresolved — and the final state"), + "started": toolkit.Str("when the incident started (RFC3339) or how long ago (e.g. 45m); default: this session's start"), + }, "title", "root_cause", "outcome") +} +func (RecordIncident) Tier() toolkit.Tier { return toolkit.TierDiagnose } // writes only cometcli's own records + +func (RecordIncident) Run(c *toolkit.Context, a toolkit.Args) (*toolkit.Result, error) { + if c.Profile == nil { + return nil, fmt.Errorf("no node profile — use_node first") + } + now := time.Now() + session := "" + if c.Audit != nil { + session = c.Audit.Session() + } + started := now.Add(-time.Hour) + if s := strings.TrimSpace(a.String("started", "")); s != "" { + if t, err := time.Parse(time.RFC3339, s); err == nil { + started = t + } else if d, err := parseWindow(s); err == nil { + started = now.Add(-d) + } else { + return nil, fmt.Errorf("started %q: want RFC3339 or a duration like 45m", s) + } + } else if parts := strings.SplitN(session, "-", 3); len(parts) == 3 { + // session ids start with their UTC start time; a long or resumed + // session must not stretch the incident back for days + if t, err := time.Parse("20060102-150405", parts[0]+"-"+parts[1]); err == nil { + started = t + if now.Sub(started) > 6*time.Hour { + started = now.Add(-6 * time.Hour) + } + } + } + inc := &incidents.Incident{ + Profile: c.Profile.Name, Title: a.String("title", ""), Case: a.String("case", ""), + Started: started, Resolved: now, RootCause: a.String("root_cause", ""), + Evidence: lines(a.String("evidence", "")), Actions: lines(a.String("actions", "")), + Outcome: a.String("outcome", ""), Session: session, + } + inc.Timeline, inc.TxHashes = incidents.FromAudit(c.Profile.Name, session, started.Add(-time.Minute), now) + path, err := incidents.Save(inc) + if err != nil { + return nil, err + } + return &toolkit.Result{Text: fmt.Sprintf("recorded incident %s → %s (%d timeline events, %d txs)", inc.ID, path, len(inc.Timeline), len(inc.TxHashes)), + Data: map[string]any{"id": inc.ID, "path": path}}, nil +} + +func lines(s string) []string { + var out []string + for _, l := range strings.Split(s, "\n") { + l = strings.TrimSpace(strings.TrimLeft(strings.TrimSpace(l), "-*• ")) + if l != "" { + out = append(out, l) + } + } + return out +} + +// ListIncidents is incident.list. +type ListIncidents struct{} + +func (ListIncidents) Name() string { return "incident.list" } +func (ListIncidents) Desc() string { + return "List past incidents recorded on this node (newest first): when, case, title, outcome" +} +func (ListIncidents) Schema() map[string]any { + return toolkit.ObjSchema(map[string]any{ + "since": toolkit.Str("window, e.g. 7d, 90d (default 30d)"), + }) +} +func (ListIncidents) Tier() toolkit.Tier { return toolkit.TierObserve } + +func (ListIncidents) Run(c *toolkit.Context, a toolkit.Args) (*toolkit.Result, error) { + if c.Profile == nil { + return nil, fmt.Errorf("no node profile — use_node first") + } + window, err := parseWindow(a.String("since", "30d")) + if err != nil { + return nil, err + } + past, err := incidents.List(c.Profile.Name, time.Now().Add(-window)) + if err != nil { + return nil, err + } + if len(past) == 0 { + return &toolkit.Result{Text: "no incidents recorded for " + c.Profile.Name + " in the last " + a.String("since", "30d")}, nil + } + var b strings.Builder + var ids []string + for _, p := range past { + fmt.Fprintf(&b, "%s %s %s — %s\n", p.ID, orNone(p.Case), clip(p.Title, 70), clip(p.Outcome, 40)) + ids = append(ids, p.ID) + } + return &toolkit.Result{Text: strings.TrimRight(b.String(), "\n"), Data: map[string]any{"incidents": ids}}, nil +} + +// ShowIncident is incident.show. +type ShowIncident struct{} + +func (ShowIncident) Name() string { return "incident.show" } +func (ShowIncident) Desc() string { + return "Show a recorded incident's postmortem: root cause, evidence, actions, transactions, timeline" +} +func (ShowIncident) Schema() map[string]any { + return toolkit.ObjSchema(map[string]any{"id": toolkit.Str("incident id from incident.list or triage")}, "id") +} +func (ShowIncident) Tier() toolkit.Tier { return toolkit.TierObserve } + +func (ShowIncident) Run(c *toolkit.Context, a toolkit.Args) (*toolkit.Result, error) { + if c.Profile == nil { + return nil, fmt.Errorf("no node profile — use_node first") + } + inc, err := incidents.Get(c.Profile.Name, a.String("id", "")) + if err != nil { + return nil, err + } + return &toolkit.Result{Text: incidents.Markdown(inc), Data: map[string]any{"id": inc.ID}}, nil +} + +func orNone(s string) string { + if s == "" { + return "-" + } + return s +} diff --git a/internal/tools/triage/tool.go b/internal/tools/triage/tool.go index 2f21c42..5f23810 100644 --- a/internal/tools/triage/tool.go +++ b/internal/tools/triage/tool.go @@ -19,6 +19,11 @@ func Register(r *toolkit.Registry) { r.Register(Search{}) r.Register(Show{}) r.Register(Add{}) + r.Register(History{}) + r.Register(Record{}) + r.Register(RecordIncident{}) + r.Register(ListIncidents{}) + r.Register(ShowIncident{}) } // LoadKB loads the built-in cases plus the user's and the project's. @@ -58,6 +63,7 @@ func (Triage) Run(c *toolkit.Context, a toolkit.Args) (*toolkit.Result, error) { since = d } r := Collect(c, since) + remember(c, r) base := LoadKB(c) hits := base.Match(r.Signals, r.Chain) return &toolkit.Result{Text: Render(r, hits, a.String("signals", "")), Data: map[string]any{ @@ -123,6 +129,14 @@ func Render(r *Report, hits []kb.Hit, prefix string) string { } groups[g] = append(groups[g], name+"="+fmtVal(v)) } + if len(r.Deltas) > 0 { + ago := time.Since(r.DeltaSince).Round(time.Minute) + d := r.Deltas + if len(d) > 12 { + d = append(d[:12:12], fmt.Sprintf("… %d more", len(r.Deltas)-12)) + } + fmt.Fprintf(&b, "changed since the last check (%s ago): %s\n", ago, strings.Join(d, ", ")) + } b.WriteString("signals:\n") for _, g := range order { fmt.Fprintf(&b, " %s: %s\n", g, strings.Join(groups[g], " ")) @@ -152,6 +166,13 @@ func Render(r *Report, hits []kb.Hit, prefix string) string { incidents = append(incidents, h) } } + matched := map[string]bool{} + for _, h := range hits { + matched[h.Case.ID] = true + } + if pl := pastLine(r.Past, matched); pl != "" { + b.WriteString(pl + "\n") + } if len(incidents) == 0 && len(symptoms) == 0 { b.WriteString("matched cases: none — no known failure pattern. If something is still wrong, investigate from the signals, then record what you learn with kb.add.") if len(advisories) > 0 { diff --git a/internal/tools/triage/triage_test.go b/internal/tools/triage/triage_test.go index dbfdcb0..2ded75f 100644 --- a/internal/tools/triage/triage_test.go +++ b/internal/tools/triage/triage_test.go @@ -217,3 +217,38 @@ func TestKBTools(t *testing.T) { t.Fatal("path-traversal id accepted") } } + +func TestTriageRemembersChangesIncidentsAndHistory(t *testing.T) { + c, chain := setup(t, "evmd", "db_backend = \"goleveldb\"\n", "minimum-gas-prices = \"1adex\"\n") + if _, err := (Triage{}).Run(c, toolkit.Args{}); err != nil { + t.Fatal(err) + } + chain.Set(func(c *fakenode.Comet) { c.Peers = 0 }) // something breaks between sweeps + res, err := (Triage{}).Run(c, toolkit.Args{}) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(res.Text, "changed since the last check") || !strings.Contains(res.Text, "node.peers 3→0") { + t.Fatalf("no deltas:\n%s", res.Text) + } + // the agent records the incident when it's done + rec, err := (RecordIncident{}).Run(c, toolkit.Args{"title": "Peers lost after firewall change", "case": "node-no-peers", + "root_cause": "ufw blocked 26656", "actions": "- reopened 26656\n- restarted", "outcome": "resolved"}) + if err != nil || !strings.Contains(rec.Text, "recorded incident") { + t.Fatalf("%v %v", rec, err) + } + res, _ = (Triage{}).Run(c, toolkit.Args{}) + if !strings.Contains(res.Text, "past incidents on this node (30d): 1") || !strings.Contains(res.Text, "recurring: node-no-peers ×1") { + t.Fatalf("past incident not surfaced:\n%s", res.Text) + } + list, _ := (ListIncidents{}).Run(c, toolkit.Args{}) + id := list.Data["incidents"].([]string)[0] + show, _ := (ShowIncident{}).Run(c, toolkit.Args{"id": id}) + if !strings.Contains(show.Text, "ufw blocked 26656") || !strings.Contains(show.Text, "- reopened 26656") { + t.Fatalf("show:\n%s", show.Text) + } + h, err := (History{}).Run(c, toolkit.Args{"signals": "node.peers,val.", "since": "1h"}) + if err != nil || !strings.Contains(h.Text, "node.peers: 3 → 0") || !strings.Contains(h.Text, "val.jailed") || !strings.Contains(h.Text, "3 samples") { + t.Fatalf("history:\n%s %v", h.Text, err) + } +} From 676cadde37ab1e4936072f39496ce5b742b06d8b Mon Sep 17 00:00:00 2001 From: Abhijit Date: Thu, 8 Oct 2026 18:03:43 +0530 Subject: [PATCH 2/7] /incident and /recover-jail work from general mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit '/incident val01 is down' answered 'needs node mode — /one first'. A node-only command now switches to the profile its arguments name; with none (or several) named, it runs and the model picks the profile with use_node. It only errors when there are no profiles. The welcome tip and /one hint say you can just name the node. Co-Authored-By: Claude Opus 5.5 --- docs/COMMANDS.md | 7 +++--- internal/agent/commands.go | 46 +++++++++++++++++++++++++++++++--- internal/agent/usenode_test.go | 33 ++++++++++++++++++++++++ internal/tui/chatui/chatui.go | 2 +- 4 files changed, 81 insertions(+), 7 deletions(-) diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index 128c954..9be4a28 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -99,7 +99,8 @@ down is reported, never fatal. Signals are flat names (`val.jailed`, `host.disk_used_pct`, `logs.apphash`, `node.key_mismatch`…). Each **case** says which signals point to it, how to confirm it, causes, fix steps tagged -`[read]` / `[change]` / `[tx]`, how to verify, and what never to do. In node mode, +`[read]` / `[change]` / `[tx]`, how to verify, and what never to do. In any session +(name the node — `/incident val01 is down` — and cometcli switches to it), **`/incident [what you see]`** runs the method: triage → confirm the top case → fix the root cause first (approvals as usual) → wait with `wait.until` → verify → re-triage → report. When the agent finds a cause no case covered, it offers to record one with @@ -434,8 +435,8 @@ saved after each turn to `~/.cometcli/sessions/` (0600, already redacted). | `/memory`, `/remember [--user\|--node] ` | memory files loaded; add a line to COMET.md | | `/mcp`, `/agents` | connected MCP servers; subagents for the task tool | | `/` | commands from `.cometcli/commands/*.md` — see [EXTENDING.md](EXTENDING.md) | -| `/incident [what you see]` | node mode: triage → known case → fix root cause → wait → verify → report | -| `/recover-jail [notes]` | node mode: run the full jail-recovery procedure | +| `/incident [what you see]` | triage → known case → fix root cause → wait → verify → record → report; name the node to switch to it | +| `/recover-jail [notes]` | the full jail-recovery procedure (name the node, or the agent picks it) | | `/one [profile\|off]` | switch the session to node mode for a profile, or back to general (conversation kept) | ### General tools and permissions diff --git a/internal/agent/commands.go b/internal/agent/commands.go index 37fa266..7c606ed 100644 --- a/internal/agent/commands.go +++ b/internal/agent/commands.go @@ -7,6 +7,7 @@ import ( "path/filepath" "sort" "strings" + "unicode" "github.com/abhijitkrm/cometcli/internal/config" "github.com/abhijitkrm/cometcli/internal/runbook" @@ -232,7 +233,7 @@ func RunCommand(a *Agent, c *toolkit.Context, reg *toolkit.Registry, line string return CmdResult{}, err } if len(f) == 1 { - txt := "general mode — shell and files on this machine; /one for node work" + txt := "general mode — shell and files on this machine; ask about a node by name and the agent switches to it (/one pins one)" if a.Node() { txt = "node mode: " + a.Ctx.Profile.Name + " — /one off for general mode" } @@ -468,14 +469,53 @@ func RunCommand(a *Agent, c *toolkit.Context, reg *toolkit.Registry, line string if err := needAgent(); err != nil { return CmdResult{}, err } + prompt, text := cmd.Expand(args, nil, nil), "/"+name if builtinNeedsNode[name] && !a.Node() { - return CmdResult{}, fmt.Errorf("/%s needs node mode — /one first", name) + p, err := a.nodeFor(args) + switch { + case err != nil: + return CmdResult{}, err + case p != nil: + // the operator named the node: work on it + a.SwitchScope(p) + text += " — working on " + p.Name + default: + // let the model pick the profile itself + prompt = "This is about one of the operator's nodes: call use_node with the right profile first.\n\n" + prompt + } } - return CmdResult{Text: "/" + name, Prompt: cmd.Expand(args, nil, nil)}, nil + return CmdResult{Text: text, Prompt: prompt}, nil } return CmdResult{}, ErrUnknownCommand } +// nodeFor finds the profile a node-only command is about: one named in +// its arguments, or nil when none is (the model then picks with use_node). +// It errors only when there are no profiles at all. +func (a *Agent) nodeFor(args string) (*config.Profile, error) { + if a.Ctx == nil || a.Ctx.Cfg == nil || len(a.Ctx.Cfg.Profiles) == 0 { + return nil, fmt.Errorf("no node profiles yet — add one with `cometcli init` or `cometcli profile add`") + } + words := map[string]bool{} + for _, w := range strings.FieldsFunc(strings.ToLower(args), func(r rune) bool { + return !(r == '-' || r == '_' || r == '.' || unicode.IsLetter(r) || unicode.IsDigit(r)) + }) { + words[strings.Trim(w, ".")] = true + } + var names []string + for n := range a.Ctx.Cfg.Profiles { + if words[strings.ToLower(n)] { + names = append(names, n) + } + } + if len(names) != 1 { // none, or ambiguous: the model decides + return nil, nil + } + p := a.Ctx.Cfg.Profiles[names[0]] + p.Name = names[0] + return p, nil +} + // memoryTarget picks the file /remember writes: --user, --node, or the // project's COMET.md (the user file without a project). func (a *Agent) memoryTarget(rest *string) (string, error) { diff --git a/internal/agent/usenode_test.go b/internal/agent/usenode_test.go index a058259..43a444a 100644 --- a/internal/agent/usenode_test.go +++ b/internal/agent/usenode_test.go @@ -66,3 +66,36 @@ func TestModelSwitchesToNodeItself(t *testing.T) { t.Fatal("unknown profile accepted") } } + +func TestNodeOnlyCommandsInGeneralMode(t *testing.T) { + a := newTestAgent(t, &mockProvider{}) + a.Ctx.Profile = nil + a.Ctx.Cfg = &config.Config{Profiles: map[string]*config.Profile{ + "val01": {ChainID: "mychain-1", Role: "validator"}, + "val02": {ChainID: "mychain-1", Role: "validator"}, + }} + // names a node: switch to it + res, err := RunCommand(a, a.Ctx, a.Reg, "/incident val01 is down") + if err != nil || !a.Node() || a.Ctx.Profile.Name != "val01" || !strings.Contains(res.Text, "working on val01") { + t.Fatalf("named node: %+v %v node=%v", res, err, a.Node()) + } + if !strings.Contains(res.Prompt, "node.triage") || !strings.Contains(res.Prompt, "val01 is down") { + t.Fatalf("procedure prompt: %q", res.Prompt) + } + // no node named: the model picks with use_node + a.SwitchScope(nil) + res, err = RunCommand(a, a.Ctx, a.Reg, "/incident blocks stopped") + if err != nil || a.Node() || !strings.Contains(res.Prompt, "call use_node") { + t.Fatalf("unnamed node: %+v %v", res, err) + } + // both named: ambiguous, the model decides + res, _ = RunCommand(a, a.Ctx, a.Reg, "/incident val01 and val02 disagree") + if a.Node() || !strings.Contains(res.Prompt, "call use_node") { + t.Fatal("ambiguous names switched anyway") + } + // no profiles at all: explain how to add one + a.Ctx.Cfg = &config.Config{} + if _, err := RunCommand(a, a.Ctx, a.Reg, "/incident val01 is down"); err == nil || !strings.Contains(err.Error(), "cometcli init") { + t.Fatalf("no profiles: %v", err) + } +} diff --git a/internal/tui/chatui/chatui.go b/internal/tui/chatui/chatui.go index 6add97c..7327fe2 100644 --- a/internal/tui/chatui/chatui.go +++ b/internal/tui/chatui/chatui.go @@ -1236,7 +1236,7 @@ func (m *model) banner() string { } tips := []string{ `Ask about this machine, e.g. "why is the disk filling up?"`, - "Work on a validator: " + bold.Render("/one ") + " (or start with cometcli one )", + `Ask about a node by name, e.g. "is val01 healthy?" · ` + bold.Render("/incident ") + " works a problem end to end", bold.Render("!") + " runs a command in your terminal · " + bold.Render("#") + " saves to memory · " + bold.Render("@") + " attaches a file", } if m.a.Node() { From 8de1589e69f000df5ac60e74aad1231a47455103 Mon Sep 17 00:00:00 2001 From: Abhijit Date: Thu, 8 Oct 2026 18:06:32 +0530 Subject: [PATCH 3/7] fleet.triage: every node at once, compared Triages all profiles in parallel (recording each node's history) and compares them: a node behind the others on its chain, a chain-wide halt (every node stalled), version mismatches, two nodes running one consensus key (double-sign risk), validators sharing a host, and validators whose sentries are down (new profile field: sentries). Offered in general mode (tools can now declare FleetWide); the general prompt sends fleet questions to it. Co-Authored-By: Claude Opus 5.5 --- internal/agent/scope.go | 2 +- internal/agent/toolsearch.go | 4 +- internal/cli/profile.go | 4 + internal/config/config.go | 3 + internal/toolkit/toolkit.go | 12 ++ internal/tools/triage/fleet.go | 258 ++++++++++++++++++++++++++++ internal/tools/triage/fleet_test.go | 79 +++++++++ internal/tools/triage/tool.go | 1 + 8 files changed, 360 insertions(+), 3 deletions(-) create mode 100644 internal/tools/triage/fleet.go create mode 100644 internal/tools/triage/fleet_test.go diff --git a/internal/agent/scope.go b/internal/agent/scope.go index 26ce244..682347a 100644 --- a/internal/agent/scope.go +++ b/internal/agent/scope.go @@ -47,7 +47,7 @@ ENVIRONMENT: %s/%s, host %s, working directory %s, date %s if ap, err := c.Cfg.ActiveProfile(""); err == nil && ap != nil { active = " Active profile: " + ap.Name + "." } - fmt.Fprintf(&b, "NODE PROFILES: %s.%s For any question about a node, validator or chain (jailing, sync, peers, upgrades, governance, txs), call use_node with the right profile yourself and continue — never ask the operator to switch. Any profile on a chain answers chain-wide questions; pick the active one when nothing points elsewhere.\n", strings.Join(names, "; "), active) + fmt.Fprintf(&b, "NODE PROFILES: %s.%s For any question about a node, validator or chain (jailing, sync, peers, upgrades, governance, txs), call use_node with the right profile yourself and continue — never ask the operator to switch. Any profile on a chain answers chain-wide questions; pick the active one when nothing points elsewhere. For questions about several nodes or the whole network (\"how is my fleet?\"), start with fleet.triage.\n", strings.Join(names, "; "), active) } return b.String() } diff --git a/internal/agent/toolsearch.go b/internal/agent/toolsearch.go index 45deaf3..6950296 100644 --- a/internal/agent/toolsearch.go +++ b/internal/agent/toolsearch.go @@ -22,7 +22,7 @@ var coreTools = map[string]bool{ "val.status": true, "val.signing": true, "val.jail-check": true, "val.consensus": true, "wait.until": true, "chain.validators": true, "node.triage": true, "kb.show": true, "kb.search": true, - "node.history": true, "incident.record": true, + "node.history": true, "incident.record": true, "fleet.triage": true, } var toolSearchDef = ToolDef{ @@ -67,7 +67,7 @@ func (a *Agent) advertised(t toolkit.Tool) bool { return true } if !a.Node() { - return false // node tools need a node: /one + return toolkit.IsFleetWide(t) // other node tools need a node (use_node) } if !a.deferred() || coreTools[t.Name()] { return true diff --git a/internal/cli/profile.go b/internal/cli/profile.go index 36310c9..c2e6355 100644 --- a/internal/cli/profile.go +++ b/internal/cli/profile.go @@ -130,6 +130,9 @@ func profileAddCmd() *cobra.Command { if fl.Changed("ssh-host") { np.Transport.Host = p.Transport.Host } + if fl.Changed("sentries") { + np.Sentries = p.Sentries + } if fl.Changed("ssh-user") { np.Transport.User = p.Transport.User } @@ -205,6 +208,7 @@ func profileAddCmd() *cobra.Command { f.StringVar(&p.Transport.KeyFile, "ssh-key", "", "ssh private key path") f.StringVar(&p.Service.Type, "service", "", "systemd | docker | launchd | none") f.StringVar(&p.Service.Unit, "unit", "", "service unit/container name") + f.StringSliceVar(&p.Sentries, "sentries", nil, "profiles of the sentry nodes this validator peers through (comma-separated)") f.StringVar(&signerKey, "signer", "", "ops key name in keyring") f.StringVar(&p.Signer.Backend, "signer-backend", "", "os | file | test") f.StringVar(&feeDenom, "fee-denom", "", "fee denom (e.g. atest)") diff --git a/internal/config/config.go b/internal/config/config.go index f2ce061..73e5d8c 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -127,6 +127,9 @@ type Profile struct { Agent AgentConf `yaml:"agent"` Alerts Alerts `yaml:"alerts"` Metadata map[string]string `yaml:"metadata,omitempty"` + // Sentries names the profiles of the sentry nodes this validator + // peers through (fleet triage checks it isn't cut off). + Sentries []string `yaml:"sentries,omitempty"` } // IsValidator reports whether this profile is a consensus-signing validator. diff --git a/internal/toolkit/toolkit.go b/internal/toolkit/toolkit.go index 40ffead..41966e1 100644 --- a/internal/toolkit/toolkit.go +++ b/internal/toolkit/toolkit.go @@ -334,3 +334,15 @@ func CallTimeout(t Tool, args Args) time.Duration { } return 90 * time.Second } + +// FleetWide is implemented by tools that work across all profiles and +// need no active node — they're offered in general mode too. +type FleetWide interface { + FleetWide() bool +} + +// IsFleetWide reports whether t is fleet-wide. +func IsFleetWide(t Tool) bool { + f, ok := t.(FleetWide) + return ok && f.FleetWide() +} diff --git a/internal/tools/triage/fleet.go b/internal/tools/triage/fleet.go new file mode 100644 index 0000000..aeb2747 --- /dev/null +++ b/internal/tools/triage/fleet.go @@ -0,0 +1,258 @@ +package triage + +import ( + "fmt" + "sort" + "strings" + "sync" + "time" + + "github.com/abhijitkrm/cometcli/internal/config" + "github.com/abhijitkrm/cometcli/internal/kb" + "github.com/abhijitkrm/cometcli/internal/toolkit" +) + +// FleetTriage is fleet.triage: triage every profile at once and compare +// them — what a single node's view can't show. +type FleetTriage struct{} + +func (FleetTriage) Name() string { return "fleet.triage" } +func (FleetTriage) Desc() string { + return "Triage every node at once and compare them: one row per node (sync lag, peers, signing, jail, top case) plus " + + "fleet findings — a node behind the others, a chain-wide halt, version mismatches, two nodes on one consensus key, " + + "validators sharing a host, sentries down. Use for 'how is my network/fleet?' or before picking a node to work on." +} +func (FleetTriage) Schema() map[string]any { + return toolkit.ObjSchema(map[string]any{ + "profiles": toolkit.Str("comma-separated profiles (default: all)"), + "chain": toolkit.Str("only profiles on this chain id"), + }) +} +func (FleetTriage) Tier() toolkit.Tier { return toolkit.TierDiagnose } +func (FleetTriage) FleetWide() bool { return true } +func (FleetTriage) Timeout(toolkit.Args) time.Duration { return 4 * time.Minute } + +// NodeReport is one node's part of a fleet sweep. +type NodeReport struct { + Profile *config.Profile + Report *Report + Hits []kb.Hit +} + +// FleetConcurrency bounds parallel node sweeps. +var FleetConcurrency = 4 + +// CollectFleet triages profiles in parallel, recording each node's history. +func CollectFleet(c *toolkit.Context, profiles []*config.Profile, since time.Duration) []NodeReport { + out := make([]NodeReport, len(profiles)) + sem := make(chan struct{}, FleetConcurrency) + var wg sync.WaitGroup + for i, p := range profiles { + wg.Add(1) + go func(i int, p *config.Profile) { + defer wg.Done() + sem <- struct{}{} + defer func() { <-sem }() + pc := &toolkit.Context{Context: c.Context, Profile: p, Cfg: c.Cfg, Audit: c.Audit, WorkRoot: c.WorkRoot} + defer pc.Close() + r := Collect(pc, since) + remember(pc, r) + out[i] = NodeReport{Profile: p, Report: r, Hits: LoadKB(pc).Match(r.Signals, r.Chain)} + }(i, p) + } + wg.Wait() + return out +} + +func (FleetTriage) Run(c *toolkit.Context, a toolkit.Args) (*toolkit.Result, error) { + if c.Cfg == nil || len(c.Cfg.Profiles) == 0 { + return nil, fmt.Errorf("no profiles configured — cometcli init") + } + want := map[string]bool{} + for _, n := range strings.Split(a.String("profiles", ""), ",") { + if n = strings.TrimSpace(n); n != "" { + want[n] = true + } + } + chain := a.String("chain", "") + var profiles []*config.Profile + var names []string + for n := range c.Cfg.Profiles { + names = append(names, n) + } + sort.Strings(names) + for _, n := range names { + p := c.Cfg.Profiles[n] + if (len(want) > 0 && !want[n]) || (chain != "" && p.ChainID != chain) { + continue + } + p.Name = n + profiles = append(profiles, p) + } + if len(profiles) == 0 { + return nil, fmt.Errorf("no profiles match") + } + reports := CollectFleet(c, profiles, 15*time.Minute) + findings := FleetFindings(reports) + return &toolkit.Result{Text: RenderFleet(reports, findings), Data: map[string]any{"findings": findings, "nodes": len(reports)}}, nil +} + +// FleetFindings compares nodes: per-chain lag and halts, versions, shared +// consensus keys, shared hosts, sentries. +func FleetFindings(nodes []NodeReport) []string { + var out []string + byChain := map[string][]NodeReport{} + byName := map[string]NodeReport{} + for _, n := range nodes { + byChain[n.Profile.ChainID] = append(byChain[n.Profile.ChainID], n) + byName[n.Profile.Name] = n + } + f := func(n NodeReport, k string) (float64, bool) { v, ok := n.Report.Signals[k].(float64); return v, ok } + var chains []string + for ch := range byChain { + chains = append(chains, ch) + } + sort.Strings(chains) + for _, ch := range chains { + group := byChain[ch] + var top float64 + reachable, stalled := 0, 0 + versions := map[string][]string{} + for _, n := range group { + if n.Report.Signals["node.reachable"] != true { + continue + } + reachable++ + if h, ok := f(n, "node.height"); ok && h > top { + top = h + } + if age, ok := f(n, "node.block_age_s"); ok && age > 60 { + stalled++ + } + if v, _ := n.Report.Signals["node.version"].(string); v != "" { + versions[v] = append(versions[v], n.Profile.Name) + } + } + if reachable >= 2 && stalled == reachable { + out = append(out, fmt.Sprintf("CHAIN HALT on %s: all %d reachable nodes are stalled near height %d — a chain-wide problem (upgrade height? consensus failure?), not one node", orNone(ch), reachable, int64(top))) + } + for _, n := range group { + if n.Report.Signals["node.reachable"] != true { + out = append(out, fmt.Sprintf("%s is unreachable", n.Profile.Name)) + continue + } + if h, ok := f(n, "node.height"); ok && top-h > 20 && stalled != reachable { + out = append(out, fmt.Sprintf("%s is %d blocks behind the other %s nodes", n.Profile.Name, int64(top-h), orNone(ch))) + } + } + if len(versions) > 1 { + var parts []string + for v, ns := range versions { + parts = append(parts, v+": "+strings.Join(ns, ",")) + } + sort.Strings(parts) + out = append(out, fmt.Sprintf("version mismatch on %s — %s", orNone(ch), strings.Join(parts, "; "))) + } + } + // one consensus key on two running nodes: double-sign risk + keys := map[string][]string{} + for _, n := range nodes { + vp, _ := n.Report.Signals["node.voting_power"].(float64) + if k, _ := n.Report.Signals["node.cons_addr_hex"].(string); k != "" && vp > 0 { + keys[k] = append(keys[k], n.Profile.Name) + } + } + for k, ns := range keys { + if len(ns) > 1 { + sort.Strings(ns) + out = append(out, fmt.Sprintf("DOUBLE-SIGN RISK: %s run the same consensus key (%s…) — stop all but one now", strings.Join(ns, " and "), k[:min(8, len(k))])) + } + } + // validators sharing one remote host + hosts := map[string][]string{} + for _, n := range nodes { + p := n.Profile + if p.IsValidator() && p.Transport.Type == "ssh" && p.Transport.Host != "" { + hosts[p.Transport.Host] = append(hosts[p.Transport.Host], p.Name) + } + } + for h, ns := range hosts { + if len(ns) > 1 { + sort.Strings(ns) + out = append(out, fmt.Sprintf("validators %s share host %s — one machine failure takes them all down", strings.Join(ns, ", "), h)) + } + } + // sentries: a validator whose sentries are all down is cut off + for _, n := range nodes { + if len(n.Profile.Sentries) == 0 { + continue + } + var down []string + known := 0 + for _, s := range n.Profile.Sentries { + sr, ok := byName[s] + if !ok { + continue + } + known++ + if sr.Report.Signals["node.reachable"] != true || sr.Report.Signals["node.peers"] == 0.0 { + down = append(down, s) + } + } + switch { + case known > 0 && len(down) == known: + out = append(out, fmt.Sprintf("%s: all its sentries are down (%s) — it's cut off from the network", n.Profile.Name, strings.Join(down, ", "))) + case len(down) > 0: + out = append(out, fmt.Sprintf("%s: sentry %s is down — one failure from being cut off", n.Profile.Name, strings.Join(down, ", "))) + } + } + return out +} + +// RenderFleet is the compact fleet report. +func RenderFleet(nodes []NodeReport, findings []string) string { + var b strings.Builder + fmt.Fprintf(&b, "fleet triage — %d nodes\n", len(nodes)) + fmt.Fprintf(&b, "%-14s %-9s %-10s %-6s %-8s %-7s %s\n", "node", "role", "height", "peers", "signing", "jailed", "top case") + for _, n := range nodes { + s := n.Report.Signals + val := func(k string) string { + v, ok := s[k] + if !ok { + return "-" + } + return fmtVal(v) + } + height := val("node.height") + if s["node.reachable"] != true { + height = "DOWN" + } else if s["node.catching_up"] == true { + height += "↻" + } else if s["node.stalled"] == true { + height += "!" + } + signing := val("val.signed_recent_pct") + if signing != "-" { + signing += "%" + } + top := "ok" + for _, h := range n.Hits { + if h.Case.Kind != "advisory" && h.SymptomOf == "" { + top = h.Case.ID + " [" + h.Case.Severity + "]" + break + } + } + fmt.Fprintf(&b, "%-14s %-9s %-10s %-6s %-8s %-7s %s\n", clip(n.Profile.Name, 14), clip(n.Profile.Role, 9), height, + val("node.peers"), signing, val("val.jailed"), top) + } + if len(findings) == 0 { + b.WriteString("fleet findings: none — nodes agree with each other") + } else { + b.WriteString("fleet findings:\n") + for _, f := range findings { + b.WriteString(" - " + f + "\n") + } + } + b.WriteString("\n→ use_node then node.triage / kb.show for a node's details") + return strings.TrimRight(b.String(), "\n") +} diff --git a/internal/tools/triage/fleet_test.go b/internal/tools/triage/fleet_test.go new file mode 100644 index 0000000..f139b39 --- /dev/null +++ b/internal/tools/triage/fleet_test.go @@ -0,0 +1,79 @@ +package triage + +import ( + "strings" + "testing" + + "github.com/abhijitkrm/cometcli/internal/config" + "github.com/abhijitkrm/cometcli/internal/kb" + "github.com/abhijitkrm/cometcli/internal/toolkit" +) + +func node(name, chain, role string, sig kb.Signals) NodeReport { + return NodeReport{Profile: &config.Profile{Name: name, ChainID: chain, Role: role}, Report: &Report{Signals: sig}} +} + +func up(h float64, extra ...any) kb.Signals { + s := kb.Signals{"node.reachable": true, "node.height": h, "node.block_age_s": 2.0, "node.version": "0.39.0", "node.peers": 5.0} + for i := 0; i+1 < len(extra); i += 2 { + s[extra[i].(string)] = extra[i+1] + } + return s +} + +func TestFleetFindings(t *testing.T) { + sentry := node("sentry1", "c-1", "sentry", kb.Signals{"node.reachable": false}) + val1 := node("val1", "c-1", "validator", up(1000, "node.cons_addr_hex", "AAAA1111", "node.voting_power", 10.0)) + val1.Profile.Sentries = []string{"sentry1"} + val2 := node("val2", "c-1", "validator", up(700, "node.cons_addr_hex", "AAAA1111", "node.voting_power", 10.0, "node.version", "0.38.0")) + val3 := node("val3", "c-1", "validator", up(1000)) + val2.Profile.Transport = config.Transport{Type: "ssh", Host: "10.0.0.5"} + val3.Profile.Transport = config.Transport{Type: "ssh", Host: "10.0.0.5"} + got := strings.Join(FleetFindings([]NodeReport{sentry, val1, val2, val3}), "\n") + for _, want := range []string{ + "sentry1 is unreachable", + "val2 is 300 blocks behind the other c-1 nodes", + "version mismatch on c-1", + "DOUBLE-SIGN RISK: val1 and val2 run the same consensus key", + "validators val2, val3 share host 10.0.0.5", + "val1: all its sentries are down (sentry1)", + } { + if !strings.Contains(got, want) { + t.Errorf("missing %q in:\n%s", want, got) + } + } + if strings.Contains(got, "CHAIN HALT") { + t.Error("halt reported while nodes advance") + } + // every reachable node stalled: the chain halted, nobody is "behind" + a := node("a", "c-2", "validator", up(3644, "node.block_age_s", 300.0)) + b := node("b", "c-2", "validator", up(3640, "node.block_age_s", 310.0)) + got = strings.Join(FleetFindings([]NodeReport{a, b}), "\n") + if !strings.Contains(got, "CHAIN HALT on c-2") || strings.Contains(got, "behind") { + t.Fatalf("halt:\n%s", got) + } + // healthy fleet: nothing to say + if f := FleetFindings([]NodeReport{node("x", "c-3", "validator", up(10)), node("y", "c-3", "validator", up(10))}); len(f) != 0 { + t.Fatalf("healthy fleet findings: %v", f) + } +} + +func TestFleetTriageRunsEveryProfile(t *testing.T) { + c, chain := setup(t, "evmd", "db_backend = \"goleveldb\"\n", "minimum-gas-prices = \"1adex\"\n") + _ = chain + p := c.Profile + other := *p + other.Name, other.Role = "rpc1", "rpc" + c.Cfg.Profiles = map[string]*config.Profile{p.Name: p, "rpc1": &other} + general := &toolkit.Context{Context: c.Context, Cfg: c.Cfg} // general mode: no active profile + res, err := (FleetTriage{}).Run(general, toolkit.Args{}) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(res.Text, "fleet triage — 2 nodes") || !strings.Contains(res.Text, "rpc1") || !strings.Contains(res.Text, p.Name) { + t.Fatalf("report:\n%s", res.Text) + } + if !toolkit.IsFleetWide(FleetTriage{}) { + t.Fatal("fleet.triage must be offered in general mode") + } +} diff --git a/internal/tools/triage/tool.go b/internal/tools/triage/tool.go index 5f23810..be4439f 100644 --- a/internal/tools/triage/tool.go +++ b/internal/tools/triage/tool.go @@ -20,6 +20,7 @@ func Register(r *toolkit.Registry) { r.Register(Show{}) r.Register(Add{}) r.Register(History{}) + r.Register(FleetTriage{}) r.Register(Record{}) r.Register(RecordIncident{}) r.Register(ListIncidents{}) From 55244fc927ef2d5a1f1a17771586e8cce7c8eb4e Mon Sep 17 00:00:00 2001 From: Abhijit Date: Thu, 8 Oct 2026 18:12:27 +0530 Subject: [PATCH 4/7] cometcli watch: autonomous fleet watcher with Telegram approvals Sweeps the fleet on an interval (fleet triage). A newly appearing critical/high case, or a halt / double-sign / cut-off fleet finding, becomes an incident: notify (report it), diagnose (default: the agent works it read-only and reports), or fix (the agent may act; every approval goes to Telegram with Approve/Deny buttons, unanswered within watch.approval_timeout = denied, presses only from the configured chat and users). Transactions are refused unless watch.allow_tx. Recoveries are announced; an issue isn't worked again within the cooldown; state survives restarts. Reports go to Slack, Discord and Telegram. Co-Authored-By: Claude Opus 5.5 --- internal/cli/root.go | 2 +- internal/cli/watch.go | 172 ++++++++++++++++++++ internal/config/config.go | 23 +++ internal/watch/telegram.go | 159 +++++++++++++++++++ internal/watch/watch.go | 296 +++++++++++++++++++++++++++++++++++ internal/watch/watch_test.go | 181 +++++++++++++++++++++ 6 files changed, 832 insertions(+), 1 deletion(-) create mode 100644 internal/cli/watch.go create mode 100644 internal/watch/telegram.go create mode 100644 internal/watch/watch.go create mode 100644 internal/watch/watch_test.go diff --git a/internal/cli/root.go b/internal/cli/root.go index 4457758..71ea6a7 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -76,7 +76,7 @@ status, cometcli doctor, cometcli tx unjail) and a tool the agent calls.`, for _, c := range toolGroupCommands(reg) { root.AddCommand(c) } - root.AddCommand(profileCmd(), auditCmd(), versionCmd(), initCmd(), sessionsCmd(), oneCmd(reg), configCmd(), trustCmd()) + root.AddCommand(profileCmd(), auditCmd(), versionCmd(), initCmd(), sessionsCmd(), oneCmd(reg), configCmd(), trustCmd(), watchCmd(reg)) root.AddCommand(extra...) return root } diff --git a/internal/cli/watch.go b/internal/cli/watch.go new file mode 100644 index 0000000..443c078 --- /dev/null +++ b/internal/cli/watch.go @@ -0,0 +1,172 @@ +package cli + +import ( + "context" + "fmt" + "sort" + "strings" + "time" + + "github.com/spf13/cobra" + + "github.com/abhijitkrm/cometcli/internal/agent" + "github.com/abhijitkrm/cometcli/internal/audit" + "github.com/abhijitkrm/cometcli/internal/config" + "github.com/abhijitkrm/cometcli/internal/monitor" + "github.com/abhijitkrm/cometcli/internal/toolkit" + "github.com/abhijitkrm/cometcli/internal/tools/triage" + "github.com/abhijitkrm/cometcli/internal/watch" +) + +func watchCmd(reg *toolkit.Registry) *cobra.Command { + var interval, mode, cooldown string + var profiles []string + var once bool + cmd := &cobra.Command{ + Use: "watch", + Short: "Watch the fleet on your behalf: triage on an interval, work new incidents, report and ask for approvals remotely", + Long: `Sweeps every node (fleet triage) on an interval. A newly appearing critical or +high problem becomes an incident: + + notify report the finding + diagnose the agent investigates read-only and reports (default) + fix the agent may act; every approval is sent to Telegram with + Approve / Deny buttons (unanswered = denied) + +Findings, reports and recoveries go to the watch.alerts channels (Slack, +Discord, Telegram). Transactions are refused unless watch.allow_tx is set. +Configure under 'watch:' in ~/.cometcli/config.yaml; flags override it.`, + RunE: func(cmd *cobra.Command, _ []string) error { + cfg, err := config.Load() + if err != nil { + return err + } + wc := cfg.Watch + pick := func(flag, conf, def string) string { + if cmd.Flags().Changed(flag) { + return map[string]string{"interval": interval, "mode": mode, "cooldown": cooldown}[flag] + } + if conf != "" { + return conf + } + return def + } + iv, err := time.ParseDuration(pick("interval", wc.Interval, "5m")) + if err != nil || iv < 30*time.Second { + return fmt.Errorf("interval: want a duration ≥ 30s") + } + cd, err := time.ParseDuration(pick("cooldown", wc.Cooldown, "1h")) + if err != nil { + return fmt.Errorf("cooldown: %w", err) + } + m, err := watch.ParseMode(pick("mode", wc.Mode, "")) + if err != nil { + return err + } + names := wc.Profiles + if cmd.Flags().Changed("profiles") { + names = profiles + } + selected := map[string]*config.Profile{} + var list []*config.Profile + var all []string + for n := range cfg.Profiles { + all = append(all, n) + } + sort.Strings(all) + for _, n := range all { + if len(names) > 0 && !contains(names, n) { + continue + } + p := cfg.Profiles[n] + p.Name = n + selected[n] = p + list = append(list, p) + } + if len(list) == 0 { + return fmt.Errorf("no profiles to watch — cometcli init") + } + + alerts := wc.Alerts + if alerts.TelegramToken == "" { + alerts.TelegramToken = config.Credential("TELEGRAM_BOT_TOKEN") + } + var notify []watch.Notifier + var channels []string + for _, s := range monitor.Sinks(alerts) { + notify = append(notify, s) + channels = append(channels, s.Name()) + } + var approver toolkit.Approver + if m == watch.ModeFix { + if alerts.TelegramToken == "" || alerts.TelegramChatID == "" { + fmt.Fprintln(cmd.ErrOrStderr(), "fix mode needs Telegram for approvals (watch.alerts.telegram_token or the TELEGRAM_BOT_TOKEN credential, and telegram_chat_id) — running read-only") + } else { + at, _ := time.ParseDuration(wc.ApprovalTimeout) + users := map[int64]bool{} + for _, u := range wc.TelegramUsers { + users[u] = true + } + tg := &watch.Telegram{Token: alerts.TelegramToken, ChatID: alerts.TelegramChatID, Users: users, Timeout: at} + approver = watch.RemoteApprover(tg.Ask, wc.AllowTx) + } + } + statePath, _ := config.Path("watch-state.json") + glog, _ := audit.Open("watch") + w := &watch.Watcher{ + Mode: m, Cooldown: cd, Profiles: selected, Notify: notify, Approver: approver, + StatePath: statePath, Out: cmd.OutOrStdout(), + Collect: func(ctx context.Context) []triage.NodeReport { + return triage.CollectFleet(&toolkit.Context{Context: ctx, Cfg: cfg, Audit: glog}, list, iv) + }, + Run: func(ctx context.Context, p *config.Profile, desc string, readOnly bool, approve toolkit.Approver) (string, error) { + lg, _ := audit.Open(p.Name) + if lg != nil { + defer lg.Close() + } + c := &toolkit.Context{Context: ctx, Profile: p, Cfg: cfg, Audit: lg, Approver: approve} + a, err := agent.New(c, reg) + if err != nil { + return "", err + } + defer a.Close() + if readOnly { + a.Policy.Mode = agent.ModeReadOnly + } + res, err := agent.RunCommand(a, a.Ctx, reg, "/incident "+desc) + if err != nil { + return "", err + } + return a.Run(ctx, res.Prompt) + }, + } + if len(channels) == 0 { + channels = []string{"this terminal only"} + } + var watched []string + for _, p := range list { + watched = append(watched, p.Name) + } + fmt.Fprintf(cmd.OutOrStdout(), "watching %d nodes (%s) every %s · mode %s · cooldown %s · reports to %s\n", + len(list), strings.Join(watched, ", "), iv, m, cd, strings.Join(channels, ", ")) + w.Loop(cmd.Context(), iv, once) + return nil + }, + } + f := cmd.Flags() + f.StringVar(&interval, "interval", "", "time between sweeps (default 5m, or watch.interval)") + f.StringVar(&mode, "mode", "", "notify | diagnose | fix (default diagnose, or watch.mode)") + f.StringVar(&cooldown, "cooldown", "", "before the same issue is worked again (default 1h)") + f.StringSliceVar(&profiles, "profiles", nil, "profiles to watch (default all, or watch.profiles)") + f.BoolVar(&once, "once", false, "run one sweep and exit") + return cmd +} + +func contains(list []string, s string) bool { + for _, x := range list { + if x == s { + return true + } + } + return false +} diff --git a/internal/config/config.go b/internal/config/config.go index 73e5d8c..6c24f01 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -44,6 +44,29 @@ type Config struct { // (no node) and as the base a profile's agent section overrides. Agent AgentConf `yaml:"agent,omitempty"` Profiles map[string]*Profile `yaml:"profiles"` + // Watch configures `cometcli watch`, the autonomous fleet watcher. + Watch WatchConf `yaml:"watch,omitempty"` +} + +// WatchConf configures the autonomous watcher. +type WatchConf struct { + Interval string `yaml:"interval,omitempty"` // between sweeps (default 5m) + Mode string `yaml:"mode,omitempty"` // notify | diagnose (default) | fix + Profiles []string `yaml:"profiles,omitempty"` // default: all + Cooldown string `yaml:"cooldown,omitempty"` // before the same issue is worked again (default 1h) + // ApprovalTimeout is how long a remote approval waits before it's a + // "no" (default 15m). + ApprovalTimeout string `yaml:"approval_timeout,omitempty"` + // AllowTx lets fix mode request transaction approvals remotely; + // without it, transactions are always refused in watch mode. + AllowTx bool `yaml:"allow_tx,omitempty"` + // Alerts are where findings, reports and approvals go. Telegram is + // the approval channel (token: here or the TELEGRAM_BOT_TOKEN + // credential); Slack and Discord are notify-only. + Alerts Alerts `yaml:"alerts,omitempty"` + // TelegramUsers, when set, are the only Telegram user ids whose + // button presses count. + TelegramUsers []int64 `yaml:"telegram_users,omitempty"` } // AgentFor resolves the agent settings for a profile (nil = general diff --git a/internal/watch/telegram.go b/internal/watch/telegram.go new file mode 100644 index 0000000..8eb567e --- /dev/null +++ b/internal/watch/telegram.go @@ -0,0 +1,159 @@ +package watch + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "net/http" + "strings" + "sync" + "time" +) + +// Telegram asks for approvals with inline Approve / Deny buttons and +// waits for a press from the configured chat (and, if set, users). +// Anything unanswered by Timeout is a "no". +type Telegram struct { + Base string // API base (tests); default https://api.telegram.org + Token string + ChatID string + Users map[int64]bool // allowed approvers; empty = anyone in the chat + Timeout time.Duration + HTTP *http.Client + + mu sync.Mutex // one approval at a time; getUpdates offsets are shared + offset int64 +} + +func (t *Telegram) api(ctx context.Context, method string, body any, out any) error { + base := t.Base + if base == "" { + base = "https://api.telegram.org" + } + raw, _ := json.Marshal(body) + req, err := http.NewRequestWithContext(ctx, "POST", fmt.Sprintf("%s/bot%s/%s", base, t.Token, method), bytes.NewReader(raw)) + if err != nil { + return err + } + req.Header.Set("content-type", "application/json") + hc := t.HTTP + if hc == nil { + hc = &http.Client{Timeout: 60 * time.Second} + } + resp, err := hc.Do(req) + if err != nil { + return fmt.Errorf("telegram %s: %w", method, err) // never echo the URL: it holds the token + } + defer resp.Body.Close() + var env struct { + OK bool `json:"ok"` + Description string `json:"description"` + Result json.RawMessage `json:"result"` + } + if err := json.NewDecoder(resp.Body).Decode(&env); err != nil { + return fmt.Errorf("telegram %s: %w", method, err) + } + if !env.OK { + return fmt.Errorf("telegram %s: %s", method, env.Description) + } + if out != nil { + return json.Unmarshal(env.Result, out) + } + return nil +} + +// Send posts a plain message. +func (t *Telegram) Send(ctx context.Context, text string) error { + return t.api(ctx, "sendMessage", map[string]any{"chat_id": t.ChatID, "text": clip(text, 4000)}, nil) +} + +// Ask posts the question with buttons and waits for the answer. +func (t *Telegram) Ask(ctx context.Context, text string) (bool, string, error) { + t.mu.Lock() + defer t.mu.Unlock() + b := make([]byte, 4) + _, _ = rand.Read(b) + id := hex.EncodeToString(b) + var msg struct { + MessageID int64 `json:"message_id"` + } + err := t.api(ctx, "sendMessage", map[string]any{ + "chat_id": t.ChatID, "text": clip(text, 3800), + "reply_markup": map[string]any{"inline_keyboard": [][]map[string]string{{ + {"text": "✅ Approve", "callback_data": "a:" + id}, + {"text": "❌ Deny", "callback_data": "d:" + id}, + }}}, + }, &msg) + if err != nil { + return false, "", err + } + timeout := t.Timeout + if timeout <= 0 { + timeout = 15 * time.Minute + } + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if ctx.Err() != nil { + return false, "", ctx.Err() + } + wait := int(time.Until(deadline).Seconds()) + if wait > 25 { + wait = 25 + } + var updates []struct { + UpdateID int64 `json:"update_id"` + Callback *struct { + ID string `json:"id"` + Data string `json:"data"` + From struct { + ID int64 `json:"id"` + Username string `json:"username"` + } `json:"from"` + Message struct { + Chat struct { + ID int64 `json:"id"` + } `json:"chat"` + } `json:"message"` + } `json:"callback_query"` + } + if err := t.api(ctx, "getUpdates", map[string]any{"offset": t.offset, "timeout": wait, "allowed_updates": []string{"callback_query"}}, &updates); err != nil { + time.Sleep(2 * time.Second) + continue + } + for _, u := range updates { + t.offset = u.UpdateID + 1 + cq := u.Callback + if cq == nil || len(cq.Data) < 3 || cq.Data[2:] != id { + continue + } + if fmt.Sprint(cq.Message.Chat.ID) != t.ChatID || (len(t.Users) > 0 && !t.Users[cq.From.ID]) { + _ = t.api(ctx, "answerCallbackQuery", map[string]any{"callback_query_id": cq.ID, "text": "you can't approve this"}, nil) + continue + } + ok := strings.HasPrefix(cq.Data, "a:") + who := cq.From.Username + if who == "" { + who = fmt.Sprint(cq.From.ID) + } + verdict := "❌ denied by " + who + if ok { + verdict = "✅ approved by " + who + } + _ = t.api(ctx, "answerCallbackQuery", map[string]any{"callback_query_id": cq.ID, "text": verdict}, nil) + _ = t.api(ctx, "editMessageText", map[string]any{"chat_id": t.ChatID, "message_id": msg.MessageID, "text": clip(text, 3800) + "\n\n" + verdict}, nil) + return ok, who, nil + } + } + _ = t.api(ctx, "editMessageText", map[string]any{"chat_id": t.ChatID, "message_id": msg.MessageID, "text": clip(text, 3800) + "\n\n⌛ no answer in " + timeout.String() + " — denied"}, nil) + return false, "", nil +} + +func clip(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n-1] + "…" +} diff --git a/internal/watch/watch.go b/internal/watch/watch.go new file mode 100644 index 0000000..5791691 --- /dev/null +++ b/internal/watch/watch.go @@ -0,0 +1,296 @@ +// Package watch is cometcli's autonomous watcher: it sweeps the fleet on +// an interval, turns newly appearing problems into incidents, has the +// agent work them (read-only, or fixing with remote approvals), and +// reports findings, results and recoveries to Slack, Discord or Telegram. +package watch + +import ( + "context" + "encoding/json" + "fmt" + "io" + "os" + "regexp" + "sort" + "strings" + "time" + + "github.com/abhijitkrm/cometcli/internal/config" + "github.com/abhijitkrm/cometcli/internal/toolkit" + "github.com/abhijitkrm/cometcli/internal/tools/triage" +) + +// Mode is what the watcher does with a new incident. +type Mode string + +const ( + ModeNotify Mode = "notify" // report the finding only + ModeDiagnose Mode = "diagnose" // the agent investigates read-only and reports + ModeFix Mode = "fix" // the agent may act; every approval goes to the operator +) + +// ParseMode validates a mode name ("" = diagnose). +func ParseMode(s string) (Mode, error) { + switch Mode(strings.ToLower(strings.TrimSpace(s))) { + case "", ModeDiagnose: + return ModeDiagnose, nil + case ModeNotify: + return ModeNotify, nil + case ModeFix: + return ModeFix, nil + } + return "", fmt.Errorf("watch mode %q: want notify, diagnose or fix", s) +} + +// Notifier delivers a message somewhere (Slack, Discord, Telegram, stdout). +type Notifier interface { + Send(ctx context.Context, text string) error +} + +// Runner has the agent work an incident on a profile and returns its +// report. readOnly forbids any change; approve gates the rest. +type Runner func(ctx context.Context, p *config.Profile, description string, readOnly bool, approve toolkit.Approver) (string, error) + +// Issue is a problem seen in a sweep. +type Issue struct { + Key string // stable identity: profile|case, or fleet| + Profile string + Case string + Severity string + Text string +} + +// Watcher runs sweeps. +type Watcher struct { + Mode Mode + Cooldown time.Duration + Collect func(ctx context.Context) []triage.NodeReport + Profiles map[string]*config.Profile + Run Runner + Notify []Notifier + Approver toolkit.Approver // remote approvals (fix mode); nil = none + StatePath string + Out io.Writer + Now func() time.Time + + state state +} + +type state struct { + Active map[string]time.Time `json:"active"` // issues seen in the last sweep + Handled map[string]time.Time `json:"handled"` // when each issue was last worked +} + +func (w *Watcher) now() time.Time { + if w.Now != nil { + return w.Now() + } + return time.Now() +} + +func (w *Watcher) logf(format string, a ...any) { + if w.Out != nil { + fmt.Fprintf(w.Out, "%s %s\n", w.now().Format("15:04:05"), fmt.Sprintf(format, a...)) + } +} + +func (w *Watcher) load() { + w.state = state{Active: map[string]time.Time{}, Handled: map[string]time.Time{}} + if w.StatePath == "" { + return + } + if raw, err := os.ReadFile(w.StatePath); err == nil { + _ = json.Unmarshal(raw, &w.state) + } + if w.state.Active == nil { + w.state.Active = map[string]time.Time{} + } + if w.state.Handled == nil { + w.state.Handled = map[string]time.Time{} + } +} + +func (w *Watcher) save() { + if w.StatePath == "" { + return + } + raw, _ := json.MarshalIndent(w.state, "", " ") + _ = os.WriteFile(w.StatePath, raw, 0o600) +} + +func (w *Watcher) send(ctx context.Context, text string) { + w.logf("%s", strings.ReplaceAll(text, "\n", "\n ")) + for _, n := range w.Notify { + if err := n.Send(ctx, text); err != nil { + w.logf("notify failed: %v", err) + } + } +} + +var digits = regexp.MustCompile(`[0-9]+`) + +// Issues extracts the problems worth acting on from a sweep: each node's +// most severe root case (critical or high), and fleet findings that mean +// trouble (halts, double-sign risk, nodes cut off or down). +func Issues(nodes []triage.NodeReport, findings []string) []Issue { + var out []Issue + for _, n := range nodes { + for _, h := range n.Hits { + if h.Case.Kind == "advisory" || h.SymptomOf != "" { + continue + } + if h.Case.Severity == "critical" || h.Case.Severity == "high" { + out = append(out, Issue{Key: n.Profile.Name + "|" + h.Case.ID, Profile: n.Profile.Name, Case: h.Case.ID, + Severity: h.Case.Severity, Text: h.Case.Title + " — " + strings.Join(h.Matched, ", ")}) + } + break // the top root case speaks for the node + } + } + for _, f := range findings { + if strings.Contains(f, "CHAIN HALT") || strings.Contains(f, "DOUBLE-SIGN") || strings.Contains(f, "cut off") { + // numbers move between sweeps; the finding's shape is its identity + out = append(out, Issue{Key: "fleet|" + digits.ReplaceAllString(f, "#"), Severity: "critical", Text: f}) + } + } + return out +} + +// Sweep runs one round: collect, find new and cleared issues, act. +func (w *Watcher) Sweep(ctx context.Context) { + if w.state.Active == nil { + w.load() + } + nodes := w.Collect(ctx) + issues := Issues(nodes, triage.FleetFindings(nodes)) + now := w.now() + current := map[string]Issue{} + for _, is := range issues { + current[is.Key] = is + } + // recovered: active last time, gone now + var cleared []string + for k := range w.state.Active { + if _, still := current[k]; !still { + cleared = append(cleared, k) + } + } + sort.Strings(cleared) + for _, k := range cleared { + delete(w.state.Active, k) + w.send(ctx, "✅ resolved: "+strings.Replace(k, "|", ": ", 1)) + } + keys := make([]string, 0, len(current)) + for k := range current { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + is := current[k] + _, wasActive := w.state.Active[k] + w.state.Active[k] = now + if wasActive { + continue // already known and being (or been) handled + } + if last, ok := w.state.Handled[k]; ok && now.Sub(last) < w.Cooldown { + continue // worked recently: don't loop on it + } + w.state.Handled[k] = now + w.save() + w.handle(ctx, is) + } + w.save() + w.logf("sweep: %d nodes, %d open issues, %d resolved", len(nodes), len(current), len(cleared)) +} + +func (w *Watcher) handle(ctx context.Context, is Issue) { + where := is.Profile + if where == "" { + where = "fleet" + } + w.send(ctx, fmt.Sprintf("🚨 %s: %s [%s]\n%s", where, orDash(is.Case), is.Severity, is.Text)) + if w.Mode == ModeNotify || is.Profile == "" || w.Run == nil { + return + } + p := w.Profiles[is.Profile] + if p == nil { + return + } + readOnly := w.Mode == ModeDiagnose || w.Approver == nil + approve := w.Approver + if readOnly { + approve = toolkit.DenyApprover + } + verb := "investigating (read-only)" + if !readOnly { + verb = "working it — approvals will be sent here" + } + w.send(ctx, fmt.Sprintf("🔎 %s: %s", is.Profile, verb)) + report, err := w.Run(ctx, p, fmt.Sprintf("%s on %s detected by the watcher: %s", orDash(is.Case), is.Profile, is.Text), readOnly, approve) + if err != nil { + w.send(ctx, fmt.Sprintf("⚠️ %s: the agent stopped: %v", is.Profile, err)) + return + } + w.send(ctx, fmt.Sprintf("📋 %s report:\n%s", is.Profile, clip(strings.TrimSpace(report), 3000))) +} + +// Loop sweeps every interval until ctx ends; once runs a single sweep. +func (w *Watcher) Loop(ctx context.Context, interval time.Duration, once bool) { + w.load() + for { + w.Sweep(ctx) + if once { + return + } + select { + case <-ctx.Done(): + return + case <-time.After(interval): + } + } +} + +// RemoteApprover adapts an asker (Telegram) into a toolkit.Approver: +// transactions are refused unless allowTx, and the question carries the +// tool's detail (command, diff, tx document). +func RemoteApprover(ask func(ctx context.Context, text string) (bool, string, error), allowTx bool) toolkit.Approver { + return func(c *toolkit.Context, prompt string, tier toolkit.Tier, detail map[string]any) (bool, error) { + if tier == toolkit.TierOnChain && !allowTx { + return false, fmt.Errorf("transactions are disabled in watch mode (set watch.allow_tx to approve them remotely)") + } + node := "fleet" + if c != nil && c.Profile != nil { + node = c.Profile.Name + } + var b strings.Builder + fmt.Fprintf(&b, "🔐 approval needed on %s [%s]\n%s", node, tier, prompt) + var keys []string + for k := range detail { + if !strings.HasPrefix(k, "_") { + keys = append(keys, k) + } + } + sort.Strings(keys) + for _, k := range keys { + fmt.Fprintf(&b, "\n%s: %v", k, detail[k]) + } + ctx := context.Background() + if c != nil { + ctx = c + } + ok, _, err := ask(ctx, b.String()) + if err != nil { + return false, fmt.Errorf("remote approval failed: %w", err) + } + if !ok { + return false, fmt.Errorf("denied by the operator (or no answer in time)") + } + return true, nil + } +} + +func orDash(s string) string { + if s == "" { + return "-" + } + return s +} diff --git a/internal/watch/watch_test.go b/internal/watch/watch_test.go new file mode 100644 index 0000000..a777ef8 --- /dev/null +++ b/internal/watch/watch_test.go @@ -0,0 +1,181 @@ +package watch + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/abhijitkrm/cometcli/internal/config" + "github.com/abhijitkrm/cometcli/internal/kb" + "github.com/abhijitkrm/cometcli/internal/toolkit" + "github.com/abhijitkrm/cometcli/internal/tools/triage" +) + +type memNotify struct{ msgs []string } + +func (m *memNotify) Send(_ context.Context, t string) error { m.msgs = append(m.msgs, t); return nil } + +func jailedNode(jailed bool) []triage.NodeReport { + p := &config.Profile{Name: "val01", ChainID: "c-1", Role: "validator"} + r := &triage.Report{Signals: kb.Signals{"node.reachable": true, "node.height": 10.0, "node.block_age_s": 1.0}} + var hits []kb.Hit + if jailed { + hits = []kb.Hit{{Case: &kb.Case{ID: "val-jailed-downtime", Title: "Validator jailed", Severity: "critical", Kind: "incident"}, Matched: []string{"val.jailed == true"}}} + } + return []triage.NodeReport{{Profile: p, Report: r, Hits: hits}} +} + +func TestSweepLifecycle(t *testing.T) { + jailed := true + now := time.Now() + var runs []bool + n := &memNotify{} + w := &Watcher{ + Mode: ModeDiagnose, Cooldown: time.Hour, StatePath: filepath.Join(t.TempDir(), "state.json"), + Profiles: map[string]*config.Profile{"val01": {Name: "val01"}}, + Collect: func(context.Context) []triage.NodeReport { return jailedNode(jailed) }, + Run: func(_ context.Context, p *config.Profile, desc string, readOnly bool, approve toolkit.Approver) (string, error) { + runs = append(runs, readOnly) + if ok, _ := approve(nil, "restart", toolkit.TierLocalChange, nil); ok { + t.Error("diagnose mode approved a change") + } + return "cause: container stopped", nil + }, + Notify: []Notifier{n}, Now: func() time.Time { return now }, + } + ctx := context.Background() + w.Sweep(ctx) // new incident: report + read-only investigation + all := strings.Join(n.msgs, "\n") + if len(runs) != 1 || !runs[0] || !strings.Contains(all, "🚨 val01: val-jailed-downtime [critical]") || !strings.Contains(all, "📋 val01 report:\ncause: container stopped") { + t.Fatalf("first sweep: runs=%v\n%s", runs, all) + } + w.Sweep(ctx) // still jailed: nothing new + if len(runs) != 1 { + t.Fatal("worked the same incident twice") + } + jailed = false + w.Sweep(ctx) // cleared + if !strings.Contains(n.msgs[len(n.msgs)-1], "✅ resolved: val01: val-jailed-downtime") { + t.Fatalf("no recovery message: %v", n.msgs) + } + jailed = true + w.Sweep(ctx) // back within the cooldown: not worked again + if len(runs) != 1 { + t.Fatal("cooldown ignored") + } + now = now.Add(2 * time.Hour) + jailed = false + w.Sweep(ctx) + jailed = true + w.Sweep(ctx) // after the cooldown: worked again + if len(runs) != 2 { + t.Fatalf("runs after cooldown = %d", len(runs)) + } + // state survives a restart + w2 := &Watcher{StatePath: w.StatePath, Mode: ModeNotify, Collect: w.Collect, Notify: []Notifier{&memNotify{}}, Cooldown: time.Hour, Now: w.Now} + w2.Sweep(ctx) + if len(w2.Notify[0].(*memNotify).msgs) != 0 { + t.Fatal("restart re-announced an open incident") + } +} + +func TestIssuesFromFleetFindings(t *testing.T) { + a := Issues(nil, []string{"CHAIN HALT on c-1: all 4 reachable nodes are stalled near height 3644 — …"}) + b := Issues(nil, []string{"CHAIN HALT on c-1: all 4 reachable nodes are stalled near height 3651 — …"}) + if len(a) != 1 || a[0].Key != b[0].Key { + t.Fatalf("a moving height changed the halt's identity: %v %v", a, b) + } + if got := Issues(nil, []string{"val2 is 30 blocks behind the other c-1 nodes"}); len(got) != 0 { + t.Fatalf("minor finding became an incident: %v", got) + } +} + +func TestRemoteApproverRefusesTxUnlessAllowed(t *testing.T) { + asked := 0 + ask := func(context.Context, string) (bool, string, error) { asked++; return true, "op", nil } + ap := RemoteApprover(ask, false) + if ok, err := ap(nil, "unjail", toolkit.TierOnChain, nil); ok || err == nil || asked != 0 { + t.Fatalf("tx approved without allow_tx: %v %v asked=%d", ok, err, asked) + } + if ok, _ := ap(nil, "restart node", toolkit.TierLocalChange, map[string]any{"command": "docker restart val01"}); !ok || asked != 1 { + t.Fatal("local change not sent for approval") + } + if ok, _ := RemoteApprover(ask, true)(nil, "unjail", toolkit.TierOnChain, nil); !ok { + t.Fatal("allow_tx ignored") + } +} + +// fakeTelegram serves the Bot API methods the approver uses. +type fakeTelegram struct { + mu sync.Mutex + pressBy []int64 // users who press "approve", in order + sent []string + edited []string + answered []string + cbData string +} + +func (f *fakeTelegram) handler(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + defer f.mu.Unlock() + raw, _ := io.ReadAll(r.Body) + var body map[string]any + _ = json.Unmarshal(raw, &body) + method := r.URL.Path[strings.LastIndex(r.URL.Path, "/")+1:] + reply := func(v any) { _ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "result": v}) } + switch method { + case "sendMessage": + f.sent = append(f.sent, body["text"].(string)) + if rm, ok := body["reply_markup"].(map[string]any); ok { + f.cbData = rm["inline_keyboard"].([]any)[0].([]any)[0].(map[string]any)["callback_data"].(string) + } + reply(map[string]any{"message_id": 7}) + case "getUpdates": + var ups []any + for i, u := range f.pressBy { + ups = append(ups, map[string]any{"update_id": i + 1, "callback_query": map[string]any{ + "id": "cb", "data": f.cbData, "from": map[string]any{"id": u, "username": "user" + string(rune('0'+i))}, + "message": map[string]any{"chat": map[string]any{"id": 42}}}}) + } + f.pressBy = nil + reply(ups) + case "answerCallbackQuery": + f.answered = append(f.answered, body["text"].(string)) + reply(true) + case "editMessageText": + f.edited = append(f.edited, body["text"].(string)) + reply(true) + } +} + +func TestTelegramApproval(t *testing.T) { + f := &fakeTelegram{pressBy: []int64{999, 111}} // a stranger first, then the operator + srv := httptest.NewServer(http.HandlerFunc(f.handler)) + defer srv.Close() + tg := &Telegram{Base: srv.URL, Token: "tok", ChatID: "42", Users: map[int64]bool{111: true}, Timeout: 5 * time.Second} + ok, who, err := tg.Ask(context.Background(), "restart val01?") + if err != nil || !ok || who != "user1" { + t.Fatalf("ok=%v who=%q err=%v", ok, who, err) + } + if len(f.answered) < 2 || !strings.Contains(f.answered[0], "can't approve") { + t.Fatalf("stranger's press not refused: %v", f.answered) + } + if len(f.edited) != 1 || !strings.Contains(f.edited[0], "✅ approved by user1") { + t.Fatalf("message not updated: %v", f.edited) + } + // nobody answers: denied when the timeout passes + f2 := &fakeTelegram{} + srv2 := httptest.NewServer(http.HandlerFunc(f2.handler)) + defer srv2.Close() + tg2 := &Telegram{Base: srv2.URL, Token: "tok", ChatID: "42", Timeout: 1500 * time.Millisecond} + if ok, _, _ := tg2.Ask(context.Background(), "restart?"); ok || len(f2.edited) == 0 || !strings.Contains(f2.edited[0], "denied") { + t.Fatalf("timeout: ok=%v edited=%v", ok, f2.edited) + } +} From 9b40d3ead25df10d2e13c794755a3c3e653f27d1 Mon Sep 17 00:00:00 2001 From: Abhijit Date: Thu, 8 Oct 2026 18:16:24 +0530 Subject: [PATCH 5/7] cometcli drill: incident drills on a throwaway docker network - drill testnet up --image : a local validator network (fast slashing, 60s votes) with drill profiles (metadata drill=true, container signer, fallback gRPC); testnet down removes everything - drill run: per scenario inject a real fault (node stopped until jailed, broken config, halt-height, OOM limit, network partition), score whether triage named an expected root case, let the agent work it via /incident with approvals auto-granted only on drill profiles, verify on the node, then recover it (files, limits, network, unjail) - scoreboard (triage x/y, fixed x/y, steps, tokens, time) and JSON results in ~/.cometcli/drills/ Co-Authored-By: Claude Opus 5.5 --- internal/cli/drill.go | 133 +++++++++++++++ internal/cli/root.go | 2 +- internal/drill/drill_test.go | 77 +++++++++ internal/drill/run.go | 179 ++++++++++++++++++++ internal/drill/scenarios.go | 229 +++++++++++++++++++++++++ internal/drill/testnet.go | 314 +++++++++++++++++++++++++++++++++++ 6 files changed, 933 insertions(+), 1 deletion(-) create mode 100644 internal/cli/drill.go create mode 100644 internal/drill/drill_test.go create mode 100644 internal/drill/run.go create mode 100644 internal/drill/scenarios.go create mode 100644 internal/drill/testnet.go diff --git a/internal/cli/drill.go b/internal/cli/drill.go new file mode 100644 index 0000000..ab8ed56 --- /dev/null +++ b/internal/cli/drill.go @@ -0,0 +1,133 @@ +package cli + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/spf13/cobra" + + "github.com/abhijitkrm/cometcli/internal/agent" + "github.com/abhijitkrm/cometcli/internal/audit" + "github.com/abhijitkrm/cometcli/internal/config" + "github.com/abhijitkrm/cometcli/internal/drill" + "github.com/abhijitkrm/cometcli/internal/toolkit" +) + +func drillCmd(reg *toolkit.Registry) *cobra.Command { + cmd := &cobra.Command{ + Use: "drill", + Short: "Incident drills: inject real faults on a throwaway docker network, score how the agent handles them", + } + logf := func(cmd *cobra.Command) func(string) { + return func(s string) { fmt.Fprintf(cmd.OutOrStdout(), "%s %s\n", time.Now().Format("15:04:05"), s) } + } + + var image string + var validators int + tn := &cobra.Command{Use: "testnet", Short: "Create or remove the throwaway drill network"} + up := &cobra.Command{ + Use: "up", + Short: "Create a local validator network in docker (fast slashing: 20-block window, 30s jail) with drill profiles", + RunE: func(cmd *cobra.Command, _ []string) error { + cfg, err := config.Load() + if err != nil { + return err + } + n := &drill.Net{Image: image, Validators: validators} + if err := n.Up(cmd.Context(), cfg, logf(cmd)); err != nil { + return err + } + fmt.Fprintf(cmd.OutOrStdout(), "drill network up: profiles drill0..drill%d on chain %s — run `cometcli drill run`; remove with `cometcli drill testnet down`\n", n.Validators-1, n.ChainID) + return nil + }, + } + up.Flags().StringVar(&image, "image", "", "evmd docker image to run (required)") + up.Flags().IntVar(&validators, "validators", 4, "validators (4 keeps the chain live with one node down)") + down := &cobra.Command{ + Use: "down", + Short: "Remove the drill network, its data and its profiles", + RunE: func(cmd *cobra.Command, _ []string) error { + cfg, err := config.Load() + if err != nil { + return err + } + if err := (&drill.Net{}).Down(cmd.Context(), cfg); err != nil { + return err + } + fmt.Fprintln(cmd.OutOrStdout(), "drill network removed") + return nil + }, + } + tn.AddCommand(up, down) + + list := &cobra.Command{ + Use: "list", + Short: "List the drill scenarios", + Run: func(cmd *cobra.Command, _ []string) { + for _, s := range drill.Scenarios { + fmt.Fprintf(cmd.OutOrStdout(), "%-14s %s (expects %s)\n", s.Name, s.Desc, strings.Join(s.Expect, " | ")) + } + }, + } + + var only []string + var node int + var model string + run := &cobra.Command{ + Use: "run", + Short: "Run drills: inject each fault, let the agent work it with /incident, verify on the node, score", + RunE: func(cmd *cobra.Command, _ []string) error { + cfg, err := config.Load() + if err != nil { + return err + } + n := &drill.Net{} + n.Defaults() + if node < 0 { + node = 3 + } + p := cfg.Profiles[drill.Profile(node)] + if !drill.IsDrill(p) { + return fmt.Errorf("no drill profile %s — `cometcli drill testnet up --image ` first", drill.Profile(node)) + } + p.Name = drill.Profile(node) + factory := func(ctx context.Context, approve toolkit.Approver) (*agent.Agent, error) { + lg, _ := audit.Open("drill") + c := &toolkit.Context{Context: ctx, Cfg: cfg, Audit: lg, Approver: approve} + a, err := agent.New(c, reg) + if err != nil { + return nil, err + } + if model != "" { + a.Model = model + } + return a, nil + } + var results []drill.Result + for _, s := range drill.Scenarios { + if len(only) > 0 && !contains(only, s.Name) { + continue + } + fmt.Fprintf(cmd.OutOrStdout(), "\n== %s on %s\n", s.Name, p.Name) + e := &drill.Env{Net: n, Cfg: cfg, Reg: reg, Node: node, Profile: p} + r := drill.RunScenario(cmd.Context(), e, s, factory, logf(cmd)) + results = append(results, r) + fmt.Fprintf(cmd.OutOrStdout(), " triage %s (%v) · fixed %v · %d steps · %d tokens · %s\n", r.TriageCase, r.TriageOK, r.Fixed, r.Steps, r.Tokens, r.Duration.Round(time.Second)) + } + if len(results) == 0 { + return fmt.Errorf("no scenarios matched — cometcli drill list") + } + path, _ := drill.Save(results) + fmt.Fprintf(cmd.OutOrStdout(), "\n%s\nresults: %s\n", drill.Scoreboard(results), path) + return nil + }, + } + run.Flags().StringSliceVar(&only, "scenario", nil, "scenarios to run (default all)") + run.Flags().IntVar(&node, "node", -1, "validator index to break (default the last)") + run.Flags().StringVar(&model, "model", "", "model for the agent (default: your agent settings)") + + cmd.AddCommand(tn, list, run) + return cmd +} diff --git a/internal/cli/root.go b/internal/cli/root.go index 71ea6a7..919cd7d 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -76,7 +76,7 @@ status, cometcli doctor, cometcli tx unjail) and a tool the agent calls.`, for _, c := range toolGroupCommands(reg) { root.AddCommand(c) } - root.AddCommand(profileCmd(), auditCmd(), versionCmd(), initCmd(), sessionsCmd(), oneCmd(reg), configCmd(), trustCmd(), watchCmd(reg)) + root.AddCommand(profileCmd(), auditCmd(), versionCmd(), initCmd(), sessionsCmd(), oneCmd(reg), configCmd(), trustCmd(), watchCmd(reg), drillCmd(reg)) root.AddCommand(extra...) return root } diff --git a/internal/drill/drill_test.go b/internal/drill/drill_test.go new file mode 100644 index 0000000..1bd0751 --- /dev/null +++ b/internal/drill/drill_test.go @@ -0,0 +1,77 @@ +package drill + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/abhijitkrm/cometcli/internal/config" + "github.com/abhijitkrm/cometcli/internal/toolkit" +) + +func TestGenesisAndConfigPatching(t *testing.T) { + n := &Net{Dir: t.TempDir(), Validators: 2, Image: "evmd:test"} + n.Defaults() + for i := 0; i < 2; i++ { + _ = os.MkdirAll(filepath.Join(n.home(i), "config"), 0o755) + _ = os.WriteFile(filepath.Join(n.home(i), "config", "config.toml"), []byte("addr_book_strict = true\n"), 0o644) + _ = os.WriteFile(filepath.Join(n.home(i), "config", "app.toml"), []byte("minimum-gas-prices = \"0stake\"\naddress = \"localhost:9090\"\n"), 0o644) + } + g := `{"app_state":{"slashing":{"params":{"signed_blocks_window":"100","downtime_jail_duration":"600s"}}, + "gov":{"params":{"voting_period":"172800s"}},"evm":{"params":{"evm_denom":"aevmos"}}}}` + _ = os.WriteFile(filepath.Join(n.home(0), "config", "genesis.json"), []byte(g), 0o644) + denom, err := n.patchGenesis() + if err != nil || denom != "aevmos" { + t.Fatalf("denom %q %v", denom, err) + } + raw, _ := os.ReadFile(filepath.Join(n.home(1), "config", "genesis.json")) + var got map[string]any + _ = json.Unmarshal(raw, &got) + sl := got["app_state"].(map[string]any)["slashing"].(map[string]any)["params"].(map[string]any) + if sl["signed_blocks_window"] != "20" || sl["downtime_jail_duration"] != "30s" { + t.Fatalf("slashing not patched (or genesis not copied): %v", sl) + } + if err := n.patchConfigs(1, denom); err != nil { + t.Fatal(err) + } + app, _ := os.ReadFile(filepath.Join(n.home(1), "config", "app.toml")) + if !strings.Contains(string(app), `"1000000000aevmos"`) || !strings.Contains(string(app), "0.0.0.0:9090") { + t.Fatalf("app.toml:\n%s", app) + } + compose := n.composeFile("501:20") + for _, want := range []string{"container_name: cometcli-drill-1", "127.0.0.1:47067:26657", "image: evmd:test", "172.30.9.3"} { + if !strings.Contains(compose, want) { + t.Errorf("compose lacks %q", want) + } + } +} + +func TestApprovalsOnlyForDrillProfiles(t *testing.T) { + if ok, err := drillApprover(&toolkit.Context{Profile: &config.Profile{Name: "val01"}}, "restart", toolkit.TierLocalChange, nil); ok || err == nil { + t.Fatal("approved a change on a real profile") + } + if ok, err := drillApprover(&toolkit.Context{}, "rm -rf", toolkit.TierLocalChange, nil); ok || err == nil { + t.Fatal("approved a change on this machine (no profile)") + } + drill := &config.Profile{Name: "drill3", Metadata: map[string]string{"drill": "true"}} + if ok, _ := drillApprover(&toolkit.Context{Profile: drill}, "unjail", toolkit.TierOnChain, nil); !ok { + t.Fatal("drill profile not approved") + } + res := RunScenario(nil, &Env{Profile: &config.Profile{Name: "val01"}}, Scenarios[0], nil, func(string) {}) + if !strings.Contains(res.Err, "not a drill profile") { + t.Fatalf("ran a scenario on a real profile: %+v", res) + } +} + +func TestScoreboard(t *testing.T) { + s := Scoreboard([]Result{ + {Scenario: "jail-downtime", TriageCase: "val-jailed-downtime", TriageOK: true, Fixed: true, Steps: 9, Tokens: 41000, Duration: 3 * time.Minute}, + {Scenario: "oom", TriageCase: "node-crash-loop", Fixed: false, FixDetail: "memory limit still 120 MB", Steps: 30}, + }) + if !strings.Contains(s, "score: triage 1/2 · fixed 1/2") || !strings.Contains(s, "memory limit still 120 MB") { + t.Fatalf("scoreboard:\n%s", s) + } +} diff --git a/internal/drill/run.go b/internal/drill/run.go new file mode 100644 index 0000000..dc21533 --- /dev/null +++ b/internal/drill/run.go @@ -0,0 +1,179 @@ +package drill + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "time" + + "github.com/abhijitkrm/cometcli/internal/agent" + "github.com/abhijitkrm/cometcli/internal/config" + "github.com/abhijitkrm/cometcli/internal/toolkit" + "github.com/abhijitkrm/cometcli/internal/tools/triage" +) + +// Result is one scored drill. +type Result struct { + Scenario string `json:"scenario"` + Model string `json:"model"` + TriageCase string `json:"triage_case"` + TriageOK bool `json:"triage_ok"` // triage named an expected root case + Fixed bool `json:"fixed"` // verified on the node after the agent finished + FixDetail string `json:"fix_detail"` + Steps int `json:"steps"` + Tokens int `json:"tokens"` + Duration time.Duration `json:"duration_ns"` + Report string `json:"report"` + Err string `json:"error,omitempty"` +} + +// AgentFactory builds an agent in general mode with the given approver. +type AgentFactory func(ctx context.Context, approve toolkit.Approver) (*agent.Agent, error) + +// drillApprover approves everything — but only on drill profiles. +func drillApprover(c *toolkit.Context, prompt string, tier toolkit.Tier, _ map[string]any) (bool, error) { + if c == nil || !IsDrill(c.Profile) { + return false, fmt.Errorf("drill approvals only apply to drill profiles") + } + return true, nil +} + +// topCase is triage's leading root case for the node. +func topCase(ctx context.Context, e *Env) string { + c := &toolkit.Context{Context: ctx, Profile: e.Profile, Cfg: e.Cfg} + defer c.Close() + r := triage.Collect(c, 10*time.Minute) + for _, h := range triage.LoadKB(c).Match(r.Signals, r.Chain) { + if h.Case.Kind != "advisory" && h.SymptomOf == "" { + return h.Case.ID + } + } + return "" +} + +// RunScenario injects the fault, has the agent work it, scores the +// outcome and recovers the node. log reports progress. +func RunScenario(ctx context.Context, e *Env, s Scenario, newAgent AgentFactory, log func(string)) Result { + res := Result{Scenario: s.Name} + if !IsDrill(e.Profile) { + res.Err = "refusing: " + e.Profile.Name + " is not a drill profile" + return res + } + start := time.Now() + defer func() { + res.Duration = time.Since(start) + log("recovering " + e.Profile.Name) + if err := e.Recover(ctx, func(ctx context.Context) error { return e.unjail(ctx) }); err != nil { + res.Err = strings.TrimSpace(res.Err + "; recovery: " + err.Error()) + } + }() + log(fmt.Sprintf("injecting: %s", s.Desc)) + if err := s.Inject(ctx, e); err != nil { + res.Err = "inject: " + err.Error() + return res + } + if s.Ready != nil && !s.Ready(ctx, e) { + res.Err = "the fault never became visible" + return res + } + res.TriageCase = topCase(ctx, e) + for _, want := range s.Expect { + res.TriageOK = res.TriageOK || res.TriageCase == want + } + log(fmt.Sprintf("triage says %s (%v); the agent takes over", orDash(res.TriageCase), res.TriageOK)) + + a, err := newAgent(ctx, drillApprover) + if err != nil { + res.Err = "agent: " + err.Error() + return res + } + defer a.Close() + res.Model = a.Provider.Name() + "/" + a.Model + cmd, err := agent.RunCommand(a, a.Ctx, a.Reg, "/incident "+fmt.Sprintf(s.Prompt, e.Profile.Name)) + if err != nil { + res.Err = "incident: " + err.Error() + return res + } + runCtx, cancel := context.WithTimeout(ctx, 15*time.Minute) + report, err := a.Run(runCtx, cmd.Prompt) + cancel() + res.Report = report + res.Steps = a.Rounds() + u, _ := a.Usage() + res.Tokens = u.Input + u.Output + if err != nil { + res.Err = "agent run: " + err.Error() + } + log("verifying on the node") + ok := waitFor(ctx, 3*time.Minute, func() bool { v, _ := s.Verify(ctx, e); return v }) + res.Fixed = ok + _, res.FixDetail = s.Verify(ctx, e) + return res +} + +// unjail sends MsgUnjail for the drill node through the normal tool. +func (e *Env) unjail(ctx context.Context) error { + t, ok := e.Reg.Get("val.unjail") + if !ok { + return fmt.Errorf("val.unjail not registered") + } + c := &toolkit.Context{Context: ctx, Profile: e.Profile, Cfg: e.Cfg, Approver: drillApprover} + defer c.Close() + _, err := t.Run(c, toolkit.Args{}) + return err +} + +// Save writes results to ~/.cometcli/drills/