B4 tests: probe store, agent prober, controller + wire

Baseline note: TestMetricHistoryMinuteFold fails identically on
the pristine B3 tip in the same wall-clock window (minute-parity
sensitive seed), passes in others — pre-existing, untouched path.

internal/agent/ping_test.go

@@ -0,0 +1,169 @@
+package agent
+
+import (
+	"encoding/json"
+	"net"
+	"testing"
+	"time"
+)
+
+// probeTCP against a live listener yields a real non-negative
+// latency; a closed port and a malformed target both file -1; 0 is
+// never a latency (contract: failures are negative or absent).
+func TestProbeTCPOutcomes(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ln.Close()
+	go func() {
+		for {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			conn.Close()
+		}
+	}()
+
+	ms, sampled := probeTCP(ln.Addr().String(), nil)
+	if !sampled || ms < 0 {
+		t.Fatalf("live target: ms=%d sampled=%v", ms, sampled)
+	}
+	ms, sampled = probeTCP("127.0.0.1:1", nil)
+	if !sampled || ms != -1 {
+		t.Fatalf("closed port: ms=%d sampled=%v", ms, sampled)
+	}
+	ms, sampled = probeTCP("not-a-target", nil)
+	if !sampled || ms != -1 {
+		t.Fatalf("malformed: ms=%d sampled=%v", ms, sampled)
+	}
+	ms, sampled = probeTCP("256.256.256.256:80", nil)
+	if !sampled || ms != -1 {
+		t.Fatalf("unresolvable: ms=%d sampled=%v", ms, sampled)
+	}
+}
+
+// applyTasks: an unchanged task keeps its loop; a changed target or
+// interval restarts it; a removed task is cancelled; the set is
+// truncated at the cap with intervals clamped into range.
+func TestProbeRunnerApplyTasks(t *testing.T) {
+	r := newProbeRunner()
+	defer r.stop()
+
+	r.applyTasks([]ProbeTask{{ID: 1, Target: "127.0.0.1:1", Interval: 60}})
+	if len(r.tasks) != 1 {
+		t.Fatalf("tasks %v", r.tasks)
+	}
+	first := r.tasks[1]
+
+	// Unchanged: same loop object survives.
+	r.applyTasks([]ProbeTask{{ID: 1, Target: "127.0.0.1:1", Interval: 60}})
+	if r.tasks[1] != first {
+		t.Fatal("unchanged task restarted")
+	}
+	// Changed interval: replaced, clamped values are what runs.
+	r.applyTasks([]ProbeTask{{ID: 1, Target: "127.0.0.1:1", Interval: 120}})
+	if r.tasks[1] == first || r.tasks[1].interval != 120 {
+		t.Fatal("changed task not restarted")
+	}
+	// Removed: cancelled.
+	r.applyTasks(nil)
+	if len(r.tasks) != 0 {
+		t.Fatalf("tasks after clear %v", r.tasks)
+	}
+
+	// Clamp + cap.
+	many := make([]ProbeTask, 0, 70)
+	for i := 1; i <= 70; i++ {
+		many = append(many, ProbeTask{ID: int64(i), Target: "127.0.0.1:1", Interval: 1})
+	}
+	r.applyTasks(many)
+	if len(r.tasks) != maxProbeTasksRun {
+		t.Fatalf("cap: %d running", len(r.tasks))
+	}
+	if r.tasks[1].interval != minProbeIntervalSec {
+		t.Fatalf("clamp: interval %d", r.tasks[1].interval)
+	}
+}
+
+// A running loop files samples into the buffer; drain takes them and
+// requeue restores order ahead of newer samples.
+func TestProbeRunnerResultFlow(t *testing.T) {
+	r := newProbeRunner()
+	defer r.stop()
+	r.applyTasks([]ProbeTask{{ID: 9, Target: "127.0.0.1:1", Interval: 3600}})
+	// The loop probes immediately on start.
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		if got := r.drainResults(); len(got) > 0 {
+			if got[0].TaskID != 9 || got[0].LatencyMs != -1 {
+				t.Fatalf("sample %+v", got[0])
+			}
+			goto drained
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	t.Fatal("no immediate sample from a fresh task")
+drained:
+
+	r.addResult(ProbeResult{TaskID: 1, LatencyMs: 5})
+	batch := r.drainResults()
+	r.addResult(ProbeResult{TaskID: 1, LatencyMs: 7})
+	r.requeueResults(batch)
+	got := r.drainResults()
+	if len(got) != 2 || got[0].LatencyMs != 5 || got[1].LatencyMs != 7 {
+		t.Fatalf("requeue order %+v", got)
+	}
+	if again := r.drainResults(); again != nil {
+		t.Fatalf("second drain %+v", again)
+	}
+}
+
+// Buffer overflow drops the oldest samples, keeping the newest cap.
+func TestProbeRunnerBufferCap(t *testing.T) {
+	r := newProbeRunner()
+	defer r.stop()
+	for i := 0; i < probeResultCap+10; i++ {
+		r.addResult(ProbeResult{TaskID: 1, LatencyMs: int64(i)})
+	}
+	got := r.drainResults()
+	if len(got) != probeResultCap {
+		t.Fatalf("buffer %d", len(got))
+	}
+	if got[0].LatencyMs != 10 || got[len(got)-1].LatencyMs != int64(probeResultCap+9) {
+		t.Fatalf("kept range %d..%d", got[0].LatencyMs, got[len(got)-1].LatencyMs)
+	}
+	if r.dropped != 10 {
+		t.Fatalf("dropped %d", r.dropped)
+	}
+}
+
+// The wire shapes: a result marshals as taskId/latencyMs, and the
+// config decode distinguishes an absent probe_tasks (old master,
+// keep current set) from an empty one (clear all).
+func TestProbeWireShapes(t *testing.T) {
+	body, err := json.Marshal(ProbeResult{TaskID: 3, LatencyMs: -1})
+	if err != nil || string(body) != `{"taskId":3,"latencyMs":-1}` {
+		t.Fatalf("result wire: %s %v", body, err)
+	}
+	decode := func(s string) *[]ProbeTask {
+		var result struct {
+			ProbeTasks *[]ProbeTask `json:"probe_tasks"`
+		}
+		if err := json.Unmarshal([]byte(s), &result); err != nil {
+			t.Fatal(err)
+		}
+		return result.ProbeTasks
+	}
+	if got := decode(`{"config_hash":"x"}`); got != nil {
+		t.Fatalf("absent field decoded as %v", got)
+	}
+	if got := decode(`{"probe_tasks":[]}`); got == nil || len(*got) != 0 {
+		t.Fatalf("empty list decoded as %v", got)
+	}
+	got := decode(`{"probe_tasks":[{"id":2,"target":"1.1.1.1:443","interval":30}]}`)
+	if got == nil || len(*got) != 1 || (*got)[0].ID != 2 || (*got)[0].Target != "1.1.1.1:443" || (*got)[0].Interval != 30 {
+		t.Fatalf("task decode %+v", got)
+	}
+}

internal/web/controller/probe_public_test.go

@@ -78,6 +78,7 @@ func newProbeTestRig(t *testing.T) *probeTestRig {
 	agentAdmin := engine.Group("/panel/api/agent")
 	agentAdmin.Use(a.checkAPIAuth)
 	NewProbeMetaController(agentAdmin)
+	NewProbeTasksController(agentAdmin)
 	engine.NoRoute(func(c *gin.Context) {
 		if pc.HandleFallback(c) {
 			return

internal/web/controller/probe_tasks_agent_test.go

@@ -0,0 +1,138 @@
+package controller
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+
+	"github.com/mhsanaei/3x-ui/v3/internal/database"
+	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+	"github.com/mhsanaei/3x-ui/v3/internal/web/service/probe"
+)
+
+// Item 7 B4: the agent-facing wire — probe_tasks rides the config
+// response (always present, empty list clears), and probeResults on
+// the heartbeat are filed stamped at arrival, with junk rows
+// (unknown/absent latency, non-positive task ids) skipped and
+// unassigned tasks dropped by the store's assignment gate.
+func TestAgentProbeTasksWire(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	dbDir := t.TempDir()
+	t.Setenv("XUI_DB_FOLDER", dbDir)
+	if err := database.InitDB(filepath.Join(dbDir, "x-ui.db")); err != nil {
+		t.Fatalf("InitDB: %v", err)
+	}
+	t.Cleanup(func() { _ = database.CloseDB() })
+	db := database.GetDB()
+
+	node := &model.Node{Name: "b4-wire"}
+	if err := db.Create(node).Error; err != nil {
+		t.Fatal(err)
+	}
+	const token = "b4-wire-token"
+	agent := &model.AgentNode{
+		TokenHash: hashAgentToken(token), TokenPrefix: "b4-",
+		Name: "b4-wire-agent", NodeID: &node.Id,
+	}
+	if err := db.Create(agent).Error; err != nil {
+		t.Fatal(err)
+	}
+	agentID := int64(agent.Id)
+	probe.DeleteAgentData(agentID)
+
+	taskID, err := probe.SavePingTask(probe.PingTask{
+		Name: "wire", Target: "1.1.1.1:443", Interval: 30,
+		AgentIDs: []int64{agentID},
+	}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	engine := gin.New()
+	NewAgentController(engine.Group("/panel/api/agent"))
+	do := func(method, path, body string) (int, []byte) {
+		var rd *bytes.Reader
+		if body != "" {
+			rd = bytes.NewReader([]byte(body))
+		} else {
+			rd = bytes.NewReader(nil)
+		}
+		req := httptest.NewRequest(method, path, rd)
+		req.Header.Set("Content-Type", "application/json")
+		req.Header.Set("Authorization", "Bearer "+token)
+		rec := httptest.NewRecorder()
+		engine.ServeHTTP(rec, req)
+		return rec.Code, rec.Body.Bytes()
+	}
+
+	const cfgPath = "/panel/api/agent/config"
+	const hbPath = "/panel/api/agent/heartbeat"
+
+	// Config carries the task set.
+	code, body := do(http.MethodGet, cfgPath, "")
+	if code != http.StatusOK {
+		t.Fatalf("config: %d %s", code, body)
+	}
+	var cfg struct {
+		ProbeTasks []probe.PingTaskRef `json:"probe_tasks"`
+	}
+	if err := json.Unmarshal(body, &cfg); err != nil {
+		t.Fatal(err)
+	}
+	if len(cfg.ProbeTasks) != 1 || cfg.ProbeTasks[0].ID != taskID ||
+		cfg.ProbeTasks[0].Target != "1.1.1.1:443" || cfg.ProbeTasks[0].Interval != 30 {
+		t.Fatalf("config probe_tasks: %s", body)
+	}
+
+	// Heartbeat with a mixed batch: only the valid, assigned sample
+	// lands. (All rows share the arrival stamp, so the two filings of
+	// the same task collapse — assert on the surviving value.)
+	hb := `{"probeResults":[` +
+		`{"taskId":` + itoa(taskID) + `,"latencyMs":42},` +
+		`{"taskId":999999,"latencyMs":5},` +
+		`{"taskId":-3,"latencyMs":5},` +
+		`{"taskId":` + itoa(taskID) + `}` +
+		`]}`
+	code, body = do(http.MethodPost, hbPath, hb)
+	if code != http.StatusOK {
+		t.Fatalf("heartbeat: %d %s", code, body)
+	}
+	st := probe.Default()
+	var count int64
+	var latency int64
+	if err := st.DB().QueryRow(`SELECT COUNT(*), COALESCE(MAX(latency), -999) FROM probe_ping_record WHERE agent_id = ?`, agentID).Scan(&count, &latency); err != nil {
+		t.Fatal(err)
+	}
+	if count != 1 || latency != 42 {
+		t.Fatalf("filed rows %d latency %d", count, latency)
+	}
+
+	// After the task is unassigned, config reports an empty (present)
+	// list and further results are dropped by the assignment gate.
+	if err := probe.SetNodePingTasks(agentID, []int64{}, []int64{taskID}); err != nil {
+		t.Fatal(err)
+	}
+	code, body = do(http.MethodGet, cfgPath, "")
+	if code != http.StatusOK {
+		t.Fatalf("config 2: %d", code)
+	}
+	if !bytes.Contains(body, []byte(`"probe_tasks":[]`)) {
+		t.Fatalf("empty probe_tasks not present: %s", body)
+	}
+	code, _ = do(http.MethodPost, hbPath, `{"probeResults":[{"taskId":`+itoa(taskID)+`,"latencyMs":77}]}`)
+	if code != http.StatusOK {
+		t.Fatalf("heartbeat 2: %d", code)
+	}
+	if err := st.DB().QueryRow(`SELECT COUNT(*) FROM probe_ping_record WHERE agent_id = ? AND latency = 77`, agentID).Scan(&count); err != nil {
+		t.Fatal(err)
+	}
+	if count != 0 {
+		t.Fatal("unassigned result was filed")
+	}
+	probe.DeletePingTask(taskID)
+}

internal/web/controller/probe_tasks_test.go

@@ -0,0 +1,281 @@
+package controller
+
+import (
+	"encoding/json"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/mhsanaei/3x-ui/v3/internal/database"
+	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+	"github.com/mhsanaei/3x-ui/v3/internal/web/service/probe"
+)
+
+// Item 7 B4 controller tests: probe task management endpoints (hub
+// validation chain, node-id translation, base-diff, reorder), the
+// metrics series=ping read going live, and heartbeat ingestion of
+// probeResults + config delivery of probe_tasks.
+
+type envelope struct {
+	Success bool            `json:"success"`
+	Msg     string          `json:"msg"`
+	Obj     json.RawMessage `json:"obj"`
+}
+
+func decodeEnvelope(t *testing.T, body []byte) envelope {
+	t.Helper()
+	var env envelope
+	if err := json.Unmarshal(body, &env); err != nil {
+		t.Fatalf("envelope: %v (%s)", err, body)
+	}
+	return env
+}
+
+func cleanPingTasks(t *testing.T) {
+	t.Helper()
+	tasks, err := probe.PingTasks()
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, task := range tasks {
+		if err := probe.DeletePingTask(task.ID); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+const tasksPath = "/panel/api/agent/probe-tasks"
+
+func TestProbeTasksAuthAndValidation(t *testing.T) {
+	rig := newProbeTestRig(t)
+	cleanPingTasks(t)
+
+	// Anonymous callers are refused by the group auth.
+	rec := rig.do(http.MethodPut, tasksPath, []byte(`{"name":"x","target":"1.1.1.1:443","interval":60}`), false)
+	if rec.Code == http.StatusOK && decodeEnvelope(t, rec.Body.Bytes()).Success {
+		t.Fatal("anonymous save succeeded")
+	}
+
+	cases := []struct {
+		name string
+		body string
+		want string
+	}{
+		{"empty name", `{"name":" ","target":"1.1.1.1:443","interval":60}`, "请填写名称和目标"},
+		{"empty target", `{"name":"x","target":"","interval":60}`, "请填写名称和目标"},
+		{"long name", `{"name":"` + strings.Repeat("名", 65) + `","target":"1.1.1.1:443","interval":60}`, "名称最多 64 个字"},
+		{"no port", `{"name":"x","target":"1.1.1.1","interval":60}`, "目标要写成"},
+		{"empty port", `{"name":"x","target":"1.1.1.1:","interval":60}`, "目标要写成"},
+		{"port range", `{"name":"x","target":"1.1.1.1:65536","interval":60}`, "目标要写成"},
+		{"bare v6", `{"name":"x","target":"2606:4700:4700::1111:443","interval":60}`, "目标要写成"},
+		{"interval low", `{"name":"x","target":"1.1.1.1:443","interval":3}`, "间隔要在 5 到 3600 秒之间"},
+		{"interval high", `{"name":"x","target":"1.1.1.1:443","interval":3601}`, "间隔要在 5 到 3600 秒之间"},
+		{"long region", `{"name":"x","target":"1.1.1.1:443","interval":60,"region":"` + strings.Repeat("区", 14) + `"}`, "地区最多 13 个字"},
+	}
+	for _, tc := range cases {
+		rec := rig.do(http.MethodPut, tasksPath, []byte(tc.body), true)
+		env := decodeEnvelope(t, rec.Body.Bytes())
+		if env.Success {
+			t.Errorf("%s: save succeeded, want refusal", tc.name)
+			continue
+		}
+		if !strings.Contains(env.Msg, tc.want) {
+			t.Errorf("%s: msg %q, want fragment %q", tc.name, env.Msg, tc.want)
+		}
+	}
+	if tasks, _ := probe.PingTasks(); len(tasks) != 0 {
+		t.Fatalf("refused saves leaked %d tasks", len(tasks))
+	}
+
+	// Valid forms: bracketed v6, boundary interval, 13-rune region.
+	for _, body := range []string{
+		`{"name":"v6","target":"[2606:4700:4700::1111]:443","interval":5}`,
+		`{"name":"edge","target":"example.com:65535","interval":3600,"region":"` + strings.Repeat("区", 13) + `"}`,
+	} {
+		rec := rig.do(http.MethodPut, tasksPath, []byte(body), true)
+		if env := decodeEnvelope(t, rec.Body.Bytes()); !env.Success {
+			t.Fatalf("valid save refused: %s (%s)", env.Msg, body)
+		}
+	}
+	cleanPingTasks(t)
+}
+
+func TestProbeTasksNodeMappingAndLifecycle(t *testing.T) {
+	rig := newProbeTestRig(t)
+	cleanPingTasks(t)
+	agentID, nodeID := rig.seedAgent(t, "b4node", true)
+
+	// A node with no agent bound cannot take assignments.
+	db := database.GetDB()
+	lone := &model.Node{Name: "b4-lone"}
+	if err := db.Create(lone).Error; err != nil {
+		t.Fatal(err)
+	}
+	body := `{"name":"t1","target":"1.1.1.1:443","interval":60,"nodeIds":[` + itoa(int64(lone.Id)) + `]}`
+	rec := rig.do(http.MethodPut, tasksPath, []byte(body), true)
+	if env := decodeEnvelope(t, rec.Body.Bytes()); env.Success || !strings.Contains(env.Msg, "未绑定 agent") {
+		t.Fatalf("unbound node save: %+v", env)
+	}
+
+	// Create assigned to the bound node.
+	body = `{"name":"t1","target":"1.1.1.1:443","interval":60,"region":"亚洲","nodeIds":[` + itoa(int64(nodeID)) + `]}`
+	rec = rig.do(http.MethodPut, tasksPath, []byte(body), true)
+	env := decodeEnvelope(t, rec.Body.Bytes())
+	if !env.Success {
+		t.Fatalf("create: %s", env.Msg)
+	}
+	var created struct {
+		ID int64 `json:"id"`
+	}
+	if err := json.Unmarshal(env.Obj, &created); err != nil || created.ID == 0 {
+		t.Fatalf("create obj: %s", env.Obj)
+	}
+	refs := probe.PingTasksFor(agentID)
+	if len(refs) != 1 || refs[0].ID != created.ID {
+		t.Fatalf("agent view: %v", refs)
+	}
+
+	// The list speaks node ids.
+	rec = rig.do(http.MethodGet, tasksPath, nil, true)
+	env = decodeEnvelope(t, rec.Body.Bytes())
+	var list []probeTaskForm
+	if err := json.Unmarshal(env.Obj, &list); err != nil || len(list) != 1 {
+		t.Fatalf("list: %s", env.Obj)
+	}
+	if len(list[0].NodeIDs) != 1 || list[0].NodeIDs[0] != int64(nodeID) || list[0].Region != "亚洲" {
+		t.Fatalf("list form %+v", list[0])
+	}
+
+	// by-node edit: clear the assignment with a base diff.
+	byNode := `{"tasks":[],"base":[` + itoa64(created.ID) + `]}`
+	rec = rig.do(http.MethodPut, tasksPath+"/by-node/"+itoa(int64(nodeID)), []byte(byNode), true)
+	if env := decodeEnvelope(t, rec.Body.Bytes()); !env.Success {
+		t.Fatalf("by-node clear: %s", env.Msg)
+	}
+	if refs := probe.PingTasksFor(agentID); len(refs) != 0 {
+		t.Fatalf("after by-node clear: %v", refs)
+	}
+	// by-node on an unknown node 404s.
+	rec = rig.do(http.MethodPut, tasksPath+"/by-node/999999", []byte(`{"tasks":[],"base":[]}`), true)
+	if rec.Code != http.StatusNotFound {
+		t.Fatalf("by-node unknown node: %d", rec.Code)
+	}
+
+	// Reorder: second task, flip, duplicates refused.
+	rec = rig.do(http.MethodPut, tasksPath, []byte(`{"name":"t2","target":"8.8.8.8:53","interval":30}`), true)
+	env = decodeEnvelope(t, rec.Body.Bytes())
+	var created2 struct {
+		ID int64 `json:"id"`
+	}
+	_ = json.Unmarshal(env.Obj, &created2)
+	flip := `{"ids":[` + itoa64(created2.ID) + `,` + itoa64(created.ID) + `]}`
+	rec = rig.do(http.MethodPost, tasksPath+"/reorder", []byte(flip), true)
+	if env := decodeEnvelope(t, rec.Body.Bytes()); !env.Success {
+		t.Fatalf("reorder: %s", env.Msg)
+	}
+	dup := `{"ids":[` + itoa64(created.ID) + `,` + itoa64(created.ID) + `]}`
+	rec = rig.do(http.MethodPost, tasksPath+"/reorder", []byte(dup), true)
+	if env := decodeEnvelope(t, rec.Body.Bytes()); env.Success {
+		t.Fatal("duplicate reorder accepted")
+	}
+	tasks, _ := probe.PingTasks()
+	if len(tasks) != 2 || tasks[0].ID != created2.ID {
+		t.Fatalf("order after flip %+v", tasks)
+	}
+
+	// Delete clears the agent's view too.
+	rec = rig.do(http.MethodDelete, tasksPath+"/"+itoa64(created.ID), nil, true)
+	if env := decodeEnvelope(t, rec.Body.Bytes()); !env.Success {
+		t.Fatalf("delete: %s", env.Msg)
+	}
+	if tasks, _ := probe.PingTasks(); len(tasks) != 1 {
+		t.Fatalf("after delete %v", tasks)
+	}
+	cleanPingTasks(t)
+}
+
+// series=ping serves real rows now: medians from filed samples, the
+// probes label map, and whole-window loss for a fully-missed task.
+func TestProbeMetricsPingSeriesLive(t *testing.T) {
+	rig := newProbeTestRig(t)
+	cleanPingTasks(t)
+	rig.setSetting(t, "public_page", "on")
+	agentID, _ := rig.seedAgent(t, "b4metrics", true)
+
+	good, err := probe.SavePingTask(probe.PingTask{Name: "good", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{agentID}}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	bad, err := probe.SavePingTask(probe.PingTask{Name: "bad", Target: "127.0.0.1:1", Interval: 60, AgentIDs: []int64{agentID}}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	st := probe.Default()
+	now := time.Now().Unix()
+	for i := 0; i < 3; i++ {
+		if err := st.InsertPings(agentID, []probe.PingResult{{TaskID: good, LatencyMs: 10}}, now-int64(i)*10); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := st.InsertPings(agentID, []probe.PingResult{{TaskID: bad, LatencyMs: -1}}, now); err != nil {
+		t.Fatal(err)
+	}
+
+	rec := rig.do(http.MethodGet, "/api/nodes/"+itoa64(agentID)+"/metrics?series=ping&hours=1", nil, false)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("metrics: %d %s", rec.Code, rec.Body.String())
+	}
+	var resp struct {
+		Ping []struct {
+			TaskID  int64  `json:"task_id"`
+			Latency *int64 `json:"latency"`
+			Loss    *int64 `json:"loss"`
+		} `json:"ping"`
+		Probes map[string]string  `json:"probes"`
+		Loss   map[string]float64 `json:"loss"`
+	}
+	if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
+		t.Fatal(err)
+	}
+	var goodRow, badRow bool
+	for _, row := range resp.Ping {
+		switch row.TaskID {
+		case good:
+			goodRow = true
+			if row.Latency == nil || *row.Latency != 10 {
+				t.Fatalf("good latency %v", row.Latency)
+			}
+		case bad:
+			badRow = true
+			if row.Latency != nil {
+				t.Fatalf("bad latency %v, want null", *row.Latency)
+			}
+			if row.Loss == nil || *row.Loss != 100 {
+				t.Fatalf("bad bucket loss %v", row.Loss)
+			}
+		}
+	}
+	if !goodRow || !badRow {
+		t.Fatalf("rows %+v", resp.Ping)
+	}
+	if resp.Probes[itoa64(good)] != "good" || resp.Probes[itoa64(bad)] != "bad" {
+		t.Fatalf("probes %v", resp.Probes)
+	}
+	if resp.Loss[itoa64(bad)] != 100.0 {
+		t.Fatalf("window loss %v", resp.Loss)
+	}
+	if _, ok := resp.Loss[itoa64(good)]; ok {
+		t.Fatalf("good task in loss map %v", resp.Loss)
+	}
+	cleanPingTasks(t)
+}
+
+func itoa64(n int64) string {
+	return strings.TrimSpace(jsonNumber(n))
+}
+
+func jsonNumber(n int64) string {
+	b, _ := json.Marshal(n)
+	return string(b)
+}

internal/web/service/probe/ping_test.go

@@ -0,0 +1,411 @@
+package probe
+
+import (
+	"errors"
+	"strconv"
+	"testing"
+	"time"
+)
+
+func saveTask(t *testing.T, st *Store, task PingTask) int64 {
+	t.Helper()
+	id, err := st.SavePingTask(task, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return id
+}
+
+// Create assigns an id from the high-water mark, lists back with its
+// assignments, and a deleted id is never reissued (hub next_id).
+func TestPingTaskCRUDAndIDReuse(t *testing.T) {
+	st := newTestStore(t)
+	id1 := saveTask(t, st, PingTask{Name: "CF", Target: "1.1.1.1:443", Interval: 60, Region: "全球", AgentIDs: []int64{7}})
+	id2 := saveTask(t, st, PingTask{Name: "G", Target: "8.8.8.8:53", Interval: 30})
+	if id1 == 0 || id2 == 0 || id1 == id2 {
+		t.Fatalf("ids %d %d", id1, id2)
+	}
+	tasks, err := st.PingTasks()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(tasks) != 2 || tasks[0].ID != id1 || tasks[1].ID != id2 {
+		t.Fatalf("list %+v", tasks)
+	}
+	if len(tasks[0].AgentIDs) != 1 || tasks[0].AgentIDs[0] != 7 || tasks[0].Region != "全球" {
+		t.Fatalf("task0 %+v", tasks[0])
+	}
+	if tasks[1].Sort <= tasks[0].Sort {
+		t.Fatalf("sort not appended: %d %d", tasks[0].Sort, tasks[1].Sort)
+	}
+	if err := st.DeletePingTask(id1); err != nil {
+		t.Fatal(err)
+	}
+	id3 := saveTask(t, st, PingTask{Name: "Q", Target: "9.9.9.9:443", Interval: 60})
+	if id3 == id1 || id3 == id2 {
+		t.Fatalf("id reissued: %d", id3)
+	}
+	// Updating a task another session deleted reports it (hub).
+	if _, err := st.SavePingTask(PingTask{ID: id1, Name: "x", Target: "1.1.1.1:443", Interval: 60}, nil); !errors.Is(err, ErrPingTaskGone) {
+		t.Fatalf("gone update: %v", err)
+	}
+}
+
+// Base-diff saving: an assignment made after the editor's base snapshot
+// survives the save; base members removed from the new set go away.
+func TestPingTaskSaveBaseDiff(t *testing.T) {
+	st := newTestStore(t)
+	id := saveTask(t, st, PingTask{Name: "A", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{1, 2}})
+	// Another session assigns agent 3 meanwhile.
+	other := saveTaskPingAssign(t, st, id, []int64{1, 2, 3})
+	_ = other
+	base := []int64{1, 2}
+	if _, err := st.SavePingTask(PingTask{ID: id, Name: "A", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{2}}, &base); err != nil {
+		t.Fatal(err)
+	}
+	tasks, _ := st.PingTasks()
+	got := map[int64]bool{}
+	for _, a := range tasks[0].AgentIDs {
+		got[a] = true
+	}
+	if got[1] || !got[2] || !got[3] {
+		t.Fatalf("base-diff result %v", tasks[0].AgentIDs)
+	}
+	// Without base the assignment is replaced wholesale.
+	if _, err := st.SavePingTask(PingTask{ID: id, Name: "A", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{9}}, nil); err != nil {
+		t.Fatal(err)
+	}
+	tasks, _ = st.PingTasks()
+	if len(tasks[0].AgentIDs) != 1 || tasks[0].AgentIDs[0] != 9 {
+		t.Fatalf("wholesale replace %v", tasks[0].AgentIDs)
+	}
+}
+
+func saveTaskPingAssign(t *testing.T, st *Store, taskID int64, agents []int64) error {
+	t.Helper()
+	_, err := st.db.Exec(`DELETE FROM probe_ping_node WHERE task_id = ?`, taskID)
+	if err != nil {
+		return err
+	}
+	for _, a := range agents {
+		if _, err := st.db.Exec(`INSERT INTO probe_ping_node (task_id, agent_id) VALUES (?, ?)`, taskID, a); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Crowding: the 65th task on one agent refuses the whole save, and so
+// does a 65th auto_join task (a fresh agent would be over its cap the
+// moment it appears).
+func TestPingTaskCrowding(t *testing.T) {
+	st := newTestStore(t)
+	for i := 0; i < MaxProbesPerNode; i++ {
+		saveTask(t, st, PingTask{Name: "t", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{5}})
+	}
+	_, err := st.SavePingTask(PingTask{Name: "one too many", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{5}}, nil)
+	var crowded *PingCrowdedError
+	if !errors.As(err, &crowded) || crowded.AgentID != 5 {
+		t.Fatalf("crowded: %v", err)
+	}
+	if tasks, _ := st.PingTasks(); len(tasks) != MaxProbesPerNode {
+		t.Fatalf("refused save leaked: %d tasks", len(tasks))
+	}
+	// The same count on a different agent is fine.
+	saveTask(t, st, PingTask{Name: "ok", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{6}})
+
+	st2 := newTestStore(t)
+	for i := 0; i < MaxProbesPerNode; i++ {
+		saveTask(t, st2, PingTask{Name: "t", Target: "1.1.1.1:443", Interval: 60, AutoJoin: true})
+	}
+	_, err = st2.SavePingTask(PingTask{Name: "join too many", Target: "1.1.1.1:443", Interval: 60, AutoJoin: true}, nil)
+	var aj *PingAutoJoinError
+	if !errors.As(err, &aj) {
+		t.Fatalf("autoJoin crowded: %v", err)
+	}
+}
+
+// SetNodePingTasks applies only the base difference and refuses when
+// THIS agent would overflow; another agent's overflow is not its fault.
+func TestSetNodePingTasks(t *testing.T) {
+	st := newTestStore(t)
+	a := saveTask(t, st, PingTask{Name: "a", Target: "1.1.1.1:443", Interval: 60})
+	b := saveTask(t, st, PingTask{Name: "b", Target: "1.1.1.1:443", Interval: 60})
+	// Assigned meanwhile by the task-side editor: survives.
+	saveTaskPingAssign(t, st, b, []int64{3})
+	if err := st.SetNodePingTasks(3, []int64{a}, []int64{}); err != nil {
+		t.Fatal(err)
+	}
+	if refs, _ := st.PingTasksFor(3); len(refs) != 2 {
+		t.Fatalf("node tasks %v", refs)
+	}
+	// A task deleted since the base makes the edit refuse.
+	if err := st.SetNodePingTasks(3, []int64{a, 9999}, []int64{a}); err == nil {
+		t.Fatal("deleted task accepted")
+	}
+}
+
+// Reorder: duplicates and stale lists refuse; a full list renumbers
+// sort and PingTasks reflects it.
+func TestReorderPingTasks(t *testing.T) {
+	st := newTestStore(t)
+	a := saveTask(t, st, PingTask{Name: "a", Target: "1.1.1.1:443", Interval: 60})
+	b := saveTask(t, st, PingTask{Name: "b", Target: "1.1.1.1:443", Interval: 60})
+	if err := st.ReorderPingTasks([]int64{a, a}); err == nil {
+		t.Fatal("duplicate accepted")
+	}
+	if err := st.ReorderPingTasks([]int64{a}); err == nil {
+		t.Fatal("short list accepted")
+	}
+	if err := st.ReorderPingTasks([]int64{b, a}); err != nil {
+		t.Fatal(err)
+	}
+	tasks, _ := st.PingTasks()
+	if tasks[0].ID != b || tasks[1].ID != a || tasks[0].Sort != 0 || tasks[1].Sort != 1 {
+		t.Fatalf("after reorder %+v", tasks)
+	}
+}
+
+// PingTasksFor is id-ordered (the agent's cap keeps the first N) and
+// PingTaskNames covers only assigned tasks.
+func TestPingTasksForAndNames(t *testing.T) {
+	st := newTestStore(t)
+	a := saveTask(t, st, PingTask{Name: "first", Target: "1.1.1.1:443", Interval: 45, AgentIDs: []int64{2}})
+	b := saveTask(t, st, PingTask{Name: "second", Target: "8.8.8.8:53", Interval: 90, AgentIDs: []int64{2}})
+	saveTask(t, st, PingTask{Name: "elsewhere", Target: "9.9.9.9:443", Interval: 60, AgentIDs: []int64{3}})
+	refs, err := st.PingTasksFor(2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(refs) != 2 || refs[0].ID != a || refs[1].ID != b || refs[0].Target != "1.1.1.1:443" || refs[0].Interval != 45 {
+		t.Fatalf("refs %+v", refs)
+	}
+	names, err := st.PingTaskNames(2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(names) != 2 || names[strconv.FormatInt(a, 10)] != "first" {
+		t.Fatalf("names %v", names)
+	}
+}
+
+// InsertPings files only under live assignments: a result for an
+// unassigned or deleted task is dropped silently; a second sample at
+// the same stamp replaces the first (one session, one stamp).
+func TestInsertPingsAssignmentGate(t *testing.T) {
+	st := newTestStore(t)
+	id := saveTask(t, st, PingTask{Name: "a", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{4}})
+	ts := time.Now().Unix()
+	err := st.InsertPings(4, []PingResult{
+		{TaskID: id, LatencyMs: 12},
+		{TaskID: id + 100, LatencyMs: 5}, // never assigned
+	}, ts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// From an agent the task is not assigned to: dropped too.
+	if err := st.InsertPings(9, []PingResult{{TaskID: id, LatencyMs: 99}}, ts); err != nil {
+		t.Fatal(err)
+	}
+	var count int64
+	var latency int64
+	st.db.QueryRow(`SELECT COUNT(*), COALESCE(MAX(latency), -999) FROM probe_ping_record`).Scan(&count, &latency)
+	if count != 1 || latency != 12 {
+		t.Fatalf("rows %d latency %d", count, latency)
+	}
+	if err := st.InsertPings(4, []PingResult{{TaskID: id, LatencyMs: 20}}, ts); err != nil {
+		t.Fatal(err)
+	}
+	st.db.QueryRow(`SELECT COUNT(*), COALESCE(MAX(latency), -999) FROM probe_ping_record`).Scan(&count, &latency)
+	if count != 1 || latency != 20 {
+		t.Fatalf("after replace rows %d latency %d", count, latency)
+	}
+}
+
+// PingHistory over minute rows: per-bucket medians, band only when the
+// bucket varied, bucket loss rounded up, window loss unrounded and
+// only for tasks that lost something.
+func TestPingHistoryMinuteRows(t *testing.T) {
+	st := newTestStore(t)
+	a := saveTask(t, st, PingTask{Name: "a", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{1}})
+	b := saveTask(t, st, PingTask{Name: "b", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{1}})
+	now := time.Now().Unix()
+	base := (now/60)*60 - 600 // ten minutes back, bucket aligned
+	span := Span{Since: base, Step: 60, Hourly: false}
+	// Task A: one bucket with answers 10, 20, 40 and one miss.
+	for i, lat := range []int64{10, 20, 40, -1} {
+		seedPing(t, st, 1, base+int64(i)*7, a, lat)
+	}
+	// Task B: one bucket, all misses.
+	seedPing(t, st, 1, base+5, b, -1)
+	seedPing(t, st, 1, base+15, b, -1)
+
+	rows, loss, err := st.PingHistory(1, span)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(rows) != 2 {
+		t.Fatalf("rows %+v", rows)
+	}
+	ra, rb := rows[0], rows[1]
+	if ra.TaskID != a || rb.TaskID != b {
+		t.Fatalf("order %+v", rows)
+	}
+	if ra.Latency == nil || *ra.Latency != 20 {
+		t.Fatalf("median %v", ra.Latency)
+	}
+	if ra.Band == nil || (*ra.Band)[0] != 10 || (*ra.Band)[1] != 40 {
+		t.Fatalf("band %v", ra.Band)
+	}
+	if ra.Loss == nil || *ra.Loss != 25 { // 1 of 4, exact
+		t.Fatalf("bucket loss %v", ra.Loss)
+	}
+	if rb.Latency != nil || rb.Band != nil {
+		t.Fatalf("all-miss row %+v", rb)
+	}
+	if rb.Loss == nil || *rb.Loss != 100 {
+		t.Fatalf("all-miss loss %v", rb.Loss)
+	}
+	if loss[strconv.FormatInt(a, 10)] != 25.0 {
+		t.Fatalf("window loss a %v", loss)
+	}
+	if loss[strconv.FormatInt(b, 10)] != 100.0 {
+		t.Fatalf("window loss b %v", loss)
+	}
+	// An agent without the assignment sees none of it.
+	rows2, loss2, err := st.PingHistory(2, span)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(rows2) != 0 || len(loss2) != 0 {
+		t.Fatalf("unassigned view %v %v", rows2, loss2)
+	}
+}
+
+// PingHistory over an hourly span merges folded hours (weighted by
+// their answer counts) with the minute rows after the watermark.
+func TestPingHistoryHourlyMerge(t *testing.T) {
+	st := newTestStore(t)
+	a := saveTask(t, st, PingTask{Name: "a", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{1}})
+	now := time.Now().Unix()
+	hourStart := hourFloor(now) - 3*3600
+	span := Span{Since: hourStart, Step: 3600, Hourly: true}
+	// Folded hour: 3 answers (median 30, range 10..50), 1 lost.
+	if _, err := st.db.Exec(`INSERT INTO probe_ping_hour (agent_id, ts, task_id, answered, lost, latency, lo, hi)
+		VALUES (1, ?, ?, 3, 1, 30, 10, 50)`, hourStart, a); err != nil {
+		t.Fatal(err)
+	}
+	// Watermark at the hour boundary: minute rows at/after it count.
+	if _, err := st.db.Exec(`INSERT INTO probe_setting (key, value) VALUES (?, ?)`, rolledKey, strconv.FormatInt(hourStart+3600, 10)); err != nil {
+		t.Fatal(err)
+	}
+	minuteTs := hourStart + 3600 + 120 // inside the NEXT hour bucket
+	seedPing(t, st, 1, minuteTs, a, 100)
+	// A stale minute row below the watermark must not double count.
+	seedPing(t, st, 1, hourStart+60, a, 999)
+
+	rows, loss, err := st.PingHistory(1, span)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(rows) != 2 {
+		t.Fatalf("rows %+v", rows)
+	}
+	if rows[0].Ts != hourStart || rows[0].Latency == nil || *rows[0].Latency != 30 {
+		t.Fatalf("hour row %+v", rows[0])
+	}
+	if rows[0].Band == nil || (*rows[0].Band)[0] != 10 || (*rows[0].Band)[1] != 50 {
+		t.Fatalf("hour band %+v", rows[0].Band)
+	}
+	if rows[0].Loss == nil || *rows[0].Loss != 25 {
+		t.Fatalf("hour loss %v", rows[0].Loss)
+	}
+	if rows[1].Ts != hourStart+3600 || rows[1].Latency == nil || *rows[1].Latency != 100 {
+		t.Fatalf("minute row %+v", rows[1])
+	}
+	// Window loss counts the hour tally plus the minute row: 1 lost of 5.
+	if loss[strconv.FormatInt(a, 10)] != 20.0 {
+		t.Fatalf("window loss %v", loss)
+	}
+}
+
+// Weighted median: with answers weighted by counts, the middle ranks
+// decide — two middle values average (hub Tally::median).
+func TestPingWeightedMedianAveragesMiddleTwo(t *testing.T) {
+	st := newTestStore(t)
+	a := saveTask(t, st, PingTask{Name: "a", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{1}})
+	now := time.Now().Unix()
+	base := (now / 3600) * 3600
+	span := Span{Since: base, Step: 3600, Hourly: true}
+	// Two folded hours inside one bucket: medians 10 (×1) and 21 (×1)
+	// → total 2, ranks 1 and 2 → (10+21)/2 = 15 (integer).
+	for i, m := range []int64{10, 21} {
+		if _, err := st.db.Exec(`INSERT INTO probe_ping_hour (agent_id, ts, task_id, answered, lost, latency, lo, hi)
+			VALUES (1, ?, ?, 1, 0, ?, ?, ?)`, base+int64(i)*60, a, m, m, m); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := st.db.Exec(`INSERT INTO probe_setting (key, value) VALUES (?, ?)`, rolledKey, strconv.FormatInt(base+3600, 10)); err != nil {
+		t.Fatal(err)
+	}
+	rows, _, err := st.PingHistory(1, span)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(rows) != 1 || rows[0].Latency == nil || *rows[0].Latency != 15 {
+		t.Fatalf("weighted median %+v", rows)
+	}
+}
+
+// Auto-join: an agent's first recorded sample assigns it every
+// auto_join task; agents already known are NOT retroactively joined
+// when a task is flagged later (hub semantics).
+func TestAutoJoinOnFirstRecord(t *testing.T) {
+	st := newTestStore(t)
+	auto := saveTask(t, st, PingTask{Name: "auto", Target: "1.1.1.1:443", Interval: 60, AutoJoin: true})
+	saveTask(t, st, PingTask{Name: "manual", Target: "1.1.1.1:443", Interval: 60})
+
+	// Existing agent 1 records before any task existed... (tasks were
+	// created first here, so use a pre-existing agent row instead:)
+	st.Record(1, Sample{BootID: "b1"}, time.Now(), 0)
+	if refs, _ := st.PingTasksFor(1); len(refs) != 1 || refs[0].ID != auto {
+		t.Fatalf("first record join %v", refs)
+	}
+	// Agent 2's first record joins too.
+	st.Record(2, Sample{BootID: "b2"}, time.Now(), 0)
+	if refs, _ := st.PingTasksFor(2); len(refs) != 1 {
+		t.Fatalf("agent2 join %v", refs)
+	}
+	// A task flagged auto_join afterwards does not reach agent 1/2.
+	late := saveTask(t, st, PingTask{Name: "late", Target: "1.1.1.1:443", Interval: 60, AutoJoin: true})
+	st.Record(1, Sample{BootID: "b1"}, time.Now(), time.Now().UnixMilli())
+	if refs, _ := st.PingTasksFor(1); len(refs) != 1 {
+		t.Fatalf("retroactive join happened: %v", refs)
+	}
+	// But a brand-new agent 3 picks up both.
+	st.Record(3, Sample{BootID: "b3"}, time.Now(), 0)
+	if refs, _ := st.PingTasksFor(3); len(refs) != 2 || refs[1].ID != late {
+		t.Fatalf("agent3 join %v", refs)
+	}
+}
+
+// DeletePingTask clears records, hours and assignments with the task.
+func TestDeletePingTaskClearsAll(t *testing.T) {
+	st := newTestStore(t)
+	id := saveTask(t, st, PingTask{Name: "a", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{1}})
+	seedPing(t, st, 1, time.Now().Unix(), id, 5)
+	if _, err := st.db.Exec(`INSERT INTO probe_ping_hour (agent_id, ts, task_id, answered, lost, latency, lo, hi)
+		VALUES (1, 0, ?, 1, 0, 5, 5, 5)`, id); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.DeletePingTask(id); err != nil {
+		t.Fatal(err)
+	}
+	for _, table := range []string{"probe_ping_record", "probe_ping_hour", "probe_ping_node", "probe_ping_task"} {
+		var n int64
+		st.db.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&n)
+		if n != 0 {
+			t.Fatalf("%s still has %d rows", table, n)
+		}
+	}
+}