3x-ui-agent/internal/agent/singbox_e2e_test.go

package agent

import (
	"context"
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"crypto/sha256"
	"crypto/tls"
	"crypto/x509"
	"crypto/x509/pkix"
	"encoding/hex"
	"encoding/json"
	"encoding/pem"
	"fmt"
	"io"
	"math/big"
	"net"
	"net/http"
	"net/http/httptest"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"golang.org/x/net/proxy"
)

// ITEM6 B3 end-to-end, env-gated: XUI_SINGBOX_BIN must point at the
// self-built sing-box v1.13.21 (with_v2ray_api with_utls with_quic).
// Skipped otherwise so the package suite stays green anywhere.
//
// Topology: a stub master serves a B2-shaped singbox_config; the real
// Agent applies it and runs the real core; a second core instance is
// the client (vless-reality / ss2022 / mixed / trojan / hysteria2 /
// a forwarding inbound), a third is the forwarding landing.
// Assertions cover: apply+serve, per-user deltas against known byte
// counts, online users, forwarding, flush-before-change + epoch
// rebase across a core restart (no double counting), the check gate.
//
// Environment limits of the build sandbox, probed at runtime: direct
// internet is unreachable (the reality handshake target is a local
// TLS server — production points it at real sites), and UDP sends
// are blocked outright (EPERM), so the hysteria2 leg only joins when
// a UDP loopback probe succeeds; its on-node verification belongs
// to the B5 pilot, like trojan+flow (N5) and wireguard (N7).

type e2eMaster struct {
	mu      sync.Mutex
	cfg     []byte
	hash    string
	reports []trafficReport
}

func (m *e2eMaster) setConfig(cfg []byte) {
	sum := sha256.Sum256(cfg)
	m.mu.Lock()
	m.cfg = cfg
	m.hash = hex.EncodeToString(sum[:])
	m.mu.Unlock()
}

func (m *e2eMaster) handler() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("/panel/api/agent/register", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
	})
	mux.HandleFunc("/panel/api/agent/heartbeat", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
	})
	mux.HandleFunc("/panel/api/agent/config", func(w http.ResponseWriter, r *http.Request) {
		m.mu.Lock()
		defer m.mu.Unlock()
		resp := map[string]any{
			"config_hash":         "xray-placeholder",
			"xray_config":         json.RawMessage(`{}`),
			"singbox_config":      json.RawMessage(m.cfg),
			"singbox_config_hash": m.hash,
		}
		_ = json.NewEncoder(w).Encode(resp)
	})
	mux.HandleFunc("/panel/api/agent/traffic", func(w http.ResponseWriter, r *http.Request) {
		var rep trafficReport
		if err := json.NewDecoder(r.Body).Decode(&rep); err == nil {
			m.mu.Lock()
			m.reports = append(m.reports, rep)
			m.mu.Unlock()
		}
		w.WriteHeader(http.StatusOK)
	})
	return mux
}

// totals folds every report the master received into per-user and
// per-inbound cumulative deltas plus the last online set.
func (m *e2eMaster) totals() (users map[string][2]int64, inbounds map[string][2]int64, onlines []string) {
	m.mu.Lock()
	defer m.mu.Unlock()
	users = map[string][2]int64{}
	inbounds = map[string][2]int64{}
	for _, rep := range m.reports {
		for _, u := range rep.Users {
			v := users[u.Email]
			users[u.Email] = [2]int64{v[0] + u.Up, v[1] + u.Down}
		}
		for _, in := range rep.Inbounds {
			v := inbounds[in.Tag]
			inbounds[in.Tag] = [2]int64{v[0] + in.Up, v[1] + in.Down}
		}
		if len(rep.Onlines) > 0 {
			onlines = rep.Onlines
		}
	}
	return users, inbounds, onlines
}

func freePort(t *testing.T) int {
	t.Helper()
	l, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	defer l.Close()
	return l.Addr().(*net.TCPAddr).Port
}

func freeUDPPort(t *testing.T) int {
	t.Helper()
	c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0})
	if err != nil {
		t.Fatal(err)
	}
	defer c.Close()
	return c.LocalAddr().(*net.UDPAddr).Port
}

