feat(agent): node agent — enroll/mTLS, heartbeat, command stream, sing-box 用户表 (tsk__R8M4jEw43JR)

与控制面同仓同 go.mod，新增节点 agent 实现： - proto/agent/v1/agent.proto + internal/pb/agentv1：冻结的控制面↔agent gRPC 契约（Enroll/Register/Heartbeat/Subscribe/Ack/ReportUsage）。仓库尚无 protoc 流水线，暂以手写 Go 类型 + JSON gRPC codec 实现，与 proto 1:1 对应，待 protoc 接入即可替换。 - internal/agentd： - enroll.go：首启生成 EC 密钥+CSR，持 bootstrap token 调 Enroll 换 90d 节点证书（CN=node_uuid），落 /etc/pangolin-agent/，此后 mTLS。 - conn.go(agent.go)+creds.go：mTLS 主动拨号 + 指数退避重连；重连携带 last_command_id； Register 取 ConfigSnapshot 全量配置覆盖本地。 - heartbeat.go：30s 上报 peer/带宽/CPU + config_version；need_full_resync→全量同步。 - command.go：消费 Subscribe，Upsert/Revoke/Rotate/ApplyConfig/Lifecycle 幂等处理后 Ack（at-least-once，按 command_id 去重）。 - singbox.go+render.go：内存用户表 + 落盘 state.json（仅 dp_uuid+expires_at）；任何变更渲染完整 sing-box 配置（REALITY users[uuid,flow] + Hy2 users[派生口令]）→ 500ms 去抖合并 → systemd 重启。 - ttl.go：凭证 TTL 定时移除并上报。 - usage.go：按 dp_uuid 聚合上报，绝无 user_id/email/目的地址。 - derive.go：Hy2 口令 = HMAC-SHA256(key, dp_uuid)，与控制面同源派生。 - cmd/agent：入口（flag/env 配置）。 - infra/cloud-init/{node.yaml.tmpl,install-node.sh,README.md}：一段式安装，下载锁定版本二进制并校验 SHA-256，systemd 拉管，首启即 Enroll/Register。shellcheck -S warning 通过。测试（bufconn mock 控制面，无需 docker）：Enroll→Register→Heartbeat 全流转；Upsert/Revoke 渲染正确；Rotate 宽限期新旧并存到点移除；TTL 自动移除并上报；断流重连 last_command_id 续发不丢不重；need_full_resync 触发重注册；state.json 恢复；去抖合并；扫描确认无身份字段。 go test -race ./internal/agentd/... ./internal/pb/... 通过；go vet ./... 通过。落实 doc/04 §2 节点无状态化与 doc/06 §3 数据面红线（节点仅见 dp_uuid）。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 12:26:58 +08:00
parent 787151245e
commit f03d2dc8a6
25 changed files with 2963 additions and 0 deletions
@@ -12,6 +12,9 @@ build:
 build-codegen: ## 激活码批次生成 CLI
 	go build -o bin/codegen ./cmd/codegen

+build-agent: ## 节点 agent 二进制（部署到加速节点）
+	go build -o bin/pangolin-agent ./cmd/agent
+
 # ── test ───────────────────────────────────────────────────────────────────────
 test:
 	go test ./...
@@ -19,6 +22,9 @@ test:
 test-unit: ## 单测（不依赖外部服务）
 	go test -count=1 -race ./internal/codes/... -run 'Test[^I][^n][^t]'

+test-agent: ## 节点 agent 单测 + 集成（bufconn mock 控制面，无需 docker）
+	go test -count=1 -race ./internal/agentd/... ./internal/pb/...
+
 test-integration: ## 集成测试（需本机 docker，testcontainers）
 	go test -count=1 ./internal/codes/... -run TestInt

