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

// Package agent implements the lightweight 3x-ui node agent.
// In agent mode, the node does not run the web panel or database.
// It registers with the master via token, receives xray config, manages the
// xray process, and reports traffic back to the master.
package agent

import (
	"bytes"
	"crypto/tls"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/mhsanaei/3x-ui/v3/internal/config"
	"github.com/mhsanaei/3x-ui/v3/internal/logger"
	"github.com/mhsanaei/3x-ui/v3/internal/xray"
)

const (
	// Default poll interval for config updates
	defaultPollInterval = 30 * time.Second
	// Heartbeat interval
	defaultHeartbeatInterval = 60 * time.Second
	// Traffic report interval (own ticker; 10s keeps agent inbounds'
	// traffic/onlines freshness in the same order of magnitude as the
	// master's seconds-level polling of traditional panel nodes)
	defaultTrafficInterval = 10 * time.Second
)

// DefaultBinFolder is used when XUI_BIN_FOLDER is unset. It matches the
// panel's directory layout: xray binary, geo files and config.json live here.
const DefaultBinFolder = "/usr/local/x-ui/bin"

// Config holds the agent configuration
type Config struct {
	MasterURL    string        `json:"master_url"`
	Token        string        `json:"token"`
	NodeName     string        `json:"node_name"`
	PollInterval time.Duration `json:"poll_interval"`
	// ApiPort overrides the xray stats API inbound port from the master's
	// config. Needed when agent and master share a host (e.g. testing),
	// otherwise both xrays fight over the default 62789. 0 = use master's.
	ApiPort int `json:"api_port"`
	// BaselinePath is where the last master-acknowledged traffic counters
	// are persisted, so an agent restart keeps diffing against them
	// instead of swallowing unreported traffic into a fresh baseline.
	// Empty: main derives it next to the agent config file; the agent
	// falls back to the bin folder.
	BaselinePath string `json:"baseline_path,omitempty"`
}

// Agent represents the node agent
type Agent struct {
	config         *Config
	httpClient     *http.Client
	stopCh         chan struct{}
	lastConfigHash string
	xrayProc       *xray.Process
	xrayAPI        xray.XrayAPI
	apiPort        int
	baselineLoaded bool

	// Heartbeat telemetry state (P13): last measured master-request RTT
	// and the mtime-keyed xray version cache.
	lastRTTms  atomic.Int64
	hasRTT     atomic.Bool
	xrayVerMu  sync.Mutex
	xrayVer    string
	xrayVerFor string // binary path the cached version was read from
	xrayVerMod time.Time

	// One-click update bookkeeping (item 20, see update.go).
	updateMu        sync.Mutex
	updateRunning   bool
	updateHandled   int64
	updateFailedReq int64
	updateFailedAt  time.Time
	updateReport    *agentUpdateStatus
	exePathOverride string

	// sing-box runtime (ITEM6 B3, see singbox.go). sbMu serializes
	// config activation, the sing-box traffic cycle and shutdown: a
	// flush-before-change must never race a periodic snapshot.
	sbMu             sync.Mutex
	sbProc           *singboxProcess
	sbStats          *singboxStatsClient
	sbOnline         *singboxOnlineTracker
	sbStatsAddr      string
	lastSingboxHash  string
	sbBaselineLoaded bool

	// Switchover state (ITEM6 B5): set once the master serves a
	// sing-box config — from then on sing-box is the node's only
	// runtime and no xray payload is ever applied again, even if a
	// later poll somehow lacks the sing-box keys. lastXrayConfig
	// keeps the last applied xray payload so a failed sing-box
	// activation during the switch can restart xray and keep the
	// node serving while the switch retries on the next poll.
	sbSwitched     bool
	lastXrayConfig json.RawMessage

	// sing-box version cache (telemetry), mirroring the xray one.
	sbVerMu  sync.Mutex
	sbVer    string
	sbVerFor string
	sbVerMod time.Time

	// Core fetch backoff (V2.0.1 ①, see singbox_fetch.go).
	sbFetchMu       sync.Mutex
	sbFetchFailedAt time.Time

	// Fast-report state (V2.0.1 probe batch): the cadence and
	// counted-interface pin delivered on the config poll. A zero
	// fastInterval means the master never offered one (pre-V2.0.1
	// master) and metrics keep riding the 60 s heartbeat.
	fastIntervalSec atomic.Int64
	fastUnsupported atomic.Bool

	// Latency probe runner (item 7 B4, see ping.go): task set pushed
	// on the config poll, samples returned on the heartbeat.
	probes *probeRunner
}