// udpLoopbackWorks probes whether UDP over loopback supports a
// full round trip in this environment. A one-way check is not
// enough: the build sandbox lets an unbound client socket send but
// refuses (EPERM) the bound server socket's reply, which no
// QUIC-based protocol can survive.
func udpLoopbackWorks() bool {
	srv, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0})
	if err != nil {
		return false
	}
	defer srv.Close()
	go func() {
		buf := make([]byte, 16)
		_ = srv.SetReadDeadline(time.Now().Add(3 * time.Second))
		_, addr, err := srv.ReadFromUDP(buf)
		if err == nil {
			_, _ = srv.WriteToUDP([]byte("pong"), addr)
		}
	}()
	cli, err := net.DialUDP("udp", nil, srv.LocalAddr().(*net.UDPAddr))
	if err != nil {
		return false
	}
	defer cli.Close()
	if _, err := cli.Write([]byte("ping")); err != nil {
		return false
	}
	_ = cli.SetReadDeadline(time.Now().Add(3 * time.Second))
	buf := make([]byte, 16)
	_, err = cli.Read(buf)
	return err == nil
}

func selfSignedCert(t *testing.T, dir string, dnsNames ...string) (certPath, keyPath string) {
	t.Helper()
	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	tmpl := &x509.Certificate{
		SerialNumber: big.NewInt(1),
		Subject:      pkix.Name{CommonName: dnsNames[0]},
		DNSNames:     dnsNames,
		NotBefore:    time.Now().Add(-time.Hour),
		NotAfter:     time.Now().Add(24 * time.Hour),
		KeyUsage:     x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
	}
	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
	if err != nil {
		t.Fatal(err)
	}
	certPath = filepath.Join(dir, "cert-"+dnsNames[0]+".pem")
	keyPath = filepath.Join(dir, "key-"+dnsNames[0]+".pem")
	certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
	keyDER, err := x509.MarshalECPrivateKey(key)
	if err != nil {
		t.Fatal(err)
	}
	keyPEM := pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER})
	if err := os.WriteFile(certPath, certPEM, 0o644); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(keyPath, keyPEM, 0o600); err != nil {
		t.Fatal(err)
	}
	return certPath, keyPath
}

// startHandshakeServer runs the local TLS endpoint the reality
// server mirrors.
func startHandshakeServer(t *testing.T, certPath, keyPath string) (port int, stop func()) {
	t.Helper()
	cert, err := tls.LoadX509KeyPair(certPath, keyPath)
	if err != nil {
		t.Fatal(err)
	}
	ln, err := tls.Listen("tcp", "127.0.0.1:0", &tls.Config{Certificates: []tls.Certificate{cert}})
	if err != nil {
		t.Fatal(err)
	}
	go func() {
		for {
			conn, err := ln.Accept()
			if err != nil {
				return
			}
			go func(c net.Conn) {
				defer c.Close()
				if tc, ok := c.(*tls.Conn); ok {
					_ = tc.Handshake()
				}
				_, _ = io.Copy(io.Discard, c)
			}(conn)
		}
	}()
	return ln.Addr().(*net.TCPAddr).Port, func() { _ = ln.Close() }
}

func runCore(t *testing.T, bin, name string, cfg any, dir string) *exec.Cmd {
	t.Helper()
	raw, err := json.Marshal(cfg)
	if err != nil {
		t.Fatal(err)
	}
	path := filepath.Join(dir, name+".json")
	if err := os.WriteFile(path, raw, 0o600); err != nil {
		t.Fatal(err)
	}
	logFile, err := os.Create(filepath.Join(dir, name+".log"))
	if err != nil {
		t.Fatal(err)
	}
	cmd := exec.Command(bin, "run", "-c", path)
	cmd.Stdout = logFile
	cmd.Stderr = logFile
	if err := cmd.Start(); err != nil {
		t.Fatalf("start %s: %v", name, err)
	}
	t.Cleanup(func() {
		_ = cmd.Process.Kill()
		_ = cmd.Wait()
		_ = logFile.Close()
	})
	return cmd
}

func waitPort(t *testing.T, port int, what string) {
	t.Helper()
	deadline := time.Now().Add(10 * time.Second)
	for time.Now().Before(deadline) {
		conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), 200*time.Millisecond)
		if err == nil {
			_ = conn.Close()
			return
		}
		time.Sleep(100 * time.Millisecond)
	}
	t.Fatalf("port %d (%s) never came up", port, what)
}

