M4 and M5 gates passed: gate logs

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Lx9FNLG2CqeBbVGGd9bFsP

.golangci.yml

@@ -108,13 +108,18 @@ linters:
             - pkg: github.com/v10networks/nmon/internal/traps
               desc: collection paths share code via a lower package, not each other
         devtools-not-in-nmon:
-          # profile_record.go (`nmon profile record`, M2) already uses the recording types and writer; nothing else may.
-          files: ["**/cmd/nmon/**", "!**/cmd/nmon/profile_record.go", "!$test"]
+          # The product binary never links dev/test machinery. `nmon profile record` needs only the
+          # .snmprec format, which lives in internal/snmprec.
+          files: ["**/cmd/nmon/**", "!$test"]
           deny:
             - pkg: github.com/v10networks/nmon/internal/simagent
               desc: the simulator belongs to cmd/nmon-devtools and tests, never the shipped binary
             - pkg: github.com/v10networks/nmon/internal/fixturegen
               desc: the fixture generator belongs to cmd/nmon-devtools and tests, never the shipped binary
+            - pkg: github.com/v10networks/nmon/internal/fixtures
+              desc: fixtures are test data, never the shipped binary
+            - pkg: github.com/v10networks/nmon/internal/testutil
+              desc: test helpers are never in the shipped binary
         devtools-no-core:
           files: ["**/cmd/nmon-devtools/**"]
           deny:

cmd/nmon-devtools/backfilllong.go

@@ -0,0 +1,80 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/v10networks/nmon/internal/longq"
+	"github.com/v10networks/nmon/internal/metrics"
+	"github.com/v10networks/nmon/internal/rollup"
+)
+
+// backfillLongCmd writes rollup series straight into the long store, as if the rollup job had been running
+// for years: hourly average and maximum rates for --months months and 5-minute rates for --days days, flat
+// at the given rates so a gate can assert exact values. The data ends --end-offset ago, leaving the recent
+// hours to the real rollup job (its repair writes higher versions, which win).
+func backfillLongCmd(ctx context.Context, args []string) error {
+	fs := devFlagSet("backfill-long")
+	var ifaces stringList
+	long := fs.String("long-url", "", "base URL of the long VictoriaMetrics (see \"metric stores up\" in the serve log)")
+	device := fs.String("device", "", "device slug (the series' device label)")
+	fs.Var(&ifaces, "iface", "interface ULID; repeat the flag or separate with commas")
+	months := fs.Int("months", 25, "months of hourly rollups to write")
+	days := fs.Int("days", 70, "days of 5-minute rates to write")
+	offset := fs.Duration("end-offset", 3*time.Hour, "stop this long before now")
+	in := fs.Float64("in-bps", 8e6, "constant inbound rate")
+	out := fs.Float64("out-bps", 2e6, "constant outbound rate")
+	if err := fs.Parse(args); err != nil {
+		return flagErr(err)
+	}
+	if *long == "" || *device == "" || len(ifaces) == 0 {
+		return fmt.Errorf("backfill-long needs --long-url, --device and --iface: %w", errUsage)
+	}
+	c := &backfillConfig{hotURL: strings.TrimRight(*long, "/"), hc: &http.Client{Timeout: 5 * time.Minute}}
+	end := time.Now().Add(-*offset).Truncate(time.Hour)
+	for _, id := range ifaces {
+		s := longSamples(*device, id, end, *months, *days, map[string]float64{"in": *in, "out": *out})
+		if err := c.send(ctx, s); err != nil {
+			return errors.Join(fmt.Errorf("iface %s", id), err)
+		}
+		fmt.Printf("backfilled long %s/%s: %d months hourly, %d days 5m (%d samples)\n", *device, id, *months, *days, len(s))
+	}
+	return c.flush(ctx)
+}
+
+// flush makes what was just imported searchable at once: VictoriaMetrics shows new rows only after its
+// buffers flush, and a gate that queries right after the import would otherwise see a partial history.
+func (c *backfillConfig) flush(ctx context.Context) error {
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.hotURL+"/internal/force_flush", http.NoBody)
+	if err != nil {
+		return err
+	}
+	resp, err := c.hc.Do(req)
+	if err != nil {
+		return err
+	}
+	_ = resp.Body.Close() // nothing in the reply matters
+	return nil
+}
+
+// longSamples builds the hourly series from end back months months and the 5-minute series from end back days.
+func longSamples(device, iface string, end time.Time, months, days int, rate map[string]float64) []metrics.Sample {
+	var out []metrics.Sample
+	labels := func(dir string, ws time.Time) []string {
+		return []string{"device", device, "iface", iface, "dir", dir, "v", rollup.VersionLabel(ws, 1)}
+	}
+	for dir, bps := range rate {
+		for h := end; h.After(end.AddDate(0, -months, 0)); h = h.Add(-time.Hour) {
+			out = append(out, metrics.New(longq.MetricRateAvg1h, bps, h, labels(dir, h)...),
+				metrics.New(longq.MetricRateMax1h, bps*1.25, h, labels(dir, h)...))
+		}
+		for t := end; t.After(end.AddDate(0, 0, -days)); t = t.Add(-5 * time.Minute) {
+			out = append(out, metrics.New(longq.MetricRate5m, bps, t, labels(dir, t)...))
+		}
+	}
+	return out
+}

cmd/nmon-devtools/fleet.go