// ensureSingboxState lazily creates the sing-box runtime state.
// Callers hold sbMu (or are in single-threaded setup).
func (a *Agent) ensureSingboxState() {
	if a.sbProc == nil {
		a.sbProc = &singboxProcess{}
	}
	if a.sbStats == nil {
		a.sbStats = newSingboxStatsClient(a.sbProc.epoch)
	}
	if a.sbOnline == nil {
		a.sbOnline = newSingboxOnlineTracker(singboxLogPath())
	}
}

// New creates a new agent instance
func New(cfg *Config) *Agent {
	if cfg.PollInterval == 0 {
		cfg.PollInterval = defaultPollInterval
	}
	return &Agent{
		config: cfg,
		httpClient: &http.Client{
			Timeout: 30 * time.Second,
			Transport: &http.Transport{
				TLSClientConfig: &tls.Config{InsecureSkipVerify: false},
			},
		},
		stopCh: make(chan struct{}),
		probes: newProbeRunner(),
	}
}

// statusError is a non-200 response from the master. It keeps the status
// code addressable so the registration retry loop can tell a wrong-path
// 404 (master_url missing the panel base path) apart from other failures.
type statusError struct {
	code int
	body string
}

func (e *statusError) Error() string {
	return fmt.Sprintf("register failed with status %d: %s", e.code, e.body)
}

// Register registers the agent with the master
func (a *Agent) Register() error {
	url := fmt.Sprintf("%s/panel/api/agent/register", a.config.MasterURL)
	payload := map[string]any{
		"token": a.config.Token,
		"name":  a.config.NodeName,
		// Probe facts ride along (B1): old masters ignore the extra field.
		"probeFacts": collectFacts(),
	}
	body, _ := json.Marshal(payload)
	req, err := http.NewRequest("POST", url, bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+a.config.Token)

	resp, err := a.do(req)
	if err != nil {
		return fmt.Errorf("register failed: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		data, _ := io.ReadAll(resp.Body)
		return &statusError{code: resp.StatusCode, body: string(data)}
	}

	logger.Info("Agent registered successfully with master")
	return nil
}

// registrationBackoff returns the delay before the next registration
// attempt: 5s, 15s, 30s, then capped at 60s.
func registrationBackoff(failures int) time.Duration {
	switch {
	case failures <= 1:
		return 5 * time.Second
	case failures == 2:
		return 15 * time.Second
	case failures == 3:
		return 30 * time.Second
	default:
		return 60 * time.Second
	}
}

// registerWait sleeps between registration attempts, returning false when
// the agent is stopping. A package var so tests can drive the retry loop
// without real delays.
var registerWait = func(stopCh chan struct{}, d time.Duration) bool {
	select {
	case <-stopCh:
		return false
	case <-time.After(d):
		return true
	}
}

// errRegistrationAborted marks a registration loop cut short by Stop.
var errRegistrationAborted = errors.New("registration aborted: agent stopping")

// registerWithRetry keeps trying to register until the master accepts or
// the agent is stopped. A one-shot failure used to kill the agent
// outright — a typo'd master_url (classically: the panel base path
// missing, so the master answers 404) left the node dead and "unknown"
// on the panel with nothing retrying. Registration is idempotent on the
// master, so retrying with backoff is safe.
func (a *Agent) registerWithRetry() error {
	failures := 0
	for {
		err := a.Register()
		if err == nil {
			return nil
		}
		failures++
		var st *statusError
		if errors.As(err, &st) && st.code == http.StatusNotFound {
			logger.Warningf("主控返回 404：检查 master_url 是否包含面板基础路径（如 https://host/rogue），当前 master_url=%s", a.config.MasterURL)
		}
		delay := registrationBackoff(failures)
		logger.Warningf("Registration attempt %d failed: %v; retrying in %s", failures, err, delay)
		if !registerWait(a.stopCh, delay) {
			return errRegistrationAborted
		}
	}
}

// Run starts the agent main loop
func (a *Agent) Run() error {
	logger.Info("Starting 3x-ui agent...")
	logger.Infof("Master: %s, Node: %s", a.config.MasterURL, a.config.NodeName)

	// Initial registration, retried with backoff: a node that cannot
	// register yet (master down, URL just fixed, network still coming up)
	// must not die on the first failure.
	if err := a.registerWithRetry(); err != nil {
		if errors.Is(err, errRegistrationAborted) {
			logger.Info("Agent stopping...")
			return nil
		}
		return err
	}

	// Start heartbeat ticker
	heartbeatTicker := time.NewTicker(defaultHeartbeatInterval)
	defer heartbeatTicker.Stop()

	// Start config poll ticker
	pollTicker := time.NewTicker(a.config.PollInterval)
	defer pollTicker.Stop()

	// Start traffic report ticker
	trafficTicker := time.NewTicker(defaultTrafficInterval)
	defer trafficTicker.Stop()

	// Fast-report timer (V2.0.1): armed by the config poll once the
	// master offers a cadence; disarmed while the master is older
	// (no cadence offered) or answered 404 (endpoint unknown).
	fastTimer := time.NewTimer(time.Hour)
	fastTimer.Stop()
	defer fastTimer.Stop()

	// Initial config fetch
	if err := a.fetchAndApplyConfig(); err != nil {
		logger.Warningf("Initial config fetch failed: %v", err)
	}
	a.armFastTimer(fastTimer)

	for {
		select {
		case <-a.stopCh:
			logger.Info("Agent stopping...")
			a.stopXray()
			a.stopSingbox()
			return nil
		case <-heartbeatTicker.C:
			if err := a.sendHeartbeat(); err != nil {
				logger.Warningf("Heartbeat failed: %v", err)
			}
		case <-pollTicker.C:
			if err := a.fetchAndApplyConfig(); err != nil {
				logger.Warningf("Config fetch failed: %v", err)
			}
			a.armFastTimer(fastTimer)
		case <-trafficTicker.C:
			if err := a.collectAndReportTraffic(); err != nil {
				logger.Warningf("Traffic report failed: %v", err)
			}
		case <-fastTimer.C:
			a.sendFastReport()
			a.armFastTimer(fastTimer)
		}
	}
}

// armFastTimer (re)schedules the fast report from the current
// master-provided cadence. Not offered (0) or unsupported (404
// seen) leaves the timer stopped: metrics then ride the heartbeat,
// the pre-V2.0.1 arrangement.
func (a *Agent) armFastTimer(t *time.Timer) {
	sec := a.fastIntervalSec.Load()
	if sec <= 0 || a.fastUnsupported.Load() {
		t.Stop()
		return
	}
	t.Reset(time.Duration(sec) * time.Second)
}

// fastReportActive reports whether runtime metrics currently ride
// the fast channel (and so must stay off the heartbeat).
func (a *Agent) fastReportActive() bool {
	return a.fastIntervalSec.Load() > 0 && !a.fastUnsupported.Load()
}

// sendFastReport posts one metrics-only report (V2.0.1). Failures
// are quiet by design — the stream is high-cadence and the next
// tick retries; a 404 marks the endpoint unsupported (pre-V2.0.1
// master) and metrics stay on the heartbeat from then on.
func (a *Agent) sendFastReport() {
	m := collectMetrics()
	if m == nil {
		return
	}
	body, err := json.Marshal(map[string]any{"probeMetrics": m})
	if err != nil {
		return
	}
	url := a.config.MasterURL + "/panel/api/agent/probe-report"
	req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(body))
	if err != nil {
		return
	}
	req.Header.Set("Authorization", "Bearer "+a.config.Token)
	req.Header.Set("Content-Type", "application/json")
	resp, err := a.httpClient.Do(req)
	if err != nil {
		return
	}
	defer resp.Body.Close()
	_, _ = io.Copy(io.Discard, resp.Body)
	if resp.StatusCode == http.StatusNotFound {
		a.fastUnsupported.Store(true)
		logger.Warning("master does not serve the probe fast-report endpoint; metrics stay on the heartbeat")
	}
}