// fetchViaSocks GETs url through the socks5 listener at port and
// returns the body length.
func fetchViaSocks(t *testing.T, port int, url string) int {
	t.Helper()
	dialer, err := proxy.SOCKS5("tcp", fmt.Sprintf("127.0.0.1:%d", port), nil, proxy.Direct)
	if err != nil {
		t.Fatal(err)
	}
	tr := &http.Transport{DialContext: func(_ context.Context, network, addr string) (net.Conn, error) {
		return dialer.Dial(network, addr)
	}}
	client := &http.Client{Transport: tr, Timeout: 15 * time.Second}
	resp, err := client.Get(url)
	if err != nil {
		t.Fatalf("GET via socks %d: %v", port, err)
	}
	defer resp.Body.Close()
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		t.Fatal(err)
	}
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("GET via socks %d: status %d", port, resp.StatusCode)
	}
	return len(body)
}

func TestSingboxEndToEnd(t *testing.T) {
	bin := os.Getenv("XUI_SINGBOX_BIN")
	if bin == "" {
		t.Skip("XUI_SINGBOX_BIN not set; skipping real-core E2E")
	}
	if _, err := os.Stat(bin); err != nil {
		t.Skipf("sing-box binary unusable: %v", err)
	}
	dir := t.TempDir()
	udpOK := udpLoopbackWorks()
	if !udpOK {
		t.Log("UDP loopback unavailable in this environment; hysteria2 leg excluded (pilot-node item)")
	}

	// --- fixtures ---------------------------------------------------
	gen := func(args ...string) string {
		out, err := exec.Command(bin, args...).Output()
		if err != nil {
			t.Fatalf("sing-box %v: %v", args, err)
		}
		return strings.TrimSpace(string(out))
	}
	var privKey, pubKey string
	for _, line := range strings.Split(gen("generate", "reality-keypair"), "\n") {
		if strings.HasPrefix(line, "PrivateKey:") {
			privKey = strings.TrimSpace(strings.TrimPrefix(line, "PrivateKey:"))
		}
		if strings.HasPrefix(line, "PublicKey:") {
			pubKey = strings.TrimSpace(strings.TrimPrefix(line, "PublicKey:"))
		}
	}
	aliceUUID := gen("generate", "uuid")
	ssServerPass := strings.TrimSpace(gen("generate", "rand", "--base64", "16"))
	ssBobPass := strings.TrimSpace(gen("generate", "rand", "--base64", "16"))

	hsCert, hsKey := selfSignedCert(t, dir, "www.microsoft.com")
	hsPort, stopHS := startHandshakeServer(t, hsCert, hsKey)
	defer stopHS()
	tlsCert, tlsKey := selfSignedCert(t, dir, "example.com")

	targetBody := strings.Repeat("0123456789", 500) // 5000 bytes
	target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = io.WriteString(w, targetBody)
	}))
	defer target.Close()
	targetURL := target.URL + "/"

	pVless := freePort(t)
	pSS := freePort(t)
	pMixed := freePort(t)
	pFwd := freePort(t)
	pTrojan := freePort(t)
	pHy2 := freeUDPPort(t)
	pStats := freePort(t)
	pLanding := freePort(t)
	cVless := freePort(t)
	cSS := freePort(t)
	cMixed := freePort(t)
	cFwd := freePort(t)
	cTrojan := freePort(t)
	cHy2 := freePort(t)

	type leg struct {
		name       string
		clientPort int
		user       string
		serverTag  string
	}
	legs := []leg{
		{"vless-reality", cVless, "alice@example.com", "vless-reality"},
		{"ss2022", cSS, "bob@example.com", "ss22"},
		{"mixed", cMixed, "carol@example.com", "mixed-in"},
		{"trojan-tls", cTrojan, "frank@example.com", "trojan-in"},
		{"forwarding", cFwd, "dave@example.com", "mixed-fwd"},
	}
	if udpOK {
		legs = append(legs, leg{"hysteria2", cHy2, "erin@example.com", "hy2-in"})
	}

	serverInbounds := []any{
		map[string]any{
			"type": "vless", "tag": "vless-reality", "listen": "127.0.0.1", "listen_port": pVless,
			"users": []any{map[string]any{"name": "alice@example.com", "uuid": aliceUUID, "flow": "xtls-rprx-vision"}},
			"tls": map[string]any{
				"enabled": true, "server_name": "www.microsoft.com",
				"reality": map[string]any{
					"enabled":     true,
					"handshake":   map[string]any{"server": "127.0.0.1", "server_port": hsPort},
					"private_key": privKey, "short_id": []string{"abcd"},
				},
			},
		},
		map[string]any{
			"type": "shadowsocks", "tag": "ss22", "listen": "127.0.0.1", "listen_port": pSS,
			"method": "2022-blake3-aes-128-gcm", "password": ssServerPass,
			"users": []any{map[string]any{"name": "bob@example.com", "password": ssBobPass}},
		},
		map[string]any{
			"type": "mixed", "tag": "mixed-in", "listen": "127.0.0.1", "listen_port": pMixed,
			"users": []any{map[string]any{"username": "carol@example.com", "password": "carolpass"}},
		},
		map[string]any{
			"type": "mixed", "tag": "mixed-fwd", "listen": "127.0.0.1", "listen_port": pFwd,
			"users": []any{map[string]any{"username": "dave@example.com", "password": "davepass"}},
		},
		map[string]any{
			"type": "trojan", "tag": "trojan-in", "listen": "127.0.0.1", "listen_port": pTrojan,
			"users": []any{map[string]any{"name": "frank@example.com", "password": "frankpass"}},
			"tls":   map[string]any{"enabled": true, "server_name": "example.com", "certificate_path": tlsCert, "key_path": tlsKey},
		},
	}
	statsInbounds := []string{"vless-reality", "ss22", "mixed-in", "mixed-fwd", "trojan-in"}
	statsUsers := []string{"alice@example.com", "bob@example.com", "carol@example.com", "dave@example.com", "frank@example.com"}
	if udpOK {
		serverInbounds = append(serverInbounds, map[string]any{
			"type": "hysteria2", "tag": "hy2-in", "listen": "127.0.0.1", "listen_port": pHy2,
			"users": []any{map[string]any{"name": "erin@example.com", "password": "erinpass"}},
			"tls":   map[string]any{"enabled": true, "server_name": "example.com", "certificate_path": tlsCert, "key_path": tlsKey},
		})
		statsInbounds = append(statsInbounds, "hy2-in")
		statsUsers = append(statsUsers, "erin@example.com")
	}

	buildServerCfg := func(logLevel string) map[string]any {
		return map[string]any{
			"log":      map[string]any{"level": logLevel, "timestamp": true},
			"inbounds": serverInbounds,
			"outbounds": []any{
				map[string]any{"type": "direct", "tag": "direct"},
				map[string]any{"type": "socks", "tag": "fwd-1", "server": "127.0.0.1", "server_port": pLanding, "version": "5"},
			},
			"route": map[string]any{
				"final": "direct",
				"rules": []any{
					map[string]any{"inbound": []string{"mixed-fwd"}, "action": "route", "outbound": "fwd-1"},
				},
			},
			"experimental": map[string]any{
				"v2ray_api": map[string]any{
					"listen": fmt.Sprintf("127.0.0.1:%d", pStats),
					"stats": map[string]any{
						"enabled":   true,
						"inbounds":  statsInbounds,
						"outbounds": []string{"direct", "fwd-1"},
						"users":     statsUsers,
					},
				},
			},
		}
	}

	// Landing for the forwarding leg: a bare no-auth mixed inbound.
	runCore(t, bin, "landing", map[string]any{
		"log":       map[string]any{"level": "warn"},
		"inbounds":  []any{map[string]any{"type": "mixed", "tag": "landing-in", "listen": "127.0.0.1", "listen_port": pLanding}},
		"outbounds": []any{map[string]any{"type": "direct", "tag": "direct"}},
	}, dir)
	waitPort(t, pLanding, "landing")

	// Client core: one local socks port per protocol leg.
	clientInbounds := []any{
		map[string]any{"type": "mixed", "tag": "in-vless", "listen": "127.0.0.1", "listen_port": cVless},
		map[string]any{"type": "mixed", "tag": "in-ss", "listen": "127.0.0.1", "listen_port": cSS},
		map[string]any{"type": "mixed", "tag": "in-mixed", "listen": "127.0.0.1", "listen_port": cMixed},
		map[string]any{"type": "mixed", "tag": "in-fwd", "listen": "127.0.0.1", "listen_port": cFwd},
		map[string]any{"type": "mixed", "tag": "in-trojan", "listen": "127.0.0.1", "listen_port": cTrojan},
	}
	clientOutbounds := []any{
		map[string]any{
			"type": "vless", "tag": "out-vless", "server": "127.0.0.1", "server_port": pVless,
			"uuid": aliceUUID, "flow": "xtls-rprx-vision",
			"tls": map[string]any{
				"enabled": true, "server_name": "www.microsoft.com",
				"utls":    map[string]any{"enabled": true, "fingerprint": "chrome"},
				"reality": map[string]any{"enabled": true, "public_key": pubKey, "short_id": "abcd"},
			},
		},
		map[string]any{
			"type": "shadowsocks", "tag": "out-ss", "server": "127.0.0.1", "server_port": pSS,
			"method": "2022-blake3-aes-128-gcm", "password": ssServerPass + ":" + ssBobPass,
		},
		map[string]any{"type": "socks", "tag": "out-mixed", "server": "127.0.0.1", "server_port": pMixed, "version": "5",
			"username": "carol@example.com", "password": "carolpass"},
		map[string]any{"type": "socks", "tag": "out-fwd", "server": "127.0.0.1", "server_port": pFwd, "version": "5",
			"username": "dave@example.com", "password": "davepass"},
		map[string]any{
			"type": "trojan", "tag": "out-trojan", "server": "127.0.0.1", "server_port": pTrojan, "password": "frankpass",
			"tls": map[string]any{"enabled": true, "server_name": "example.com", "insecure": true},
		},
	}
	clientRules := []any{
		map[string]any{"inbound": []string{"in-vless"}, "action": "route", "outbound": "out-vless"},
		map[string]any{"inbound": []string{"in-ss"}, "action": "route", "outbound": "out-ss"},
		map[string]any{"inbound": []string{"in-mixed"}, "action": "route", "outbound": "out-mixed"},
		map[string]any{"inbound": []string{"in-fwd"}, "action": "route", "outbound": "out-fwd"},
		map[string]any{"inbound": []string{"in-trojan"}, "action": "route", "outbound": "out-trojan"},
	}
	clientPorts := []int{cVless, cSS, cMixed, cFwd, cTrojan}
	if udpOK {
		clientInbounds = append(clientInbounds,
			map[string]any{"type": "mixed", "tag": "in-hy2", "listen": "127.0.0.1", "listen_port": cHy2})
		clientOutbounds = append(clientOutbounds, map[string]any{
			"type": "hysteria2", "tag": "out-hy2", "server": "127.0.0.1", "server_port": pHy2, "password": "erinpass",
			"tls": map[string]any{"enabled": true, "server_name": "example.com", "insecure": true, "alpn": []string{"h3"}},
		})
		clientRules = append(clientRules,
			map[string]any{"inbound": []string{"in-hy2"}, "action": "route", "outbound": "out-hy2"})
		clientPorts = append(clientPorts, cHy2)
	}
	clientOutbounds = append(clientOutbounds, map[string]any{"type": "direct", "tag": "direct"})
	runCore(t, bin, "client", map[string]any{
		"log":       map[string]any{"level": "warn"},
		"inbounds":  clientInbounds,
		"outbounds": clientOutbounds,
		"route":     map[string]any{"final": "direct", "rules": clientRules},
	}, dir)
	for _, p := range clientPorts {
		waitPort(t, p, "client")
	}

	// --- agent under test --------------------------------------------
	binDir := filepath.Join(dir, "bin")
	if err := os.MkdirAll(binDir, 0o755); err != nil {
		t.Fatal(err)
	}
	linkPath := filepath.Join(binDir, singboxBinaryName())
	if err := os.Symlink(bin, linkPath); err != nil {
		data, rerr := os.ReadFile(bin)
		if rerr != nil {
			t.Fatal(rerr)
		}
		if werr := os.WriteFile(linkPath, data, 0o755); werr != nil {
			t.Fatal(werr)
		}
	}
	t.Setenv("XUI_BIN_FOLDER", binDir)

	master := &e2eMaster{}
	cfgA, _ := json.Marshal(buildServerCfg("info"))
	master.setConfig(cfgA)
	srv := httptest.NewServer(master.handler())
	defer srv.Close()

	agt := New(&Config{
		MasterURL:    srv.URL,
		Token:        "e2e-token",
		NodeName:     "e2e",
		BaselinePath: filepath.Join(dir, "agent-traffic-baseline.json"),
	})
	t.Cleanup(agt.stopSingbox)

	// Apply. The xray half of the poll fails by design (no xray
	// binary in the sandbox bin folder); the sing-box half must
	// have succeeded before that error is returned.
	_ = agt.fetchAndApplyConfig()
	agt.sbMu.Lock()
	running := agt.sbProc != nil && agt.sbProc.isRunning()
	agt.sbMu.Unlock()
	if !running {
		t.Fatal("agent did not start sing-box from the delivered config")
	}
	for _, p := range []int{pVless, pSS, pMixed, pFwd, pTrojan, pStats} {
		waitPort(t, p, "agent sing-box")
	}

	cycle := func() {
		t.Helper()
		if err := agt.collectAndReportSingboxTraffic(); err != nil {
			t.Fatalf("traffic cycle: %v", err)
		}
	}

	// First cycle only establishes the baseline: traffic generated
	// before it must NOT show up as deltas.
	if n := fetchViaSocks(t, cSS, targetURL); n != len(targetBody) {
		t.Fatalf("ss leg body = %d, want %d", n, len(targetBody))
	}
	cycle()
	users, _, _ := master.totals()
	if len(users) != 0 {
		t.Fatalf("first cycle reported user deltas before baseline: %v", users)
	}

	// --- per-protocol traffic, then a reporting cycle -----------------
	downloads := map[string]int{}
	for _, lg := range legs {
		n := fetchViaSocks(t, lg.clientPort, targetURL)
		if n != len(targetBody) {
			t.Fatalf("%s leg body = %d, want %d", lg.name, n, len(targetBody))
		}
		downloads[lg.user] += n
	}
	cycle()

	users, inbounds, onlines := master.totals()
	for _, lg := range legs {
		got := users[lg.user]
		want := int64(downloads[lg.user])
		if got[1] < want || got[1] > want*3/2+2048 {
			t.Fatalf("%s (%s): reported down = %d, want within [%d, %d]", lg.name, lg.user, got[1], want, want*3/2+2048)
		}
		if got[0] <= 0 {
			t.Fatalf("%s (%s): reported up = %d, want > 0", lg.name, lg.user, got[0])
		}
		found := false
		for _, e := range onlines {
			if e == lg.user {
				found = true
			}
		}
		if !found {
			t.Fatalf("%s (%s) missing from onlines %v", lg.name, lg.user, onlines)
		}
		if got := inbounds[lg.serverTag]; got[1] <= 0 {
			t.Fatalf("inbound %s reported down = %d, want > 0", lg.serverTag, got[1])
		}
	}

	// --- config change: flush + epoch rebase, no double counting -----
	if n := fetchViaSocks(t, cSS, targetURL); n != len(targetBody) {
		t.Fatalf("pre-change ss body = %d", n)
	}
	downloads["bob@example.com"] += len(targetBody)
	cfgB, _ := json.Marshal(buildServerCfg("debug"))
	master.setConfig(cfgB)
	agt.sbMu.Lock()
	pidBefore := agt.sbProc.pid()
	agt.sbMu.Unlock()
	_ = agt.fetchAndApplyConfig()
	agt.sbMu.Lock()
	pidAfter := agt.sbProc.pid()
	running = agt.sbProc.isRunning()
	agt.sbMu.Unlock()
	if !running || pidAfter == pidBefore {
		t.Fatalf("config change did not restart the core: pid %d -> %d running=%v", pidBefore, pidAfter, running)
	}
	for _, p := range []int{pVless, pSS, pMixed, pFwd, pTrojan, pStats} {
		waitPort(t, p, "agent sing-box after restart")
	}

	if n := fetchViaSocks(t, cSS, targetURL); n != len(targetBody) {
		t.Fatalf("post-change ss body = %d", n)
	}
	downloads["bob@example.com"] += len(targetBody)
	cycle()
	cycle() // a second cycle must add nothing new

	users, _, _ = master.totals()
	bob := users["bob@example.com"]
	want := int64(downloads["bob@example.com"])
	if bob[1] < want || bob[1] > want*3/2+4096 {
		t.Fatalf("bob total down after restart = %d, want within [%d, %d] (double counting or loss)", bob[1], want, want*3/2+4096)
	}

	// --- the check gate -------------------------------------------------
	badCfg := []byte(`{"inbounds":[{"type":"no-such-type","tag":"x"}],"outbounds":[{"type":"direct","tag":"direct"}]}`)
	badSum := sha256.Sum256(badCfg)
	master.setConfig(badCfg)
	_ = agt.fetchAndApplyConfig()
	if agt.lastSingboxHash == hex.EncodeToString(badSum[:]) {
		t.Fatal("bad config hash must not be recorded as applied")
	}
	agt.sbMu.Lock()
	pidNow := agt.sbProc.pid()
	running = agt.sbProc.isRunning()
	agt.sbMu.Unlock()
	if !running || pidNow != pidAfter {
		t.Fatalf("check-failing config disturbed the core: pid %d -> %d running=%v", pidAfter, pidNow, running)
	}
	applied, err := os.ReadFile(singboxConfigPath())
	if err != nil {
		t.Fatal(err)
	}
	if string(applied) != string(cfgB) {
		t.Fatal("active config file changed after a rejected apply")
	}
}