@@ -0,0 +1,58 @@
+// Command agent is the Pangolin node agent. It runs on every acceleration node,
+// self-enrolls over mTLS, dials the control plane, applies the streamed command
+// feed by managing the local sing-box user table, and reports heartbeat/usage.
+//
+// Configuration comes from flags or PANGOLIN_AGENT_* environment variables
+// (cloud-init injects the latter). The node holds zero user identities on disk.
+package main
+
+import (
+	"context"
+	"flag"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/wangjia/pangolin/server/internal/agentd"
+)
+
+func main() {
+	cfg := agentd.Config{}
+	var singboxUnit string
+
+	flag.StringVar(&cfg.ControlPlaneAddr, "control-plane", env("PANGOLIN_AGENT_CONTROL_PLANE", ""), "control plane gRPC address host:port")
+	flag.StringVar(&cfg.ServerName, "server-name", env("PANGOLIN_AGENT_SERVER_NAME", ""), "TLS SNI to validate (defaults to control-plane host)")
+	flag.StringVar(&cfg.BootstrapToken, "bootstrap-token", env("PANGOLIN_AGENT_BOOTSTRAP_TOKEN", ""), "one-time enrollment token")
+	flag.StringVar(&cfg.StateDir, "state-dir", env("PANGOLIN_AGENT_STATE_DIR", agentd.DefaultStateDir), "directory for node key/cert/ca/state")
+	flag.StringVar(&cfg.SingboxConfigPath, "singbox-config", env("PANGOLIN_AGENT_SINGBOX_CONFIG", agentd.DefaultSingboxCfg), "path to rendered sing-box config")
+	flag.StringVar(&cfg.DeriveKey, "derive-key", env("PANGOLIN_AGENT_DERIVE_KEY", ""), "Hy2 password derivation key")
+	flag.StringVar(&cfg.AgentVersion, "agent-version", env("PANGOLIN_AGENT_VERSION", "dev"), "reported agent version")
+	flag.StringVar(&singboxUnit, "singbox-unit", env("PANGOLIN_AGENT_SINGBOX_UNIT", "sing-box"), "systemd unit to restart on config change")
+	flag.BoolVar(&cfg.Insecure, "insecure", env("PANGOLIN_AGENT_INSECURE", "") == "1", "dial without mTLS (dev only)")
+	flag.Parse()
+
+	if cfg.ControlPlaneAddr == "" {
+		log.Fatal("agent: --control-plane (or PANGOLIN_AGENT_CONTROL_PLANE) is required")
+	}
+
+	agent := agentd.New(cfg, agentd.WithRestarter(agentd.SystemdRestarter{Unit: singboxUnit}))
+
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	log.Printf("[pangolin-agent] starting (control-plane=%s, version=%s)", cfg.ControlPlaneAddr, cfg.AgentVersion)
+	start := time.Now()
+	if err := agent.Run(ctx); err != nil {
+		log.Fatalf("[pangolin-agent] exited after %s: %v", time.Since(start), err)
+	}
+	log.Printf("[pangolin-agent] shut down cleanly after %s", time.Since(start))
+}
+
+func env(key, def string) string {
+	if v, ok := os.LookupEnv(key); ok {
+		return v
+	}
+	return def
+}
@@ -0,0 +1,202 @@
+package agentd
+
+import (
+	"context"
+	"errors"
+	"math/rand"
+	"sync/atomic"
+	"time"
+
+	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
+)
+
+// errResync is returned within a session to force a reconnect + full Register.
+var errResync = errors.New("agentd: full resync required")
+
+// Agent is the top-level node daemon. Construct with New and drive with Run.
+type Agent struct {
+	cfg      Config
+	sb       *SingBox
+	nodeUUID atomic.Value // string; written once by Run, read by all loops + tests
+
+	dial       Dialer
+	enrollDial EnrollDialer
+	load       LoadSource
+	usage      UsageSource
+	clock      func() time.Time
+
+	lastCommandID atomic.Int64
+	shutdown      context.CancelFunc
+
+	restarter Restarter
+}
+
+// Option customises an Agent (primarily for tests / dependency injection).
+type Option func(*Agent)
+
+func WithRestarter(r Restarter) Option       { return func(a *Agent) { a.restarter = r } }
+func WithDialer(d Dialer) Option             { return func(a *Agent) { a.dial = d } }
+func WithEnrollDialer(d EnrollDialer) Option { return func(a *Agent) { a.enrollDial = d } }
+func WithLoadSource(l LoadSource) Option     { return func(a *Agent) { a.load = l } }
+func WithUsageSource(u UsageSource) Option   { return func(a *Agent) { a.usage = u } }
+func WithClock(f func() time.Time) Option    { return func(a *Agent) { a.clock = f } }
+
+// New builds an Agent. Production callers pass nothing extra and get mTLS dialers;
+// tests inject bufconn dialers, fake restarters, etc.
+func New(cfg Config, opts ...Option) *Agent {
+	cfg = cfg.withDefaults()
+	a := &Agent{cfg: cfg, clock: time.Now}
+	for _, o := range opts {
+		o(a)
+	}
+	if a.dial == nil {
+		if cfg.Insecure {
+			a.dial = insecureDialer(cfg.ControlPlaneAddr)
+		} else {
+			a.dial = NewMTLSDialer(cfg)
+		}
+	}
+	if a.enrollDial == nil {
+		if cfg.Insecure {
+			a.enrollDial = EnrollDialer(insecureDialer(cfg.ControlPlaneAddr))
+		} else {
+			a.enrollDial = NewEnrollDialer(cfg)
+		}
+	}
+	a.sb = NewSingBox(cfg, a.restarter)
+	a.sb.clock = a.clock
+	if a.load == nil {
+		a.load = defaultLoadSource{sb: a.sb}
+	}
+	if a.usage == nil {
+		a.usage = nopUsageSource{}
+	}
+	return a
+}
+
+// SingBox exposes the underlying manager (tests / introspection).
+func (a *Agent) SingBox() *SingBox { return a.sb }
+
+// NodeUUID returns the enrolled node UUID (empty until Run enrolls).
+func (a *Agent) NodeUUID() string {
+	v, _ := a.nodeUUID.Load().(string)
+	return v
+}
+
+// Run enrolls (if needed), restores state, then maintains the control-plane
+// connection with exponential-backoff reconnect until ctx is cancelled.
+func (a *Agent) Run(ctx context.Context) error {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	a.shutdown = cancel
+
+	uuid, err := EnsureEnrolled(ctx, a.cfg, a.enrollDial)
+	if err != nil {
+		return err
+	}
+	a.nodeUUID.Store(uuid)
+
+	if err := a.sb.LoadState(); err != nil {
+		return err
+	}
+
+	go a.sb.Run(ctx)
+	go a.runTTL(ctx)
+	a.sb.markDirty() // render whatever we restored before the first connect
+
+	backoff := a.cfg.BackoffMin
+	for {
+		if ctx.Err() != nil {
+			return nil
+		}
+		registered, err := a.runSession(ctx)
+		if ctx.Err() != nil {
+			return nil
+		}
+		if registered {
+			backoff = a.cfg.BackoffMin // healthy session → reset backoff
+		}
+		if err != nil && !errors.Is(err, errResync) {
+			logf("session ended: %v; reconnecting in %s", err, backoff)
+		}
+		if !sleepCtx(ctx, jitter(backoff)) {
+			return nil
+		}
+		backoff = nextBackoff(backoff, a.cfg.BackoffMax)
+	}
+}
+
+// runSession establishes one connection: Register (full snapshot), then runs
+// heartbeat/subscribe/usage until the first of them fails. Returns whether
+// Register succeeded (used to reset backoff).
+func (a *Agent) runSession(ctx context.Context) (registered bool, err error) {
+	conn, err := a.dial(ctx)
+	if err != nil {
+		return false, err
+	}
+	defer conn.Close()
+
+	client := agentv1.NewAgentServiceClient(conn)
+
+	snap, err := client.Register(ctx, &agentv1.RegisterRequest{
+		NodeUUID:           a.NodeUUID(),
+		AgentVersion:       a.cfg.AgentVersion,
+		LocalConfigVersion: a.sb.ConfigVersion(),
+	})
+	if err != nil {
+		return false, err
+	}
+	// Full snapshot is authoritative: overwrite local credential table + inbounds.
+	a.sb.ApplyConfig(snap, true)
+	if snap.LastCommandID > a.lastCommandID.Load() {
+		a.lastCommandID.Store(snap.LastCommandID)
+	}
+
+	sctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	errc := make(chan error, 3)
+	go func() { errc <- a.runHeartbeat(sctx, client) }()
+	go func() { errc <- a.runSubscribe(sctx, client) }()
+	go func() { errc <- a.runUsage(sctx, client) }()
+
+	first := <-errc
+	cancel()
+	<-errc
+	<-errc
+	return true, first
+}
+
+// ─── backoff helpers ─────────────────────────────────────────────────────────
+
+func nextBackoff(cur, max time.Duration) time.Duration {
+	next := cur * 2
+	if next > max {
+		next = max
+	}
+	return next
+}
+
+// jitter applies ±20% randomisation to avoid thundering-herd reconnects.
+func jitter(d time.Duration) time.Duration {
+	if d <= 0 {
+		return 0
+	}
+	delta := float64(d) * 0.2
+	return d + time.Duration((rand.Float64()*2-1)*delta)
+}
+
+// sleepCtx sleeps for d unless ctx is cancelled first. Returns false if cancelled.
+func sleepCtx(ctx context.Context, d time.Duration) bool {
+	if d <= 0 {
+		return ctx.Err() == nil
+	}
+	t := time.NewTimer(d)
+	defer t.Stop()
+	select {
+	case <-ctx.Done():
+		return false
+	case <-t.C:
+		return true
+	}
+}
@@ -0,0 +1,105 @@
+package agentd
+
+import (
+	"context"
+
+	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
+)
+
+// runSubscribe opens the Command stream (resuming from the last durably processed
+// command id) and applies each command idempotently, Ack'ing after apply
+// (at-least-once). It returns when ctx is cancelled or the stream errors, so the
+// session can reconnect and resume.
+func (a *Agent) runSubscribe(ctx context.Context, client agentv1.AgentServiceClient) error {
+	nodeUUID := a.NodeUUID()
+	stream, err := client.Subscribe(ctx, &agentv1.SubscribeRequest{
+		NodeUUID:      nodeUUID,
+		LastCommandID: a.lastCommandID.Load(),
+	})
+	if err != nil {
+		return err
+	}
+	for {
+		cmd, err := stream.Recv()
+		if err != nil {
+			return err
+		}
+		a.handleCommand(cmd)
+		// Ack after apply. A failed Ack just causes redelivery, which the
+		// dedup-by-command-id + idempotent apply below tolerates.
+		if _, err := client.Ack(ctx, &agentv1.AckRequest{
+			NodeUUID:  nodeUUID,
+			CommandID: cmd.CommandID,
+		}); err != nil {
+			return err
+		}
+	}
+}
+
+// handleCommand applies one command. Commands at or below the high-water mark are
+// skipped (already applied) — this makes redelivery after a dropped Ack safe.
+func (a *Agent) handleCommand(cmd *agentv1.Command) {
+	if cmd.CommandID != 0 && cmd.CommandID <= a.lastCommandID.Load() {
+		return
+	}
+	switch cmd.Type {
+	case agentv1.CommandTypeUpsert:
+		if cmd.Upsert != nil && cmd.Upsert.Credential != nil {
+			a.sb.Upsert(credFromPB(cmd.Upsert.Credential))
+		}
+	case agentv1.CommandTypeRevoke:
+		if cmd.Revoke != nil {
+			a.sb.Revoke(cmd.Revoke.DpUUID)
+		}
+	case agentv1.CommandTypeRotateCredential:
+		if cmd.Rotate != nil && cmd.Rotate.NewCredential != nil {
+			a.sb.Rotate(cmd.Rotate.OldDpUUID, credFromPB(cmd.Rotate.NewCredential), cmd.Rotate.GraceUntilUnix)
+		}
+	case agentv1.CommandTypeApplyConfig:
+		if cmd.ApplyConfig != nil {
+			a.sb.ApplyConfig(cmd.ApplyConfig, cmd.ApplyConfig.Credentials != nil)
+		}
+	case agentv1.CommandTypeLifecycle:
+		a.handleLifecycle(cmd.Lifecycle)
+	default:
+		logf("ignoring command id=%d with unknown type=%d", cmd.CommandID, cmd.Type)
+	}
+
+	if cmd.CommandID > a.lastCommandID.Load() {
+		a.lastCommandID.Store(cmd.CommandID)
+	}
+}
+
+func (a *Agent) handleLifecycle(p *agentv1.LifecyclePayload) {
+	if p == nil {
+		return
+	}
+	switch p.Action {
+	case agentv1.LifecycleActionDrain:
+		// Draining is steered by control-plane node weight; the data plane keeps
+		// serving existing sessions. Nothing to mutate locally.
+		logf("lifecycle: drain")
+	case agentv1.LifecycleActionResume:
+		logf("lifecycle: resume")
+	case agentv1.LifecycleActionShutdown:
+		logf("lifecycle: shutdown requested")
+		if a.shutdown != nil {
+			a.shutdown()
+		}
+	default:
+		logf("lifecycle: unknown action %d", p.Action)
+	}
+}
+
+func credFromPB(c *agentv1.Credential) *Cred {
+	flow := c.Flow
+	if flow == "" {
+		flow = DefaultFlow
+	}
+	return &Cred{
+		DpUUID:    c.DpUUID,
+		Protocol:  c.Protocol,
+		Flow:      flow,
+		ExpiresAt: c.ExpiresAtUnix,
+	}
+}
@@ -0,0 +1,66 @@
+package agentd
+
+import (
+	"testing"
+
+	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
+)
+
+func TestHandleCommandIdempotentAndOrdered(t *testing.T) {
+	a := New(testConfig(t), WithRestarter(&fakeRestarter{}))
+	a.sb.ApplyConfig(sampleSnapshot(), true) // inbounds present
+
+	upsert := func(id int64, dp string) *agentv1.Command {
+		return &agentv1.Command{
+			CommandID: id,
+			Type:      agentv1.CommandTypeUpsert,
+			Upsert:    &agentv1.UpsertPayload{Credential: &agentv1.Credential{DpUUID: dp, Protocol: agentv1.ProtocolBoth}},
+		}
+	}
+
+	// Apply id=1, then redelivered id=1 (dropped Ack) — applied once, no regression.
+	a.handleCommand(upsert(1, "alpha"))
+	a.handleCommand(upsert(1, "alpha"))
+	if a.lastCommandID.Load() != 1 {
+		t.Fatalf("lastCommandID = %d, want 1", a.lastCommandID.Load())
+	}
+
+	// A stale command at or below the high-water mark must be skipped (idempotent
+	// at-least-once redelivery): revoke at id=1 should NOT remove alpha.
+	a.handleCommand(&agentv1.Command{
+		CommandID: 1,
+		Type:      agentv1.CommandTypeRevoke,
+		Revoke:    &agentv1.RevokePayload{DpUUID: "alpha"},
+	})
+	if !a.sb.Has("alpha") {
+		t.Error("revoke with stale id=1 should have been skipped, but alpha was removed")
+	}
+
+	// Forward progress: id=2 revokes alpha.
+	a.handleCommand(&agentv1.Command{
+		CommandID: 2,
+		Type:      agentv1.CommandTypeRevoke,
+		Revoke:    &agentv1.RevokePayload{DpUUID: "alpha"},
+	})
+	if a.sb.Has("alpha") {
+		t.Error("alpha should be revoked by id=2")
+	}
+	if a.lastCommandID.Load() != 2 {
+		t.Fatalf("lastCommandID = %d, want 2", a.lastCommandID.Load())
+	}
+}
+
+func TestHandleApplyConfigUpdatesInbounds(t *testing.T) {
+	a := New(testConfig(t), WithRestarter(&fakeRestarter{}))
+	a.handleCommand(&agentv1.Command{
+		CommandID:   1,
+		Type:        agentv1.CommandTypeApplyConfig,
+		ApplyConfig: sampleSnapshot(&agentv1.Credential{DpUUID: "z", Protocol: agentv1.ProtocolReality}),
+	})
+	if a.sb.ConfigVersion() != 7 {
+		t.Errorf("config version = %d, want 7", a.sb.ConfigVersion())
+	}
+	if !a.sb.Has("z") {
+		t.Error("apply_config credential z not applied")
+	}
+}
@@ -0,0 +1,112 @@
+// Package agentd implements the Pangolin node agent: a lightweight daemon that
+// runs on every acceleration node. It self-enrolls over mTLS, dials the control
+// plane (the agent is always the dialer), keeps a long-lived gRPC connection with
+// exponential-backoff reconnect, reports heartbeat/usage, and applies the streamed
+// command feed by managing the local sing-box user table.
+//
+// No-state invariant (doc/04 §2, doc/06 §3): the agent persists ONLY the
+// credential table (dp_uuid + expires_at) to disk. It keeps zero user identities,
+// zero destination/DNS data and writes no access logs. A seized node leaks only
+// opaque dp_uuids, never accounts.
+package agentd
+
+import (
+	"path/filepath"
+	"time"
+)
+
+// Default tuning values. The intervals match doc/06 (30s heartbeat) and the
+// 500ms render debounce that batches bursts of credential changes into a single
+// sing-box restart (sing-box has no hot-reload; a restart is a ~1-2s blip).
+const (
+	DefaultHeartbeatInterval = 30 * time.Second
+	DefaultUsageInterval     = 60 * time.Second
+	DefaultTTLScanInterval   = 15 * time.Second
+	DefaultDebounceWindow    = 500 * time.Millisecond
+	DefaultDialTimeout       = 10 * time.Second
+
+	DefaultBackoffMin = 1 * time.Second
+	DefaultBackoffMax = 60 * time.Second
+
+	DefaultStateDir   = "/etc/pangolin-agent"
+	DefaultSingboxCfg = "/etc/sing-box/config.json"
+
+	// DefaultFlow is the REALITY VLESS flow (doc/02 §3.1).
+	DefaultFlow = "xtls-rprx-vision"
+)
+
+// Config holds everything the agent needs. Most fields come from cloud-init
+// (control-plane address + bootstrap token) or have safe defaults.
+type Config struct {
+	// ControlPlaneAddr is the control plane gRPC endpoint (host:port).
+	ControlPlaneAddr string
+	// ServerName overrides the TLS SNI used when dialing (defaults to the host
+	// part of ControlPlaneAddr). The control plane cert must match it.
+	ServerName string
+
+	// BootstrapToken is the one-time cloud-init token consumed during Enroll.
+	// Ignored once the node already holds a certificate.
+	BootstrapToken string
+
+	// StateDir holds node.key/node.crt/ca.crt and state.json.
+	StateDir string
+	// SingboxConfigPath is where the rendered sing-box config is written.
+	SingboxConfigPath string
+
+	// DeriveKey keys the Hy2 password derivation (see DeriveHy2Password).
+	DeriveKey string
+
+	AgentVersion string
+
+	HeartbeatInterval time.Duration
+	UsageInterval     time.Duration
+	TTLScanInterval   time.Duration
+	DebounceWindow    time.Duration
+	DialTimeout       time.Duration
+	BackoffMin        time.Duration
+	BackoffMax        time.Duration
+
+	// Insecure dials without mTLS. ONLY for local dev / integration tests.
+	Insecure bool
+}
+
+// withDefaults returns a copy of c with zero-valued tunables filled in.
+func (c Config) withDefaults() Config {
+	if c.StateDir == "" {
+		c.StateDir = DefaultStateDir
+	}
+	if c.SingboxConfigPath == "" {
+		c.SingboxConfigPath = DefaultSingboxCfg
+	}
+	if c.HeartbeatInterval == 0 {
+		c.HeartbeatInterval = DefaultHeartbeatInterval
+	}
+	if c.UsageInterval == 0 {
+		c.UsageInterval = DefaultUsageInterval
+	}
+	if c.TTLScanInterval == 0 {
+		c.TTLScanInterval = DefaultTTLScanInterval
+	}
+	if c.DebounceWindow == 0 {
+		c.DebounceWindow = DefaultDebounceWindow
+	}
+	if c.DialTimeout == 0 {
+		c.DialTimeout = DefaultDialTimeout
+	}
+	if c.BackoffMin == 0 {
+		c.BackoffMin = DefaultBackoffMin
+	}
+	if c.BackoffMax == 0 {
+		c.BackoffMax = DefaultBackoffMax
+	}
+	if c.AgentVersion == "" {
+		c.AgentVersion = "dev"
+	}
+	return c
+}
+
+// Paths to the on-disk PKI + state material.
+func (c Config) KeyPath() string   { return filepath.Join(c.StateDir, "node.key") }
+func (c Config) CertPath() string  { return filepath.Join(c.StateDir, "node.crt") }
+func (c Config) CAPath() string    { return filepath.Join(c.StateDir, "ca.crt") }
+func (c Config) StatePath() string { return filepath.Join(c.StateDir, "state.json") }
@@ -0,0 +1,101 @@
+package agentd
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
+	"fmt"
+	"net"
+	"os"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// Dialer opens an authenticated long-lived connection to the control plane.
+type Dialer func(ctx context.Context) (*grpc.ClientConn, error)
+
+// mtlsClientConfig builds the client *tls.Config from the persisted node key/cert
+// and pinned CA.
+func mtlsClientConfig(cfg Config) (*tls.Config, error) {
+	cert, err := tls.LoadX509KeyPair(cfg.CertPath(), cfg.KeyPath())
+	if err != nil {
+		return nil, fmt.Errorf("agentd: load client keypair: %w", err)
+	}
+	caPEM, err := os.ReadFile(cfg.CAPath())
+	if err != nil {
+		return nil, fmt.Errorf("agentd: read CA: %w", err)
+	}
+	pool := x509.NewCertPool()
+	if !pool.AppendCertsFromPEM(caPEM) {
+		return nil, errors.New("agentd: CA file contains no certificate")
+	}
+	return &tls.Config{
+		Certificates: []tls.Certificate{cert},
+		RootCAs:      pool,
+		ServerName:   cfg.tlsServerName(),
+		MinVersion:   tls.VersionTLS13,
+	}, nil
+}
+
+// tlsServerName returns the SNI to validate the control-plane cert against.
+func (c Config) tlsServerName() string {
+	if c.ServerName != "" {
+		return c.ServerName
+	}
+	host, _, err := net.SplitHostPort(c.ControlPlaneAddr)
+	if err != nil {
+		return c.ControlPlaneAddr
+	}
+	return host
+}
+
+// NewMTLSDialer returns a Dialer that connects to the control plane over mTLS,
+// presenting the node's client certificate.
+func NewMTLSDialer(cfg Config) Dialer {
+	return func(ctx context.Context) (*grpc.ClientConn, error) {
+		tlsCfg, err := mtlsClientConfig(cfg)
+		if err != nil {
+			return nil, err
+		}
+		return grpc.NewClient(cfg.ControlPlaneAddr,
+			grpc.WithTransportCredentials(credentials.NewTLS(tlsCfg)),
+		)
+	}
+}
+
+// NewEnrollDialer returns an EnrollDialer for the bootstrap (pre-certificate)
+// Enroll call. It pins the control-plane CA if ca.crt is already present
+// (cloud-init may inject it); otherwise it falls back to a server-unauthenticated
+// TLS handshake — the bootstrap token is the trust anchor for that single call.
+func NewEnrollDialer(cfg Config) EnrollDialer {
+	return func(ctx context.Context) (*grpc.ClientConn, error) {
+		var creds credentials.TransportCredentials
+		if caPEM, err := os.ReadFile(cfg.CAPath()); err == nil {
+			pool := x509.NewCertPool()
+			if !pool.AppendCertsFromPEM(caPEM) {
+				return nil, errors.New("agentd: pinned CA file invalid")
+			}
+			creds = credentials.NewTLS(&tls.Config{
+				RootCAs:    pool,
+				ServerName: cfg.tlsServerName(),
+				MinVersion: tls.VersionTLS13,
+			})
+		} else {
+			creds = credentials.NewTLS(&tls.Config{
+				InsecureSkipVerify: true, //nolint:gosec // bootstrap-token-authenticated enroll only
+				MinVersion:         tls.VersionTLS13,
+			})
+		}
+		return grpc.NewClient(cfg.ControlPlaneAddr, grpc.WithTransportCredentials(creds))
+	}
+}
+
+// insecureDialer is used only by tests/dev (cfg.Insecure).
+func insecureDialer(addr string) Dialer {
+	return func(ctx context.Context) (*grpc.ClientConn, error) {
+		return grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	}
+}
@@ -0,0 +1,34 @@
+package agentd
+
+import (
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/base64"
+)
+
+// DeriveHy2Password derives a node-agnostic Hysteria2 password from the opaque
+// data-plane credential id (dp_uuid). The REALITY inbound uses dp_uuid directly as
+// the VLESS user uuid; the Hy2 inbound cannot reuse a UUID as a password verbatim
+// (it must look like an opaque secret), so it is derived from the SAME source —
+// "password = dp_uuid 同源派生" — via a keyed HMAC.
+//
+// The derivation is deterministic given (dp_uuid, key): the control plane runs the
+// exact same function when it builds the client's connect config (doc/02 §3.1), so
+// both sides agree without the password ever crossing the agent contract.
+//
+// Risk note (carried from task #5): the derivation key is shared material. If a
+// node is seized the attacker still only sees dp_uuids and this node's key, which
+// lets them recompute Hy2 passwords for dp_uuids THEY ALREADY HOLD — it does not
+// reveal other subscribers' credentials or any account identity. Rotating the key
+// rotates every Hy2 password. When key == "" the password falls back to the raw
+// dp_uuid (acceptable for dev; production cloud-init always injects a key).
+func DeriveHy2Password(dpUUID, key string) string {
+	if key == "" {
+		return dpUUID
+	}
+	mac := hmac.New(sha256.New, []byte(key))
+	mac.Write([]byte(dpUUID))
+	sum := mac.Sum(nil)
+	// base64url without padding → URL/JSON-safe, 43 chars.
+	return base64.RawURLEncoding.EncodeToString(sum)
+}
@@ -0,0 +1,30 @@
+package agentd
+
+import "testing"
+
+func TestDeriveHy2Password(t *testing.T) {
+	const dp = "11111111-1111-1111-1111-111111111111"
+
+	// Deterministic for a given (dp, key).
+	a := DeriveHy2Password(dp, "secret")
+	b := DeriveHy2Password(dp, "secret")
+	if a != b {
+		t.Fatalf("derivation not deterministic: %q != %q", a, b)
+	}
+	// Different key → different password.
+	if DeriveHy2Password(dp, "other") == a {
+		t.Fatal("different key produced identical password")
+	}
+	// Different dp_uuid → different password.
+	if DeriveHy2Password("22222222-2222-2222-2222-222222222222", "secret") == a {
+		t.Fatal("different dp_uuid produced identical password")
+	}
+	// The derived value is not the raw dp_uuid (must look like an opaque secret).
+	if a == dp {
+		t.Fatal("derived password equals raw dp_uuid")
+	}
+	// Empty key → raw dp_uuid fallback (dev mode).
+	if got := DeriveHy2Password(dp, ""); got != dp {
+		t.Fatalf("empty key fallback = %q, want raw dp_uuid", got)
+	}
+}
@@ -0,0 +1,139 @@
+package agentd
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"errors"
+	"fmt"
+	"io/fs"
+	"os"
+
+	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
+	"google.golang.org/grpc"
+)
+
+// EnrollDialer opens a connection suitable for the (un-authenticated) Enroll RPC.
+// The agent has no client certificate yet, so this connection presents none; the
+// bootstrap token authenticates the call instead.
+type EnrollDialer func(ctx context.Context) (*grpc.ClientConn, error)
+
+// EnsureEnrolled guarantees the node holds a valid client certificate.
+//
+//   - If node.crt/node.key already exist, it loads and returns the certificate's
+//     CN (the node UUID) without contacting the control plane (idempotent first
+//     start vs. restart).
+//   - Otherwise it generates an EC P-256 key + CSR, calls Enroll with the
+//     bootstrap token, and persists node.key (0600), node.crt and ca.crt.
+//
+// The private key is generated on the node and never leaves it.
+func EnsureEnrolled(ctx context.Context, cfg Config, dial EnrollDialer) (string, error) {
+	cfg = cfg.withDefaults()
+
+	if uuid, ok, err := loadEnrolledUUID(cfg); err != nil {
+		return "", err
+	} else if ok {
+		return uuid, nil
+	}
+
+	if cfg.BootstrapToken == "" {
+		return "", errors.New("agentd: not enrolled and no bootstrap token provided")
+	}
+
+	keyPEM, csrPEM, err := generateKeyAndCSR()
+	if err != nil {
+		return "", err
+	}
+
+	conn, err := dial(ctx)
+	if err != nil {
+		return "", fmt.Errorf("agentd: dial for enroll: %w", err)
+	}
+	defer conn.Close()
+
+	client := agentv1.NewAgentServiceClient(conn)
+	resp, err := client.Enroll(ctx, &agentv1.EnrollRequest{
+		BootstrapToken: cfg.BootstrapToken,
+		CSRPEM:         csrPEM,
+		AgentVersion:   cfg.AgentVersion,
+	})
+	if err != nil {
+		return "", fmt.Errorf("agentd: enroll rpc: %w", err)
+	}
+	if resp.NodeUUID == "" || len(resp.CertPEM) == 0 {
+		return "", errors.New("agentd: enroll response missing node_uuid/cert")
+	}
+
+	if err := os.MkdirAll(cfg.StateDir, 0o700); err != nil {
+		return "", fmt.Errorf("agentd: mkdir state dir: %w", err)
+	}
+	if err := atomicWrite(cfg.KeyPath(), keyPEM, 0o600); err != nil {
+		return "", err
+	}
+	if err := atomicWrite(cfg.CertPath(), resp.CertPEM, 0o644); err != nil {
+		return "", err
+	}
+	if len(resp.CAPEM) > 0 {
+		if err := atomicWrite(cfg.CAPath(), resp.CAPEM, 0o644); err != nil {
+			return "", err
+		}
+	}
+	logf("enrolled successfully (cert valid until unix=%d)", resp.NotAfterUnix)
+	return resp.NodeUUID, nil
+}
+
+// loadEnrolledUUID returns the CN of the persisted node certificate, if present.
+func loadEnrolledUUID(cfg Config) (string, bool, error) {
+	certPEM, err := os.ReadFile(cfg.CertPath())
+	if errors.Is(err, fs.ErrNotExist) {
+		return "", false, nil
+	}
+	if err != nil {
+		return "", false, fmt.Errorf("agentd: read node cert: %w", err)
+	}
+	if _, err := os.Stat(cfg.KeyPath()); err != nil {
+		// Cert without key is unusable — treat as not enrolled.
+		return "", false, nil
+	}
+	block, _ := pem.Decode(certPEM)
+	if block == nil {
+		return "", false, errors.New("agentd: node cert has no PEM block")
+	}
+	cert, err := x509.ParseCertificate(block.Bytes)
+	if err != nil {
+		return "", false, fmt.Errorf("agentd: parse node cert: %w", err)
+	}
+	if cert.Subject.CommonName == "" {
+		return "", false, errors.New("agentd: node cert has empty CN")
+	}
+	return cert.Subject.CommonName, true, nil
+}
+
+// generateKeyAndCSR creates an EC P-256 private key and a PKCS#10 CSR. The CSR's
+// CN is a placeholder; the CA overrides it with the authoritative node UUID.
+func generateKeyAndCSR() (keyPEM, csrPEM []byte, err error) {
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		return nil, nil, fmt.Errorf("agentd: generate key: %w", err)
+	}
+	keyDER, err := x509.MarshalECPrivateKey(key)
+	if err != nil {
+		return nil, nil, fmt.Errorf("agentd: marshal key: %w", err)
+	}
+	keyPEM = pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER})
+
+	tmpl := &x509.CertificateRequest{
+		Subject:            pkix.Name{CommonName: "pangolin-node-pending"},
+		SignatureAlgorithm: x509.ECDSAWithSHA256,
+	}
+	csrDER, err := x509.CreateCertificateRequest(rand.Reader, tmpl, key)
+	if err != nil {
+		return nil, nil, fmt.Errorf("agentd: create CSR: %w", err)
+	}
+	csrPEM = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE REQUEST", Bytes: csrDER})
+	return keyPEM, csrPEM, nil
+}
@@ -0,0 +1,63 @@
+package agentd
+
+import (
+	"context"
+	"time"
+
+	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
+)
+
+// LoadSource provides the runtime metrics reported in each heartbeat. The
+// production implementation can read sing-box stats + host CPU; the default only
+// reports the online peer count (number of provisioned credentials), which never
+// reveals identities.
+type LoadSource interface {
+	Load() (onlinePeers int32, upBps, downBps int64, cpuPercent float64)
+}
+
+// defaultLoadSource reports peer count from the SingBox table and zeros for the
+// rest. Bandwidth/CPU collection is a documented extension point.
+type defaultLoadSource struct{ sb *SingBox }
+
+func (d defaultLoadSource) Load() (int32, int64, int64, float64) {
+	return int32(d.sb.OnlinePeers()), 0, 0, 0
+}
+
+// runHeartbeat sends a heartbeat every cfg.HeartbeatInterval until ctx is
+// cancelled or an RPC error occurs. A need_full_resync response returns
+// errResync so the session re-Registers and overwrites local state.
+func (a *Agent) runHeartbeat(ctx context.Context, client agentv1.AgentServiceClient) error {
+	ticker := time.NewTicker(a.cfg.HeartbeatInterval)
+	defer ticker.Stop()
+	for {
+		if err := a.sendHeartbeat(ctx, client); err != nil {
+			return err
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		}
+	}
+}
+
+func (a *Agent) sendHeartbeat(ctx context.Context, client agentv1.AgentServiceClient) error {
+	peers, up, down, cpu := a.load.Load()
+	resp, err := client.Heartbeat(ctx, &agentv1.HeartbeatRequest{
+		NodeUUID:         a.NodeUUID(),
+		ConfigVersion:    a.sb.ConfigVersion(),
+		OnlinePeers:      peers,
+		BandwidthUpBps:   up,
+		BandwidthDownBps: down,
+		CPUPercent:       cpu,
+		TimestampUnix:    a.clock().Unix(),
+	})
+	if err != nil {
+		return err
+	}
+	if resp.NeedFullResync {
+		logf("control plane requested full resync")
+		return errResync
+	}
+	return nil
+}
@@ -0,0 +1,370 @@
+package agentd
+
+import (
+	"context"
+	"net"
+	"os"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/wangjia/pangolin/server/internal/mtls"
+	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/test/bufconn"
+)
+
+const testNodeUUID = "node-test-uuid"
+
+// mockCP is an in-memory control plane implementing AgentServiceServer.
+type mockCP struct {
+	ca *mtls.CA
+
+	mu          sync.Mutex
+	snapCreds   map[string]*agentv1.Credential // converged credential set for Register
+	configVer   int64
+	lastCmdID   int64
+	issued      []*agentv1.Command // backlog for resume-after-drop
+	acked       map[int64]bool
+	registerN   int
+	heartbeatN  int
+	subReqs     []int64
+	usage       []*agentv1.UsageReport
+	resyncOnce  bool
+	enrollToken string
+
+	liveCh chan *agentv1.Command
+	dropCh chan struct{}
+}
+
+func newMockCP(t *testing.T) *mockCP {
+	t.Helper()
+	dir := t.TempDir()
+	ca, err := mtls.NewCA(mtls.CAConfig{
+		KeyPath:  dir + "/ca.key",
+		CertPath: dir + "/ca.crt",
+	})
+	if err != nil {
+		t.Fatalf("new CA: %v", err)
+	}
+	return &mockCP{
+		ca:          ca,
+		snapCreds:   map[string]*agentv1.Credential{},
+		acked:       map[int64]bool{},
+		enrollToken: "test-token",
+		liveCh:      make(chan *agentv1.Command),
+		dropCh:      make(chan struct{}, 1),
+	}
+}
+
+func (m *mockCP) Enroll(_ context.Context, req *agentv1.EnrollRequest) (*agentv1.EnrollResponse, error) {
+	if req.BootstrapToken != m.enrollToken {
+		return nil, status.Error(codes.Unauthenticated, "bad bootstrap token")
+	}
+	certPEM, err := m.ca.SignCSR(req.CSRPEM, testNodeUUID)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "sign csr: %v", err)
+	}
+	return &agentv1.EnrollResponse{
+		NodeUUID:     testNodeUUID,
+		CertPEM:      certPEM,
+		CAPEM:        m.ca.CAPEM(),
+		NotAfterUnix: time.Now().Add(90 * 24 * time.Hour).Unix(),
+	}, nil
+}
+
+func (m *mockCP) Register(_ context.Context, _ *agentv1.RegisterRequest) (*agentv1.ConfigSnapshot, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.registerN++
+	creds := make([]*agentv1.Credential, 0, len(m.snapCreds))
+	for _, c := range m.snapCreds {
+		creds = append(creds, c)
+	}
+	sort.Slice(creds, func(i, j int) bool { return creds[i].DpUUID < creds[j].DpUUID })
+	return &agentv1.ConfigSnapshot{
+		ConfigVersion: m.configVer,
+		Credentials:   creds,
+		Reality: &agentv1.RealityInbound{
+			ListenPort: 11443, PrivateKey: "pk", ShortID: "deadbeef",
+			ServerName: "www.apple.com", HandshakeServer: "www.apple.com", HandshakePort: 443,
+		},
+		Hy2:           &agentv1.Hy2Inbound{ListenPort: 443, Masquerade: "https://www.bing.com", CertPath: "/c", KeyPath: "/k"},
+		LastCommandID: m.lastCmdID,
+	}, nil
+}
+
+func (m *mockCP) Heartbeat(_ context.Context, _ *agentv1.HeartbeatRequest) (*agentv1.HeartbeatResponse, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.heartbeatN++
+	resync := false
+	if m.resyncOnce {
+		m.resyncOnce = false
+		resync = true
+	}
+	return &agentv1.HeartbeatResponse{NeedFullResync: resync, ServerTimeUnix: time.Now().Unix()}, nil
+}
+
+func (m *mockCP) Ack(_ context.Context, req *agentv1.AckRequest) (*agentv1.AckResponse, error) {
+	m.mu.Lock()
+	m.acked[req.CommandID] = true
+	m.mu.Unlock()
+	return &agentv1.AckResponse{}, nil
+}
+
+func (m *mockCP) ReportUsage(_ context.Context, req *agentv1.UsageReport) (*agentv1.UsageAck, error) {
+	m.mu.Lock()
+	m.usage = append(m.usage, req)
+	m.mu.Unlock()
+	return &agentv1.UsageAck{}, nil
+}
+
+func (m *mockCP) Subscribe(req *agentv1.SubscribeRequest, stream agentv1.AgentService_SubscribeServer) error {
+	m.mu.Lock()
+	m.subReqs = append(m.subReqs, req.LastCommandID)
+	backlog := make([]*agentv1.Command, 0)
+	for _, c := range m.issued {
+		if c.CommandID > req.LastCommandID {
+			backlog = append(backlog, c)
+		}
+	}
+	m.mu.Unlock()
+
+	for _, c := range backlog {
+		if err := stream.Send(c); err != nil {
+			return err
+		}
+	}
+	for {
+		select {
+		case <-stream.Context().Done():
+			return stream.Context().Err()
+		case <-m.dropCh:
+			return status.Error(codes.Unavailable, "simulated stream drop")
+		case c := <-m.liveCh:
+			if err := stream.Send(c); err != nil {
+				return err
+			}
+		}
+	}
+}
+
+// ─── mock helpers (test goroutine) ───────────────────────────────────────────
+
+// seedCred adds a credential to the converged Register snapshot before start.
+func (m *mockCP) seedCred(c *agentv1.Credential) {
+	m.mu.Lock()
+	m.snapCreds[c.DpUUID] = c
+	if m.configVer == 0 {
+		m.configVer = 1
+	}
+	m.mu.Unlock()
+}
+
+// push issues a live command AND folds it into the Register snapshot, so a later
+// reconnect's Register stays consistent with the command stream.
+func (m *mockCP) push(cmd *agentv1.Command) {
+	m.applyToSnapshot(cmd, true) // updates snapCreds + lastCmdID
+	m.mu.Lock()
+	m.issued = append(m.issued, cmd)
+	m.mu.Unlock()
+	m.liveCh <- cmd
+}
+
+// enqueueOffline appends a command to the resume backlog WITHOUT touching the
+// Register snapshot — so the only way the agent learns it is via Subscribe resume.
+func (m *mockCP) enqueueOffline(cmd *agentv1.Command) {
+	m.mu.Lock()
+	// Intentionally does NOT advance lastCmdID: this command is not yet reflected
+	// in the Register snapshot, so the agent must learn it via Subscribe resume.
+	m.issued = append(m.issued, cmd)
+	m.mu.Unlock()
+}
+
+func (m *mockCP) applyToSnapshot(cmd *agentv1.Command, bumpVer bool) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	switch cmd.Type {
+	case agentv1.CommandTypeUpsert:
+		m.snapCreds[cmd.Upsert.Credential.DpUUID] = cmd.Upsert.Credential
+	case agentv1.CommandTypeRevoke:
+		delete(m.snapCreds, cmd.Revoke.DpUUID)
+	}
+	if cmd.CommandID > m.lastCmdID {
+		m.lastCmdID = cmd.CommandID
+	}
+	if bumpVer {
+		m.configVer++
+	}
+}
+
+func (m *mockCP) drop()                 { m.dropCh <- struct{}{} }
+func (m *mockCP) registerCount() int    { m.mu.Lock(); defer m.mu.Unlock(); return m.registerN }
+func (m *mockCP) heartbeatCount() int   { m.mu.Lock(); defer m.mu.Unlock(); return m.heartbeatN }
+func (m *mockCP) isAcked(id int64) bool { m.mu.Lock(); defer m.mu.Unlock(); return m.acked[id] }
+func (m *mockCP) subReqList() []int64 {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return append([]int64(nil), m.subReqs...)
+}
+
+// ─── test harness ─────────────────────────────────────────────────────────────
+
+func startMock(t *testing.T, m *mockCP) func(ctx context.Context) (*grpc.ClientConn, error) {
+	t.Helper()
+	lis := bufconn.Listen(1024 * 1024)
+	srv := grpc.NewServer()
+	agentv1.RegisterAgentServiceServer(srv, m)
+	go func() { _ = srv.Serve(lis) }()
+	t.Cleanup(func() { srv.Stop() })
+	return func(ctx context.Context) (*grpc.ClientConn, error) {
+		return grpc.NewClient("passthrough:///bufnet",
+			grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) {
+				return lis.DialContext(ctx)
+			}),
+			grpc.WithTransportCredentials(insecure.NewCredentials()),
+		)
+	}
+}
+
+func fastConfig(t *testing.T) Config {
+	cfg := testConfig(t)
+	cfg.BootstrapToken = "test-token"
+	cfg.HeartbeatInterval = 25 * time.Millisecond
+	cfg.UsageInterval = time.Hour
+	cfg.TTLScanInterval = time.Hour
+	cfg.DebounceWindow = 10 * time.Millisecond
+	cfg.BackoffMin = 10 * time.Millisecond
+	cfg.BackoffMax = 40 * time.Millisecond
+	return cfg
+}
+
+func eventually(t *testing.T, timeout time.Duration, fn func() bool, msg string) {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		if fn() {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatalf("timeout waiting for: %s", msg)
+}
+
+func readRender(path string) string {
+	b, err := os.ReadFile(path)
+	if err != nil {
+		return ""
+	}
+	return string(b)
+}
+
+// TestIntegrationFullFlow exercises Enroll → Register → Heartbeat → Subscribe/Ack,
+// command application (upsert/revoke), render correctness, heartbeat-triggered
+// resync, and reconnect resume via last_command_id.
+func TestIntegrationFullFlow(t *testing.T) {
+	m := newMockCP(t)
+	m.seedCred(&agentv1.Credential{DpUUID: "boot", Protocol: agentv1.ProtocolBoth})
+	dial := startMock(t, m)
+
+	cfg := fastConfig(t)
+	a := New(cfg,
+		WithDialer(Dialer(dial)),
+		WithEnrollDialer(EnrollDialer(dial)),
+		WithRestarter(&fakeRestarter{}),
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	runErr := make(chan error, 1)
+	go func() { runErr <- a.Run(ctx) }()
+
+	// 1) Enrollment persisted the cert/key/ca and the node UUID is set.
+	eventually(t, 2*time.Second, func() bool {
+		_, err1 := os.Stat(cfg.CertPath())
+		_, err2 := os.Stat(cfg.KeyPath())
+		_, err3 := os.Stat(cfg.CAPath())
+		return err1 == nil && err2 == nil && err3 == nil && a.NodeUUID() == testNodeUUID
+	}, "enrollment to persist cert/key/ca and set node UUID")
+
+	// 2) Initial Register snapshot rendered with the seeded credential.
+	eventually(t, 2*time.Second, func() bool {
+		return strings.Contains(readRender(cfg.SingboxConfigPath), `"uuid": "boot"`)
+	}, "initial snapshot credential 'boot' to be rendered")
+
+	// 3) Heartbeats flowing.
+	eventually(t, 2*time.Second, func() bool { return m.heartbeatCount() > 0 }, "heartbeats")
+
+	// 4) Stream a command: upsert 'extra'. Applied + Ack'd + rendered.
+	m.push(&agentv1.Command{
+		CommandID: 1, Type: agentv1.CommandTypeUpsert,
+		Upsert: &agentv1.UpsertPayload{Credential: &agentv1.Credential{DpUUID: "extra", Protocol: agentv1.ProtocolBoth}},
+	})
+	eventually(t, 2*time.Second, func() bool {
+		return strings.Contains(readRender(cfg.SingboxConfigPath), `"uuid": "extra"`) && m.isAcked(1)
+	}, "upsert 'extra' applied and acked")
+
+	// 5) Stream a revoke for 'boot'. Removed from render + Ack'd.
+	m.push(&agentv1.Command{
+		CommandID: 2, Type: agentv1.CommandTypeRevoke,
+		Revoke: &agentv1.RevokePayload{DpUUID: "boot"},
+	})
+	eventually(t, 2*time.Second, func() bool {
+		r := readRender(cfg.SingboxConfigPath)
+		return !strings.Contains(r, `"uuid": "boot"`) && m.isAcked(2)
+	}, "revoke 'boot' applied and acked")
+
+	// 6) Reconnect resume: enqueue an OFFLINE command (only reachable via backlog),
+	//    then drop the stream. After reconnect the agent must resume from
+	//    last_command_id and receive id=3 without redelivering 1/2.
+	m.enqueueOffline(&agentv1.Command{
+		CommandID: 3, Type: agentv1.CommandTypeUpsert,
+		Upsert: &agentv1.UpsertPayload{Credential: &agentv1.Credential{DpUUID: "resumed", Protocol: agentv1.ProtocolBoth}},
+	})
+	m.drop()
+	eventually(t, 3*time.Second, func() bool {
+		return strings.Contains(readRender(cfg.SingboxConfigPath), `"uuid": "resumed"`) && m.isAcked(3)
+	}, "offline command 3 delivered after reconnect resume")
+
+	// The second Subscribe must have resumed from a non-zero high-water mark.
+	reqs := m.subReqList()
+	if len(reqs) < 2 {
+		t.Fatalf("expected at least 2 Subscribe calls (reconnect), got %v", reqs)
+	}
+	if reqs[len(reqs)-1] < 2 {
+		t.Errorf("reconnect Subscribe last_command_id = %d, want >= 2 (resume, no redelivery)", reqs[len(reqs)-1])
+	}
+
+	cancel()
+	select {
+	case <-runErr:
+	case <-time.After(2 * time.Second):
+		t.Fatal("agent did not shut down after context cancel")
+	}
+}
+
+// TestIntegrationHeartbeatResync verifies a need_full_resync heartbeat forces a
+// reconnect + re-Register so the node converges on the authoritative snapshot.
+func TestIntegrationHeartbeatResync(t *testing.T) {
+	m := newMockCP(t)
+	m.seedCred(&agentv1.Credential{DpUUID: "boot", Protocol: agentv1.ProtocolBoth})
+	m.resyncOnce = true
+	dial := startMock(t, m)
+
+	a := New(fastConfig(t),
+		WithDialer(Dialer(dial)),
+		WithEnrollDialer(EnrollDialer(dial)),
+		WithRestarter(&fakeRestarter{}),
+	)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() { _ = a.Run(ctx) }()
+
+	eventually(t, 3*time.Second, func() bool { return m.registerCount() >= 2 }, "re-Register after need_full_resync")
+}
@@ -0,0 +1,95 @@
+package agentd
+
+import (
+	"encoding/json"
+
+	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
+)
+
+// renderSingboxConfig produces a complete sing-box SERVER config JSON for the node:
+// a VLESS+REALITY inbound and a Hysteria2 inbound, each carrying one user per
+// provisioned credential. REALITY users key on the dp_uuid; Hy2 users key on the
+// derived password (DeriveHy2Password) — both from the same dp_uuid source.
+//
+// Only the opaque dp_uuid is ever written; no account identity touches the node.
+func renderSingboxConfig(creds []Cred, reality *agentv1.RealityInbound, hy2 *agentv1.Hy2Inbound, deriveKey string) ([]byte, error) {
+	cfg := map[string]any{
+		"log":       map[string]any{"level": "warn", "timestamp": true},
+		"inbounds":  buildInbounds(creds, reality, hy2, deriveKey),
+		"outbounds": []any{map[string]any{"type": "direct", "tag": "direct"}},
+	}
+	return json.MarshalIndent(cfg, "", "  ")
+}
+
+func buildInbounds(creds []Cred, reality *agentv1.RealityInbound, hy2 *agentv1.Hy2Inbound, deriveKey string) []any {
+	inbounds := make([]any, 0, 2)
+
+	if reality != nil {
+		users := make([]any, 0, len(creds))
+		for _, c := range creds {
+			if c.Protocol == agentv1.ProtocolReality || c.Protocol == agentv1.ProtocolBoth {
+				flow := c.Flow
+				if flow == "" {
+					flow = DefaultFlow
+				}
+				users = append(users, map[string]any{
+					"name": c.DpUUID,
+					"uuid": c.DpUUID,
+					"flow": flow,
+				})
+			}
+		}
+		realityTLS := map[string]any{
+			"enabled":     true,
+			"server_name": reality.ServerName,
+			"reality": map[string]any{
+				"enabled":     true,
+				"private_key": reality.PrivateKey,
+				"short_id":    []string{reality.ShortID},
+				"handshake": map[string]any{
+					"server":      reality.HandshakeServer,
+					"server_port": reality.HandshakePort,
+				},
+			},
+		}
+		inbounds = append(inbounds, map[string]any{
+			"type":        "vless",
+			"tag":         "reality-in",
+			"listen":      "::",
+			"listen_port": reality.ListenPort,
+			"users":       users,
+			"tls":         realityTLS,
+		})
+	}
+
+	if hy2 != nil {
+		users := make([]any, 0, len(creds))
+		for _, c := range creds {
+			if c.Protocol == agentv1.ProtocolHy2 || c.Protocol == agentv1.ProtocolBoth {
+				users = append(users, map[string]any{
+					"name":     c.DpUUID,
+					"password": DeriveHy2Password(c.DpUUID, deriveKey),
+				})
+			}
+		}
+		hy2In := map[string]any{
+			"type":        "hysteria2",
+			"tag":         "hy2-in",
+			"listen":      "::",
+			"listen_port": hy2.ListenPort,
+			"users":       users,
+			"tls": map[string]any{
+				"enabled":          true,
+				"alpn":             []string{"h3"},
+				"certificate_path": hy2.CertPath,
+				"key_path":         hy2.KeyPath,
+			},
+		}
+		if hy2.Masquerade != "" {
+			hy2In["masquerade"] = hy2.Masquerade
+		}
+		inbounds = append(inbounds, hy2In)
+	}
+
+	return inbounds
+}
@@ -0,0 +1,354 @@
+package agentd
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"sort"
+	"sync"
+	"time"
+
+	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
+)
+
+// Restarter applies a freshly-rendered sing-box config. sing-box has no hot
+// reload, so the production implementation restarts the systemd unit (a ~1-2s
+// data-plane blip). Tests inject a fake.
+type Restarter interface {
+	Restart(ctx context.Context) error
+}
+
+// Cred is the agent's in-memory view of one data-plane credential. It is the
+// authoritative local state together with the rendered sing-box config.
+type Cred struct {
+	DpUUID    string
+	Protocol  agentv1.Protocol
+	Flow      string
+	ExpiresAt int64 // unix seconds; 0 = no expiry
+}
+
+// persistedCred is the on-disk shape. Per the no-state invariant we persist ONLY
+// dp_uuid + expires_at — protocol/flow are reconstructed (defaults) on load and
+// then corrected by the next Register reconciliation.
+type persistedCred struct {
+	DpUUID    string `json:"dp_uuid"`
+	ExpiresAt int64  `json:"expires_at"`
+}
+
+type persistedState struct {
+	Creds []persistedCred `json:"creds"`
+}
+
+// SingBox owns the credential table, state.json persistence, sing-box config
+// rendering and the debounced restart loop.
+type SingBox struct {
+	cfg       Config
+	restarter Restarter
+	clock     func() time.Time
+
+	// OnExpire, if set, is invoked (outside the lock) when the TTL sweep removes
+	// a credential, so the agent can surface the removal upstream.
+	OnExpire func(dpUUID string)
+
+	mu            sync.Mutex
+	creds         map[string]*Cred
+	reality       *agentv1.RealityInbound
+	hy2           *agentv1.Hy2Inbound
+	configVersion int64
+
+	// debounce machinery
+	dirty   chan struct{}
+	done    chan struct{}
+	closed  bool
+	startWG sync.WaitGroup
+}
+
+// NewSingBox builds a manager. restarter may be nil (renders only, no restart).
+func NewSingBox(cfg Config, restarter Restarter) *SingBox {
+	if restarter == nil {
+		restarter = noopRestarter{}
+	}
+	return &SingBox{
+		cfg:       cfg.withDefaults(),
+		restarter: restarter,
+		clock:     time.Now,
+		creds:     make(map[string]*Cred),
+		dirty:     make(chan struct{}, 1),
+		done:      make(chan struct{}),
+	}
+}
+
+// LoadState restores the credential table from state.json after a restart. Only
+// dp_uuid + expires_at survive; protocol defaults to BOTH and flow to the
+// configured default until the next Register overwrites them. Expired entries are
+// dropped on load.
+func (s *SingBox) LoadState() error {
+	data, err := os.ReadFile(s.cfg.StatePath())
+	if errors.Is(err, fs.ErrNotExist) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("agentd: read state: %w", err)
+	}
+	var ps persistedState
+	if err := json.Unmarshal(data, &ps); err != nil {
+		return fmt.Errorf("agentd: parse state: %w", err)
+	}
+	now := s.clock().Unix()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for _, pc := range ps.Creds {
+		if pc.ExpiresAt != 0 && pc.ExpiresAt <= now {
+			continue
+		}
+		s.creds[pc.DpUUID] = &Cred{
+			DpUUID:    pc.DpUUID,
+			Protocol:  agentv1.ProtocolBoth,
+			Flow:      DefaultFlow,
+			ExpiresAt: pc.ExpiresAt,
+		}
+	}
+	return nil
+}
+
+// saveStateLocked writes state.json atomically. Caller holds s.mu.
+func (s *SingBox) saveStateLocked() error {
+	ps := persistedState{Creds: make([]persistedCred, 0, len(s.creds))}
+	for _, c := range s.creds {
+		ps.Creds = append(ps.Creds, persistedCred{DpUUID: c.DpUUID, ExpiresAt: c.ExpiresAt})
+	}
+	sort.Slice(ps.Creds, func(i, j int) bool { return ps.Creds[i].DpUUID < ps.Creds[j].DpUUID })
+	data, err := json.MarshalIndent(ps, "", "  ")
+	if err != nil {
+		return err
+	}
+	return atomicWrite(s.cfg.StatePath(), data, 0o600)
+}
+
+// ─── mutations (all idempotent, all schedule a render) ───────────────────────────
+
+// Upsert adds or replaces a credential.
+func (s *SingBox) Upsert(c *Cred) {
+	s.mu.Lock()
+	cp := *c
+	if cp.Flow == "" {
+		cp.Flow = DefaultFlow
+	}
+	s.creds[cp.DpUUID] = &cp
+	_ = s.saveStateLocked()
+	s.mu.Unlock()
+	s.markDirty()
+}
+
+// Revoke removes a credential. No-op if absent (idempotent).
+func (s *SingBox) Revoke(dpUUID string) {
+	s.mu.Lock()
+	_, existed := s.creds[dpUUID]
+	delete(s.creds, dpUUID)
+	if existed {
+		_ = s.saveStateLocked()
+	}
+	s.mu.Unlock()
+	if existed {
+		s.markDirty()
+	}
+}
+
+// Rotate installs new alongside old, keeping old alive until graceUntil. The TTL
+// sweep removes old when its grace expires, giving "宽限期内新旧并存".
+func (s *SingBox) Rotate(oldDpUUID string, newCred *Cred, graceUntil int64) {
+	s.mu.Lock()
+	if old, ok := s.creds[oldDpUUID]; ok && oldDpUUID != newCred.DpUUID {
+		// Clamp old credential's expiry to the grace deadline.
+		if graceUntil > 0 && (old.ExpiresAt == 0 || graceUntil < old.ExpiresAt) {
+			old.ExpiresAt = graceUntil
+		}
+	}
+	nc := *newCred
+	if nc.Flow == "" {
+		nc.Flow = DefaultFlow
+	}
+	s.creds[nc.DpUUID] = &nc
+	_ = s.saveStateLocked()
+	s.mu.Unlock()
+	s.markDirty()
+}
+
+// ApplyConfig replaces the node-wide inbound parameters and (if the snapshot
+// carries credentials) the full credential table. Used by APPLY_CONFIG commands
+// and full resync.
+func (s *SingBox) ApplyConfig(snap *agentv1.ConfigSnapshot, replaceCreds bool) {
+	s.mu.Lock()
+	if snap.Reality != nil {
+		s.reality = snap.Reality
+	}
+	if snap.Hy2 != nil {
+		s.hy2 = snap.Hy2
+	}
+	if snap.ConfigVersion != 0 {
+		s.configVersion = snap.ConfigVersion
+	}
+	if replaceCreds {
+		s.creds = make(map[string]*Cred, len(snap.Credentials))
+		for _, c := range snap.Credentials {
+			flow := c.Flow
+			if flow == "" {
+				flow = DefaultFlow
+			}
+			s.creds[c.DpUUID] = &Cred{
+				DpUUID:    c.DpUUID,
+				Protocol:  c.Protocol,
+				Flow:      flow,
+				ExpiresAt: c.ExpiresAtUnix,
+			}
+		}
+	}
+	_ = s.saveStateLocked()
+	s.mu.Unlock()
+	s.markDirty()
+}
+
+// SetConfigVersion records the authoritative config version (e.g. from Register).
+func (s *SingBox) SetConfigVersion(v int64) {
+	s.mu.Lock()
+	s.configVersion = v
+	s.mu.Unlock()
+}
+
+// ─── read accessors ──────────────────────────────────────────────────────────
+
+func (s *SingBox) ConfigVersion() int64 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.configVersion
+}
+
+func (s *SingBox) OnlinePeers() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return len(s.creds)
+}
+
+// Has reports whether a dp_uuid is currently provisioned.
+func (s *SingBox) Has(dpUUID string) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	_, ok := s.creds[dpUUID]
+	return ok
+}
+
+// Snapshot returns a copy of the current credential set (sorted by dp_uuid).
+func (s *SingBox) Snapshot() []Cred {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := make([]Cred, 0, len(s.creds))
+	for _, c := range s.creds {
+		out = append(out, *c)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].DpUUID < out[j].DpUUID })
+	return out
+}
+
+// ─── TTL sweep ───────────────────────────────────────────────────────────────
+
+// sweepExpired removes every credential whose expires_at has passed. Returns the
+// removed dp_uuids and whether anything changed.
+func (s *SingBox) sweepExpired() []string {
+	now := s.clock().Unix()
+	s.mu.Lock()
+	var removed []string
+	for id, c := range s.creds {
+		if c.ExpiresAt != 0 && c.ExpiresAt <= now {
+			delete(s.creds, id)
+			removed = append(removed, id)
+		}
+	}
+	if len(removed) > 0 {
+		_ = s.saveStateLocked()
+	}
+	s.mu.Unlock()
+	if len(removed) > 0 {
+		s.markDirty()
+		if s.OnExpire != nil {
+			for _, id := range removed {
+				s.OnExpire(id)
+			}
+		}
+	}
+	sort.Strings(removed)
+	return removed
+}
+
+// ─── rendering ───────────────────────────────────────────────────────────────
+
+// RenderConfig builds the sing-box server config JSON from the current state.
+// Exported (and pure w.r.t. the snapshot) so tests can assert its contents.
+func (s *SingBox) RenderConfig() ([]byte, error) {
+	s.mu.Lock()
+	creds := make([]Cred, 0, len(s.creds))
+	for _, c := range s.creds {
+		creds = append(creds, *c)
+	}
+	reality := s.reality
+	hy2 := s.hy2
+	s.mu.Unlock()
+	sort.Slice(creds, func(i, j int) bool { return creds[i].DpUUID < creds[j].DpUUID })
+	return renderSingboxConfig(creds, reality, hy2, s.cfg.DeriveKey)
+}
+
+// writeAndRestart renders, writes the config file and restarts sing-box.
+func (s *SingBox) writeAndRestart(ctx context.Context) error {
+	data, err := s.RenderConfig()
+	if err != nil {
+		return err
+	}
+	if err := os.MkdirAll(filepath.Dir(s.cfg.SingboxConfigPath), 0o755); err != nil {
+		return fmt.Errorf("agentd: mkdir singbox cfg: %w", err)
+	}
+	if err := atomicWrite(s.cfg.SingboxConfigPath, data, 0o644); err != nil {
+		return err
+	}
+	return s.restarter.Restart(ctx)
+}
+
+// markDirty signals the debounce loop that a render is pending (non-blocking).
+func (s *SingBox) markDirty() {
+	select {
+	case s.dirty <- struct{}{}:
+	default:
+	}
+}
+
+// Run drives the debounce loop: it coalesces bursts of changes within
+// DebounceWindow into a single write+restart. It blocks until ctx is cancelled.
+func (s *SingBox) Run(ctx context.Context) {
+	timer := time.NewTimer(time.Hour)
+	timer.Stop()
+	pending := false
+	for {
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		case <-s.dirty:
+			if !pending {
+				pending = true
+				timer.Reset(s.cfg.DebounceWindow)
+			}
+		case <-timer.C:
+			pending = false
+			if err := s.writeAndRestart(ctx); err != nil {
+				logf("singbox render/restart failed: %v", err)
+			}
+		}
+	}
+}
+
+// noopRestarter renders without restarting (dev / config-check only).
+type noopRestarter struct{}
+
+func (noopRestarter) Restart(context.Context) error { return nil }
@@ -0,0 +1,229 @@
+package agentd
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
+)
+
+// fakeRestarter counts Restart calls.
+type fakeRestarter struct {
+	mu sync.Mutex
+	n  int
+}
+
+func (f *fakeRestarter) Restart(context.Context) error {
+	f.mu.Lock()
+	f.n++
+	f.mu.Unlock()
+	return nil
+}
+
+func (f *fakeRestarter) count() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.n
+}
+
+func testConfig(t *testing.T) Config {
+	t.Helper()
+	dir := t.TempDir()
+	return Config{
+		ControlPlaneAddr:  "bufnet",
+		StateDir:          dir,
+		SingboxConfigPath: filepath.Join(dir, "singbox", "config.json"),
+		DeriveKey:         "test-derive-key",
+		DebounceWindow:    20 * time.Millisecond,
+		Insecure:          true,
+	}.withDefaults()
+}
+
+func sampleSnapshot(creds ...*agentv1.Credential) *agentv1.ConfigSnapshot {
+	return &agentv1.ConfigSnapshot{
+		ConfigVersion: 7,
+		Credentials:   creds,
+		Reality: &agentv1.RealityInbound{
+			ListenPort: 11443, PrivateKey: "pk", ShortID: "deadbeef",
+			ServerName: "www.apple.com", HandshakeServer: "www.apple.com", HandshakePort: 443,
+		},
+		Hy2:           &agentv1.Hy2Inbound{ListenPort: 443, Masquerade: "https://www.bing.com", CertPath: "/c", KeyPath: "/k"},
+		LastCommandID: 3,
+	}
+}
+
+func TestRenderContainsCredentials(t *testing.T) {
+	sb := NewSingBox(testConfig(t), nil)
+	sb.ApplyConfig(sampleSnapshot(
+		&agentv1.Credential{DpUUID: "aaaa", Protocol: agentv1.ProtocolBoth},
+		&agentv1.Credential{DpUUID: "bbbb", Protocol: agentv1.ProtocolReality},
+		&agentv1.Credential{DpUUID: "cccc", Protocol: agentv1.ProtocolHy2},
+	), true)
+
+	data, err := sb.RenderConfig()
+	if err != nil {
+		t.Fatal(err)
+	}
+	s := string(data)
+
+	// REALITY inbound: aaaa + bbbb as VLESS uuids, with flow; cccc must NOT be there.
+	for _, want := range []string{`"uuid": "aaaa"`, `"uuid": "bbbb"`, `"flow": "xtls-rprx-vision"`} {
+		if !strings.Contains(s, want) {
+			t.Errorf("rendered config missing %q\n%s", want, s)
+		}
+	}
+	if strings.Contains(s, `"uuid": "cccc"`) {
+		t.Errorf("hy2-only credential cccc leaked into REALITY users")
+	}
+
+	// Hy2 inbound: derived passwords for aaaa + cccc, never the raw dp_uuid as pw.
+	var cfg map[string]any
+	if err := json.Unmarshal(data, &cfg); err != nil {
+		t.Fatalf("rendered config is not valid JSON: %v", err)
+	}
+	if !strings.Contains(s, DeriveHy2Password("aaaa", "test-derive-key")) {
+		t.Error("hy2 password for aaaa not found")
+	}
+	if !strings.Contains(s, DeriveHy2Password("cccc", "test-derive-key")) {
+		t.Error("hy2 password for cccc not found")
+	}
+}
+
+func TestRevokeRemovesFromRender(t *testing.T) {
+	sb := NewSingBox(testConfig(t), nil)
+	sb.ApplyConfig(sampleSnapshot(&agentv1.Credential{DpUUID: "aaaa", Protocol: agentv1.ProtocolBoth}), true)
+	sb.Revoke("aaaa")
+	data, _ := sb.RenderConfig()
+	if strings.Contains(string(data), "aaaa") {
+		t.Errorf("revoked credential still present:\n%s", data)
+	}
+	if sb.Has("aaaa") {
+		t.Error("Has reports revoked credential as present")
+	}
+}
+
+func TestRotateGraceBothPresentThenOldExpires(t *testing.T) {
+	cfg := testConfig(t)
+	sb := NewSingBox(cfg, nil)
+	now := time.Unix(1000, 0)
+	sb.clock = func() time.Time { return now }
+	sb.ApplyConfig(sampleSnapshot(&agentv1.Credential{DpUUID: "old", Protocol: agentv1.ProtocolBoth}), true)
+
+	grace := int64(1005)
+	sb.Rotate("old", &Cred{DpUUID: "new", Protocol: agentv1.ProtocolBoth}, grace)
+
+	// During grace both coexist.
+	if !sb.Has("old") || !sb.Has("new") {
+		t.Fatalf("expected both old+new during grace; old=%v new=%v", sb.Has("old"), sb.Has("new"))
+	}
+
+	// After grace, sweep removes old, keeps new.
+	now = time.Unix(1006, 0)
+	removed := sb.sweepExpired()
+	if len(removed) != 1 || removed[0] != "old" {
+		t.Fatalf("expected old removed by sweep, got %v", removed)
+	}
+	if sb.Has("old") || !sb.Has("new") {
+		t.Fatalf("after grace: old should be gone, new present; old=%v new=%v", sb.Has("old"), sb.Has("new"))
+	}
+}
+
+func TestStateRecovery(t *testing.T) {
+	cfg := testConfig(t)
+	now := time.Unix(2000, 0)
+
+	// First manager: add creds (one already expired), persist via mutation.
+	sb1 := NewSingBox(cfg, nil)
+	sb1.clock = func() time.Time { return now }
+	sb1.Upsert(&Cred{DpUUID: "live", Protocol: agentv1.ProtocolBoth, ExpiresAt: 3000})
+	sb1.Upsert(&Cred{DpUUID: "dead", Protocol: agentv1.ProtocolBoth, ExpiresAt: 1000}) // already expired
+
+	// state.json must contain ONLY dp_uuid + expires_at — no identity fields.
+	raw, err := os.ReadFile(cfg.StatePath())
+	if err != nil {
+		t.Fatal(err)
+	}
+	assertNoIdentityFields(t, raw)
+
+	// Second manager (simulated kill -9 + restart) restores from disk.
+	sb2 := NewSingBox(cfg, nil)
+	sb2.clock = func() time.Time { return now }
+	if err := sb2.LoadState(); err != nil {
+		t.Fatal(err)
+	}
+	if !sb2.Has("live") {
+		t.Error("live credential not recovered from state.json")
+	}
+	if sb2.Has("dead") {
+		t.Error("expired credential should be dropped on load")
+	}
+	// Recovered creds default to BOTH + default flow until Register reconciles.
+	for _, c := range sb2.Snapshot() {
+		if c.DpUUID == "live" && (c.Protocol != agentv1.ProtocolBoth || c.Flow != DefaultFlow) {
+			t.Errorf("recovered cred has unexpected defaults: %+v", c)
+		}
+	}
+}
+
+func TestTTLSweepReportsExpiry(t *testing.T) {
+	cfg := testConfig(t)
+	sb := NewSingBox(cfg, nil)
+	now := time.Unix(5000, 0)
+	sb.clock = func() time.Time { return now }
+
+	var mu sync.Mutex
+	var reported []string
+	sb.OnExpire = func(dp string) { mu.Lock(); reported = append(reported, dp); mu.Unlock() }
+
+	sb.Upsert(&Cred{DpUUID: "x", Protocol: agentv1.ProtocolBoth, ExpiresAt: 4999})
+	sb.Upsert(&Cred{DpUUID: "y", Protocol: agentv1.ProtocolBoth, ExpiresAt: 9999})
+
+	removed := sb.sweepExpired()
+	if len(removed) != 1 || removed[0] != "x" {
+		t.Fatalf("sweep removed = %v, want [x]", removed)
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if len(reported) != 1 || reported[0] != "x" {
+		t.Fatalf("OnExpire reported = %v, want [x]", reported)
+	}
+}
+
+func TestDebounceCoalescesRestarts(t *testing.T) {
+	cfg := testConfig(t)
+	cfg.DebounceWindow = 40 * time.Millisecond
+	fr := &fakeRestarter{}
+	sb := NewSingBox(cfg, fr)
+	sb.ApplyConfig(sampleSnapshot(), true) // set inbounds so render succeeds
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go sb.Run(ctx)
+
+	// Burst of changes within one debounce window.
+	for i := 0; i < 5; i++ {
+		sb.Upsert(&Cred{DpUUID: string(rune('a' + i)), Protocol: agentv1.ProtocolBoth})
+	}
+	time.Sleep(150 * time.Millisecond)
+
+	if n := fr.count(); n != 1 {
+		t.Fatalf("expected 1 coalesced restart, got %d", n)
+	}
+}
+
+// assertNoIdentityFields fails if the blob contains any user-identity token.
+func assertNoIdentityFields(t *testing.T, blob []byte) {
+	t.Helper()
+	lower := strings.ToLower(string(blob))
+	for _, banned := range []string{"email", "user_id", "userid", "phone", "device_id", "deviceid", "account", "destination", "dns"} {
+		if strings.Contains(lower, banned) {
+			t.Errorf("forbidden identity field %q found in persisted/rendered data:\n%s", banned, blob)
+		}
+	}
+}
@@ -0,0 +1,30 @@
+package agentd
+
+import (
+	"context"
+	"time"
+)
+
+// runTTL periodically sweeps expired credentials out of the local table. It runs
+// for the whole agent lifetime (independent of the control-plane connection) so
+// credentials expire on time even during a network partition.
+//
+// Per doc/06, sing-box has no native per-user TTL: free-plan credentials carry the
+// current day's remaining minutes as expires_at, and paid (24h) credentials are
+// refreshed by RotateCredential/Upsert commands from the control plane. When a
+// credential expires the agent removes it locally, re-renders sing-box and (via
+// SingBox.OnExpire) surfaces the removal upstream.
+func (a *Agent) runTTL(ctx context.Context) {
+	ticker := time.NewTicker(a.cfg.TTLScanInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if removed := a.sb.sweepExpired(); len(removed) > 0 {
+				logf("ttl sweep removed %d expired credential(s)", len(removed))
+			}
+		}
+	}
+}
@@ -0,0 +1,54 @@
+package agentd
+
+import (
+	"context"
+	"time"
+
+	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
+)
+
+// UsageSource yields per-dp_uuid aggregate counters for the elapsed window. It
+// MUST return only dp_uuids and byte/minute counts — never user ids, addresses or
+// DNS. The default returns nothing (no stats backend wired); production plugs in a
+// sing-box stats reader.
+type UsageSource interface {
+	// Collect drains and returns usage accumulated since the previous call.
+	Collect() []*agentv1.UsageEntry
+}
+
+// nopUsageSource reports nothing.
+type nopUsageSource struct{}
+
+func (nopUsageSource) Collect() []*agentv1.UsageEntry { return nil }
+
+// runUsage periodically aggregates and uploads usage. Reporting failures are
+// logged but do not tear down the session (usage is best-effort, at-least-once).
+func (a *Agent) runUsage(ctx context.Context, client agentv1.AgentServiceClient) error {
+	ticker := time.NewTicker(a.cfg.UsageInterval)
+	defer ticker.Stop()
+	windowStart := a.clock().Unix()
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		}
+		entries := a.usage.Collect()
+		if len(entries) == 0 {
+			windowStart = a.clock().Unix()
+			continue
+		}
+		now := a.clock().Unix()
+		_, err := client.ReportUsage(ctx, &agentv1.UsageReport{
+			NodeUUID:        a.NodeUUID(),
+			WindowStartUnix: windowStart,
+			WindowEndUnix:   now,
+			Entries:         entries,
+		})
+		if err != nil {
+			// Connection-level errors should restart the session; surface them.
+			return err
+		}
+		windowStart = now
+	}
+}
@@ -0,0 +1,62 @@
+package agentd
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"os/exec"
+	"path/filepath"
+)
+
+// logf is the agent's structured-ish logger. It deliberately never logs
+// dp_uuids, peers' addresses or any payload — only operational events — to honour
+// the no-log policy.
+func logf(format string, args ...any) {
+	log.Printf("[pangolin-agent] "+format, args...)
+}
+
+// atomicWrite writes data to a temp file in the same directory and renames it into
+// place, so a crash mid-write never leaves a truncated config/state file.
+func atomicWrite(path string, data []byte, perm os.FileMode) error {
+	dir := filepath.Dir(path)
+	tmp, err := os.CreateTemp(dir, ".tmp-*")
+	if err != nil {
+		return fmt.Errorf("agentd: temp file: %w", err)
+	}
+	tmpName := tmp.Name()
+	defer os.Remove(tmpName) // no-op if the rename succeeded
+	if _, err := tmp.Write(data); err != nil {
+		tmp.Close()
+		return fmt.Errorf("agentd: write temp: %w", err)
+	}
+	if err := tmp.Chmod(perm); err != nil {
+		tmp.Close()
+		return fmt.Errorf("agentd: chmod temp: %w", err)
+	}
+	if err := tmp.Close(); err != nil {
+		return fmt.Errorf("agentd: close temp: %w", err)
+	}
+	if err := os.Rename(tmpName, path); err != nil {
+		return fmt.Errorf("agentd: rename temp: %w", err)
+	}
+	return nil
+}
+
+// SystemdRestarter restarts the sing-box unit via `systemctl restart`.
+type SystemdRestarter struct {
+	Unit string // e.g. "sing-box"
+}
+
+// Restart runs `systemctl restart <unit>`.
+func (r SystemdRestarter) Restart(ctx context.Context) error {
+	unit := r.Unit
+	if unit == "" {
+		unit = "sing-box"
+	}
+	cmd := exec.CommandContext(ctx, "systemctl", "restart", unit)
+	if out, err := cmd.CombinedOutput(); err != nil {
+		return fmt.Errorf("agentd: restart %s: %w: %s", unit, err, out)
+	}
+	return nil
+}
@@ -0,0 +1,30 @@
+package agentv1
+
+import (
+	"encoding/json"
+
+	"google.golang.org/grpc/encoding"
+)
+
+// CodecName is the gRPC content-subtype used by the agent contract. Because this
+// repository has no protoc pipeline yet, messages are encoded as JSON rather than
+// protobuf wire format. Clients select it with grpc.CallContentSubtype(CodecName)
+// (wired automatically by the client/server helpers in service.go); the server
+// echoes the same subtype on responses.
+const CodecName = "pangolinagentjson"
+
+func init() { encoding.RegisterCodec(jsonCodec{}) }
+
+// jsonCodec implements google.golang.org/grpc/encoding.Codec over encoding/json.
+type jsonCodec struct{}
+
+func (jsonCodec) Marshal(v any) ([]byte, error) { return json.Marshal(v) }
+
+func (jsonCodec) Unmarshal(data []byte, v any) error {
+	if len(data) == 0 {
+		return nil // empty message (e.g. AckResponse / UsageAck)
+	}
+	return json.Unmarshal(data, v)
+}
+
+func (jsonCodec) Name() string { return CodecName }
@@ -0,0 +1,238 @@
+package agentv1
+
+import (
+	"context"
+
+	"google.golang.org/grpc"
+)
+
+// ServiceName is the fully-qualified gRPC service name. It MUST match the value
+// the mTLS interceptor whitelists for Enroll (server/internal/mtls/identity.go).
+const ServiceName = "pangolin.agent.v1.AgentService"
+
+// Fully-qualified method names.
+const (
+	MethodEnroll      = "/" + ServiceName + "/Enroll"
+	MethodRegister    = "/" + ServiceName + "/Register"
+	MethodHeartbeat   = "/" + ServiceName + "/Heartbeat"
+	MethodSubscribe   = "/" + ServiceName + "/Subscribe"
+	MethodAck         = "/" + ServiceName + "/Ack"
+	MethodReportUsage = "/" + ServiceName + "/ReportUsage"
+)
+
+// withCodec forces the JSON content-subtype on every call so the contract does
+// not depend on the caller remembering to set dial-level call options.
+func withCodec(opts []grpc.CallOption) []grpc.CallOption {
+	return append([]grpc.CallOption{grpc.CallContentSubtype(CodecName)}, opts...)
+}
+
+// ─── server interface ──────────────────────────────────────────────────────────
+
+// AgentServiceServer is implemented by the control plane.
+type AgentServiceServer interface {
+	Enroll(context.Context, *EnrollRequest) (*EnrollResponse, error)
+	Register(context.Context, *RegisterRequest) (*ConfigSnapshot, error)
+	Heartbeat(context.Context, *HeartbeatRequest) (*HeartbeatResponse, error)
+	Subscribe(*SubscribeRequest, AgentService_SubscribeServer) error
+	Ack(context.Context, *AckRequest) (*AckResponse, error)
+	ReportUsage(context.Context, *UsageReport) (*UsageAck, error)
+}
+
+// AgentService_SubscribeServer is the server side of the Command stream.
+type AgentService_SubscribeServer interface {
+	Send(*Command) error
+	grpc.ServerStream
+}
+
+type subscribeServer struct{ grpc.ServerStream }
+
+func (s *subscribeServer) Send(m *Command) error { return s.ServerStream.SendMsg(m) }
+
+// RegisterAgentServiceServer wires srv into a gRPC server (or any ServiceRegistrar).
+func RegisterAgentServiceServer(s grpc.ServiceRegistrar, srv AgentServiceServer) {
+	s.RegisterService(&serviceDesc, srv)
+}
+
+func handlerEnroll(srv any, ctx context.Context, dec func(any) error, interceptor grpc.UnaryServerInterceptor) (any, error) {
+	in := new(EnrollRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(AgentServiceServer).Enroll(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: MethodEnroll}
+	return interceptor(ctx, in, info, func(ctx context.Context, req any) (any, error) {
+		return srv.(AgentServiceServer).Enroll(ctx, req.(*EnrollRequest))
+	})
+}
+
+func handlerRegister(srv any, ctx context.Context, dec func(any) error, interceptor grpc.UnaryServerInterceptor) (any, error) {
+	in := new(RegisterRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(AgentServiceServer).Register(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: MethodRegister}
+	return interceptor(ctx, in, info, func(ctx context.Context, req any) (any, error) {
+		return srv.(AgentServiceServer).Register(ctx, req.(*RegisterRequest))
+	})
+}
+
+func handlerHeartbeat(srv any, ctx context.Context, dec func(any) error, interceptor grpc.UnaryServerInterceptor) (any, error) {
+	in := new(HeartbeatRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(AgentServiceServer).Heartbeat(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: MethodHeartbeat}
+	return interceptor(ctx, in, info, func(ctx context.Context, req any) (any, error) {
+		return srv.(AgentServiceServer).Heartbeat(ctx, req.(*HeartbeatRequest))
+	})
+}
+
+func handlerAck(srv any, ctx context.Context, dec func(any) error, interceptor grpc.UnaryServerInterceptor) (any, error) {
+	in := new(AckRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(AgentServiceServer).Ack(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: MethodAck}
+	return interceptor(ctx, in, info, func(ctx context.Context, req any) (any, error) {
+		return srv.(AgentServiceServer).Ack(ctx, req.(*AckRequest))
+	})
+}
+
+func handlerReportUsage(srv any, ctx context.Context, dec func(any) error, interceptor grpc.UnaryServerInterceptor) (any, error) {
+	in := new(UsageReport)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(AgentServiceServer).ReportUsage(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: MethodReportUsage}
+	return interceptor(ctx, in, info, func(ctx context.Context, req any) (any, error) {
+		return srv.(AgentServiceServer).ReportUsage(ctx, req.(*UsageReport))
+	})
+}
+
+func handlerSubscribe(srv any, stream grpc.ServerStream) error {
+	m := new(SubscribeRequest)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(AgentServiceServer).Subscribe(m, &subscribeServer{stream})
+}
+
+var serviceDesc = grpc.ServiceDesc{
+	ServiceName: ServiceName,
+	HandlerType: (*AgentServiceServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{MethodName: "Enroll", Handler: handlerEnroll},
+		{MethodName: "Register", Handler: handlerRegister},
+		{MethodName: "Heartbeat", Handler: handlerHeartbeat},
+		{MethodName: "Ack", Handler: handlerAck},
+		{MethodName: "ReportUsage", Handler: handlerReportUsage},
+	},
+	Streams: []grpc.StreamDesc{
+		{StreamName: "Subscribe", Handler: handlerSubscribe, ServerStreams: true},
+	},
+	Metadata: "proto/agent/v1/agent.proto",
+}
+
+// ─── client ─────────────────────────────────────────────────────────────────────
+
+// AgentServiceClient is consumed by the node agent.
+type AgentServiceClient interface {
+	Enroll(ctx context.Context, in *EnrollRequest, opts ...grpc.CallOption) (*EnrollResponse, error)
+	Register(ctx context.Context, in *RegisterRequest, opts ...grpc.CallOption) (*ConfigSnapshot, error)
+	Heartbeat(ctx context.Context, in *HeartbeatRequest, opts ...grpc.CallOption) (*HeartbeatResponse, error)
+	Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (AgentService_SubscribeClient, error)
+	Ack(ctx context.Context, in *AckRequest, opts ...grpc.CallOption) (*AckResponse, error)
+	ReportUsage(ctx context.Context, in *UsageReport, opts ...grpc.CallOption) (*UsageAck, error)
+}
+
+// AgentService_SubscribeClient is the client side of the Command stream.
+type AgentService_SubscribeClient interface {
+	Recv() (*Command, error)
+	grpc.ClientStream
+}
+
+type subscribeClient struct{ grpc.ClientStream }
+
+func (c *subscribeClient) Recv() (*Command, error) {
+	m := new(Command)
+	if err := c.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+type agentServiceClient struct{ cc grpc.ClientConnInterface }
+
+// NewAgentServiceClient returns a client bound to cc. All calls use the JSON codec.
+func NewAgentServiceClient(cc grpc.ClientConnInterface) AgentServiceClient {
+	return &agentServiceClient{cc}
+}
+
+func (c *agentServiceClient) Enroll(ctx context.Context, in *EnrollRequest, opts ...grpc.CallOption) (*EnrollResponse, error) {
+	out := new(EnrollResponse)
+	if err := c.cc.Invoke(ctx, MethodEnroll, in, out, withCodec(opts)...); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *agentServiceClient) Register(ctx context.Context, in *RegisterRequest, opts ...grpc.CallOption) (*ConfigSnapshot, error) {
+	out := new(ConfigSnapshot)
+	if err := c.cc.Invoke(ctx, MethodRegister, in, out, withCodec(opts)...); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *agentServiceClient) Heartbeat(ctx context.Context, in *HeartbeatRequest, opts ...grpc.CallOption) (*HeartbeatResponse, error) {
+	out := new(HeartbeatResponse)
+	if err := c.cc.Invoke(ctx, MethodHeartbeat, in, out, withCodec(opts)...); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *agentServiceClient) Ack(ctx context.Context, in *AckRequest, opts ...grpc.CallOption) (*AckResponse, error) {
+	out := new(AckResponse)
+	if err := c.cc.Invoke(ctx, MethodAck, in, out, withCodec(opts)...); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *agentServiceClient) ReportUsage(ctx context.Context, in *UsageReport, opts ...grpc.CallOption) (*UsageAck, error) {
+	out := new(UsageAck)
+	if err := c.cc.Invoke(ctx, MethodReportUsage, in, out, withCodec(opts)...); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *agentServiceClient) Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (AgentService_SubscribeClient, error) {
+	stream, err := c.cc.NewStream(ctx, &serviceDesc.Streams[0], MethodSubscribe, withCodec(opts)...)
+	if err != nil {
+		return nil, err
+	}
+	x := &subscribeClient{stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
@@ -0,0 +1,176 @@
+// Package agentv1 is the control-plane <-> node-agent gRPC contract.
+//
+// These types mirror server/proto/agent/v1/agent.proto 1:1. The repository has no
+// protoc/buf pipeline wired yet (Makefile `generate` is a stub from task #5), so
+// the messages are hand-written Go structs and the gRPC stubs (service.go) use a
+// JSON codec (codec.go) instead of protobuf wire format. The JSON field names are
+// the proto field names so a future protoc-generated package is drop-in
+// compatible at the API surface.
+//
+// Data-plane red line: nothing in this package carries user_id, email, device id,
+// destination address or DNS data — only the opaque dp_uuid and aggregate counters.
+package agentv1
+
+// Protocol selects which sing-box inbound(s) a credential is provisioned on.
+type Protocol int32
+
+const (
+	ProtocolUnspecified Protocol = 0
+	ProtocolReality     Protocol = 1
+	ProtocolHy2         Protocol = 2
+	ProtocolBoth        Protocol = 3
+)
+
+// CommandType discriminates the Command payload.
+type CommandType int32
+
+const (
+	CommandTypeUnspecified      CommandType = 0
+	CommandTypeUpsert           CommandType = 1
+	CommandTypeRevoke           CommandType = 2
+	CommandTypeRotateCredential CommandType = 3
+	CommandTypeApplyConfig      CommandType = 4
+	CommandTypeLifecycle        CommandType = 5
+)
+
+// LifecycleAction is the verb of a LIFECYCLE command.
+type LifecycleAction int32
+
+const (
+	LifecycleActionUnspecified LifecycleAction = 0
+	LifecycleActionDrain       LifecycleAction = 1
+	LifecycleActionResume      LifecycleAction = 2
+	LifecycleActionShutdown    LifecycleAction = 3
+)
+
+// ─── enroll ──────────────────────────────────────────────────────────────────
+
+type EnrollRequest struct {
+	BootstrapToken string `json:"bootstrap_token,omitempty"`
+	CSRPEM         []byte `json:"csr_pem,omitempty"`
+	AgentVersion   string `json:"agent_version,omitempty"`
+}
+
+type EnrollResponse struct {
+	NodeUUID     string `json:"node_uuid,omitempty"`
+	CertPEM      []byte `json:"cert_pem,omitempty"`
+	CAPEM        []byte `json:"ca_pem,omitempty"`
+	NotAfterUnix int64  `json:"not_after_unix,omitempty"`
+}
+
+// ─── register / config ─────────────────────────────────────────────────────────
+
+type RegisterRequest struct {
+	NodeUUID           string `json:"node_uuid,omitempty"`
+	AgentVersion       string `json:"agent_version,omitempty"`
+	LocalConfigVersion int64  `json:"local_config_version,omitempty"`
+}
+
+// Credential is the only per-subscriber object a node ever sees.
+type Credential struct {
+	DpUUID        string   `json:"dp_uuid,omitempty"`
+	Protocol      Protocol `json:"protocol,omitempty"`
+	Flow          string   `json:"flow,omitempty"`
+	ExpiresAtUnix int64    `json:"expires_at_unix,omitempty"`
+}
+
+type ConfigSnapshot struct {
+	ConfigVersion int64           `json:"config_version,omitempty"`
+	Credentials   []*Credential   `json:"credentials,omitempty"`
+	Reality       *RealityInbound `json:"reality,omitempty"`
+	Hy2           *Hy2Inbound     `json:"hy2,omitempty"`
+	LastCommandID int64           `json:"last_command_id,omitempty"`
+}
+
+type RealityInbound struct {
+	ListenPort      int32  `json:"listen_port,omitempty"`
+	PrivateKey      string `json:"private_key,omitempty"`
+	ShortID         string `json:"short_id,omitempty"`
+	ServerName      string `json:"server_name,omitempty"`
+	HandshakeServer string `json:"handshake_server,omitempty"`
+	HandshakePort   int32  `json:"handshake_port,omitempty"`
+}
+
+type Hy2Inbound struct {
+	ListenPort int32  `json:"listen_port,omitempty"`
+	Masquerade string `json:"masquerade,omitempty"`
+	CertPath   string `json:"cert_path,omitempty"`
+	KeyPath    string `json:"key_path,omitempty"`
+}
+
+// ─── heartbeat ───────────────────────────────────────────────────────────────
+
+type HeartbeatRequest struct {
+	NodeUUID         string  `json:"node_uuid,omitempty"`
+	ConfigVersion    int64   `json:"config_version,omitempty"`
+	OnlinePeers      int32   `json:"online_peers,omitempty"`
+	BandwidthUpBps   int64   `json:"bandwidth_up_bps,omitempty"`
+	BandwidthDownBps int64   `json:"bandwidth_down_bps,omitempty"`
+	CPUPercent       float64 `json:"cpu_percent,omitempty"`
+	TimestampUnix    int64   `json:"timestamp_unix,omitempty"`
+}
+
+type HeartbeatResponse struct {
+	NeedFullResync bool  `json:"need_full_resync,omitempty"`
+	ServerTimeUnix int64 `json:"server_time_unix,omitempty"`
+}
+
+// ─── command stream ──────────────────────────────────────────────────────────
+
+type Command struct {
+	CommandID   int64             `json:"command_id,omitempty"`
+	Type        CommandType       `json:"type,omitempty"`
+	Upsert      *UpsertPayload    `json:"upsert,omitempty"`
+	Revoke      *RevokePayload    `json:"revoke,omitempty"`
+	Rotate      *RotatePayload    `json:"rotate,omitempty"`
+	ApplyConfig *ConfigSnapshot   `json:"apply_config,omitempty"`
+	Lifecycle   *LifecyclePayload `json:"lifecycle,omitempty"`
+}
+
+type UpsertPayload struct {
+	Credential *Credential `json:"credential,omitempty"`
+}
+
+type RevokePayload struct {
+	DpUUID string `json:"dp_uuid,omitempty"`
+}
+
+type RotatePayload struct {
+	OldDpUUID      string      `json:"old_dp_uuid,omitempty"`
+	NewCredential  *Credential `json:"new_credential,omitempty"`
+	GraceUntilUnix int64       `json:"grace_until_unix,omitempty"`
+}
+
+type LifecyclePayload struct {
+	Action LifecycleAction `json:"action,omitempty"`
+}
+
+type SubscribeRequest struct {
+	NodeUUID      string `json:"node_uuid,omitempty"`
+	LastCommandID int64  `json:"last_command_id,omitempty"`
+}
+
+type AckRequest struct {
+	NodeUUID  string `json:"node_uuid,omitempty"`
+	CommandID int64  `json:"command_id,omitempty"`
+}
+
+type AckResponse struct{}
+
+// ─── usage ───────────────────────────────────────────────────────────────────
+
+type UsageEntry struct {
+	DpUUID         string `json:"dp_uuid,omitempty"`
+	BytesUp        int64  `json:"bytes_up,omitempty"`
+	BytesDown      int64  `json:"bytes_down,omitempty"`
+	SessionMinutes int64  `json:"session_minutes,omitempty"`
+}
+
+type UsageReport struct {
+	NodeUUID        string        `json:"node_uuid,omitempty"`
+	WindowStartUnix int64         `json:"window_start_unix,omitempty"`
+	WindowEndUnix   int64         `json:"window_end_unix,omitempty"`
+	Entries         []*UsageEntry `json:"entries,omitempty"`
+}
+
+type UsageAck struct{}
@@ -0,0 +1,212 @@
+// Package pangolin.agent.v1 is the frozen control-plane <-> node-agent contract.
+//
+// CONNECTION DIRECTION: the node agent ALWAYS dials the control plane (never the
+// reverse). Every RPC except Enroll requires a verified mTLS client certificate
+// whose Subject CN equals the node UUID (see server/internal/mtls). Enroll is the
+// single bootstrap exception, authenticated by a one-time bootstrap token instead.
+//
+// DATA-PLANE RED LINE (doc/06 §3, doc/04 §2): messages on this contract carry ONLY
+// the opaque data-plane credential id (dp_uuid) and aggregate counters. They MUST
+// NOT carry user_id, email, device id, destination address, SNI or any DNS query.
+//
+// NOTE ON CODE GENERATION: this repository has no protoc/buf toolchain wired yet
+// (Makefile `generate` is a stub). Until task #5's proto pipeline lands, the Go
+// types and gRPC stubs in server/internal/pb/agentv1 are hand-written to mirror
+// this file 1:1 and are wire-encoded with a JSON gRPC codec. When protoc is
+// introduced this file is the source of truth; field numbers are reserved here so
+// the generated code stays compatible.
+syntax = "proto3";
+
+package pangolin.agent.v1;
+
+option go_package = "github.com/wangjia/pangolin/server/internal/pb/agentv1;agentv1";
+
+// AgentService is implemented by the control plane and consumed by node agents.
+service AgentService {
+  // Enroll exchanges a one-time bootstrap token + CSR for a node client
+  // certificate (CN = node_uuid, 90d). Exempt from mTLS — this is the only RPC an
+  // un-enrolled agent may call. Idempotent only within the token TTL.
+  rpc Enroll(EnrollRequest) returns (EnrollResponse);
+
+  // Register is called after (re)connecting over mTLS. It returns the authoritative
+  // full ConfigSnapshot so the agent can converge local state.
+  rpc Register(RegisterRequest) returns (ConfigSnapshot);
+
+  // Heartbeat reports liveness + load every ~30s and the agent's local
+  // config_version. The control plane flags need_full_resync when it detects drift.
+  rpc Heartbeat(HeartbeatRequest) returns (HeartbeatResponse);
+
+  // Subscribe is a server-streaming feed of Commands. The agent passes the last
+  // command_id it durably processed so the control plane can resume without gaps
+  // (at-least-once; commands are idempotent and Ack'd individually).
+  rpc Subscribe(SubscribeRequest) returns (stream Command);
+
+  // Ack confirms a command was applied. Safe to retry.
+  rpc Ack(AckRequest) returns (AckResponse);
+
+  // ReportUsage uploads per-dp_uuid aggregate byte/minute counters. No identities.
+  rpc ReportUsage(UsageReport) returns (UsageAck);
+}
+
+// ─── enroll ──────────────────────────────────────────────────────────────────
+
+message EnrollRequest {
+  string bootstrap_token = 1; // one-time, injected via cloud-init
+  bytes csr_pem = 2;          // PKCS#10 CSR, agent-generated key never leaves node
+  string agent_version = 3;
+}
+
+message EnrollResponse {
+  string node_uuid = 1;  // authoritative; becomes the cert CN
+  bytes cert_pem = 2;    // signed leaf, 90d, EKU=clientAuth
+  bytes ca_pem = 3;      // CA to pin for server validation
+  int64 not_after_unix = 4;
+}
+
+// ─── register / config ─────────────────────────────────────────────────────────
+
+message RegisterRequest {
+  string node_uuid = 1;
+  string agent_version = 2;
+  int64 local_config_version = 3; // 0 on a fresh node
+}
+
+// Protocol selects which inbound(s) a credential is provisioned on.
+enum Protocol {
+  PROTOCOL_UNSPECIFIED = 0;
+  PROTOCOL_REALITY = 1; // VLESS+REALITY inbound, user = {uuid, flow}
+  PROTOCOL_HY2 = 2;     // Hysteria2 inbound, user = {password}
+  PROTOCOL_BOTH = 3;    // both inbounds share the same dp_uuid
+}
+
+// Credential is the ONLY per-subscriber object a node ever sees.
+message Credential {
+  string dp_uuid = 1;          // opaque data-plane id; REALITY user uuid
+  Protocol protocol = 2;
+  string flow = 3;             // e.g. "xtls-rprx-vision"
+  int64 expires_at_unix = 4;   // agent-side TTL; 0 = no expiry
+}
+
+// ConfigSnapshot is the full authoritative node state.
+message ConfigSnapshot {
+  int64 config_version = 1;
+  repeated Credential credentials = 2;
+
+  // Inbound parameters (node-wide, not per-user).
+  RealityInbound reality = 3;
+  Hy2Inbound hy2 = 4;
+
+  // Baseline command id already reflected in this snapshot; the agent resumes
+  // Subscribe from here.
+  int64 last_command_id = 5;
+}
+
+message RealityInbound {
+  int32 listen_port = 1;
+  string private_key = 2; // REALITY server private key (base64)
+  string short_id = 3;
+  string server_name = 4; // masquerade SNI handshake target
+  string handshake_server = 5;
+  int32 handshake_port = 6;
+}
+
+message Hy2Inbound {
+  int32 listen_port = 1;
+  string masquerade = 2;
+  string cert_path = 3;
+  string key_path = 4;
+}
+
+// ─── heartbeat ───────────────────────────────────────────────────────────────
+
+message HeartbeatRequest {
+  string node_uuid = 1;
+  int64 config_version = 2;
+  int32 online_peers = 3;
+  int64 bandwidth_up_bps = 4;
+  int64 bandwidth_down_bps = 5;
+  double cpu_percent = 6;
+  int64 timestamp_unix = 7;
+}
+
+message HeartbeatResponse {
+  bool need_full_resync = 1; // agent must Register + overwrite local state
+  int64 server_time_unix = 2;
+}
+
+// ─── command stream ──────────────────────────────────────────────────────────
+
+enum CommandType {
+  COMMAND_TYPE_UNSPECIFIED = 0;
+  COMMAND_TYPE_UPSERT = 1;            // add/update a credential
+  COMMAND_TYPE_REVOKE = 2;            // remove a credential
+  COMMAND_TYPE_ROTATE_CREDENTIAL = 3; // new replaces old, old kept until grace
+  COMMAND_TYPE_APPLY_CONFIG = 4;      // replace inbound params / full snapshot
+  COMMAND_TYPE_LIFECYCLE = 5;         // drain / resume / shutdown
+}
+
+enum LifecycleAction {
+  LIFECYCLE_ACTION_UNSPECIFIED = 0;
+  LIFECYCLE_ACTION_DRAIN = 1;
+  LIFECYCLE_ACTION_RESUME = 2;
+  LIFECYCLE_ACTION_SHUTDOWN = 3;
+}
+
+message Command {
+  int64 command_id = 1; // monotonically increasing per node
+  CommandType type = 2;
+
+  UpsertPayload upsert = 3;
+  RevokePayload revoke = 4;
+  RotatePayload rotate = 5;
+  ConfigSnapshot apply_config = 6;
+  LifecyclePayload lifecycle = 7;
+}
+
+message UpsertPayload {
+  Credential credential = 1;
+}
+
+message RevokePayload {
+  string dp_uuid = 1;
+}
+
+message RotatePayload {
+  string old_dp_uuid = 1;
+  Credential new_credential = 2;
+  int64 grace_until_unix = 3; // old credential kept active until this time
+}
+
+message LifecyclePayload {
+  LifecycleAction action = 1;
+}
+
+message SubscribeRequest {
+  string node_uuid = 1;
+  int64 last_command_id = 2; // resume point; 0 = from the beginning
+}
+
+message AckRequest {
+  string node_uuid = 1;
+  int64 command_id = 2;
+}
+
+message AckResponse {}
+
+// ─── usage ───────────────────────────────────────────────────────────────────
+
+message UsageEntry {
+  string dp_uuid = 1; // never a user id
+  int64 bytes_up = 2;
+  int64 bytes_down = 3;
+  int64 session_minutes = 4;
+}
+
+message UsageReport {
+  string node_uuid = 1;
+  int64 window_start_unix = 2;
+  int64 window_end_unix = 3;
+  repeated UsageEntry entries = 4;
+}
+
+message UsageAck {}