// Stop stops the agent
func (a *Agent) Stop() {
	if a.probes != nil {
		a.probes.stop()
	}
	close(a.stopCh)
}

// do performs one master request and records its round-trip time as the
// node->master latency sample reported by the next heartbeat (P13).
func (a *Agent) do(req *http.Request) (*http.Response, error) {
	start := time.Now()
	resp, err := a.httpClient.Do(req)
	if err == nil {
		a.lastRTTms.Store(time.Since(start).Milliseconds())
		a.hasRTT.Store(true)
	}
	return resp, err
}

// sendHeartbeat sends a heartbeat to the master
func (a *Agent) sendHeartbeat() error {
	url := fmt.Sprintf("%s/panel/api/agent/heartbeat", a.config.MasterURL)
	payload := map[string]interface{}{
		"token":     a.config.Token,
		"timestamp": time.Now().Unix(),
		"telemetry": a.collectTelemetry(),
		// Probe snapshot (B1): facts are process-cached so re-sending them
		// costs nothing and heals a master that missed the register-time
		// copy; metrics are sampled fresh each heartbeat. Old masters
		// ignore both fields.
		"probeFacts": collectFacts(),
	}
	// V2.0.1: runtime metrics ride the fast report whenever the
	// master offered a cadence and the endpoint answered; the
	// heartbeat then carries only the heavy, slow-moving content
	// (facts, telemetry, probe results, update status). Without a
	// fast channel the metrics stay here — a pre-V2.0.1 master must
	// keep receiving them exactly as before.
	if !a.fastReportActive() {
		payload["probeMetrics"] = collectMetrics()
	}
	if rep := a.currentUpdateReport(); rep != nil {
		payload["updateStatus"] = rep
	}
	// Latency probe samples (B4) ride the heartbeat — deliberately not
	// the traffic report, which stops when xray is down while probing
	// must not. Drained here, requeued on any failure below.
	var probeBatch []ProbeResult
	if a.probes != nil {
		probeBatch = a.probes.drainResults()
		if len(probeBatch) > 0 {
			payload["probeResults"] = probeBatch
		}
	}
	body, _ := json.Marshal(payload)
	req, err := http.NewRequest("POST", url, bytes.NewReader(body))
	if err != nil {
		a.requeueProbeBatch(probeBatch)
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+a.config.Token)

	resp, err := a.do(req)
	if err != nil {
		a.requeueProbeBatch(probeBatch)
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		a.requeueProbeBatch(probeBatch)
		return fmt.Errorf("heartbeat failed with status %d", resp.StatusCode)
	}
	return nil
}