@@ -9,14 +9,14 @@ import (
 
 	"github.com/v10networks/nmon/internal/fixturegen"
 	"github.com/v10networks/nmon/internal/profile"
-	"github.com/v10networks/nmon/internal/simagent"
+	"github.com/v10networks/nmon/internal/snmprec"
 )
 
 // demoDevice is one simulated device of `devtools seed-demo`.
 type demoDevice struct {
 	name, site, role string
 	kind             string // counters, big, router, ups, v1, down: what the gate and screenshots look for
-	rec              *simagent.Recording
+	rec              *snmprec.Recording
 	community        string
 	version          string // v2c unless set
 	hc               bool   // 64-bit counters, so history can be back-filled
@@ -36,8 +36,8 @@ const (
 // animate returns a copy of rec whose 64-bit octet counters advance at a steady
 // per-port rate from counterBase and whose uptime ticks, so successive polls see
 // traffic. Counters that already advance keep their rate and only gain the base.
-func animate(rec *simagent.Recording) *simagent.Recording {
-	recs := append([]simagent.Record(nil), rec.Records()...)
+func animate(rec *snmprec.Recording) *snmprec.Recording {
+	recs := append([]snmprec.Record(nil), rec.Records()...)
 	for i := range recs {
 		r := &recs[i]
 		switch {
@@ -45,14 +45,14 @@ func animate(rec *simagent.Recording) *simagent.Recording {
 			recs[i] = animateCounter(*r)
 		case r.OID == oidUp && r.Dyn == nil:
 			if v, ok := r.Value.(uint32); ok {
-				r.Dyn, r.Value = &simagent.Dynamic{Rate: 100, Initial: uint64(v)}, nil
+				r.Dyn, r.Value = &snmprec.Dynamic{Rate: 100, Initial: uint64(v)}, nil
 			}
 		}
 	}
-	return simagent.NewRecording(recs)
+	return snmprec.NewRecording(recs)
 }
 
-func animateCounter(r simagent.Record) simagent.Record {
+func animateCounter(r snmprec.Record) snmprec.Record {
 	if r.Dyn != nil {
 		d := *r.Dyn
 		d.Initial += counterBase
@@ -65,7 +65,7 @@ func animateCounter(r simagent.Record) simagent.Record {
 		dir = 2
 	}
 	mbps := 1 + (k*(31+7*dir))%90
-	r.Dyn = &simagent.Dynamic{Rate: float64(mbps) * 125_000, Initial: counterBase + uint64(k)*1_000_000_000}
+	r.Dyn = &snmprec.Dynamic{Rate: float64(mbps) * 125_000, Initial: counterBase + uint64(k)*1_000_000_000}
 	r.Value = nil
 	return r
 }
@@ -95,7 +95,7 @@ func profileDevice(set *profile.Set, prof, name, site, role, kind string, ports
 // routers, a firewall, a UPS, one 32-bit device, the counters fixture and one
 // device whose agent is stopped after seeding, so it reads as down.
 func buildFleet(set *profile.Set, countersRec string) ([]demoDevice, error) {
-	counters, err := simagent.LoadRecording(countersRec)
+	counters, err := snmprec.Load(countersRec)
 	if err != nil {
 		return nil, err
 	}

cmd/nmon-devtools/gatedump.go

@@ -0,0 +1,163 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"time"
+)
+
+// gateDumpCmd prints what the M4 gate asserts on, one line per item, so the shell
+// script can count and match lines instead of parsing JSON.
+//
+//	gate-dump sink --file F          what the webhook sink received, in arrival order
+//	gate-dump alerts --api URL       every alert, newest first
+//	gate-dump devices --api URL      every device's reachability
+func gateDumpCmd(ctx context.Context, args []string) error {
+	if len(args) == 0 {
+		return fmt.Errorf("gate-dump sink|alerts|devices")
+	}
+	fs := devFlagSet("gate-dump " + args[0])
+	file := fs.String("file", "", "sink file (sink)")
+	api := fs.String("api", "", "nmon API base URL (alerts, devices)")
+	mapFile := fs.String("map", "", "slug map written by topology-add; slugs are printed as topology names")
+	if err := fs.Parse(args[1:]); err != nil {
+		return flagErr(err)
+	}
+	names, err := readSlugMap(*mapFile)
+	if err != nil {
+		return err
+	}
+	switch args[0] {
+	case "sink":
+		return dumpSink(*file, names)
+	case "alerts":
+		return dumpAlerts(ctx, *api, names)
+	case "devices":
+		return dumpDevices(ctx, *api, names)
+	}
+	return fmt.Errorf("gate-dump: unknown %q", args[0])
+}
+
+func orDash(s []string) string {
+	if len(s) == 0 {
+		return "-"
+	}
+	return strings.Join(s, ",")
+}
+
+// dumpSink prints "rule state device parents=a,b children=N incident=ID" per request received.
+func dumpSink(file string, names slugNames) error {
+	f, err := os.Open(file)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = f.Close() }() // read only
+	sc := bufio.NewScanner(f)
+	sc.Buffer(make([]byte, 1<<20), 1<<20)
+	for sc.Scan() {
+		var l sinkLine
+		var n struct {
+			Rule, State        string
+			IncidentID         string `json:"incident_id"`
+			Device             struct{ Slug string }
+			Parents            []string
+			SuppressedChildren int `json:"suppressed_children"`
+		}
+		if json.Unmarshal(sc.Bytes(), &l) != nil || json.Unmarshal(l.Body, &n) != nil {
+			fmt.Println("unparsable - - parents=- children=0 incident=-")
+			continue
+		}
+		fmt.Printf("%s %s %s parents=%s children=%d incident=%s\n", n.Rule, n.State, names.one(n.Device.Slug), orDash(names.of(n.Parents)),
+			n.SuppressedChildren, n.IncidentID)
+	}
+	return sc.Err()
+}
+
+func getJSON(ctx context.Context, api, path string, out any) error {
+	hc := &http.Client{Timeout: 30 * time.Second}
+	return httpJSON(ctx, hc, http.MethodGet, api+path, nil, out)
+}
+
+// dumpAlerts prints "id rule state device suppressed_by=a,b incident=ID acked=yes|no" per alert.
+func dumpAlerts(ctx context.Context, api string, names slugNames) error {
+	var page struct {
+		Items []struct {
+			ID, Rule, State string
+			Device          struct{ Slug string }
+			SuppressedBy    []string  `json:"suppressed_by"`
+			IncidentID      string    `json:"incident_id"`
+			Ack             *struct{} `json:"ack"`
+		} `json:"items"`
+	}
+	if err := getJSON(ctx, api, "/api/v1/alerts?per_page=1000", &page); err != nil {
+		return err
+	}
+	for _, a := range page.Items {
+		acked := "no"
+		if a.Ack != nil {
+			acked = "yes"
+		}
+		fmt.Printf("%s %s %s %s suppressed_by=%s incident=%s acked=%s\n", a.ID, a.Rule, a.State, names.one(a.Device.Slug),
+			orDash(names.of(a.SuppressedBy)), orDash([]string{a.IncidentID}), acked)
+	}
+	return nil
+}
+
+// dumpDevices prints "slug reachability poll_state warnings=a,b" per device.
+func dumpDevices(ctx context.Context, api string, names slugNames) error {
+	var page struct {
+		Items []struct {
+			Slug, Reachability string
+			PollState          string   `json:"poll_state"`
+			Warnings           []string `json:"warnings"`
+		} `json:"items"`
+	}
+	if err := getJSON(ctx, api, "/api/v1/devices?per_page=1000", &page); err != nil {
+		return err
+	}
+	for _, d := range page.Items {
+		fmt.Printf("%s %s %s warnings=%s\n", url.PathEscape(names.one(d.Slug)), d.Reachability, d.PollState, orDash(d.Warnings))
+	}
+	return nil
+}
+
+// slugNames maps nmon slugs back to the names in the topology file.
+type slugNames map[string]string
+
+func readSlugMap(path string) (slugNames, error) {
+	m := slugNames{}
+	if path == "" {
+		return m, nil
+	}
+	b, err := os.ReadFile(path)
+	if err != nil {
+		return nil, err
+	}
+	for _, l := range strings.Split(string(b), "\n") {
+		if f := strings.Fields(l); len(f) == 2 {
+			m[f[0]] = f[1]
+		}
+	}
+	return m, nil
+}
+
+func (m slugNames) one(slug string) string {
+	if n, ok := m[slug]; ok {
+		return n
+	}
+	return slug
+}
+
+func (m slugNames) of(slugs []string) []string {
+	out := make([]string, len(slugs))
+	for i, s := range slugs {
+		out[i] = m.one(s)
+	}
+	return out
+}

cmd/nmon-devtools/m5_test.go

@@ -0,0 +1,59 @@
+package main
+
+import (
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestRebaseInfluxMakesTheNewestLineNowAndKeepsSpacing(t *testing.T) {
+	in := "m,host=a v=1 1000000000000000000\n# c\nm,host=a v=2 1000000060000000000\n"
+	now := time.Unix(2_000_000_000, 0)
+	out, n, err := rebaseInflux(in, now)
+	if err != nil || n != 2 {
+		t.Fatalf("n=%d err=%v", n, err)
+	}
+	want := "m,host=a v=1 1999999940000000000\nm,host=a v=2 2000000000000000000\n"
+	if out != want {
+		t.Fatalf("got %q want %q", out, want)
+	}
+	if _, _, err := rebaseInflux("\n# only comments\n", now); err == nil {
+		t.Fatal("empty input accepted")
+	}
+	if !strings.Contains(out, "host=a") {
+		t.Fatal("tags lost")
+	}
+}
+
+func TestProxmoxMockAuthShape(t *testing.T) {
+	for h, want := range map[string]bool{
+		"PVEAPIToken=root@pam!nmon=s3cret": true, "PVEAPIToken=root@pam!nmon=": false, "Bearer x": false, "": false,
+	} {
+		if validPVEAuth(h, "") != want {
+			t.Errorf("%q: want %v", h, want)
+		}
+	}
+	if validPVEAuth("PVEAPIToken=a@b!c=other", "s3cret") {
+		t.Error("wrong secret accepted")
+	}
+}
+
+func TestLongSamplesCarryVersionedHourlyAndFiveMinuteSeries(t *testing.T) {
+	end := time.Date(2026, 10, 20, 12, 0, 0, 0, time.UTC)
+	s := longSamples("d", "I", end, 1, 1, map[string]float64{"in": 8e6})
+	var h, f int
+	for _, x := range s {
+		switch x.Name {
+		case "nmon_if_rate_bps_5m":
+			f++
+		default:
+			h++
+		}
+	}
+	if f != 288 || h < 2*28*24 {
+		t.Fatalf("5m=%d hourly=%d", f, h)
+	}
+	if s[0].Labels[3].Name != "v" || s[0].Labels[3].Value != "202610001" {
+		t.Fatalf("labels %+v", s[0].Labels)
+	}
+}

cmd/nmon-devtools/main.go

@@ -18,7 +18,9 @@ import (
 	"time"
 )
 
-var errUsage = errors.New("usage: nmon-devtools backfill|seed-demo [flags] (see docs/dev.md)")
+var errUsage = errors.New("usage: nmon-devtools backfill|seed-demo|topology-agents|topology-add|webhook-sink|gate-dump|" +
+	"proxmox-mock|push-replay|backfill-long [flags]" +
+	" (see docs/dev.md)")
 
 func main() {
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
@@ -38,6 +40,20 @@ func run(ctx context.Context, args []string) error {
 		return backfillCmd(ctx, args[1:])
 	case "seed-demo":
 		return seedDemoCmd(ctx, args[1:])
+	case "topology-agents":
+		return topologyAgentsCmd(ctx, args[1:])
+	case "topology-add":
+		return topologyAddCmd(ctx, args[1:])
+	case "webhook-sink":
+		return webhookSinkCmd(ctx, args[1:])
+	case "gate-dump":
+		return gateDumpCmd(ctx, args[1:])
+	case "proxmox-mock":
+		return proxmoxMockCmd(ctx, args[1:])
+	case "push-replay":
+		return pushReplayCmd(ctx, args[1:])
+	case "backfill-long":
+		return backfillLongCmd(ctx, args[1:])
 	default:
 		return fmt.Errorf("unknown command %q: %w", args[0], errUsage)
 	}

cmd/nmon-devtools/proxmoxmock.go

@@ -0,0 +1,75 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"os"
+	"strings"
+	"time"
+)
+
+// proxmoxMockCmd serves a recorded cluster/resources response the way a Proxmox VE API does, for the M5 gate:
+// GET /api2/json/cluster/resources with "Authorization: PVEAPIToken=<id>=<secret>". Any other path is a 404 and
+// a request without that header shape is a 401, so a client that forgot the secret fails like the real thing.
+func proxmoxMockCmd(ctx context.Context, args []string) error {
+	fs := devFlagSet("proxmox-mock")
+	listen := fs.String("listen", "[::1]:0", "listen address")
+	file := fs.String("file", "", "recorded cluster/resources JSON (testdata/proxmox/cluster-resources.json)")
+	secret := fs.String("secret", "", "token secret to require (empty: any non-empty secret)")
+	portFile := fs.String("port-file", "", "file that receives the bound port")
+	if err := fs.Parse(args); err != nil {
+		return flagErr(err)
+	}
+	body, err := os.ReadFile(*file)
+	if err != nil || *file == "" {
+		return fmt.Errorf("proxmox-mock needs --file: %w", errors.Join(err, errUsage))
+	}
+	ln, err := net.Listen("tcp", *listen)
+	if err != nil {
+		return err
+	}
+	if err := writePort(*portFile, ln); err != nil {
+		return err
+	}
+	mux := http.NewServeMux()
+	mux.HandleFunc("GET /api2/json/cluster/resources", func(w http.ResponseWriter, r *http.Request) {
+		if !validPVEAuth(r.Header.Get("Authorization"), *secret) {
+			http.Error(w, `{"data":null}`, http.StatusUnauthorized)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write(body) // a client that hung up has nobody to tell
+	})
+	srv := &http.Server{Handler: mux, ReadHeaderTimeout: 10 * time.Second}
+	go func() { <-ctx.Done(); _ = srv.Close() }()
+	fmt.Printf("proxmox-mock listening on %s\n", ln.Addr())
+	if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) {
+		return err
+	}
+	return nil
+}
+
+// validPVEAuth accepts "PVEAPIToken=user@realm!id=secret" with the wanted secret (or any, when want is empty).
+func validPVEAuth(h, want string) bool {
+	rest, ok := strings.CutPrefix(h, "PVEAPIToken=")
+	if !ok {
+		return false
+	}
+	_, got, ok := strings.Cut(rest, "=")
+	return ok && got != "" && (want == "" || got == want)
+}
+
+// writePort records the bound port for a shell script to read.
+func writePort(file string, ln net.Listener) error {
+	if file == "" {
+		return nil
+	}
+	tcp, ok := ln.Addr().(*net.TCPAddr)
+	if !ok {
+		return fmt.Errorf("listener is not TCP: %s", ln.Addr())
+	}
+	return os.WriteFile(file, []byte(fmt.Sprint(tcp.Port)), 0o600)
+}

cmd/nmon-devtools/pushreplay.go

@@ -0,0 +1,95 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// pushReplayCmd sends a recorded Influx line-protocol file to a running nmon's POST /write, shifting every
+// timestamp so the newest line is "now": a recording is days old by the time a gate replays it, and ingest
+// refuses timestamps far from the clock.
+func pushReplayCmd(ctx context.Context, args []string) error {
+	fs := devFlagSet("push-replay")
+	url := fs.String("url", "", "ingest URL, for example http://127.0.0.1:8443/write")
+	token := fs.String("token", "", "push token secret (omit to test the unauthenticated case)")
+	file := fs.String("file", "", "Influx line-protocol file")
+	extra := fs.String("append", "", "one more line to send after the file (a raw line, rebased like the others)")
+	if err := fs.Parse(args); err != nil {
+		return flagErr(err)
+	}
+	raw, err := os.ReadFile(*file)
+	if err != nil || *url == "" {
+		return fmt.Errorf("push-replay needs --url and --file: %w", errors.Join(err, errUsage))
+	}
+	body, n, err := rebaseInflux(string(raw)+"\n"+*extra, time.Now())
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, *url, bytes.NewReader([]byte(body)))
+	if err != nil {
+		return err
+	}
+	if *token != "" {
+		req.Header.Set("Authorization", "Bearer "+*token)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = resp.Body.Close() }() // body read below
+	msg, _ := io.ReadAll(io.LimitReader(resp.Body, 500))
+	fmt.Printf("%d lines -> HTTP %d %s\n", n, resp.StatusCode, strings.TrimSpace(string(msg)))
+	if resp.StatusCode/100 != 2 {
+		return fmt.Errorf("push refused: HTTP %d", resp.StatusCode)
+	}
+	return nil
+}
+
+// rebaseInflux shifts the nanosecond timestamp at the end of every line by the same amount so the newest one
+// equals now. Lines without a timestamp, blank lines and comments pass through.
+func rebaseInflux(text string, now time.Time) (string, int, error) {
+	lines := strings.Split(text, "\n")
+	var newest int64
+	for _, l := range lines {
+		if ts, ok := lineTS(l); ok {
+			newest = max(newest, ts)
+		}
+	}
+	shift := now.UnixNano() - newest
+	var out strings.Builder
+	n := 0
+	for _, l := range lines {
+		l = strings.TrimSpace(l)
+		if l == "" || strings.HasPrefix(l, "#") {
+			continue
+		}
+		n++
+		if ts, ok := lineTS(l); ok {
+			l = l[:strings.LastIndexByte(l, ' ')+1] + strconv.FormatInt(ts+shift, 10)
+		}
+		out.WriteString(l + "\n")
+	}
+	if n == 0 {
+		return "", 0, errors.New("no lines to send")
+	}
+	return out.String(), n, nil
+}
+
+// lineTS returns the trailing nanosecond timestamp of an Influx line, if it has one.
+func lineTS(l string) (int64, bool) {
+	l = strings.TrimSpace(l)
+	i := strings.LastIndexByte(l, ' ')
+	if l == "" || strings.HasPrefix(l, "#") || i < 0 {
+		return 0, false
+	}
+	ts, err := strconv.ParseInt(l[i+1:], 10, 64)
+	return ts, err == nil && ts > 1e15
+}

cmd/nmon-devtools/topology.go

@@ -0,0 +1,219 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/v10networks/nmon/internal/deps"
+	"github.com/v10networks/nmon/internal/fixturegen"
+	"github.com/v10networks/nmon/internal/simagent"
+)
+
+// Topology commands for the M4 gate: one simulated SNMP agent per device of a
+// topology file (deps.Topology), the devices added to a running nmon with their
+// parents, and a webhook receiver that records what nmon sends.
+
+const agentCommunity = "public"
+
+// topologyPort is the UDP port of the i-th device of a topology.
+func topologyPort(base, i int) int { return base + i }
+
+// topologyAgentsCmd runs one agent per device on [::1]:port-base+index. The
+// agent of device X is down while the file <control-dir>/stop-X exists, so a
+// shell script can take a device away and bring it back on the same port.
+func topologyAgentsCmd(ctx context.Context, args []string) error {
+	fs := devFlagSet("topology-agents")
+	file := fs.String("topology", "", "topology file (testdata/topologies/*.yaml)")
+	base := fs.Int("port-base", 0, "UDP port of the first device; device i uses port-base+i")
+	ctl := fs.String("control-dir", "", "directory watched for stop-<slug> files")
+	if err := fs.Parse(args); err != nil {
+		return flagErr(err)
+	}
+	if *file == "" || *base == 0 || *ctl == "" {
+		return fmt.Errorf("topology-agents needs --topology, --port-base and --control-dir")
+	}
+	topo, err := deps.LoadTopology(*file)
+	if err != nil {
+		return err
+	}
+	f := &agentFleet{topo: topo, base: *base, ctl: *ctl, agents: map[string]*simagent.Agent{}}
+	defer f.closeAll()
+	for {
+		if err := f.reconcile(ctx); err != nil {
+			return err
+		}
+		if !f.announced {
+			f.announced = true
+			fmt.Printf("agents ready: %d devices on [::1]:%d-%d\n", len(topo.Devices), *base, topologyPort(*base, len(topo.Devices)-1))
+		}
+		if !sleepCtx(ctx, 200*time.Millisecond) {
+			return nil
+		}
+	}
+}
+
+type agentFleet struct {
+	topo      *deps.Topology
+	base      int
+	ctl       string
+	agents    map[string]*simagent.Agent
+	announced bool
+}
+
+// reconcile starts the agents whose stop file is gone and stops those whose stop file appeared.
+func (f *agentFleet) reconcile(ctx context.Context) error {
+	for i, d := range f.topo.Devices {
+		_, stopped := os.Stat(filepath.Join(f.ctl, "stop-"+d.Slug))
+		want := stopped != nil // no stop file: the agent should run
+		a := f.agents[d.Slug]
+		switch {
+		case want && a == nil:
+			rec := fixturegen.Generic(fixturegen.Options{Ports: 4, HCCounters: true, Counters: true, SysName: d.Slug})
+			cfg := simagent.Config{Addr: fmt.Sprintf("[::1]:%d", topologyPort(f.base, i)), Recording: rec, Community: agentCommunity}
+			ag, err := simagent.Start(ctx, cfg)
+			if err != nil {
+				return fmt.Errorf("agent %s: %w", d.Slug, err)
+			}
+			f.agents[d.Slug] = ag
+		case !want && a != nil:
+			_ = a.Close() // best-effort: the port is released either way
+			delete(f.agents, d.Slug)
+		}
+	}
+	return nil
+}
+
+func (f *agentFleet) closeAll() {
+	for _, a := range f.agents {
+		_ = a.Close() // best-effort at exit
+	}
+}
+
+// topologyAddCmd adds every device of a topology to a running nmon, pointing at
+// its agent, then sets the parents by slug.
+func topologyAddCmd(ctx context.Context, args []string) error {
+	fs := devFlagSet("topology-add")
+	api := fs.String("api", "", "nmon API base URL")
+	file := fs.String("topology", "", "topology file")
+	base := fs.Int("port-base", 0, "UDP port of the first device's agent")
+	mapOut := fs.String("map-out", "", "file that receives one \"<nmon slug> <topology name>\" line per device")
+	if err := fs.Parse(args); err != nil {
+		return flagErr(err)
+	}
+	if *api == "" || *file == "" || *base == 0 {
+		return fmt.Errorf("topology-add needs --api, --topology and --port-base")
+	}
+	topo, err := deps.LoadTopology(*file)
+	if err != nil {
+		return err
+	}
+	hc := &http.Client{Timeout: 90 * time.Second}
+	slugOf := map[string]string{} // topology name -> the slug nmon generated
+	for i, d := range topo.Devices {
+		body := map[string]any{
+			"name": d.Slug, "addresses": []string{"::1"}, "role": d.Role, "kind": d.Kind,
+			"snmp": map[string]any{"version": "v2c", "community": agentCommunity, "port": topologyPort(*base, i)},
+		}
+		var got struct {
+			Slug string `json:"slug"`
+		}
+		if err := httpJSON(ctx, hc, http.MethodPost, *api+"/api/v1/devices", body, &got); err != nil {
+			return fmt.Errorf("add %s: %w", d.Slug, err)
+		}
+		slugOf[d.Slug] = got.Slug // nmon appends a random suffix to the slug it derives from the name
+	}
+	for _, d := range topo.Devices {
+		if len(d.Parents) == 0 {
+			continue
+		}
+		parents := make([]string, len(d.Parents))
+		for i, p := range d.Parents {
+			parents[i] = slugOf[p]
+		}
+		err := httpJSON(ctx, hc, http.MethodPut, *api+"/api/v1/devices/"+url.PathEscape(slugOf[d.Slug]), map[string]any{"parents": parents}, nil)
+		if err != nil {
+			return fmt.Errorf("parents of %s: %w", d.Slug, err)
+		}
+	}
+	fmt.Printf("topology-add ready: %d devices\n", len(topo.Devices))
+	return writeSlugMap(*mapOut, topo, slugOf)
+}
+
+// writeSlugMap records which nmon slug belongs to which topology name, for gate-dump --map and the gate script.
+func writeSlugMap(path string, topo *deps.Topology, slugOf map[string]string) error {
+	if path == "" {
+		return nil
+	}
+	var b strings.Builder
+	for _, d := range topo.Devices {
+		fmt.Fprintf(&b, "%s %s\n", slugOf[d.Slug], d.Slug)
+	}
+	return os.WriteFile(path, []byte(b.String()), 0o600)
+}
+
+// sinkLine is one recorded request.
+type sinkLine struct {
+	At   string          `json:"at"`
+	Path string          `json:"path"`
+	Body json.RawMessage `json:"body"`
+}
+
+// webhookSinkCmd records every POST body as one JSON line in --out, answers 200,
+// and writes the bound port to --port-file once it listens.
+func webhookSinkCmd(ctx context.Context, args []string) error {
+	fs := devFlagSet("webhook-sink")
+	listen := fs.String("listen", "[::1]:0", "listen address")
+	out := fs.String("out", "", "file that receives one JSON line per request")
+	portFile := fs.String("port-file", "", "file that receives the bound port")
+	if err := fs.Parse(args); err != nil {
+		return flagErr(err)
+	}
+	if *out == "" {
+		return fmt.Errorf("webhook-sink needs --out")
+	}
+	f, err := os.OpenFile(*out, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = f.Close() }() // lines are written whole; nothing is buffered
+	ln, err := net.Listen("tcp", *listen)
+	if err != nil {
+		return err
+	}
+	if *portFile != "" {
+		port := ln.Addr().(*net.TCPAddr).Port
+		if err := os.WriteFile(*portFile, []byte(fmt.Sprint(port)), 0o600); err != nil {
+			return err
+		}
+	}
+	var mu sync.Mutex
+	srv := &http.Server{ReadHeaderTimeout: 10 * time.Second, Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, _ := io.ReadAll(io.LimitReader(r.Body, 1<<20))
+		raw := json.RawMessage(b)
+		if !json.Valid(b) {
+			raw, _ = json.Marshal(string(b))
+		}
+		line, _ := json.Marshal(sinkLine{At: time.Now().UTC().Format(time.RFC3339Nano), Path: r.URL.Path, Body: raw})
+		mu.Lock()
+		_, _ = f.Write(append(line, '\n'))
+		mu.Unlock()
+		w.WriteHeader(http.StatusOK)
+	})}
+	go func() { <-ctx.Done(); _ = srv.Close() }()
+	fmt.Printf("webhook-sink listening on %s\n", ln.Addr())
+	if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) {
+		return err
+	}
+	return nil
+}

cmd/nmon/profile_record.go

@@ -11,9 +11,8 @@ import (
 
 	"github.com/gosnmp/gosnmp"
 
-	"github.com/v10networks/nmon/internal/fixturegen"
-	"github.com/v10networks/nmon/internal/simagent"
 	"github.com/v10networks/nmon/internal/snmp"
+	"github.com/v10networks/nmon/internal/snmprec"
 )
 
 // defaultRecordSubtrees is what `profile record` walks when --subtree is not given:
@@ -130,7 +129,7 @@ func recordDevice(ctx context.Context, o recordOpts, w io.Writer) (int, error) {
 		return 0, err
 	}
 	defer func() { _ = s.Close() }() // read-only session
-	var recs []simagent.Record
+	var recs []snmprec.Record
 	for _, root := range o.Subtrees {
 		err := s.Walk(ctx, root, o.MaxReps, func(vb snmp.Varbind) error {
 			if vb.Exception() {
@@ -146,19 +145,19 @@ func recordDevice(ctx context.Context, o recordOpts, w io.Writer) (int, error) {
 			return 0, fmt.Errorf("walk %s: %w", root, err)
 		}
 	}
-	rec := simagent.NewRecording(recs)
+	rec := snmprec.NewRecording(recs)
 	if rec.Len() == 0 {
 		return 0, fmt.Errorf("the device returned nothing under %v", o.Subtrees)
 	}
-	if err := fixturegen.WriteSnmprec(w, rec); err != nil {
+	if err := snmprec.Write(w, rec); err != nil {
 		return 0, err
 	}
 	return rec.Len(), nil
 }
 
 // redact turns a varbind into a recording row, replacing secrets with REDACTED.
-func redact(vb snmp.Varbind, community string) simagent.Record {
-	r := simagent.Record{OID: vb.OID, Type: vb.Type, Value: vb.Value}
+func redact(vb snmp.Varbind, community string) snmprec.Record {
+	r := snmprec.Record{OID: vb.OID, Type: vb.Type, Value: vb.Value}
 	if redactedOIDs[vb.OID] {
 		r.Type, r.Value = gosnmp.OctetString, []byte(redacted)
 		return r

cmd/nmon/profile_test.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/v10networks/nmon/internal/profile"
 	"github.com/v10networks/nmon/internal/simagent"
+	"github.com/v10networks/nmon/internal/snmprec"
 )
 
 func bundledMIBs(t *testing.T) string {
@@ -121,7 +122,7 @@ const recordSample = `1.3.6.1.2.1.1.1.0|4|Test device
 `
 
 func TestProfileRecordRedacts(t *testing.T) {
-	rec, err := simagent.ParseRecording(strings.NewReader(recordSample))
+	rec, err := snmprec.Parse(strings.NewReader(recordSample))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -148,7 +149,7 @@ func TestProfileRecordRedacts(t *testing.T) {
 		if n := strings.Count(s, "REDACTED"); n != 4 {
 			t.Errorf("v%s: want 4 REDACTED values, got %d:\n%s", ver, n, s)
 		}
-		got, err := simagent.ParseRecording(strings.NewReader(s))
+		got, err := snmprec.Parse(strings.NewReader(s))
 		if err != nil {
 			t.Fatalf("v%s: output does not parse: %v", ver, err)
 		}
@@ -188,7 +189,7 @@ func TestProfileTestArgs(t *testing.T) {
 }
 
 func TestProfileTestPrintsMetrics(t *testing.T) {
-	rec, err := simagent.LoadRecording("../../testdata/snmprec/juniper-junos.snmprec")
+	rec, err := snmprec.Load("../../testdata/snmprec/juniper-junos.snmprec")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -222,3 +223,20 @@ func TestNmonDoesNotImportLibreNMS(t *testing.T) {
 		}
 	}
 }
+
+// Dev and test machinery stays out of the product binary: nmon-devtools and tests own the simulator,
+// the fixture generator and the fixtures. `nmon profile record` needs only internal/snmprec.
+func TestNmonDoesNotLinkDevMachinery(t *testing.T) {
+	out, err := exec.Command("go", "list", "-deps", ".").CombinedOutput()
+	if err != nil {
+		t.Skipf("go list unavailable: %v\n%s", err, out)
+	}
+	banned := []string{"internal/simagent", "internal/fixturegen", "internal/fixtures", "internal/librenms", "internal/testutil"}
+	for _, pkg := range strings.Fields(string(out)) {
+		for _, b := range banned {
+			if strings.HasSuffix(pkg, "/"+b) {
+				t.Errorf("cmd/nmon depends on %s", pkg)
+			}
+		}
+	}
+}

cmd/nmon/rollup.go

@@ -0,0 +1,108 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/v10networks/nmon/internal/cliout"
+	"github.com/v10networks/nmon/internal/rollup"
+)
+
+func init() {
+	register("rollup", "repair rollup windows (asks a running nmon)", rollupCmd)
+}
+
+const rollupUsage = "usage: nmon rollup repair <from> <to> [--api URL] [--token TOKEN]\n\n" +
+	"Re-runs every rollup window overlapping [from, to) on the running nmon, writing a new version of\n" +
+	"each. The range must be inside hot retention (default 30 days). Times are RFC 3339 (2026-10-20T08:00:00Z)\n" +
+	"or a UTC date (2026-10-20); to is exclusive."
+
+func rollupCmd(ctx context.Context, args []string) error {
+	if len(args) == 0 || args[0] != "repair" {
+		return fmt.Errorf("%s", rollupUsage)
+	}
+	fs := profileFlagSet("rollup repair", "rollup repair <from> <to> [--api URL]")
+	var api, token string
+	var timeout time.Duration
+	fs.StringVar(&api, "api", "http://[::1]:8443", "base URL of the running nmon")
+	fs.StringVar(&token, "token", os.Getenv("NMON_TOKEN"), "admin API token (default $NMON_TOKEN)")
+	fs.DurationVar(&timeout, "timeout", 30*time.Minute, "how long to wait for the repair")
+	pos, err := parseInterspersed(fs, args[1:])
+	if err != nil {
+		return flagErr(err)
+	}
+	if len(pos) != 2 {
+		return fmt.Errorf("%s", rollupUsage)
+	}
+	from, err := parseRepairTime(pos[0])
+	if err != nil {
+		return err
+	}
+	to, err := parseRepairTime(pos[1])
+	if err != nil {
+		return err
+	}
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+	rep, err := requestRepair(ctx, api, token, rollup.RepairRequest{From: from, To: to})
+	if err != nil {
+		return err
+	}
+	w := cliout.New(os.Stdout)
+	w.Printf("repaired %s to %s: %d five-minute windows, %d hourly windows rewritten\n",
+		rep.From.Format(time.RFC3339), rep.To.Format(time.RFC3339), rep.FiveMinutes, rep.Hours)
+	return w.Err()
+}
+
+func parseRepairTime(s string) (time.Time, error) {
+	for _, layout := range []string{time.RFC3339, "2006-01-02T15:04", "2006-01-02"} {
+		if t, err := time.ParseInLocation(layout, s, time.UTC); err == nil {
+			return t.UTC(), nil
+		}
+	}
+	return time.Time{}, fmt.Errorf("cannot parse %q as RFC 3339 or YYYY-MM-DD", s)
+}
+
+// requestRepair calls POST /api/v1/rollup/repair, which internal/app mounts on
+// rollup.Job.Repair.
+func requestRepair(ctx context.Context, api, token string, rr rollup.RepairRequest) (*rollup.RepairReport, error) {
+	body, err := json.Marshal(rr)
+	if err != nil {
+		return nil, err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(api, "/")+"/api/v1/rollup/repair", bytes.NewReader(body))
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	if token != "" {
+		req.Header.Set("Authorization", "Bearer "+token)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("cannot reach nmon at %s (is it running? use --api): %w", api, err)
+	}
+	defer func() { _ = resp.Body.Close() }() // body fully read below; a close error changes nothing
+	out, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
+	if resp.StatusCode != http.StatusOK {
+		var p struct{ Title, Detail string }
+		_ = json.Unmarshal(out, &p)
+		msg := p.Detail
+		if msg == "" {
+			msg = p.Title
+		}
+		return nil, fmt.Errorf("repair refused: %d %s", resp.StatusCode, msg)
+	}
+	var rep rollup.RepairReport
+	if err := json.Unmarshal(out, &rep); err != nil {
+		return nil, err
+	}
+	return &rep, nil
+}

cmd/nmon/rollup_test.go

@@ -0,0 +1,48 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/v10networks/nmon/internal/rollup"
+)
+
+func TestRollupRepairSendsRangeAndTokenToRunningNmon(t *testing.T) {
+	var got rollup.RepairRequest
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/api/v1/rollup/repair" || r.Header.Get("Authorization") != "Bearer tok" {
+			http.Error(w, "bad request", http.StatusBadRequest)
+			return
+		}
+		_ = json.NewDecoder(r.Body).Decode(&got)
+		_ = json.NewEncoder(w).Encode(rollup.RepairReport{From: got.From, To: got.To, FiveMinutes: 12, Hours: 1})
+	}))
+	defer srv.Close()
+	from, to := time.Date(2026, 10, 20, 8, 0, 0, 0, time.UTC), time.Date(2026, 10, 20, 9, 0, 0, 0, time.UTC)
+	rep, err := requestRepair(context.Background(), srv.URL, "tok", rollup.RepairRequest{From: from, To: to})
+	if err != nil || rep.FiveMinutes != 12 || !got.From.Equal(from) || !got.To.Equal(to) {
+		t.Fatalf("rep=%+v got=%+v err=%v", rep, got, err)
+	}
+	if _, err := requestRepair(context.Background(), srv.URL, "wrong", rollup.RepairRequest{}); err == nil {
+		t.Error("a refused repair must be an error")
+	}
+}
+
+func TestParseRepairTimeAcceptsDatesAndRFC3339(t *testing.T) {
+	for in, want := range map[string]time.Time{
+		"2026-10-20":           time.Date(2026, 10, 20, 0, 0, 0, 0, time.UTC),
+		"2026-10-20T08:30":     time.Date(2026, 10, 20, 8, 30, 0, 0, time.UTC),
+		"2026-10-20T08:00:00Z": time.Date(2026, 10, 20, 8, 0, 0, 0, time.UTC),
+	} {
+		if got, err := parseRepairTime(in); err != nil || !got.Equal(want) {
+			t.Errorf("%s: %v %v", in, got, err)
+		}
+	}
+	if _, err := parseRepairTime("yesterday"); err == nil {
+		t.Error("want error")
+	}
+}

cmd/nmon/serve.go

@@ -62,6 +62,10 @@ func serve(ctx context.Context, args []string) error {
 	fs.StringVar(&f.TrapListen, "trap-listen", app.DefaultTrapListen,
 		"UDP address for SNMP traps; empty disables the receiver "+
 			"(port 162 needs CAP_NET_BIND_SERVICE; without it traps are off and polling continues)")
+	fs.StringVar(&f.GraphiteListen, "graphite-listen", app.DefaultGraphiteListen,
+		"TCP address for Graphite plaintext push (first line \"token <secret>\"); empty disables the listener")
+	fs.StringVar(&f.ICMP, "icmp", config.ICMPAuto,
+		"ICMP probing: auto, or off for hosts where ICMP is not permitted (SNMP alone then decides if a device is down)")
 	fs.IntVar(&f.MaxInflight, "max-inflight", 512, "most SNMP polls in flight at once")
 	fs.DurationVar(&f.PollInterval, "poll-interval", 0, "dev: SNMP poll interval (default 60s)")
 	fs.IntVar(&f.SNMPPort, "snmp-port", 0, "dev: UDP port of every device's agent (default 161)")

docs/DEVIATIONS.md

@@ -16,8 +16,22 @@ conscious, reviewed exception.
 | 9 | `nmon profile import-librenms <repo-path>` is an nmon subcommand | A separate binary, `nmon-librenms-import` (`cmd/nmon-librenms-import`), run offline before nmon. It writes draft profiles that the operator reviews and copies into `<data-dir>/profiles/`. `nmon` neither links nor exposes it. | User decision (2026-10-22). It keeps the importer out of the product binary and makes "the converter never runs at serve time" structural. | M2 |
 | 10 | `scripts/openapi-diff` fails on an incompatible change (M3 milestone) | The differ is the Go package `internal/openapidiff` (unit-tested, run with `go run ./internal/openapidiff/cmd/openapidiff`), driven by `scripts/openapi-check.sh`, which `make gate-m3` and the CI `openapi` job call. | A Go package is testable and linted like the rest of the code; the script only fetches the spec at the last `m*`/`v*` tag. No new dependency. | M3 |
 | 11 | Built assets are embedded from `web/dist` into `internal/webui` | `make web` copies `web/dist` to `internal/webui/dist` (git-ignored, `.gitkeep` tracked); a committed `internal/webui/placeholder/index.html` ("UI not built") is served when `dist` has no `index.html`. | A tracked `dist/index.html` would be overwritten by every build and dirty the tree, and a built UI must never be committed. `go build` still works without node. | M3 |
-| 12 | (not specified) | Developer tools (`backfill`, `seed-demo`) are a separate executable, `cmd/nmon-devtools` (`make devtools` builds `nmon-devtools`), never shipped. It talks to a running nmon over its HTTP API and to the hot store via `--hot-url`. `nmon` has no `devtools` command; a depguard rule keeps `cmd/nmon` from importing `internal/simagent` or `internal/fixturegen` beyond `profile_record.go`, which already used them for `nmon profile record` in M2. | The gate needs 30 days of history without waiting 30 days, and live screenshots need a simulated network. See `docs/dev.md`. | M3 |
+| 12 | (not specified) | Developer tools (`backfill`, `seed-demo`) are a separate executable, `cmd/nmon-devtools` (`make devtools` builds `nmon-devtools`), never shipped. It talks to a running nmon over its HTTP API and to the hot store via `--hot-url`. `nmon` has no `devtools` command; depguard and a `go list -deps` test keep `cmd/nmon` from linking `internal/simagent`, `internal/fixturegen`, `internal/fixtures` or `internal/testutil`; `nmon profile record` uses only the `.snmprec` format in `internal/snmprec`. | The gate needs 30 days of history without waiting 30 days, and live screenshots need a simulated network. See `docs/dev.md`. | M3 |
 | 13 | (not specified) `/docs` API reference | `/docs` serves a vendored, pinned swagger-ui-dist 5.17.14 (Apache-2.0, about 1.6 MB) through `embed.FS`. | huma's built-in renderers load their assets from a CDN, which an offline monitor cannot rely on. | M3 |
 | 14 | M2 defined provisional `poller/health` and `identity-reviews` responses | Both were reshaped in M3 to the reviewed UI contract, before any `openapi.json` existed, so the route freeze starts at `m3`, the first tag with a checked-in spec. | The M2 shapes had no consumer. `scripts/openapi-check.sh` passes against `m2` with a note and diffs for real from `m3`. | M3 |
 | 15 | A single-signal failure (ICMP only or SNMP only) is a warning | The warning is held as pending for 5 min before it notifies, and is closed silently if the device goes down meanwhile. | Otherwise a real outage whose second signal fails seconds later sends a spurious warning first. | M4 |
 | 16 | An unsupported probe is unknown, never failed | When ICMP is unsupported (host can't open unprivileged ICMP, or per device), SNMP alone decides down, as with `icmp_blocked`, and a one-time `icmp_unsupported` device warning is raised. | Otherwise nmon could never declare a device down on such a host, and nobody would notice. | M4 |
+| 17 | M6 gate: "the 2,000-device loadtest cycle finishes under 40 s"; the loadtest acceptance lines (200k interfaces in 40 s, memory stable over 30 min, peak RSS/disk/ingest/latency report, backup during load) | **Skipped** (approved by the owner, 2026-10-22: "skip the device load test… put this in prod asap"). The M6 gate checks only that a viewer gets 403 on PUT. Backup/restore is verified on a normal-size dataset instead of under load. Scale is unproven beyond the fixture and gate sizes (24 devices). | Time to production. | M6 |
+| 18 | Push token format not specified | The secret is `nmp_<id>_<base32 of 160 random bits>`; the row stores a salted SHA-256 (`<salt>$<hash>`). | The id selects the row, so authentication is one primary-key lookup and never scans or compares other tokens. | M5 |
+| 19 | `rollup_windows` status is `done` or `partial` | A third status, `pending`, is written before the long-store write begins. | A crash between the write and the commit still tells the next run which version number is taken, so it writes `v+1`. | M5 |
+| 20 | `v` is a version number on rollup series | `v` is on `nmon_if_rate_bps_5m`, `_avg_1h`, `_max_1h` and the gauge series, encoded `YYYYMM*1000+n` (UTC month of the window, then the version). | Numeric, so MetricsQL can compare versions; the month prefix lets retention drop whole months of 5 m series. Readers use `internal/longq.LatestVersion`. | M5 |
+| 21 | Gauge rollups: "a 5 m average, plus a 1 h avg, max and min" | Series are named `<name>_avg_5m`, `<name>_avg_1h`, `<name>_max_1h`, `<name>_min_1h`, with the source labels plus `v`. | The suffix keeps the source name searchable and cannot collide with a polled metric (such names are not rolled up). | M5 |
+| 22 | 5 m series are deleted past `retention.fiveminute` | Whole calendar months are deleted with `delete_series` once they ended before `now - retention.fiveminute`, so 5 m data lives 4 to 5 months. | Open-source VictoriaMetrics has one retention per process and no time-ranged delete; the `v` prefix makes each 5 m series one month. | M5 |
+| 23 | Rollup writes go through the write queue | Rollup long writes bypass the queue: a synchronous, acknowledged import, then `force_flush`, then a read-back count. | The queue is fire-and-forget (it can drop after 60 s); a window may be recorded done only once the store has it. | M5 |
+| 24 | Unreliable (32-bit) interfaces are excluded from p95 and group sums | They are also never rolled up, and a graph of one returns no series. | A rate from a wrapping counter is wrong whichever tier stores it. | M5 |
+| 25 | `nmon_group_p95_bps{group, dir, month, members_hash}` | Group (and interface) p95 records also carry a `coverage` label. | A figure below 90% coverage must never be shown without it. | M5 |
+| 26 | Check samples labelled by device | `nmon_check_up`, `nmon_check_latency_seconds` and `nmon_tls_expiry_seconds` also carry `check` (the check id). | A device can have several checks. | M5 |
+| 27 | Proxmox nodes are devices | A node is the existing SNMP device with that name (or sysName), else a new `check_only` device; its health is the status the API reports. Guests are `vm`/`container` devices keyed by `pve:<cluster>/<vmid>`. | `cluster/resources` has no addresses, so nothing but the name can match a node; a missing SNMP device must not orphan its guests. | M5 |
+| 28 | Pushed series carry only `device` | Pushed series also carry `push="1"` (senders cannot set it). | The device page lists what a push sender wrote without guessing from metric names. | M5 |
+| 29 | Ingest caps, late p95 months and a stalled rollup "raise an alert" | They are rows of the alerts table with rules `ingest_<kind>`, `p95_late`, `rollup_stalled` and no device, kept open until cleared (ingest alarms close 20 min after the last trip). | They are not metric rules and have no device; the metric-rule evaluator leaves these rule names alone. | M5 |
+| 30 | Proxmox cluster config could carry the token secret | Only `secret_name` exists; the secret is stored encrypted via `PUT /api/v1/secrets/{name}` and cannot be read back. | `config.yaml` is plain text. | M5 |

docs/milestones/M4-alerting.md

@@ -44,4 +44,13 @@ Alerting: all *Must* and *Must not* lines. Adversarial: redundant parents failin
 
 ## Gate log
 
-_(filled in by the supervisor when the gate passes)_
+**Passed 2026-10-23**, run by the supervisor (Opus).
+
+| Check | Result |
+|---|---|
+| `make lint` (Go + web) | 0 findings |
+| `go test -race ./...`, `-tags 'integration vmembed'`, `test-pg.sh` | all ok |
+| gates m1, m2 (LibreNMS recordings), m3, m4, m5 in one sequential run | all PASS |
+| `make gate-m4` | PASS: redundant topology (2 cores, 2 dual-homed distribution switches, 20 access switches) with `--icmp off`; (a) one core down → exactly one notice, nothing suppressed; (b) **both cores and all 22 children stopped at the same instant → exactly ONE notification** naming both cores and 22 suppressed children, zero per-child notices; (c) restart re-sends nothing; (d) core-1 back → one resolution, dist switches announced with their suppressed children; deliveries logged with redacted payloads; incident IDs stable |
+
+**Decisions in M4:** a dependency hold delays a child's notice while its parents are unsettled (proven over 100 randomised timing seeds), so simultaneous outages group correctly. Unsupported ICMP falls back to SNMP alone and shows an `icmp_unsupported` device warning. Single-signal warnings are held 5 min. `--icmp auto|off`. `alerts?state=open` (open alerts, acked or not) is used on device pages. That fix came from the gate's screenshot step, which caught an acked outage vanishing from its device page.

docs/milestones/M5-ingest-rollup-checks.md

@@ -45,4 +45,13 @@ Storage and rollups: all lines except the loadtest disk budget (M6). Ingest: all
 
 ## Gate log
 
-_(filled in by the supervisor when the gate passes)_
+**Passed 2026-10-23**, run by the supervisor (Opus).
+
+| Check | Result |
+|---|---|
+| `make lint` (Go + web) | 0 findings |
+| `go test -race ./...`, `-tags 'integration vmembed'`, `test-pg.sh` | all ok |
+| gates m1, m2 (LibreNMS recordings), m3, m4, m5 in one sequential run | all PASS |
+| `make gate-m5` | PASS: mock Proxmox synced (node pve1 + guests with hosted_on); `proxmox.influx` replayed via a push token → 93 series under pve1; **2 y graph hits the 1 h tier** (tiers `1h`+`raw`, `stitched_at`, long points exactly the back-filled values); p95 with `source`; group graph is the oriented sum; HTTP check up; ingest negatives (401s, reserved prefix dropped, series-cap alert deduped) |
+
+**Decisions in M5:** see DEVIATIONS rows 18–30 (push token format, rollup `pending` status and `v` encoding, gauge rollup names, month-partitioned 5-min retention, acked synchronous long writes, unreliable interfaces never rolled up, group `coverage` label, check `check` label, Proxmox node rule, `push="1"` label, system alarms, no inline Proxmox secret). One `internal/longq` package now owns the "highest version wins" query; this fixed M3's long-tier graph read.

docs/milestones/M6-auth-and-ops.md

@@ -1,6 +1,6 @@
 # M6: Auth and ops
 
-**Gate:** a viewer gets 403 on PUT, and the 2,000-device loadtest cycle finishes under 40 s.
+**Gate:** a viewer gets 403 on PUT. ~~and the 2,000-device loadtest cycle finishes under 40 s~~ (skipped: approved deviation, see `docs/DEVIATIONS.md`; the owner prioritised getting to production).
 **Gate command:** `make gate-m6`. It runs the viewer-fuzz test (every state-changing route with a viewer session and a viewer token → 403) and then `nmon loadtest --devices 2000 --ports 100 --duration 30m`, which must report cycle p100 < 40 s, zero deadline misses, write-queue high-water < 50%, and stable RSS. It also prints the report metrics (peak RSS, steady and peak disk, ingest samples/s, p95 device-page latency, write-queue high-water, backlog recovery after a 5 min store outage). Finally it runs backup-restore during load.
 
 ## Scope
@@ -32,7 +32,7 @@
 | WP1 OIDC, sessions, CSRF, tokens, break-glass, proxy, TLS | `internal/auth/**`, API middleware | M3 API |
 | WP2 Backup, restore, key rotate, disk guard, store restart | `internal/backup/**`, `internal/diskguard/**`, `cmd/nmon/{backup,restore,key,store,unlock}.go` | M1 vmsup, store |
 | WP3 Config reload, settings API and UI, CSV import, purge | `internal/config/**`, `internal/api/settings*.go`, `internal/importcsv/**`, `web/src/routes/settings*` | M4 deps |
-| WP4 Loadtest, viewer fuzz, gate | `internal/loadtest/**`, `cmd/nmon/loadtest.go`, `scripts/gate-m6.sh` | all |
+| WP4 Viewer fuzz, backup/restore check, gate (loadtest skipped, approved deviation) | `internal/loadtest/**`, `cmd/nmon/loadtest.go`, `scripts/gate-m6.sh` | all |
 
 ## Acceptance lines closed
 

docs/milestones/README.md

@@ -15,9 +15,9 @@ its **Gate log** is filled in when the gate passes. A milestone is done only whe
 | M1 | Skeleton and fixtures | `nmon serve` polls the counters fixture and `/api/v1/query` returns rising ifHCInOctets | [M1](M1-skeleton-and-fixtures.md) | **passed**, tagged `m1` |
 | M2 | Discovery and profiles | every fixture's expect.yaml passes, including v1 and v3, and the PDU budget holds | [M2](M2-discovery-and-profiles.md) | **passed**, tagged `m2` |
 | M3 | Graphs and UI | a port's in/out graph renders at 1 h, 24 h and 30 d from the counters fixture | [M3](M3-graphs-and-ui.md) | **passed**, tagged `m3` |
-| M4 | Alerting | stopping a parent fixture sends one notification naming it and its children | [M4](M4-alerting.md) | in progress (WP1 reach and deps, WP2 rules, WP3 notify done) |
-| M5 | Push ingest, rollup, checks | a replayed Proxmox sample shows under its node and a 2 y query hits the 1 h tier | [M5](M5-ingest-rollup-checks.md) | planned |
-| M6 | Auth and ops | a viewer gets 403 on PUT and the 2,000-device loadtest cycle finishes under 40 s | [M6](M6-auth-and-ops.md) | planned |
+| M4 | Alerting | stopping a parent fixture sends one notification naming it and its children | [M4](M4-alerting.md) | **passed**, tagged `m4` |
+| M5 | Push ingest, rollup, checks | a replayed Proxmox sample shows under its node and a 2 y query hits the 1 h tier | [M5](M5-ingest-rollup-checks.md) | **passed**, tagged `m5` |
+| M6 | Auth and ops | a viewer gets 403 on PUT (2,000-device loadtest skipped, approved deviation) | [M6](M6-auth-and-ops.md) | in progress |
 | M7 | gNMI streaming | after v1; not built in this run | — | out of scope |
 
 ## How the work is run
@@ -47,6 +47,8 @@ Departures from the spec are recorded in [`docs/DEVIATIONS.md`](../DEVIATIONS.md
 
 - **Deployment playbook**: install layout (binary, `/usr/share/nmon/mibs` sidecar, systemd unit granting port 162, key file location), TLS/ACME or proxy setup, IdP configuration, backups, and upgrades. Raised by the user on 2026-10-22; deliberately postponed.
 
+- **Web push notifications (v2)**: a `webpush` receiver kind in `internal/notify` (RFC 8291 payload encryption, RFC 8292 VAPID, standard-library crypto), per-user subscriptions (needs M6 users), a service worker in the UI, and push endpoints in the outbound allow-list. Out of v1 per the spec's non-goals ("alerts go to webhooks and email only"). Owner deferred it to v2 on 2026-10-22. Until then, a webhook receiver pointed at ntfy or Gotify gives phone push.
+
 ## Review artifacts
 
 - From M3 on, each UI milestone's gate captures Playwright screenshots of every page into `docs/screenshots/<milestone>/` for the user to review.

go.mod

@@ -3,6 +3,7 @@ module github.com/v10networks/nmon
 go 1.27.1
 
 require (
+	github.com/coreos/go-oidc/v3 v3.21.0
 	github.com/danielgtaylor/huma/v2 v2.39.1
 	github.com/go-sql-driver/mysql v1.10.1
 	github.com/gosnmp/gosnmp v1.46.0
@@ -12,7 +13,9 @@ require (
 	github.com/uptrace/bun/dialect/pgdialect v1.3.0
 	github.com/uptrace/bun/dialect/sqlitedialect v1.3.0
 	github.com/uptrace/bun/driver/pgdriver v1.3.0
+	golang.org/x/crypto v0.58.0
 	golang.org/x/net v0.61.0
+	golang.org/x/oauth2 v0.37.0
 	gopkg.in/yaml.v3 v3.0.1
 	modernc.org/sqlite v1.61.0
 )
@@ -21,6 +24,7 @@ require (
 	filippo.io/edwards25519 v1.2.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/dustin/go-humanize v1.1.0 // indirect
+	github.com/go-jose/go-jose/v4 v4.1.4 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/mattn/go-isatty v0.0.24 // indirect
@@ -32,9 +36,9 @@ require (
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	go.opentelemetry.io/otel v1.40.0 // indirect
 	go.opentelemetry.io/otel/trace v1.40.0 // indirect
-	golang.org/x/crypto v0.58.0 // indirect
 	golang.org/x/mod v0.42.0 // indirect
 	golang.org/x/sys v0.49.0 // indirect
+	golang.org/x/text v0.43.0 // indirect
 	mellium.im/sasl v0.3.2 // indirect
 	modernc.org/libc v1.79.0 // indirect
 	modernc.org/mathutil v1.7.1 // indirect

go.sum

@@ -2,6 +2,8 @@ filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
 filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/coreos/go-oidc/v3 v3.21.0 h1:wZo4Q9Pum8dYEj0eMUPrqR+kvuGkeUplbLpNCkBqoWM=
+github.com/coreos/go-oidc/v3 v3.21.0/go.mod h1:DYCf24+ncYi+XkIH97GY1+dqoRlbaSI26KVTCI9SrY4=
 github.com/danielgtaylor/huma/v2 v2.39.1 h1:0kwF4ltQoYZ+IU55VPy+BcGekzgF44R64daTGde1H+g=
 github.com/danielgtaylor/huma/v2 v2.39.1/go.mod h1:zcnQ38duIJ3VUHwFaBoZ6x8T+KN/mr33oyqxcj0HTug=
 github.com/dustin/go-humanize v1.1.0 h1:dbKTrvD0klcbBV/h4AWJdMuZogJACoMlvWIWZ5b2xWg=
@@ -10,6 +12,8 @@ github.com/fxamacker/cbor/v2 v2.9.2 h1:X4Ksno9+x3cz0TZv69ec1hxP/+tymuR8PXQJyDwfh
 github.com/fxamacker/cbor/v2 v2.9.2/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ=
 github.com/go-chi/chi/v5 v5.3.1 h1:3j4HZLGZQ3JpMCrPJF/Jl3mYJfWLKBfNJ6quurUGCf8=
 github.com/go-chi/chi/v5 v5.3.1/go.mod h1:R+tYY2hNuVUUjxoPtqUdgBqevM9s9njzkTLutVsOCto=
+github.com/go-jose/go-jose/v4 v4.1.4 h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA=
+github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
 github.com/go-sql-driver/mysql v1.10.1 h1:arlSnNLq6a5yxGxV7qg9lF4j0C+KwD6NbQyKr9QL6ME=
 github.com/go-sql-driver/mysql v1.10.1/go.mod h1:M+cqaI7+xxXGG9swrdeUIoPG3Y3KCkF0pZej+SK+nWk=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
@@ -71,10 +75,14 @@ golang.org/x/mod v0.42.0 h1:ICeL3t2L30ljSPHHxDnR4fqVcBt/zR2zJAb9juavmQg=
 golang.org/x/mod v0.42.0/go.mod h1:u5fzIrRInkL8MaO1cbRkUHT64v303YIT5KEqpNr7jRs=
 golang.org/x/net v0.61.0 h1:RR+6j/BTBCrggOLugxa+66R8FvoN6MHlzDlZIbN9YFw=
 golang.org/x/net v0.61.0/go.mod h1:WgDCOTH8iwtB66T4/GfI94AHo/hhdb7Hah9NDseQfiI=
+golang.org/x/oauth2 v0.37.0 h1:JUlcxA8oAtauLfiH8FX2/FkAWHAdi0QtGCGc+hofE98=
+golang.org/x/oauth2 v0.37.0/go.mod h1:IxwZNxUULJmpBFf9K/9NTMSIfZZuvuTy1gGxhigP/58=
 golang.org/x/sync v0.24.0 h1:iqiDrGoDgyuvAKrhLQthUEzU/R378N78gXbBjxed7oA=
 golang.org/x/sync v0.24.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0=
 golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA=
 golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
+golang.org/x/text v0.43.0 h1:1QivrlhwAsnMOcqgOdl9my7s9OyZl8O5/UGUTJVXoWc=
+golang.org/x/text v0.43.0/go.mod h1:hbSIYA/amXcRXTFSZqh8tWHEoqRvvbKPJ2uz/KU+a+s=
 golang.org/x/tools v0.52.0 h1:ryxg4U63oaUYk7uOLTF37n43f4qgtGvqV4Ioit1BaEs=
 golang.org/x/tools v0.52.0/go.mod h1:ItcP5UNaVGG+Yc4eIYmaFLNJ4soEopkx5Y2d4lcEurs=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

internal/alert/evaluator.go

@@ -197,6 +197,9 @@ func (ev *Evaluator) sweepRows(ctx context.Context, now time.Time, rows []store.
 	active := map[string][]*store.Alert{} // rule -> active rows
 	for i := range rows {
 		a := &rows[i]
+		if reachRule(a.Rule) {
+			continue // owned by the reachability evaluator, which shares the alerts table
+		}
 		ent := entityOf(a)
 		if !isActive(a.State) {
 			ev.deliver(ctx, now, a, ent) // closed row with an undelivered notification

internal/alert/evaluator_test.go

@@ -658,3 +658,12 @@ func TestMergeRules(t *testing.T) {
 	}
 	_ = fmt.Sprint
 }
+
+func TestReachabilityRowsInTheSameTableAreLeftAlone(t *testing.T) {
+	if err := (Rule{Name: "device_down", Severity: SevInfo, Op: OpGreater, Expr: "up", Labels: []string{"device"}}).Validate(); err == nil {
+		t.Fatal("a rule may not take a reachability rule's name")
+	}
+	if !reachRule("device_warning") || reachRule("cpu_high") {
+		t.Fatal("reachRule misclassifies")
+	}
+}

internal/alert/rules.go

@@ -67,6 +67,13 @@ type TemplateData struct {
 	For                                      time.Duration
 }
 
+// reachRule reports the rule names that other components own in the alerts table: the
+// reachability evaluator (internal/reach) and the subsystem alarms (SystemRule). This evaluator
+// must neither close nor deliver those rows, and no configured rule may take these names.
+func reachRule(name string) bool {
+	return name == "device_down" || name == "device_warning" || SystemRule(name)
+}
+
 var entityLabels = map[string]bool{"device": true, "iface": true, "component": true}
 
 // Validate checks a rule is usable: names, operator, label set, parseable
@@ -90,6 +97,9 @@ func (r Rule) Validate() error {
 	if r.For < 0 {
 		add("for is negative")
 	}
+	if reachRule(r.Name) {
+		add("name is reserved for reachability alerts")
+	}
 	r.validateLabels(add)
 	if err := checkExprShape(r.Expr); err != nil {
 		add("expr: %v", err)