// requeueProbeBatch returns a heartbeat's probe samples to the
// runner buffer after a failed send.
func (a *Agent) requeueProbeBatch(batch []ProbeResult) {
	if a.probes != nil {
		a.probes.requeueResults(batch)
	}
}

// fetchAndApplyConfig fetches the config poll response from the
// master and applies what changed: update intents, probe tasks, and
// the runtime config — sing-box since the B5 switchover, xray on a
// legacy master.
func (a *Agent) fetchAndApplyConfig() error {
	url := fmt.Sprintf("%s/panel/api/agent/config", a.config.MasterURL)
	req, err := http.NewRequest("GET", url, nil)
	if err != nil {
		return err
	}
	req.Header.Set("Authorization", "Bearer "+a.config.Token)

	resp, err := a.do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		data, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("config fetch failed with status %d: %s", resp.StatusCode, string(data))
	}

	var result struct {
		ConfigHash        string          `json:"config_hash"`
		XrayConfig        json.RawMessage `json:"xray_config"`
		SingboxConfig     json.RawMessage `json:"singbox_config"`
		SingboxConfigHash string          `json:"singbox_config_hash"`
		AgentUpdate       *updateIntent   `json:"agent_update"`
		// ProbeTasks is a pointer on purpose (B4): a master too old
		// to know probes omits the field and the agent must keep its
		// current set, while a present-but-empty list means "run no
		// probes" and clears it.
		ProbeTasks *[]ProbeTask `json:"probe_tasks"`
		// V2.0.1 probe batch: fast-report cadence + counted-interface
		// pin. Pointers for the same reason — an old master omits
		// both and the agent keeps its current behaviour (no fast
		// report, automatic interface selection).
		ProbeFastInterval *int64  `json:"probe_fast_interval"`
		ProbeStatIface    *string `json:"probe_stat_iface"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return err
	}

	if result.ProbeFastInterval != nil {
		sec := *result.ProbeFastInterval
		if sec < 1 {
			sec = 1
		}
		if sec > 60 {
			sec = 60
		}
		a.fastIntervalSec.Store(sec)
	}
	if result.ProbeStatIface != nil {
		SetStatIfacePin(*result.ProbeStatIface)
	}

	// One-click update intents ride this poll (item 20); handle them even
	// when the config itself is unchanged.
	a.maybeHandleUpdate(result.AgentUpdate)

	// Probe tasks change independently of any core config, so they
	// are applied before the runtime branches below.
	if result.ProbeTasks != nil && a.probes != nil {
		a.probes.applyTasks(*result.ProbeTasks)
	}

	// Switchover (ITEM6 B5): a master that serves a sing-box config
	// has made sing-box the node's formal runtime. Apply the config
	// through the switch sequence (which retires the legacy xray
	// core), and never touch the xray path again — a later poll that
	// somehow lacks the sing-box keys leaves both cores as they are
	// rather than resurrecting xray.
	//
	// V2.0.1 ① gate: the switch needs the sing-box core on disk. A
	// node whose core is missing gets one fetch attempt (throttled);
	// while it stays missing the switch is NOT latched and the xray
	// path below keeps the node current — nothing is retired for a
	// runtime that cannot start yet.
	if result.SingboxConfigHash != "" && a.singboxCoreReady() {
		a.sbSwitched = true
		if result.SingboxConfigHash != a.lastSingboxHash {
			if err := a.switchApplySingbox(result.SingboxConfig); err != nil {
				logger.Warningf("Failed to apply sing-box config: %v", err)
			} else {
				a.lastSingboxHash = result.SingboxConfigHash
				logger.Info("sing-box config applied successfully")
			}
		} else if !a.singboxRunning() {
			// The config is current but the core is not running
			// (crashed, or never started because the xray flush kept
			// failing before the first activation): re-apply through
			// the switch sequence — activation restarts a dead core
			// and is a no-op for a live one.
			if err := a.switchApplySingbox(result.SingboxConfig); err != nil {
				logger.Warningf("Failed to restart sing-box core: %v", err)
			}
		} else if a.xrayProc != nil {
			// The sing-box side is live and current, but a leftover
			// xray core is still running (its pre-switch flush kept
			// failing, or the agent was restarted mid-switch): retire
			// it now so the node converges on sing-box alone.
			a.retireXrayCore()
		}
		return nil
	}
	if a.sbSwitched {
		return nil
	}

	// Legacy path (pre-switchover master): only an xray payload is
	// served, applied exactly as before.
	if len(result.XrayConfig) == 0 {
		return nil
	}
	// Skip if config hasn't changed
	if result.ConfigHash == a.lastConfigHash && result.ConfigHash != "" {
		return nil
	}

	logger.Info("New xray config received, applying...")
	if err := a.applyXrayConfig(result.XrayConfig); err != nil {
		return fmt.Errorf("failed to apply xray config: %w", err)
	}

	a.lastConfigHash = result.ConfigHash
	a.lastXrayConfig = append(json.RawMessage(nil), result.XrayConfig...)
	logger.Info("Xray config applied successfully")
	return nil
}

// singboxRunning reports whether the managed sing-box process is
// currently alive.
func (a *Agent) singboxRunning() bool {
	a.sbMu.Lock()
	defer a.sbMu.Unlock()
	return a.sbProc != nil && a.sbProc.isRunning()
}

// switchApplySingbox activates a sing-box config in switchover mode.
// While the legacy xray core is still running it owns the node's
// inbound ports, so sing-box could never bind them: the order is
// flush xray's tail traffic (its counters die with the process),
// stop xray, then activate sing-box. A failed flush aborts the
// switch for this poll — nothing has been touched and the next poll
// retries. A failed activation restarts xray from the last applied
// payload so the node keeps serving while the switch retries.
func (a *Agent) switchApplySingbox(raw json.RawMessage) error {
	if a.xrayProc == nil {
		return a.applySingboxConfig(raw)
	}
	if err := a.collectAndReportXrayTraffic(); err != nil {
		return fmt.Errorf("flush xray traffic before switchover: %w", err)
	}
	a.stopXray()
	if err := a.applySingboxConfig(raw); err != nil {
		logger.Warningf("sing-box activation failed during switchover (%v); restarting xray with the last applied config", err)
		if len(a.lastXrayConfig) > 0 {
			if rbErr := a.applyXrayConfig(a.lastXrayConfig); rbErr != nil {
				logger.Warningf("xray restart after failed switchover also failed: %v", rbErr)
			}
		}
		return err
	}
	logger.Info("switchover complete: xray core retired, sing-box is the node's runtime")
	return nil
}

// retireXrayCore flushes and stops a leftover xray core after the
// sing-box side is already current. A failed flush keeps xray
// running (its unreported tail is worth more than a fast retire)
// and the next poll retries.
func (a *Agent) retireXrayCore() {
	if err := a.collectAndReportXrayTraffic(); err != nil {
		logger.Warningf("flush xray traffic before retiring xray core failed: %v (will retry next poll)", err)
		return
	}
	a.stopXray()
	logger.Info("xray core retired: sing-box is the node's runtime")
}

// ensureXrayBinary makes sure the xray binary exists at the panel-layout path.
// The install doc puts it at /usr/local/x-ui/bin/xray; the panel's process
// manager looks for the arch-specific name, so link one to the other.
func ensureXrayBinary() error {
	binPath := xray.GetBinaryPath()
	if _, err := os.Stat(binPath); err == nil {
		return nil
	}
	plain := filepath.Join(config.GetBinFolderPath(), "xray")
	if _, err := os.Stat(plain); err != nil {
		return fmt.Errorf("xray binary not found at %s or %s", plain, binPath)
	}
	// Best effort: link the arch-specific name the process manager expects.
	if err := os.Symlink(plain, binPath); err != nil && !os.IsExist(err) {
		return fmt.Errorf("failed to link xray binary: %w", err)
	}
	return nil
}

// applyXrayConfig writes the xray config and (re)starts xray through the
// panel's process manager.
func (a *Agent) applyXrayConfig(configJSON json.RawMessage) error {
	if err := ensureXrayBinary(); err != nil {
		return err
	}
	var cfg xray.Config
	if err := json.Unmarshal(configJSON, &cfg); err != nil {
		return fmt.Errorf("invalid xray config: %w", err)
	}
	// Same-host override: rewrite the stats API inbound port so the agent's
	// xray doesn't fight the master's xray over the default port.
	if a.config.ApiPort > 0 {
		for i := range cfg.InboundConfigs {
			if cfg.InboundConfigs[i].Tag == "api" {
				logger.Infof("Overriding xray API port %d -> %d (agent config)",
					cfg.InboundConfigs[i].Port, a.config.ApiPort)
				cfg.InboundConfigs[i].Port = a.config.ApiPort
			}
		}
		// Also move the prometheus metrics listener (default 127.0.0.1:11111)
		// which would collide with the master's xray on the same host.
		if len(cfg.Metrics) > 0 {
			var m map[string]any
			if err := json.Unmarshal(cfg.Metrics, &m); err == nil {
				if ls, ok := m["listen"].(string); ok && ls != "" {
					newListen := fmt.Sprintf("127.0.0.1:%d", a.config.ApiPort+1)
					logger.Infof("Overriding xray metrics listen %s -> %s (agent config)", ls, newListen)
					m["listen"] = newListen
					if raw, err := json.Marshal(m); err == nil {
						cfg.Metrics = raw
					}
				}
			}
		}
	}
	a.apiPort = apiPortFromConfig(&cfg)

	a.stopXray()

	// Use a dedicated config file so the agent's xray doesn't fight the
	// master's xray over the shared config.json when both run on one host.
	agentConfigPath := config.GetBinFolderPath() + "/config-agent.json"
	proc := xray.NewTestProcess(&cfg, agentConfigPath)
	if err := proc.Start(); err != nil {
		return fmt.Errorf("failed to start xray: %w", err)
	}
	a.xrayProc = proc
	return nil
}

func (a *Agent) stopXray() {
	if a.xrayProc != nil {
		_ = a.xrayProc.Stop()
		a.xrayProc = nil
	}
}

// apiPortFromConfig extracts the stats API port (the inbound tagged "api").
func apiPortFromConfig(cfg *xray.Config) int {
	for _, inbound := range cfg.InboundConfigs {
		if inbound.Tag == "api" {
			return inbound.Port
		}
	}
	return 0
}

type trafficReport struct {
	Inbounds []trafficEntry `json:"inbounds"`
	Users    []userEntry    `json:"users"`
	Onlines  []string       `json:"onlines"`
}

type trafficEntry struct {
	Tag  string `json:"tag"`
	Up   int64  `json:"up"`
	Down int64  `json:"down"`
}

type userEntry struct {
	Email string `json:"email"`
	Up    int64  `json:"up"`
	Down  int64  `json:"down"`
}

// statsClient is the slice of the xray stats API the traffic reporter
// uses. The snapshot/commit pair is the loss-proofing contract: snapshots
// never advance the baseline, commits do.
type statsClient interface {
	GetTrafficSnapshot() (*xray.TrafficSnapshot, error)
	CommitTrafficSnapshot(*xray.TrafficSnapshot)
	RestoreTrafficBaseline(map[string]int64)
	GetOnlineUsers() ([]xray.OnlineUser, error)
}

// trafficBaselineFile is the on-disk form of the last committed baseline.
type trafficBaselineFile struct {
	Version int              `json:"version"`
	Values  map[string]int64 `json:"values"`
}

// baselinePath resolves where the acknowledged traffic baseline lives.
func (a *Agent) baselinePath() string {
	if a.config.BaselinePath != "" {
		return a.config.BaselinePath
	}
	return filepath.Join(config.GetBinFolderPath(), "agent-traffic-baseline.json")
}

// loadTrafficBaseline reads a persisted baseline. A missing file is a
// normal first run; a corrupt one is reported so the caller can log it —
// both yield no values and the reporter falls back to a baseline-only
// first snapshot rather than crashing the agent.
func loadTrafficBaseline(path string) (map[string]int64, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return nil, err
	}
	var f trafficBaselineFile
	if err := json.Unmarshal(data, &f); err != nil {
		return nil, fmt.Errorf("parse traffic baseline: %w", err)
	}
	return f.Values, nil
}

// saveTrafficBaseline persists the baseline atomically (write-then-rename)
// so a crash mid-write can never leave a torn file behind.
func saveTrafficBaseline(path string, values map[string]int64) error {
	data, err := json.Marshal(trafficBaselineFile{Version: 1, Values: values})
	if err != nil {
		return err
	}
	tmp := path + ".tmp"
	if err := os.WriteFile(tmp, data, 0o600); err != nil {
		return err
	}
	return os.Rename(tmp, path)
}

// buildTrafficReport shapes a snapshot plus the freshly polled online
// users into the wire report, dropping zero-traffic entries. Onlines are
// point-in-time data: they ride along with each report and are never part
// of the baseline bookkeeping.
func buildTrafficReport(snap *xray.TrafficSnapshot, onlineUsers []xray.OnlineUser) trafficReport {
	report := trafficReport{}
	for _, t := range snap.Traffics {
		if t == nil || !t.IsInbound || t.Up+t.Down <= 0 {
			continue
		}
		report.Inbounds = append(report.Inbounds, trafficEntry{Tag: t.Tag, Up: t.Up, Down: t.Down})
	}
	for _, ct := range snap.ClientTraffics {
		if ct == nil || ct.Email == "" || ct.Up+ct.Down <= 0 {
			continue
		}
		report.Users = append(report.Users, userEntry{Email: ct.Email, Up: ct.Up, Down: ct.Down})
	}
	for _, u := range onlineUsers {
		report.Onlines = append(report.Onlines, u.Email)
	}
	return report
}

func (r trafficReport) empty() bool {
	return len(r.Inbounds) == 0 && len(r.Users) == 0 && len(r.Onlines) == 0
}

// postTrafficReport sends one report to the master; only an HTTP 200
// counts as acknowledged.
func (a *Agent) postTrafficReport(report trafficReport) error {
	url := fmt.Sprintf("%s/panel/api/agent/traffic", a.config.MasterURL)
	body, _ := json.Marshal(report)
	req, err := http.NewRequest("POST", url, bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+a.config.Token)

	resp, err := a.do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("traffic report failed with status %d", resp.StatusCode)
	}
	return nil
}

// reportTrafficWith runs one snapshot → report → commit cycle.
//
// The committed baseline advances ONLY after the master acknowledges the
// report (HTTP 200). Core counters are cumulative, so when a report fails
// nothing is lost: the next snapshot diffs against the same baseline and
// re-includes everything unreported, exactly once. The acknowledged
// baseline is persisted after every commit, and restored before the first
// snapshot of a restarted agent, so an agent restart cannot swallow the
// unreported tail into a fresh baseline either.
//
// A poll with nothing to send still commits: it establishes the very
// first baseline, and keeps the persisted copy fresh.
func (a *Agent) reportTrafficWith(sc statsClient, post func(trafficReport) error) error {
	if !a.baselineLoaded {
		a.baselineLoaded = true
		if values, err := loadTrafficBaseline(a.baselinePath()); err != nil {
			if !os.IsNotExist(err) {
				logger.Warningf("Traffic baseline at %s unusable (%v); starting from a fresh baseline", a.baselinePath(), err)
			}
		} else if len(values) > 0 {
			sc.RestoreTrafficBaseline(values)
		}
	}

	snap, err := sc.GetTrafficSnapshot()
	if err != nil {
		return err
	}
	onlineUsers, _ := sc.GetOnlineUsers()

	report := buildTrafficReport(snap, onlineUsers)
	if !report.empty() {
		if err := post(report); err != nil {
			// No commit: the baseline stays where the master last
			// acknowledged it, and these deltas are reported again
			// (grown) on the next poll.
			return err
		}
	}

	sc.CommitTrafficSnapshot(snap)
	if err := saveTrafficBaseline(a.baselinePath(), snap.BaselineValues()); err != nil {
		// The in-memory commit stands — rolling it back would re-report
		// traffic the master already counted. The stale file self-heals
		// on the next successful poll; only a crash before that could
		// re-report a single interval.
		logger.Warningf("Failed to persist traffic baseline: %v", err)
		return err
	}
	return nil
}

// collectAndReportTraffic runs one reporting cycle per running core
// (xray and, since ITEM6 B3, sing-box). The two cores keep separate
// clients and baselines; a failure in one never blocks the other.
func (a *Agent) collectAndReportTraffic() error {
	var firstErr error
	if err := a.collectAndReportXrayTraffic(); err != nil {
		firstErr = err
	}
	if err := a.collectAndReportSingboxTraffic(); err != nil && firstErr == nil {
		firstErr = err
	}
	return firstErr
}

// collectAndReportXrayTraffic snapshots the local xray counters and reports
// the deltas since the last master-acknowledged baseline. The XrayAPI
// instance is kept across polls so baselines survive; xray restarts are
// handled (counters resetting to zero are treated as fresh).
func (a *Agent) collectAndReportXrayTraffic() error {
	if a.xrayProc == nil || !a.xrayProc.IsRunning() {
		return nil
	}
	if a.apiPort <= 0 {
		return nil
	}
	if err := a.xrayAPI.Init(a.apiPort); err != nil {
		return err
	}
	defer a.xrayAPI.Close()

	return a.reportTrafficWith(&a.xrayAPI, a.postTrafficReport)
}

// singboxBaselineFile is the on-disk form of the sing-box tracker's
// last committed baseline. The epoch travels with the values: after
// an agent restart against a still-running (or since-restarted) core,
// a baseline from another core instance must not be restored.
type singboxBaselineFile struct {
	Version int              `json:"version"`
	Epoch   string           `json:"epoch"`
	Values  map[string]int64 `json:"values"`
}

// singboxBaselinePath derives the sing-box baseline location from the
// xray one (agent-traffic-baseline.json -> agent-traffic-baseline-singbox.json),
// so main's placement next to agent.json carries over.
func (a *Agent) singboxBaselinePath() string {
	p := a.baselinePath()
	if strings.HasSuffix(p, ".json") {
		return strings.TrimSuffix(p, ".json") + "-singbox.json"
	}
	return p + "-singbox"
}

func loadSingboxBaseline(path string) (*singboxBaselineFile, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return nil, err
	}
	var f singboxBaselineFile
	if err := json.Unmarshal(data, &f); err != nil {
		return nil, fmt.Errorf("parse sing-box traffic baseline: %w", err)
	}
	return &f, nil
}

func saveSingboxBaseline(path, epoch string, values map[string]int64) error {
	data, err := json.Marshal(singboxBaselineFile{Version: 1, Epoch: epoch, Values: values})
	if err != nil {
		return err
	}
	tmp := path + ".tmp"
	if err := os.WriteFile(tmp, data, 0o600); err != nil {
		return err
	}
	return os.Rename(tmp, path)
}

// collectAndReportSingboxTraffic runs one sing-box reporting cycle.
func (a *Agent) collectAndReportSingboxTraffic() error {
	a.sbMu.Lock()
	defer a.sbMu.Unlock()
	// Log hygiene rides this cycle (ITEM6 B5): bounded, and a no-op
	// until the cap is actually crossed.
	rotateSingboxLogIfOversize(singboxLogPath())
	if a.sbProc == nil || !a.sbProc.isRunning() {
		return nil
	}
	return a.reportSingboxTrafficLocked()
}

// reportSingboxTrafficLocked is the sing-box counterpart of
// reportTrafficWith, with the same contract — commit only after the
// master acknowledges (HTTP 200), persist after every commit — plus
// the epoch carried through restore/persist. The online set merges
// the access-log tracker with the users whose counters just moved.
// Callers hold sbMu.
func (a *Agent) reportSingboxTrafficLocked() error {
	a.ensureSingboxState()
	if a.sbStatsAddr == "" {
		// The applied config carries no v2ray_api stats block (the
		// panel-side stats switches are off): nothing exists to read.
		return nil
	}
	a.sbStats.SetAddr(a.sbStatsAddr)

	if !a.sbBaselineLoaded {
		a.sbBaselineLoaded = true
		if f, err := loadSingboxBaseline(a.singboxBaselinePath()); err != nil {
			if !os.IsNotExist(err) {
				logger.Warningf("sing-box traffic baseline at %s unusable (%v); starting from a fresh baseline", a.singboxBaselinePath(), err)
			}
		} else {
			a.sbStats.RestoreTrafficBaseline(f.Values, f.Epoch)
		}
	}

	a.sbOnline.Poll(time.Now())

	snap, err := a.sbStats.GetTrafficSnapshot()
	if err != nil {
		if errors.Is(err, errSingboxEpochChanged) {
			// The core restarted around this poll (config change or
			// crash): this round establishes nothing; the next poll
			// rebases onto the new instance. Not a fault.
			return nil
		}
		return err
	}

	report := buildTrafficReport(snap, mergeOnlineUsers(a.sbOnline.Online(time.Now()), snap))
	if !report.empty() {
		if err := a.postTrafficReport(report); err != nil {
			// No commit: deltas are recomputed against the same
			// baseline next poll, exactly like the xray path.
			return err
		}
	}

	a.sbStats.CommitTrafficSnapshot(snap)
	if err := saveSingboxBaseline(a.singboxBaselinePath(), a.sbProc.epoch(), snap.BaselineValues()); err != nil {
		// The in-memory commit stands (see the xray path's reasoning).
		logger.Warningf("Failed to persist sing-box traffic baseline: %v", err)
		return err
	}
	return nil
}

// LoadConfig loads agent config from file
func LoadConfig(path string) (*Config, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return nil, err
	}
	var cfg Config
	if err := json.Unmarshal(data, &cfg); err != nil {
		return nil, err
	}
	return &cfg, nil
}