feat: 实现管理后台 A3 注册群消息概览与 metrics 门禁

cmd/nixmsg/serve.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io/fs"
 	"log/slog"
 	"net"
 	"net/http"
@@ -14,7 +15,18 @@ import (
 	"syscall"
 	"time"
 
+	"git.asio.asia/nixevol/NixMsg/internal/admin"
+	"git.asio.asia/nixevol/NixMsg/internal/app/group"
+	"git.asio.asia/nixevol/NixMsg/internal/app/identity"
+	"git.asio.asia/nixevol/NixMsg/internal/app/message"
+	"git.asio.asia/nixevol/NixMsg/internal/app/port"
+	"git.asio.asia/nixevol/NixMsg/internal/app/presence"
+	"git.asio.asia/nixevol/NixMsg/internal/auth"
+	"git.asio.asia/nixevol/NixMsg/internal/broker"
 	"git.asio.asia/nixevol/NixMsg/internal/config"
+	"git.asio.asia/nixevol/NixMsg/internal/httpx"
+	"git.asio.asia/nixevol/NixMsg/internal/listener"
+	"git.asio.asia/nixevol/NixMsg/internal/metrics"
 	"git.asio.asia/nixevol/NixMsg/internal/store"
 	"git.asio.asia/nixevol/NixMsg/web"
 )
@@ -25,20 +37,16 @@ func cmdServe(_ []string) error {
 	if err != nil {
 		return err
 	}
-	// P-WIRE-BEGIN
 	if err := cfg.Validate(); err != nil {
 		return err
 	}
 	setupJSONLogger(cfg.Log)
-	// P-WIRE-END
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
 	return runServe(ctx, cfg)
 }
 
 func runServe(ctx context.Context, cfg config.Config) error {
-	deps := wire()
-
 	if err := os.MkdirAll(cfg.DataDir, 0o755); err != nil {
 		return fmt.Errorf("mkdir data_dir: %w", err)
 	}
@@ -49,7 +57,6 @@ func runServe(ctx context.Context, cfg config.Config) error {
 	}
 	defer func() { _ = db.Close() }()
 
-	// P-WIRE-BEGIN
 	ok, err := store.HasAdminPassword(ctx, db.Write)
 	if err != nil {
 		return err
@@ -57,87 +64,258 @@ func runServe(ctx context.Context, cfg config.Config) error {
 	if !ok {
 		return errors.New("admin password not initialized; run: nixmsg admin init")
 	}
-	// P-WIRE-END
 
-	// 启动恢复入口已挂上（假实现为空操作）；M 线替换 message.Service 后生效。
-	if recoverErr := deps.Messages.RecoverOnStart(ctx); recoverErr != nil {
+	hashPool := auth.NewPool()
+	sessionTokens := auth.NewSessionTokens()
+	apiTokens := auth.NewAPITokens()
+	loginLocks := auth.NewLoginLocks()
+	memConns := message.NewMemoryConns()
+	msgLim := message.LimitsFromFullConfig(cfg)
+
+	login := broker.NewLogin(broker.LoginOptions{
+		DB:       db,
+		Pool:     hashPool,
+		Tokens:   sessionTokens,
+		Locks:    loginLocks,
+		IdleDays: cfg.SessionIdleDays,
+	})
+
+	// 先用空下行建 message，拿到 App 指针；broker 建好后再 WithDownlink 重建并回填 uplink.msg。
+	msgApp := message.New(db, msgLim, hashPool,
+		message.WithLocks(loginLocks),
+		message.WithConnRegistry(memConns),
+	)
+	uplink := &appUplink{
+		msg:   msgApp,
+		conns: memConns,
+		log:   slog.Default(),
+	}
+	sess := broker.NewSession(broker.SessionOptions{
+		Login: login,
+		Inner: uplink,
+		Limits: broker.HelloLimits{
+			MaxBodyBytes:       cfg.Limits.MaxBodyBytes,
+			MaxMetaBytes:       cfg.Limits.MaxMetaBytes,
+			MaxFrameBytes:      cfg.Limits.MaxFrameBytes,
+			MaxTTLSeconds:      int64(cfg.Limits.MaxTTLSeconds),
+			MaxScheduleSeconds: int64(cfg.Limits.MaxScheduleSeconds),
+			AckTimeoutSeconds:  int64(cfg.Limits.AckTimeoutSeconds),
+			ServerVersion:      Version,
+		},
+		Logger: slog.Default(),
+	})
+
+	brk, err := broker.New(broker.Options{
+		Authenticator: login,
+		Uplink:        sess,
+		Logger:        slog.Default(),
+		OnPublishDropped: func(dropCtx context.Context, endpointID string, connID port.ConnID, payload []byte) {
+			if dropErr := msgApp.OnPublishDropped(dropCtx, endpointID, connID, payload); dropErr != nil {
+				slog.Error("on publish dropped", "endpoint", endpointID, "err", dropErr)
+			}
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("broker: %w", err)
+	}
+	defer func() { _ = brk.Close() }()
+	sess.Attach(brk)
+
+	msgApp = message.New(db, msgLim, hashPool,
+		message.WithLocks(loginLocks),
+		message.WithConnRegistry(memConns),
+		message.WithDownlink(brk),
+	)
+	uplink.msg = msgApp
+	uplink.down = brk
+
+	presApp := presence.New(presence.Config{
+		DB:       db,
+		Downlink: brk,
+		Conns:    &presenceConnTable{conns: memConns},
+	})
+	sess.SetPresence(presApp)
+	uplink.presence = presApp
+
+	idApp := identity.New(identity.Config{
+		DB:                 db,
+		Hash:               hashPool,
+		Locks:              loginLocks,
+		Sessions:           sessionTokens,
+		MaxScheduleSeconds: int64(cfg.Limits.MaxScheduleSeconds),
+		Logger:             slog.Default(),
+		ConnControl:        brk,
+	})
+	uplink.identity = idApp
+
+	groupApp := group.New(group.Config{
+		DB:              db,
+		Talk:            idApp,
+		Online:          presApp,
+		Downlink:        brk,
+		MaxGroupMembers: cfg.Limits.MaxGroupMembers,
+	})
+	uplink.groups = groupApp
+
+	if recoverErr := msgApp.RecoverOnStart(ctx); recoverErr != nil {
 		return fmt.Errorf("message recover: %w", recoverErr)
 	}
-	// 其余 deps 供后续 admin / broker / httpx 接线；此处显式引用避免未使用告警。
-	_ = deps.Identity
-	_ = deps.Groups
-	_ = deps.Presence
-	_ = deps.Downlink
-	_ = deps.Conns
-	_ = deps.Uplink
-	_ = deps.HashPool
-	_ = deps.SessionTokens
-	_ = deps.APITokens
-	_ = deps.LoginLocks
 
-	mux := http.NewServeMux()
-	mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
-		w.WriteHeader(http.StatusOK)
-		_, _ = w.Write([]byte("ok"))
+	trustedNets := httpx.ParseCIDRs(cfg.TrustedProxies)
+	adminHandler := admin.New(admin.Deps{
+		DB:             db,
+		Hash:           hashPool,
+		Tokens:         apiTokens,
+		Locks:          loginLocks,
+		Logger:         slog.Default(),
+		TrustedProxies: trustedNets,
+		Identity:       idApp,
+		Groups:         groupApp,
+		Config:         cfg,
+		Version:        Version,
+		KickEndpoint: func(kickCtx context.Context, endpointID string) (bool, error) {
+			if _, found := brk.ConnInfoOf(endpointID); !found {
+				return false, nil
+			}
+			if kickErr := sess.Kick(kickCtx, endpointID); kickErr != nil {
+				return false, kickErr
+			}
+			return true, nil
+		},
 	})
-	// P-WIRE-BEGIN
-	mux.HandleFunc("GET /readyz", func(w http.ResponseWriter, r *http.Request) {
-		if readyErr := db.Ready(r.Context()); readyErr != nil {
-			slog.Error("readyz failed", "err", readyErr)
-			http.Error(w, "not ready", http.StatusServiceUnavailable)
-			return
+
+	metricsReg := metrics.New()
+	buildHandlers := func(proxies *listener.ProxySet) listener.Handlers {
+		return listener.Handlers{
+			MQTT:         brk.WSHandler(proxies),
+			ClientAPI:    idApp.Handler(),
+			AdminAPI:     adminHandler,
+			Metrics:      metricsReg.Handler(),
+			MetricsToken: cfg.Metrics.Token,
+			Static:       staticFileHandler(),
+			Healthz: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+				w.WriteHeader(http.StatusOK)
+				_, _ = w.Write([]byte("ok"))
+			}),
+			Readyz: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if readyErr := db.Ready(r.Context()); readyErr != nil {
+					slog.Error("readyz failed", "err", readyErr)
+					http.Error(w, "not ready", http.StatusServiceUnavailable)
+					return
+				}
+				w.WriteHeader(http.StatusOK)
+				_, _ = w.Write([]byte("ok"))
+			}),
 		}
-		w.WriteHeader(http.StatusOK)
-		_, _ = w.Write([]byte("ok"))
-	})
-	// P-WIRE-END
-	// 确保前端资源被链接进二进制；完整静态托管由后续任务完善。
-	_ = web.Dist()
-
-	ln, err := net.Listen("tcp", cfg.Listen)
-	if err != nil {
-		return fmt.Errorf("listen %s: %w", cfg.Listen, err)
 	}
 
-	if err := writeListenAddr(cfg.DataDir, ln.Addr().String()); err != nil {
-		_ = ln.Close()
+	handlers := buildHandlers(nil)
+	shared := cfg.AdminListen == ""
+	var clientHandler, adminHTTP http.Handler
+	if shared {
+		clientHandler = listener.NewMux(listener.RoleShared, handlers)
+	} else {
+		clientHandler = listener.NewMux(listener.RoleClient, handlers)
+		adminHTTP = listener.NewMux(listener.RoleAdmin, handlers)
+	}
+
+	lnOpts := listener.Options{
+		Listen:         cfg.Listen,
+		AdminListen:    cfg.AdminListen,
+		DataDir:        cfg.DataDir,
+		CertFile:       cfg.TLS.CertFile,
+		KeyFile:        cfg.TLS.KeyFile,
+		AllowPlaintext: cfg.TLS.AllowPlaintext,
+		TrustedProxies: cfg.TrustedProxies,
+		ClientHandler:  clientHandler,
+		AdminHandler:   adminHTTP,
+		OnMQTT: func(conn net.Conn) {
+			_ = brk.AttachTCP(conn)
+		},
+		Logger: slog.Default(),
+	}
+	lnSrv, err := listener.New(lnOpts)
+	if err != nil {
+		return fmt.Errorf("listener: %w", err)
+	}
+	handlers = buildHandlers(lnSrv.Proxies())
+	if shared {
+		lnOpts.ClientHandler = listener.NewMux(listener.RoleShared, handlers)
+	} else {
+		lnOpts.ClientHandler = listener.NewMux(listener.RoleClient, handlers)
+		lnOpts.AdminHandler = listener.NewMux(listener.RoleAdmin, handlers)
+	}
+	lnSrv, err = listener.New(lnOpts)
+	if err != nil {
+		return fmt.Errorf("listener: %w", err)
+	}
+
+	if err := lnSrv.Start(ctx); err != nil {
 		return err
 	}
+	defer func() { _ = lnSrv.Close() }()
 
-	srv := &http.Server{
-		Handler:           mux,
-		ReadHeaderTimeout: 10 * time.Second,
+	if err := writeListenAddr(cfg.DataDir, lnSrv.ListenAddr()); err != nil {
+		return err
+	}
+	if cfg.AdminListen != "" && lnSrv.AdminAddr() != "" {
+		if err := os.WriteFile(filepath.Join(cfg.DataDir, "admin.addr"), []byte(lnSrv.AdminAddr()+"\n"), 0o644); err != nil {
+			return err
+		}
 	}
 
-	errCh := make(chan error, 1)
-	go func() {
-		errCh <- srv.Serve(ln)
-	}()
+	loopCtx, loopCancel := context.WithCancel(ctx)
+	defer loopCancel()
+	go messageLoops(loopCtx, msgApp, memConns)
 
-	select {
-	case <-ctx.Done():
-		// P-WIRE-BEGIN
-		// 先停止接受新连接，再等写队列最多 10 秒，然后断开并退出。
-		shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
-		defer cancel()
-		_ = srv.Shutdown(shutdownCtx)
-		drainCtx, drainCancel := context.WithTimeout(context.Background(), 10*time.Second)
-		defer drainCancel()
-		if drainErr := db.Queue.Drain(drainCtx); drainErr != nil && !errors.Is(drainErr, context.DeadlineExceeded) {
-			slog.Error("write queue drain", "err", drainErr)
-		}
-		// P-WIRE-END
-		serveErr := <-errCh
-		if serveErr != nil && !errors.Is(serveErr, http.ErrServerClosed) {
-			return serveErr
-		}
-		return nil
-	case serveErr := <-errCh:
-		if errors.Is(serveErr, http.ErrServerClosed) {
-			return nil
-		}
-		return serveErr
+	<-ctx.Done()
+	loopCancel()
+	_ = lnSrv.Close()
+	drainCtx, drainCancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer drainCancel()
+	if drainErr := db.Queue.Drain(drainCtx); drainErr != nil && !errors.Is(drainErr, context.DeadlineExceeded) {
+		slog.Error("write queue drain", "err", drainErr)
 	}
+	return nil
+}
+
+func messageLoops(ctx context.Context, msgApp *message.App, conns *message.MemoryConns) {
+	t := time.NewTicker(time.Second)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			nowMs := time.Now().UnixMilli()
+			if _, err := msgApp.DispatchDue(ctx, nowMs, 100); err != nil {
+				slog.Error("dispatch due", "err", err)
+			}
+			for ep, live := range conns.Snapshot() {
+				if err := msgApp.PushPending(ctx, ep, live.ConnID); err != nil {
+					slog.Debug("push pending", "endpoint", ep, "err", err)
+				}
+			}
+			if err := msgApp.CleanupOnce(ctx, nowMs); err != nil {
+				slog.Error("cleanup once", "err", err)
+			}
+		}
+	}
+}
+
+func staticFileHandler() http.Handler {
+	root := web.Dist()
+	for _, prefix := range []string{"dist", "stub"} {
+		sub, err := fs.Sub(root, prefix)
+		if err != nil {
+			continue
+		}
+		if f, err := sub.Open("index.html"); err == nil {
+			_ = f.Close()
+			return http.FileServer(http.FS(sub))
+		}
+	}
+	return http.FileServer(http.FS(root))
 }
 
 func writeListenAddr(dataDir, addr string) error {
@@ -145,7 +323,6 @@ func writeListenAddr(dataDir, addr string) error {
 	return os.WriteFile(path, []byte(addr+"\n"), 0o644)
 }
 
-// P-WIRE-BEGIN
 func setupJSONLogger(cfg config.LogConfig) {
 	level := slog.LevelInfo
 	switch strings.ToLower(strings.TrimSpace(cfg.Level)) {
@@ -159,5 +336,3 @@ func setupJSONLogger(cfg config.LogConfig) {
 	h := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: level})
 	slog.SetDefault(slog.New(h))
 }
-
-// P-WIRE-END

cmd/nixmsg/uplink.go

@@ -0,0 +1,344 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"log/slog"
+
+	"git.asio.asia/nixevol/NixMsg/internal/app/group"
+	"git.asio.asia/nixevol/NixMsg/internal/app/identity"
+	"git.asio.asia/nixevol/NixMsg/internal/app/message"
+	"git.asio.asia/nixevol/NixMsg/internal/app/port"
+	"git.asio.asia/nixevol/NixMsg/internal/app/presence"
+	"git.asio.asia/nixevol/NixMsg/internal/protocol"
+)
+
+// appUplink 把 broker 生命周期接到消息连接表，并把已握手上行帧分发到各业务服务。
+type appUplink struct {
+	msg      *message.App
+	identity *identity.App
+	presence *presence.App
+	groups   *group.App
+	conns    *message.MemoryConns
+	down     port.Downlink
+	log      *slog.Logger
+}
+
+func (u *appUplink) OnSessionEstablished(ctx context.Context, conn port.ConnInfo) error {
+	u.conns.Set(conn.EndpointID, message.LiveConn{
+		ConnID:        conn.ConnID,
+		MaxPacketSize: conn.MaxPacketSize,
+	})
+	return nil
+}
+
+func (u *appUplink) OnHandshakeComplete(ctx context.Context, hs port.HandshakeInfo) error {
+	live := message.LiveConn{
+		ConnID:          hs.ConnID,
+		MaxReceiveBytes: hs.MaxReceiveBytes,
+		MaxPacketSize:   hs.MaxPacketSize,
+	}
+	u.conns.Set(hs.EndpointID, live)
+	if err := u.msg.OnHandshakeComplete(ctx, hs.EndpointID, live); err != nil {
+		u.log.Error("message handshake", "endpoint", hs.EndpointID, "err", err)
+		return err
+	}
+	return nil
+}
+
+func (u *appUplink) OnDisconnect(ctx context.Context, conn port.ConnInfo, reason port.DisconnectReason) {
+	if u.presence != nil {
+		u.presence.ClearWatch(conn.ConnID)
+	}
+	live, ok := u.conns.Current(conn.EndpointID)
+	isCurrent := ok && live.ConnID == conn.ConnID
+	if err := u.msg.OnDisconnect(ctx, conn.EndpointID, conn.ConnID, isCurrent); err != nil {
+		u.log.Error("message disconnect", "endpoint", conn.EndpointID, "err", err)
+	}
+	u.conns.Clear(conn.EndpointID, conn.ConnID)
+}
+
+func (u *appUplink) HandleUplink(ctx context.Context, conn port.ConnInfo, payload []byte) error {
+	frame, err := protocol.Decode(payload)
+	if err != nil {
+		u.replyErr(ctx, conn, peekRID(payload), protocol.CodeBadRequest, err.Error())
+		return nil
+	}
+
+	rid, data, callErr := u.dispatch(ctx, conn, frame)
+	if callErr != nil {
+		u.replyFromErr(ctx, conn, rid, callErr)
+		return nil
+	}
+	u.replyOK(ctx, conn, rid, data)
+	return nil
+}
+
+func (u *appUplink) dispatch(ctx context.Context, conn port.ConnInfo, frame any) (rid string, data any, err error) {
+	switch f := frame.(type) {
+	case *protocol.Send:
+		rid = f.RID
+		res, e := u.msg.Submit(ctx, conn.EndpointID, conn, f)
+		if e != nil {
+			return rid, nil, e
+		}
+		return rid, protocol.SendData{ID: res.ID, SendAtMs: res.SendAtMs, State: res.State}, nil
+
+	case *protocol.Ack:
+		rid = f.RID
+		res, e := u.msg.Ack(ctx, conn.EndpointID, f)
+		if e != nil {
+			return rid, nil, e
+		}
+		return rid, map[string]any{"result": res.Result}, nil
+
+	case *protocol.ReceiptAck:
+		rid = f.RID
+		return rid, nil, u.msg.ReceiptAck(ctx, conn.EndpointID, f)
+
+	case *protocol.Recall:
+		rid = f.RID
+		res, e := u.msg.Recall(ctx, conn.EndpointID, f)
+		if e != nil {
+			return rid, nil, e
+		}
+		return rid, res, nil
+
+	case *protocol.Status:
+		rid = f.RID
+		res, e := u.msg.Status(ctx, conn.EndpointID, f)
+		return rid, res, e
+
+	case *protocol.Unlock:
+		rid = f.RID
+		if e := f.Validate(); e != nil {
+			return rid, nil, e
+		}
+		return rid, nil, u.identity.UnlockTalk(ctx, conn.EndpointID, f.EndpointID, f.TalkPassword, conn.RemoteIP)
+
+	case *protocol.SelfGet:
+		rid = f.RID
+		info, e := u.identity.SelfGet(ctx, conn.EndpointID)
+		return rid, info, e
+
+	case *protocol.SelfUpdate:
+		rid = f.RID
+		return rid, nil, u.identity.SelfUpdate(ctx, conn.EndpointID, f)
+
+	case *protocol.SelfTalkPassword:
+		rid = f.RID
+		if e := f.Validate(); e != nil {
+			return rid, nil, e
+		}
+		return rid, nil, u.identity.SelfSetTalkPassword(ctx, conn.EndpointID, f.TalkPassword)
+
+	case *protocol.SelfLoginPassword:
+		rid = f.RID
+		if e := f.Validate(); e != nil {
+			return rid, nil, e
+		}
+		tok, e := u.identity.SelfChangeLoginPassword(ctx, conn.EndpointID, f.OldPassword, f.NewPassword, conn.RemoteIP)
+		if e != nil {
+			return rid, nil, e
+		}
+		return rid, map[string]any{"session_token": tok}, nil
+
+	case *protocol.PresenceGet:
+		rid = f.RID
+		if e := f.Validate(); e != nil {
+			return rid, nil, e
+		}
+		items, e := u.presence.Get(ctx, f.IDs)
+		if e != nil {
+			return rid, nil, e
+		}
+		return rid, presence.EncodeGetData(items), nil
+
+	case *protocol.DirectoryList:
+		rid = f.RID
+		items, next, e := u.presence.Directory(ctx, f)
+		if e != nil {
+			return rid, nil, e
+		}
+		return rid, map[string]any{"items": items, "next_cursor": next}, nil
+
+	case *protocol.PresenceWatch:
+		rid = f.RID
+		return rid, nil, u.presence.Watch(ctx, conn.ConnID, conn.EndpointID, f)
+
+	case *protocol.GroupCreate:
+		rid = f.RID
+		res, e := u.groups.Create(ctx, conn.EndpointID, f)
+		return rid, res, e
+
+	case *protocol.GroupAdd:
+		rid = f.RID
+		res, e := u.groups.Add(ctx, conn.EndpointID, f)
+		return rid, res, e
+
+	case *protocol.GroupRemove:
+		rid = f.RID
+		return rid, nil, u.groups.Remove(ctx, conn.EndpointID, f)
+
+	case *protocol.GroupLeave:
+		rid = f.RID
+		return rid, nil, u.groups.Leave(ctx, conn.EndpointID, f)
+
+	case *protocol.GroupTransfer:
+		rid = f.RID
+		return rid, nil, u.groups.Transfer(ctx, conn.EndpointID, f)
+
+	case *protocol.GroupRename:
+		rid = f.RID
+		return rid, nil, u.groups.Rename(ctx, conn.EndpointID, f)
+
+	case *protocol.GroupDissolve:
+		rid = f.RID
+		return rid, nil, u.groups.Dissolve(ctx, conn.EndpointID, f)
+
+	case *protocol.GroupList:
+		rid = f.RID
+		items, next, e := u.groups.List(ctx, conn.EndpointID, f)
+		if e != nil {
+			return rid, nil, e
+		}
+		return rid, map[string]any{"items": items, "next_cursor": next}, nil
+
+	case *protocol.GroupGet:
+		rid = f.RID
+		res, e := u.groups.Get(ctx, conn.EndpointID, f)
+		return rid, res, e
+
+	case *protocol.Hello, *protocol.SelfLogout:
+		// Session 已处理；不应落到此处。
+		rid = frameRID(frame)
+		return rid, nil, &protocol.Error{Code: protocol.CodeBadRequest, Message: "unexpected frame"}
+
+	default:
+		rid = frameRID(frame)
+		return rid, nil, &protocol.Error{Code: protocol.CodeBadRequest, Message: "unsupported uplink type"}
+	}
+}
+
+func (u *appUplink) replyOK(ctx context.Context, conn port.ConnInfo, rid string, data any) {
+	if rid == "" {
+		rid = "0"
+	}
+	var raw json.RawMessage
+	if data != nil {
+		b, err := protocol.Marshal(data)
+		if err != nil {
+			u.replyErr(ctx, conn, rid, protocol.CodeBusy, "marshal resp data")
+			return
+		}
+		raw = b
+	}
+	resp := protocol.Resp{V: protocol.Version, Type: protocol.TypeResp, RID: rid, OK: true, Data: raw}
+	u.publishResp(ctx, conn, resp)
+}
+
+func (u *appUplink) replyFromErr(ctx context.Context, conn port.ConnInfo, rid string, err error) {
+	if rid == "" {
+		rid = "0"
+	}
+	var pe *protocol.Error
+	if errors.As(err, &pe) && pe != nil {
+		u.replyErr(ctx, conn, rid, pe.Code, pe.Message)
+		return
+	}
+	u.log.Error("uplink handler", "endpoint", conn.EndpointID, "err", err)
+	u.replyErr(ctx, conn, rid, protocol.CodeBusy, "internal error")
+}
+
+func (u *appUplink) replyErr(ctx context.Context, conn port.ConnInfo, rid, code, message string) {
+	if rid == "" {
+		rid = "0"
+	}
+	resp := protocol.Resp{
+		V:     protocol.Version,
+		Type:  protocol.TypeResp,
+		RID:   rid,
+		OK:    false,
+		Error: &protocol.ErrorBody{Code: code, Message: message},
+	}
+	u.publishResp(ctx, conn, resp)
+}
+
+func (u *appUplink) publishResp(ctx context.Context, conn port.ConnInfo, resp protocol.Resp) {
+	if u.down == nil {
+		return
+	}
+	b, err := protocol.Marshal(resp)
+	if err != nil {
+		u.log.Error("marshal resp", "err", err)
+		return
+	}
+	if live, ok := u.conns.Current(conn.EndpointID); ok && live.ConnID == conn.ConnID {
+		limit := respPayloadLimit(live.MaxPacketSize, live.MaxReceiveBytes)
+		if limit > 0 && len(b) > limit {
+			tooLarge := protocol.Resp{
+				V:     protocol.Version,
+				Type:  protocol.TypeResp,
+				RID:   resp.RID,
+				OK:    false,
+				Error: &protocol.ErrorBody{Code: protocol.CodeResponseTooLarge, Message: "response too large"},
+			}
+			b, err = protocol.Marshal(tooLarge)
+			if err != nil {
+				return
+			}
+		}
+	}
+	if pubErr := u.down.PublishDown(ctx, conn.EndpointID, conn.ConnID, b, port.PublishOpts{QoS: 1}); pubErr != nil {
+		u.log.Error("publish resp", "endpoint", conn.EndpointID, "err", pubErr)
+	}
+}
+
+func respPayloadLimit(maxPacketSize uint32, maxRecvBytes int) int {
+	limit := 0
+	if maxRecvBytes > 0 {
+		limit = maxRecvBytes
+	}
+	if maxPacketSize > 0 {
+		n := int(maxPacketSize)
+		if limit == 0 || n < limit {
+			limit = n
+		}
+	}
+	return limit
+}
+
+func peekRID(payload []byte) string {
+	var peek struct {
+		RID string `json:"rid"`
+	}
+	_ = json.Unmarshal(payload, &peek)
+	return peek.RID
+}
+
+func frameRID(frame any) string {
+	b, err := protocol.Marshal(frame)
+	if err != nil {
+		return ""
+	}
+	return peekRID(b)
+}
+
+// presenceConnTable 把消息连接表暴露给 presence.ConnTable。
+type presenceConnTable struct {
+	conns *message.MemoryConns
+}
+
+func (p *presenceConnTable) IsOnline(endpointID string) bool {
+	_, ok := p.conns.Current(endpointID)
+	return ok
+}
+
+func (p *presenceConnTable) CurrentConn(endpointID string) (port.ConnID, bool) {
+	live, ok := p.conns.Current(endpointID)
+	if !ok {
+		return "", false
+	}
+	return live.ConnID, true
+}

cmd/nixmsg/uplink_integration_test.go

@@ -0,0 +1,459 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"git.asio.asia/nixevol/NixMsg/internal/config"
+	"git.asio.asia/nixevol/NixMsg/internal/protocol"
+	"git.asio.asia/nixevol/NixMsg/test/harness"
+	"github.com/mochi-mqtt/server/v2/packets"
+)
+
+// TestUplinkDMOfflineGroupRecall 用真实进程验证上行分发：单聊、离线保留、群发不含发送者、延迟内撤回。
+func TestUplinkDMOfflineGroupRecall(t *testing.T) {
+	dataDir := t.TempDir()
+	cfgPath := writeTestConfig(t, dataDir)
+	initAdminForTest(t, dataDir)
+	enableRegistration(t, dataDir, "uplink-code")
+
+	cfg, err := config.Load(cfgPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if vErr := cfg.Validate(); vErr != nil {
+		t.Fatal(vErr)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	errCh := make(chan error, 1)
+	go func() { errCh <- runServe(ctx, cfg) }()
+	defer func() {
+		cancel()
+		select {
+		case err := <-errCh:
+			if err != nil {
+				t.Errorf("serve exit: %v", err)
+			}
+		case <-time.After(15 * time.Second):
+			t.Error("serve did not stop")
+		}
+	}()
+
+	addr := waitListenAddr(t, dataDir, 15*time.Second)
+	base := "http://" + addr
+
+	registerEP(t, base, "alice", "password12", "Alice")
+	registerEP(t, base, "bob", "password12", "Bob")
+	registerEP(t, base, "carol", "password12", "Carol")
+
+	alice := mqttSessionLogin(t, base, "alice", "password12")
+	defer alice.Close()
+	bob := mqttSessionLogin(t, base, "bob", "password12")
+	defer bob.Close()
+
+	// 1) 两端在线单聊：bob 收到 msg
+	delay0 := int64(0)
+	sendResp := alice.Request(t, map[string]any{
+		"v": 1, "type": "send", "rid": "s1", "id": "dm-1",
+		"to":       map[string]any{"kind": "endpoint", "id": "bob"},
+		"body":     map[string]any{"enc": "utf8", "data": "hello-bob"},
+		"delay_ms": delay0,
+	})
+	if !sendResp.OK {
+		t.Fatalf("send dm: %+v", sendResp)
+	}
+	msg := bob.WaitType(t, "msg", 8*time.Second)
+	if msg["id"] != "dm-1" || msg["from"] != "alice" {
+		t.Fatalf("bob msg=%v", msg)
+	}
+	body, _ := msg["body"].(map[string]any)
+	if body["data"] != "hello-bob" {
+		t.Fatalf("body=%v", body)
+	}
+	// 确认以免窗口占满
+	bob.Request(t, map[string]any{"v": 1, "type": "ack", "rid": "a1", "from": "alice", "id": "dm-1"})
+
+	// 2) 离线保留：carol 离线时 alice 发送，carol 稍后上线能收到
+	keep := true
+	ttl := int64(86400)
+	sendOff := alice.Request(t, map[string]any{
+		"v": 1, "type": "send", "rid": "s2", "id": "off-1",
+		"to":       map[string]any{"kind": "endpoint", "id": "carol"},
+		"body":     map[string]any{"enc": "utf8", "data": "for-carol"},
+		"delay_ms": delay0,
+		"offline":  map[string]any{"keep": keep, "ttl_seconds": ttl},
+	})
+	if !sendOff.OK {
+		t.Fatalf("send offline: %+v", sendOff)
+	}
+	carol := mqttSessionLogin(t, base, "carol", "password12")
+	defer carol.Close()
+	offMsg := carol.WaitType(t, "msg", 8*time.Second)
+	if offMsg["id"] != "off-1" {
+		t.Fatalf("carol offline msg=%v", offMsg)
+	}
+	carol.Request(t, map[string]any{"v": 1, "type": "ack", "rid": "a2", "from": "alice", "id": "off-1"})
+
+	// 3) 建群后群发：bob、carol 收到，alice 自己不收到
+	createResp := alice.Request(t, map[string]any{
+		"v": 1, "type": "group.create", "rid": "g1", "id": "g_uplink1", "name": "U",
+		"members": []map[string]any{{"id": "bob"}, {"id": "carol"}},
+	})
+	if !createResp.OK {
+		t.Fatalf("group.create: %+v", createResp)
+	}
+	// 排空建群带来的 group_event，避免干扰后续断言
+	drainEvents(t, bob, 500*time.Millisecond)
+	drainEvents(t, carol, 500*time.Millisecond)
+	drainEvents(t, alice, 300*time.Millisecond)
+
+	grpSend := alice.Request(t, map[string]any{
+		"v": 1, "type": "send", "rid": "s3", "id": "grp-1",
+		"to":       map[string]any{"kind": "group", "id": "g_uplink1"},
+		"body":     map[string]any{"enc": "utf8", "data": "hi-group"},
+		"delay_ms": delay0,
+	})
+	if !grpSend.OK {
+		t.Fatalf("group send: %+v", grpSend)
+	}
+	bobGrp := bob.WaitType(t, "msg", 8*time.Second)
+	carolGrp := carol.WaitType(t, "msg", 8*time.Second)
+	if bobGrp["id"] != "grp-1" || carolGrp["id"] != "grp-1" {
+		t.Fatalf("bob=%v carol=%v", bobGrp, carolGrp)
+	}
+	if got := alice.TryType("msg", 800*time.Millisecond); got != nil {
+		t.Fatalf("sender must not receive own group msg: %v", got)
+	}
+	bob.Request(t, map[string]any{"v": 1, "type": "ack", "rid": "a3", "from": "alice", "id": "grp-1"})
+	carol.Request(t, map[string]any{"v": 1, "type": "ack", "rid": "a4", "from": "alice", "id": "grp-1"})
+
+	// 4) 延迟内撤回：对方无 msg / revoked 回调
+	delayMs := int64(60_000)
+	sched := alice.Request(t, map[string]any{
+		"v": 1, "type": "send", "rid": "s4", "id": "rec-1",
+		"to":       map[string]any{"kind": "endpoint", "id": "bob"},
+		"body":     map[string]any{"enc": "utf8", "data": "will-recall"},
+		"delay_ms": delayMs,
+	})
+	if !sched.OK {
+		t.Fatalf("scheduled send: %+v", sched)
+	}
+	data, _ := sched.Data.(map[string]any)
+	if data["state"] != "scheduled" {
+		t.Fatalf("want scheduled got %v", data)
+	}
+	rec := alice.Request(t, map[string]any{"v": 1, "type": "recall", "rid": "r1", "id": "rec-1"})
+	if !rec.OK {
+		t.Fatalf("recall: %+v", rec)
+	}
+	if got := bob.TryType("msg", 1*time.Second); got != nil {
+		t.Fatalf("bob should not get recalled msg: %v", got)
+	}
+	if got := bob.TryType("revoked", 500*time.Millisecond); got != nil {
+		t.Fatalf("bob should not get revoked for undelivered: %v", got)
+	}
+}
+
+func registerEP(t *testing.T, base, id, password, name string) {
+	t.Helper()
+	body := `{"registration_code":"uplink-code","id":"` + id + `","login_password":"` + password + `","name":"` + name + `"}`
+	resp, err := http.Post(base+"/api/client/register", "application/json", strings.NewReader(body))
+	if err != nil {
+		t.Fatal(err)
+	}
+	raw, _ := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("register %s: %d %s", id, resp.StatusCode, raw)
+	}
+}
+
+type mqttSess struct {
+	t          *testing.T
+	mc         harness.MQTTClient
+	endpointID string
+	pktID      uint16
+	mu         sync.Mutex
+	inbox      []map[string]any
+	closed     bool
+	done       chan struct{}
+}
+
+type appResp struct {
+	OK    bool
+	Error map[string]any
+	Data  any
+	Raw   map[string]any
+}
+
+func mqttSessionLogin(t *testing.T, httpBase, endpointID, password string) *mqttSess {
+	t.Helper()
+	mc, err := harness.DialMQTTWebSocket(httpBase, 10*time.Second)
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	s := &mqttSess{t: t, mc: mc, endpointID: endpointID, pktID: 10, done: make(chan struct{})}
+	s.connectSubscribeHello(password)
+	go s.readLoop()
+	return s
+}
+
+func (s *mqttSess) Close() {
+	s.mu.Lock()
+	if s.closed {
+		s.mu.Unlock()
+		return
+	}
+	s.closed = true
+	s.mu.Unlock()
+	_ = s.mc.Close()
+	select {
+	case <-s.done:
+	case <-time.After(3 * time.Second):
+	}
+}
+
+func (s *mqttSess) nextPkt() uint16 {
+	s.pktID++
+	if s.pktID == 0 {
+		s.pktID = 1
+	}
+	return s.pktID
+}
+
+func (s *mqttSess) connectSubscribeHello(password string) {
+	t := s.t
+	pk := packets.Packet{
+		FixedHeader:     packets.FixedHeader{Type: packets.Connect},
+		ProtocolVersion: 5,
+		Connect: packets.ConnectParams{
+			ProtocolName:     []byte("MQTT"),
+			Clean:            true,
+			ClientIdentifier: s.endpointID,
+			Keepalive:        30,
+			UsernameFlag:     true,
+			Username:         []byte(s.endpointID),
+			PasswordFlag:     true,
+			Password:         []byte(password),
+		},
+	}
+	var buf bytes.Buffer
+	if err := pk.ConnectEncode(&buf); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.mc.Send(buf.Bytes()); err != nil {
+		t.Fatal(err)
+	}
+	ack, err := s.mc.Recv()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(ack) < 4 || ack[0]>>4 != packets.Connack || ack[3] != 0 {
+		t.Fatalf("connack %x", ack)
+	}
+
+	sub := packets.Packet{
+		FixedHeader:     packets.FixedHeader{Type: packets.Subscribe, Qos: 1},
+		ProtocolVersion: 5,
+		PacketID:        s.nextPkt(),
+		Filters: packets.Subscriptions{
+			{Filter: "nix/c/" + s.endpointID + "/down", Qos: 1},
+		},
+	}
+	buf.Reset()
+	if err := sub.SubscribeEncode(&buf); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.mc.Send(buf.Bytes()); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.mc.Recv(); err != nil {
+		t.Fatal(err)
+	}
+
+	hello, _ := protocol.Marshal(protocol.Hello{V: protocol.Version, Type: protocol.TypeHello, RID: "h0"})
+	s.publishRaw(hello)
+	deadline := time.Now().Add(10 * time.Second)
+	for time.Now().Before(deadline) {
+		raw, err := s.mc.Recv()
+		if err != nil {
+			t.Fatal(err)
+		}
+		m := s.handlePacket(raw)
+		if m == nil {
+			continue
+		}
+		if m["type"] == "resp" && m["ok"] == true {
+			return
+		}
+		if m["type"] == "resp" {
+			t.Fatalf("hello failed: %v", m)
+		}
+		s.push(m)
+	}
+	t.Fatal("hello timeout")
+}
+
+func (s *mqttSess) publishRaw(payload []byte) {
+	t := s.t
+	pub := packets.Packet{
+		FixedHeader:     packets.FixedHeader{Type: packets.Publish, Qos: 1},
+		ProtocolVersion: 5,
+		TopicName:       "nix/c/" + s.endpointID + "/up",
+		PacketID:        s.nextPkt(),
+		Payload:         payload,
+	}
+	var buf bytes.Buffer
+	if err := pub.PublishEncode(&buf); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.mc.Send(buf.Bytes()); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func (s *mqttSess) readLoop() {
+	defer close(s.done)
+	for {
+		raw, err := s.mc.Recv()
+		if err != nil {
+			return
+		}
+		m := s.handlePacket(raw)
+		if m != nil {
+			s.push(m)
+		}
+	}
+}
+
+func (s *mqttSess) handlePacket(raw []byte) map[string]any {
+	if len(raw) < 2 {
+		return nil
+	}
+	typ := raw[0] >> 4
+	qos := (raw[0] >> 1) & 0x3
+	switch typ {
+	case packets.Puback, packets.Pingresp, packets.Suback:
+		return nil
+	case packets.Publish:
+		payload, err := decodePublishPayload(raw)
+		if err != nil {
+			return nil
+		}
+		if qos == 1 {
+			// 回 PUBACK
+			rem, n, _ := decodeRemainingLength(raw[1:])
+			body := raw[1+n:]
+			pk := packets.Packet{ProtocolVersion: 5, FixedHeader: packets.FixedHeader{Type: packets.Publish, Remaining: rem, Qos: qos}}
+			if decErr := pk.PublishDecode(body); decErr == nil && pk.PacketID != 0 {
+				ack := packets.Packet{
+					FixedHeader:     packets.FixedHeader{Type: packets.Puback},
+					ProtocolVersion: 5,
+					PacketID:        pk.PacketID,
+				}
+				var buf bytes.Buffer
+				if encErr := ack.PubackEncode(&buf); encErr == nil {
+					_ = s.mc.Send(buf.Bytes())
+				}
+			}
+		}
+		var m map[string]any
+		if json.Unmarshal(payload, &m) != nil {
+			return nil
+		}
+		return m
+	default:
+		return nil
+	}
+}
+
+func (s *mqttSess) push(m map[string]any) {
+	s.mu.Lock()
+	s.inbox = append(s.inbox, m)
+	s.mu.Unlock()
+}
+
+func (s *mqttSess) Request(t *testing.T, frame map[string]any) appResp {
+	t.Helper()
+	rid, _ := frame["rid"].(string)
+	payload, err := protocol.Marshal(frame)
+	if err != nil {
+		t.Fatal(err)
+	}
+	s.publishRaw(payload)
+	deadline := time.Now().Add(10 * time.Second)
+	for time.Now().Before(deadline) {
+		m := s.takeMatching(func(x map[string]any) bool {
+			return x["type"] == "resp" && x["rid"] == rid
+		})
+		if m != nil {
+			r := appResp{OK: m["ok"] == true, Raw: m, Data: m["data"]}
+			if e, ok := m["error"].(map[string]any); ok {
+				r.Error = e
+			}
+			return r
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatalf("timeout waiting resp rid=%s", rid)
+	return appResp{}
+}
+
+func (s *mqttSess) WaitType(t *testing.T, typ string, timeout time.Duration) map[string]any {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		m := s.takeMatching(func(x map[string]any) bool { return x["type"] == typ })
+		if m != nil {
+			return m
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatalf("timeout waiting type=%s", typ)
+	return nil
+}
+
+func (s *mqttSess) TryType(typ string, timeout time.Duration) map[string]any {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		m := s.takeMatching(func(x map[string]any) bool { return x["type"] == typ })
+		if m != nil {
+			return m
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	return nil
+}
+
+func (s *mqttSess) takeMatching(pred func(map[string]any) bool) map[string]any {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for i, m := range s.inbox {
+		if pred(m) {
+			s.inbox = append(s.inbox[:i], s.inbox[i+1:]...)
+			return m
+		}
+	}
+	return nil
+}
+
+func drainEvents(t *testing.T, s *mqttSess, d time.Duration) {
+	t.Helper()
+	deadline := time.Now().Add(d)
+	for time.Now().Before(deadline) {
+		_ = s.takeMatching(func(x map[string]any) bool {
+			typ, _ := x["type"].(string)
+			return typ == "group_event" || typ == "presence"
+		})
+		time.Sleep(20 * time.Millisecond)
+	}
+}

cmd/nixmsg/wire.go

@@ -9,7 +9,7 @@ import (
 	"git.asio.asia/nixevol/NixMsg/internal/auth"
 )
 
-// appDeps 是 serve 组装出的模块依赖。各线在后续任务中替换假实现为真实实现。
+// appDeps 是组装出的模块依赖（测试与骨架仍可调用 wire）。
 type appDeps struct {
 	HashPool      auth.HashPool
 	SessionTokens auth.SessionTokens
@@ -26,13 +26,13 @@ type appDeps struct {
 	Conns    port.ConnControl
 }
 
-// wire 组装各业务模块的骨架依赖（T0.4：假实现；后续各线替换）。
+// wire 返回真实 auth 实现与空业务桩（无 DB 时的轻量装配；正式 serve 走 assembleRuntime）。
 func wire() appDeps {
 	return appDeps{
-		HashPool:      auth.NewStubHashPool(),
-		SessionTokens: auth.NewStubSessionTokens(),
-		APITokens:     auth.NewStubAPITokens(),
-		LoginLocks:    auth.NewStubLoginLocks(),
+		HashPool:      auth.NewPool(),
+		SessionTokens: auth.NewSessionTokens(),
+		APITokens:     auth.NewAPITokens(),
+		LoginLocks:    auth.NewLoginLocks(),
 
 		Messages: message.NewStub(),
 		Identity: identity.NewStub(),

cmd/nixmsg/wire_integration_test.go

@@ -0,0 +1,313 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/cookiejar"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"git.asio.asia/nixevol/NixMsg/internal/config"
+	"git.asio.asia/nixevol/NixMsg/internal/protocol"
+	"git.asio.asia/nixevol/NixMsg/internal/store"
+	"git.asio.asia/nixevol/NixMsg/test/harness"
+	"github.com/mochi-mqtt/server/v2/packets"
+)
+
+func TestWireAdminLoginRegisterMQTTHandshake(t *testing.T) {
+	dataDir := t.TempDir()
+	cfgPath := writeTestConfig(t, dataDir)
+	initAdminForTest(t, dataDir)
+	enableRegistration(t, dataDir, "wire-code-99")
+
+	cfg, err := config.Load(cfgPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if vErr := cfg.Validate(); vErr != nil {
+		t.Fatal(vErr)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	errCh := make(chan error, 1)
+	go func() { errCh <- runServe(ctx, cfg) }()
+
+	addr := waitListenAddr(t, dataDir, 15*time.Second)
+	base := "http://" + addr
+
+	// 1) admin init 后真实进程可管理登录
+	jar, err := cookiejar.New(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	client := &http.Client{Jar: jar, Timeout: 10 * time.Second}
+	loginBody, _ := json.Marshal(map[string]string{
+		"username": "admin",
+		"password": "test-admin-password-xx",
+	})
+	resp, err := client.Post(base+"/api/admin/login", "application/json", bytes.NewReader(loginBody))
+	if err != nil {
+		t.Fatalf("admin login: %v", err)
+	}
+	body, _ := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("admin login status=%d body=%s", resp.StatusCode, body)
+	}
+	var loginEnv struct {
+		OK bool `json:"ok"`
+	}
+	if uErr := json.Unmarshal(body, &loginEnv); uErr != nil || !loginEnv.OK {
+		t.Fatalf("admin login resp=%s", body)
+	}
+
+	// 2) 已写入注册开关与安全码后可注册
+	regBody := `{"registration_code":"wire-code-99","id":"ep_wire1","login_password":"password12","name":"接线端"}`
+	regResp, err := http.Post(base+"/api/client/register", "application/json", strings.NewReader(regBody))
+	if err != nil {
+		t.Fatalf("register: %v", err)
+	}
+	regBytes, _ := io.ReadAll(regResp.Body)
+	_ = regResp.Body.Close()
+	if regResp.StatusCode != http.StatusOK {
+		t.Fatalf("register status=%d body=%s", regResp.StatusCode, regBytes)
+	}
+	var regEnv struct {
+		OK   bool `json:"ok"`
+		Data struct {
+			ID string `json:"id"`
+		} `json:"data"`
+	}
+	if uErr := json.Unmarshal(regBytes, &regEnv); uErr != nil || !regEnv.OK || regEnv.Data.ID != "ep_wire1" {
+		t.Fatalf("register resp=%s", regBytes)
+	}
+
+	// 3) 注册出的端用密码完成 MQTT 握手并拿到 session_token
+	tok := mqttPasswordHandshake(t, base, "ep_wire1", "password12")
+	if tok == "" || !strings.HasPrefix(tok, protocol.SessionTokenPrefix) {
+		t.Fatalf("session_token=%q", tok)
+	}
+
+	cancel()
+	select {
+	case err := <-errCh:
+		if err != nil {
+			t.Fatalf("serve exit: %v", err)
+		}
+	case <-time.After(15 * time.Second):
+		t.Fatal("serve did not stop")
+	}
+}
+
+func mqttPasswordHandshake(t *testing.T, httpBase, endpointID, password string) string {
+	t.Helper()
+	mc, err := harness.DialMQTTWebSocket(httpBase, 10*time.Second)
+	if err != nil {
+		t.Fatalf("dial mqtt ws: %v", err)
+	}
+	defer func() { _ = mc.Close() }()
+
+	pk := packets.Packet{
+		FixedHeader:     packets.FixedHeader{Type: packets.Connect},
+		ProtocolVersion: 5,
+		Connect: packets.ConnectParams{
+			ProtocolName:     []byte("MQTT"),
+			Clean:            true,
+			ClientIdentifier: endpointID,
+			Keepalive:        30,
+			UsernameFlag:     true,
+			Username:         []byte(endpointID),
+			PasswordFlag:     true,
+			Password:         []byte(password),
+		},
+	}
+	var buf bytes.Buffer
+	if encErr := pk.ConnectEncode(&buf); encErr != nil {
+		t.Fatal(encErr)
+	}
+	if sendErr := mc.Send(buf.Bytes()); sendErr != nil {
+		t.Fatal(sendErr)
+	}
+	ack, err := mc.Recv()
+	if err != nil {
+		t.Fatalf("connack: %v", err)
+	}
+	if len(ack) < 2 || ack[0]>>4 != packets.Connack {
+		t.Fatalf("want CONNACK, got %x", ack)
+	}
+	// MQTT5 CONNACK: remaining length, flags, reason code
+	reason := byte(0)
+	if len(ack) >= 4 {
+		reason = ack[3]
+	}
+	if reason != 0 {
+		t.Fatalf("connack reason=%d raw=%x", reason, ack)
+	}
+
+	sub := packets.Packet{
+		FixedHeader:     packets.FixedHeader{Type: packets.Subscribe, Qos: 1},
+		ProtocolVersion: 5,
+		PacketID:        1,
+		Filters: packets.Subscriptions{
+			{Filter: "nix/c/" + endpointID + "/down", Qos: 1},
+		},
+	}
+	buf.Reset()
+	if err := sub.SubscribeEncode(&buf); err != nil {
+		t.Fatal(err)
+	}
+	if err := mc.Send(buf.Bytes()); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := mc.Recv(); err != nil { // SUBACK
+		t.Fatalf("suback: %v", err)
+	}
+
+	hello, _ := protocol.Marshal(protocol.Hello{
+		V: protocol.Version, Type: protocol.TypeHello, RID: "h1",
+	})
+	pub := packets.Packet{
+		FixedHeader:     packets.FixedHeader{Type: packets.Publish, Qos: 1},
+		ProtocolVersion: 5,
+		TopicName:       "nix/c/" + endpointID + "/up",
+		PacketID:        2,
+		Payload:         hello,
+	}
+	buf.Reset()
+	if err := pub.PublishEncode(&buf); err != nil {
+		t.Fatal(err)
+	}
+	if err := mc.Send(buf.Bytes()); err != nil {
+		t.Fatal(err)
+	}
+
+	deadline := time.Now().Add(10 * time.Second)
+	for time.Now().Before(deadline) {
+		raw, err := mc.Recv()
+		if err != nil {
+			t.Fatalf("recv down: %v", err)
+		}
+		if len(raw) < 2 {
+			continue
+		}
+		typ := raw[0] >> 4
+		if typ == packets.Puback || typ == packets.Pingresp {
+			continue
+		}
+		if typ != packets.Publish {
+			continue
+		}
+		payload, err := decodePublishPayload(raw)
+		if err != nil {
+			t.Fatalf("publish decode: %v raw=%x", err, raw)
+		}
+		var m map[string]any
+		if err := json.Unmarshal(payload, &m); err != nil {
+			t.Fatalf("json: %v payload=%s", err, payload)
+		}
+		if m["type"] == "resp" {
+			if tok := extractSessionToken(m); tok != "" {
+				return tok
+			}
+			t.Fatalf("hello resp without token: %v", m)
+		}
+	}
+	t.Fatal("timeout waiting hello resp")
+	return ""
+}
+
+func decodePublishPayload(raw []byte) ([]byte, error) {
+	rem, n, err := decodeRemainingLength(raw[1:])
+	if err != nil {
+		return nil, err
+	}
+	body := raw[1+n:]
+	if len(body) != rem {
+		return nil, io.ErrUnexpectedEOF
+	}
+	pk := packets.Packet{
+		ProtocolVersion: 5,
+		FixedHeader: packets.FixedHeader{
+			Type:      packets.Publish,
+			Remaining: rem,
+			Qos:       (raw[0] >> 1) & 0x3,
+		},
+	}
+	if err := pk.PublishDecode(body); err != nil {
+		return nil, err
+	}
+	return pk.Payload, nil
+}
+
+func decodeRemainingLength(b []byte) (value int, n int, err error) {
+	var mul uint32 = 1
+	var v uint32
+	for i := 0; i < len(b) && i < 4; i++ {
+		v += uint32(b[i]&127) * mul
+		n++
+		if b[i]&128 == 0 {
+			return int(v), n, nil
+		}
+		mul *= 128
+	}
+	return 0, 0, io.ErrUnexpectedEOF
+}
+
+func extractSessionToken(m map[string]any) string {
+	if m["ok"] != true {
+		return ""
+	}
+	data, _ := m["data"].(map[string]any)
+	tok, _ := data["session_token"].(string)
+	return tok
+}
+
+func enableRegistration(t *testing.T, dataDir, code string) {
+	t.Helper()
+	db, err := store.Open(dataDir, "FULL")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = db.Close() }()
+	now := time.Now().UnixMilli()
+	err = db.Queue.Do(context.Background(), func(tx *sql.Tx) error {
+		if _, e := tx.Exec(`INSERT INTO settings(key, value, updated_at) VALUES(?,?,?)
+ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at`,
+			"registration_enabled", "1", now); e != nil {
+			return e
+		}
+		_, e := tx.Exec(`INSERT INTO settings(key, value, updated_at) VALUES(?,?,?)
+ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at`,
+			"registration_code", code, now)
+		return e
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func waitListenAddr(t *testing.T, dataDir string, timeout time.Duration) string {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	path := filepath.Join(dataDir, "listen.addr")
+	for time.Now().Before(deadline) {
+		b, err := os.ReadFile(path)
+		if err == nil {
+			addr := strings.TrimSpace(string(b))
+			if addr != "" {
+				return addr
+			}
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	t.Fatal("listen.addr not written")
+	return ""
+}

docs/DEVIATIONS.md

@@ -43,6 +43,66 @@
    - 备选方案：docker 目标仅 echo 提示。  
    - 影响：镜像发布流程仍由 Q4 定稿。
 
+### L-WIRE 2026-09-30
+
+1. **serve 真实接线范围**  
+   - 原条款：TASKS 总控接线；listener/broker/admin/注册/消息周期循环挂到进程。  
+   - 实际做法：`cmd/nixmsg/serve.go` 注入真实 `auth.NewPool`/`NewSessionTokens`/`NewAPITokens`/`NewLoginLocks`，挂注册与管理路由（踢线调 `Session.Kick`），listener 识别 HTTP/WebSocket/`OnMQTT` 裸 TCP，broker `Login`+`Session`，握手/断线/`OnPublishDropped` 接到 `message.App`，周期 `DispatchDue`/`PushPending`/`CleanupOnce`，下行 `PublishDown`。集成测覆盖：管理登录、写 settings 后注册、密码 MQTT 握手拿 `session_token`。  
+   - 原因：第二波收尾；三件验收必须通。  
+   - 备选方案：分文件多阶段接线。  
+   - 影响：进程可对外登录/注册/握手。
+
+2. **未接线 / 未完成部分（已被 L-UPLINK 部分取代）**  
+   - 原条款：上行应用帧完整分发到 identity/group/presence/message 业务方法。  
+   - 实际做法（L-WIRE 当时）：`Session` 处理 hello/logout；其余 `HandleUplink` 仍为 Stub。管理注册设置 HTTP（A3）未实现，测试直接写 `settings` 表。  
+   - 原因：本波强制三件验收；完整协议分发属后续波次。  
+   - 备选方案：本波同时实现 HandleUplink 大 multiplex。  
+   - 影响：见 L-UPLINK。
+
+3. **listen.addr 始终写入**  
+   - 原条款：listener N1 仅端口 0 写文件；T0.1 超集为启动即写。  
+   - 实际做法：listener 按 N1 写端口 0；serve 成功后再强制写一次 `listen.addr`（及分离时的 `admin.addr`）。  
+   - 原因：与 harness / T0.1 一致。  
+   - 备选方案：改 listener 始终写。  
+   - 影响：固定端口也会有地址文件。
+
+### L-UPLINK 2026-09-30
+
+1. **HandleUplink 分发到已有业务服务**  
+   - 原条款：TASKS 总控接线；DEVELOPMENT 第 6 节上行帧由服务端处理后回 `resp`；群事件/在线通知/消息走下行。  
+   - 实际做法：`cmd/nixmsg/uplink.go` 的 `appUplink` 在已握手连接上解码并调用 `message`/`identity`/`presence`/`group` 的既有方法；结果编成 `resp` 经 `PublishDown`（QoS 1）回本连接。`Session` 仍独占 `hello`/`self.logout` 与未握手 `not_ready`。`serve` 把四个 App 与 broker Downlink 注入 uplink。不重写业务状态机。  
+   - 原因：main 上业务实现已齐，缺上行入口。  
+   - 备选方案：各包自建 MQTT 钩子（与 port 契约不符）。  
+   - 影响：端侧 send/ack/recall/status/unlock/self.*/presence.*/directory.list/group.* 可走真实进程。
+
+2. **presence.get 未知编号编码**  
+   - 原条款：DEVELOPMENT 6.5「未知编号为 not_found」；`StatusItem.NotFound` 为内部标记。  
+   - 实际做法：`presence.EncodeGetData` 薄封装，resp data 为 `{"items":[...]}`；未知项 `{"id","not_found":true}`，已知项含 `id`/`online`/`since_ms`。  
+   - 原因：Service 层未规定 JSON 形状，接线需固定可编解码形式。  
+   - 备选方案：整请求失败 `not_found`；或把 `not_found` 塞进 `online` 旁字符串字段。  
+   - 影响：SDK 若只认 items 数组两种形状均可（Go SDK 已兼容 wrap/array）。
+
+3. **resp 超限改发 response_too_large**  
+   - 原条款：DEVELOPMENT 7.5「resp 超限改发 response_too_large」。  
+   - 实际做法：发布前按连接表 `MaxReceiveBytes` 与 `MaxPacketSize` 取较小正上限；超限则改发错误 resp（不再发原 data）。未做 MQTT 包头开销扣减（与 message 推送里的 overhead 预算不完全同一函数）。  
+   - 原因：接线层最小可用检查。  
+   - 备选方案：复用 `message.effectivePayloadLimit`（未导出）。  
+   - 影响：接近包上限的大分页可能比推送路径略严或略松。
+
+4. **断线清 presence.watch**  
+   - 原条款：presence.watch 断线清空。  
+   - 实际做法：`OnDisconnect` 额外调 `ClearWatch`；`SetOffline` 路径本身也会清。顶号旧连接未走 `SetOffline(isCurrent)` 时仍能清订阅。  
+   - 原因：避免旧连接订阅泄漏。  
+   - 备选方案：仅依赖 Session 对 isCurrent 调 SetOffline。  
+   - 影响：无。
+
+5. **仍未接线 / 本波未覆盖**  
+   - 管理注册设置 HTTP（A3）仍未挂；集成测继续写 `settings` 表开注册。  
+   - 下行通知帧（`msg`/`receipt`/`revoked`/`presence`/`group_event`/`fatal`）不是上行分发对象，由既有服务经 PublishDown 发出。  
+   - `self.logout`/`hello` 仍在 Session，不经 appUplink。  
+   - 身份 I5 停用/删除级联、管理端群等非本任务范围。  
+   - 集成测覆盖：双端单聊、离线 keep 后上线、群发不含发送者、延迟内撤回对方无回调；未穷尽 presence.watch 通知与全部 group.* 变体。
+
 ### T0.2 2026-09-30
 
 1. **请求指纹规范化格式**  
@@ -499,6 +559,29 @@
    - 备选方案：测试里注入在线连接表使默认不保留也走 pending（与「离线不保留」场景重复覆盖）。  
    - 影响：仅测试期望；产品行为不变。
 
+### I5 2026-09-30
+
+1. **停用/删除级联在 identity 包内完成，admin 可选注入**  
+   - 原条款：DEVELOPMENT 7.6 / PRD F01；TASKS I5「提供给 A 线调用」。  
+   - 实际做法：`identity.App.Disable`/`Enable`/`Delete` 在一个写操作里完成启停、清令牌、作废消息/投递、发 `revoked`（可选 Downlink）、退群/转让群主/解散、清授权/回执/防重/发出记录。`admin.Deps.Identity` 非空时，`setEndpointEnabled`/`deleteEndpointBasic`（及 PATCH enabled）委托上述方法；为空时保留 A2 仅改库行为。未改 `cmd/nixmsg`、未改消息上行分发。  
+   - 原因：隔离要求不改接线；A 线已有路由，只接级联。  
+   - 备选方案：在 admin 内复制级联 SQL；或强制 Identity 必填。  
+   - 影响：生产须在挂载 admin 时注入 `identity.App`（及 KickEndpoint/ConnControl/Downlink），否则停用/删除仍无完整作废。
+
+2. **消息级作废回执 state 用 `rejected`**  
+   - 原条款：DEVELOPMENT 6.4 消息级作废写 `endpoint_id` 空、`state=rejected`；I4 解散群对 scheduled 曾写 `completed`。  
+   - 实际做法：I5 对「发给停用/删除端的 scheduled 单聊」及删除时解散群的 scheduled，回执 `state=rejected`，原因分别为 `endpoint_*` / `group_dissolved`。  
+   - 原因：与 6.4 字面一致。  
+   - 备选方案：与 I4 一样写 `completed`。  
+   - 影响：后台/SDK 若按 state 过滤回执需同时认 rejected。
+
+3. **删除时群主转让按 `joined_at` 最早，并列按编号**  
+   - 原条款：转给最早加入的其他成员。  
+   - 实际做法：`ORDER BY joined_at ASC, endpoint_id ASC LIMIT 1`。  
+   - 原因：同时加入时需稳定次序。  
+   - 备选方案：仅按 joined_at。  
+   - 影响：同毫秒加入时编号小者优先。
+
 ## 后台接口 A
 
 ### A1 2026-09-30
@@ -524,13 +607,13 @@
    - 备选方案：另加 INTEGER 列或把数字存成文本并在 JSON 里发数字。  
    - 影响：W 线类型应按 `string` 解析令牌 id。
 
-4. **尚未实现的管理路由（鉴权中间件已生效，业务返回 501）**  
+4. **A1 范围外路由当时返回 501（A2 已实现端管理）**  
    - 原条款：DEVELOPMENT 第 8 节完整路由表。  
-   - 实际做法：已实现 `login`/`logout`/`me`/`password` 与 `/api/admin/tokens` 全套。以下路由经鉴权后返回 `501 not_implemented`（属 A2/A3）：  
-     `GET /overview`；`endpoints` 列表/开通/import/batch/详情/改/删/kick/reset-login-password/talk-password/unlock；`registration` GET/PUT；`groups` 全部（含 `GET /groups/{id}`）；`messages` 列表与详情；`GET /settings`。  
+   - 实际做法（A1 当时）：已实现 `login`/`logout`/`me`/`password` 与 `/api/admin/tokens` 全套；其余经鉴权后 `501`。  
+   - A2 起：端相关路由已改为真实现，见下节；仍为 501 的有 `overview`、`registration`、`groups`、`messages`、`settings`（属 A3）。  
    - 原因：A1 范围仅鉴权、令牌、操作日志。  
    - 备选方案：无。  
-   - 影响：W/集成测试在 A2/A3 前勿依赖这些业务响应。
+   - 影响：W/集成测试在 A3 前勿依赖尚未实现的业务响应。
 
 5. **操作日志用 `slog` 结构化字段**  
    - 原条款：写结构化日志（操作者、动作、对象、结果、来源 IP）。  
@@ -539,6 +622,78 @@
    - 备选方案：独立 audit 表。  
    - 影响：日志采集需按 msg=`admin_audit` 过滤。
 
+### A2 2026-09-30
+
+1. **停用/删除完整级联留给 I5**  
+   - 原条款：PRD F01 / DEVELOPMENT 7.6：停用作废未送达消息与发送中消息；删除另含退群、群主转让/解散、清授权与回执/发出记录等。  
+   - 实际做法：A2 直接写库：停用立刻 `enabled=0` 并清空 `session_hash`；删除删 `endpoints` 行并清相关 `talk_grants`；二者均调用可注入的 `Deps.KickEndpoint` 踢连接。不作废投递/消息、不转让群主、不清理群成员与回执。  
+   - 原因：本分支尚无 I5；TASKS 允许先接现有存储并在偏差中写明。  
+   - 备选方案：阻塞等待 I5；或在 A 线内复制级联 SQL（易与 I5 重复冲突）。  
+   - 影响：合入 I5 后应由身份服务 `Disable`/`Delete` 接管级联；总控接线把 kick 钩子接到 broker。`KickEndpoint` 为 nil 时踢线为 no-op（单元测试可注入）。
+
+2. **在线状态读库字段，不依赖 presence 服务**  
+   - 原条款：列表含是否在线、最近上下线；可按在线筛选。  
+   - 实际做法：用 `endpoints.online_since` / `offline_since` 判定在线（`online_since` 非空且大于 `offline_since` 或后者为空）；列表项带 `online` / `online_since_ms` / `offline_since_ms`。  
+   - 原因：presence/N3 未在本 Handler 注入；库字段是契约字段。  
+   - 备选方案：注入 `presence.Service.IsOnline`。  
+   - 影响：上下线列未由 N/I 写入前，列表会显示离线。
+
+3. **`login_locked` 仅反映按编号的登录锁定**  
+   - 原条款：列表含 `login_locked`。  
+   - 实际做法：查 `LockLoginEndpoint`；不枚举「编号+IP」锁定。`unlock` 仍调用 `ClearEndpoint` 清两种。  
+   - 原因：`LoginLocks` 接口无「是否任一 IP 锁定」查询。  
+   - 备选方案：扩展 locks 接口。  
+   - 影响：仅 IP 档锁定时列表可能仍显示未锁定，但 unlock 可解除。
+
+4. **A2 时未挂载到 `cmd/nixmsg`；L-WIRE / A3 已接线**  
+   - A2 当时：可挂载 Handler；接线与 `KickEndpoint` 注入留给总控。  
+   - 现状：`serve` 已挂载 Admin Handler；A3 起注入 `Groups`/`Config`/`Version`/`KickEndpoint`。  
+   - 影响：无。
+
+### A3 2026-09-30
+
+1. **概览增加 `endpoints_self`（自助注册数）**  
+   - 原条款：PRD F17「端数量（其中自助注册的数量）」；`admin-api.md` 示例未列该字段。  
+   - 实际做法：`GET /api/admin/overview` 同时返回契约字段与 `endpoints_self`（`source='self'` 计数）。  
+   - 原因：以 PRD / 本任务说明为准补齐。  
+   - 备选方案：只返回 api 文档字段，自助数由前端筛端列表。  
+   - 影响：W 线可选用该字段；旧 mock 类型可增补。
+
+2. **群列表/详情直接读库；变更走 `group.Service`**  
+   - 原条款：群操作调用 `internal/app/group` 已有方法。  
+   - 实际做法：创建/加人用 `AdminCreate`/`AdminAddMembers`；改名/解散/移除/转让先查群主再以群主为 actor 调 `Rename`/`Dissolve`/`Remove`/`Transfer`。列表与详情（含 `joined_at_ms`）因 `List`/`Get` 按成员可见且 Get 无 joined_at，改为管理侧 SQL。  
+   - 原因：端协议 API 按成员视角，后台需全局列表。  
+   - 备选方案：在 group 包增加 AdminList/AdminGet。  
+   - 影响：列表不依赖 Groups 注入；写操作未注入 Groups 时返回 `503 busy`。
+
+3. **后台建群不接受自定义 `id`**  
+   - 原条款：admin-api「id 留空则生成」。  
+   - 实际做法：`AdminCreate` 无自定义 id 参数；请求带非空 id 返回 `400`。  
+   - 原因：不改身份线接口签名。  
+   - 备选方案：扩展 `AdminCreate` 接受可选 id。  
+   - 影响：后台只能服务器生成群编号。
+
+4. **`/metrics` 门禁抽到 `internal/httpx.MetricsGate`**  
+   - 原条款：A 线负责 metrics 访问规则（DEVELOPMENT 4.3）。  
+   - 实际做法：规则实现放 `httpx.MetricsGate`；`listener.NewMux` RoleShared 调用之（N 目录一行替换）。单独后台监听仍直接挂 Handler 不鉴权。  
+   - 原因：A 负责目录是 `admin`+`httpx`；listener 仅接线。  
+   - 备选方案：门禁留在 listener。  
+   - 影响：`serve` 已传 `MetricsToken`；共用端口无令牌 404、错令牌 401。
+
+5. **注册安全码写入审计不含明文**  
+   - 原条款：安全码明文返回已登录管理员，不写日志。  
+   - 实际做法：GET/PUT 响应含 `code`；`admin_audit` 的 object 为空，不记安全码。存库键 `registration_enabled`=`1`/`0`，与 I1 一致。  
+   - 原因：D14 / DEVELOPMENT 12 节。  
+   - 备选：无。  
+   - 影响：无。
+
+6. **消息查询不碰 `message_bodies`**  
+   - 原条款：只读 messages 与 deliveries；响应无正文。  
+   - 实际做法：SELECT 不含 `body`/`body_enc` 以外的正文列（messages 仍有 `body_enc` 列但查询不选它）；不 JOIN `message_bodies`。  
+   - 原因：任务硬性要求。  
+   - 备选：无。  
+   - 影响：无。
+
 ## 后台网页 W
 
 1. **W1–W3 阶段使用内存假数据，不请求真实 `/api/admin`**  
@@ -720,3 +875,26 @@
    - 原因：空命名卷属主为 root 时，distroless nonroot 无法建库（`unable to open database file`）。  
    - 备选方案：compose 增加一次性 init 服务；或文档要求宿主机目录预授权。  
    - 影响：按示例首次 `up` 前需处理权限，否则 serve 立即退出。
+
+### Q2 第一部分（已合并功能验收）2026-09-30
+
+1. **对真实进程探测，未接线则记「未测」而非改业务代码**  
+   - 原条款：TASKS Q2「PRD 第 10 节每条至少有一个集成测试」；本波只做已合并功能的第一部分。  
+   - 实际做法：`test/accept` 用 `test/harness`（随机端口 + 临时目录）起真实 `nixmsg`；对 `/api/admin/login`、`/api/client/register`、`POST /api/admin/endpoints` 先探测。404 则该条写「未测」并注明缺总控接线 / A2 / I 等，不修改 `cmd/nixmsg` 或业务包求绿。  
+   - 原因：main@d357082 上 A1/I1 等仅为可挂载 Handler，`serve` 只挂了 `/healthz`、`/readyz`（见各线 DEVIATIONS「未改 cmd」）。  
+   - 备选方案：测试进程内自行 `admin.New` 挂路由（不反映交付二进制行为，否决）。  
+   - 影响：本波 F17/F01/F23 多为未测；接线后同一测试会自动跑登录锁定、CSRF、注册与开通路径。
+
+2. **F22 只验收 init + 健康检查子集**  
+   - 原条款：F22 含备份恢复、升级迁移、证书重载、Docker、指标。  
+   - 实际做法：集成测试覆盖空目录 `admin init`、`serve`、`/healthz`、`/readyz`，并断言管理员密码不出现在 serve 的 stdout/stderr；其余 F22 子项仍标未测。  
+   - 原因：本波范围是「现在就能测的路径」。  
+   - 备选方案：本波强行跑 Docker/证书（超出第一部分）。  
+   - 影响：对照表 F22 为「通过」但备注写明未覆盖项。
+
+3. **验收报告写入方式**  
+   - 原条款：用 `test/report` 生成 F01–F23 对照表。  
+   - 实际做法：`TestQ2AcceptAndReport` 汇总探测结果；默认写临时目录。`task q:accept`（`NIXMSG_WRITE_ACCEPT_REPORT=1`）写入 `test/report/testdata/q2_results.json` 与 `test/accept/ACCEPTANCE.md`；`task q:report-q2` 可再生成 Markdown。普通 `go test`/`task check` 不改仓库文件。  
+   - 原因：避免每次单测改 `generated_at` 弄脏工作区。  
+   - 备选方案：固定时间戳始终写入仓库。  
+   - 影响：交付审阅以 `ACCEPTANCE.md` / `q2_results.json` 为准，需先跑过 `task q:accept`。

internal/admin/a3_test.go

@@ -0,0 +1,287 @@
+package admin_test
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"net/http"
+	"net/http/cookiejar"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"git.asio.asia/nixevol/NixMsg/internal/admin"
+	"git.asio.asia/nixevol/NixMsg/internal/app/group"
+	"git.asio.asia/nixevol/NixMsg/internal/auth"
+	"git.asio.asia/nixevol/NixMsg/internal/config"
+	"git.asio.asia/nixevol/NixMsg/internal/store"
+)
+
+func setupA3(t *testing.T) (*store.DB, *httptest.Server, *http.Client) {
+	t.Helper()
+	dir := t.TempDir()
+	db, err := store.Open(filepath.Join(dir, "data"), "FULL")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+
+	hash := auth.NewStubHashPool()
+	if seedErr := admin.SeedAdminPassword(context.Background(), db, hash, testPassword); seedErr != nil {
+		t.Fatal(seedErr)
+	}
+	gApp := group.New(group.Config{DB: db, MaxGroupMembers: 100})
+	h := admin.New(admin.Deps{
+		DB:      db,
+		Hash:    hash,
+		Tokens:  admin.NewRandomAPITokens(),
+		Locks:   admin.NewMemoryLoginLocks(),
+		Groups:  gApp,
+		Config:  config.Default(),
+		Version: "0.1.0-test",
+	})
+	srv := httptest.NewServer(h)
+	t.Cleanup(srv.Close)
+
+	jar, err := cookiejar.New(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	client := &http.Client{Jar: jar}
+	login(t, client, srv.URL)
+	return db, srv, client
+}
+
+func csrf() map[string]string {
+	return map[string]string{"X-Nixmsg-Request": "1"}
+}
+
+func insertEndpoint(t *testing.T, db *store.DB, id, source string, enabled bool, online bool) {
+	t.Helper()
+	now := time.Now().UnixMilli()
+	en := 1
+	if !enabled {
+		en = 0
+	}
+	var onlineSince, offlineSince any
+	if online {
+		onlineSince = now
+	} else {
+		offlineSince = now
+	}
+	err := db.Queue.Do(context.Background(), func(tx *sql.Tx) error {
+		_, e := tx.Exec(`INSERT INTO endpoints(
+  id, name, remark, source, login_hash, talk_hash, talk_version,
+  default_delay_ms, enabled, created_at, online_since, offline_since
+) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`,
+			id, id, "", source, "stub$hash", nil, 0, 0, en, now, onlineSince, offlineSince)
+		return e
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestRegistrationToggleAndGenerate(t *testing.T) {
+	_, srv, client := setupA3(t)
+	base := srv.URL
+
+	res := doReq(t, client, http.MethodGet, base+"/api/admin/registration", "", nil)
+	env := decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("get: %d %+v", res.StatusCode, env)
+	}
+	var got map[string]any
+	_ = json.Unmarshal(env.Data, &got)
+	if got["enabled"] != false {
+		t.Fatalf("default enabled=%v", got["enabled"])
+	}
+
+	res = doReq(t, client, http.MethodPut, base+"/api/admin/registration",
+		`{"enabled":true,"code":"abcdefgh"}`, csrf())
+	env = decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("put enable: %d %+v", res.StatusCode, env)
+	}
+	_ = json.Unmarshal(env.Data, &got)
+	if got["enabled"] != true || got["code"] != "abcdefgh" {
+		t.Fatalf("after put: %v", got)
+	}
+
+	res = doReq(t, client, http.MethodPut, base+"/api/admin/registration",
+		`{"generate":true}`, csrf())
+	env = decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("generate: %d %+v", res.StatusCode, env)
+	}
+	_ = json.Unmarshal(env.Data, &got)
+	code, _ := got["code"].(string)
+	if len(code) != 16 {
+		t.Fatalf("generated len=%d code=%q", len(code), code)
+	}
+
+	res = doReq(t, client, http.MethodPut, base+"/api/admin/registration",
+		`{"enabled":false}`, csrf())
+	env = decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("disable: %d %+v", res.StatusCode, env)
+	}
+	_ = json.Unmarshal(env.Data, &got)
+	if got["enabled"] != false {
+		t.Fatalf("disabled=%v", got["enabled"])
+	}
+}
+
+func TestMessageDetailHasNoBody(t *testing.T) {
+	db, srv, client := setupA3(t)
+	base := srv.URL
+	now := time.Now().UnixMilli()
+
+	err := db.Queue.Do(context.Background(), func(tx *sql.Tx) error {
+		if _, e := tx.Exec(`INSERT INTO messages(
+  seq, id, sender_id, dest_kind, dest_id, meta, content_type, body_enc,
+  send_at, keep, ttl_seconds, receipt, state, reason, created_at
+) VALUES(1,'m1','a','endpoint','b','{}','text/plain','enc',?,?,0,1,'dispatched','',?)`,
+			now, 1, now); e != nil {
+			return e
+		}
+		if _, e := tx.Exec(`INSERT INTO message_bodies(seq, body) VALUES(1, ?)`,
+			[]byte("SECRET_BODY_SHOULD_NOT_APPEAR")); e != nil {
+			return e
+		}
+		_, e := tx.Exec(`INSERT INTO deliveries(
+  seq, endpoint_id, send_at, keep, state, reason, attempts, updated_at, pushed_at
+) VALUES(1,'b',?,1,'pending','',2,?,?)`, now, now, now)
+		return e
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	res := doReq(t, client, http.MethodGet, base+"/api/admin/messages/1", "", nil)
+	env := decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("detail: %d %+v", res.StatusCode, env)
+	}
+	raw := string(env.Data)
+	if strings.Contains(raw, "body") || strings.Contains(raw, "SECRET_BODY") {
+		t.Fatalf("response contains body/secret: %s", raw)
+	}
+	var detail map[string]any
+	_ = json.Unmarshal(env.Data, &detail)
+	dels, _ := detail["deliveries"].([]any)
+	if len(dels) != 1 {
+		t.Fatalf("deliveries=%v", detail["deliveries"])
+	}
+	d0 := dels[0].(map[string]any)
+	if d0["attempts"].(float64) != 2 {
+		t.Fatalf("attempts=%v", d0["attempts"])
+	}
+
+	res = doReq(t, client, http.MethodGet, base+"/api/admin/messages", "", nil)
+	env = decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("list: %d %+v", res.StatusCode, env)
+	}
+	if strings.Contains(string(env.Data), "SECRET_BODY") || strings.Contains(string(env.Data), `"body"`) {
+		t.Fatalf("list leaked body: %s", env.Data)
+	}
+}
+
+func TestGroupsCRUD(t *testing.T) {
+	db, srv, client := setupA3(t)
+	base := srv.URL
+	insertEndpoint(t, db, "alice", "admin", true, false)
+	insertEndpoint(t, db, "bob", "admin", true, true)
+	insertEndpoint(t, db, "carol", "self", true, false)
+
+	res := doReq(t, client, http.MethodPost, base+"/api/admin/groups",
+		`{"name":"一组","owner_id":"alice","member_ids":["bob","carol"]}`, csrf())
+	env := decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("create: %d %+v", res.StatusCode, env)
+	}
+	var created struct {
+		ID      string `json:"id"`
+		OwnerID string `json:"owner_id"`
+	}
+	_ = json.Unmarshal(env.Data, &created)
+	if created.ID == "" || created.OwnerID != "alice" {
+		t.Fatalf("created=%+v", created)
+	}
+
+	res = doReq(t, client, http.MethodPatch, base+"/api/admin/groups/"+created.ID,
+		`{"name":"新名"}`, csrf())
+	env = decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("rename: %d %+v", res.StatusCode, env)
+	}
+
+	res = doReq(t, client, http.MethodPost, base+"/api/admin/groups/"+created.ID+"/transfer",
+		`{"endpoint_id":"bob"}`, csrf())
+	env = decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("transfer: %d %+v", res.StatusCode, env)
+	}
+
+	res = doReq(t, client, http.MethodDelete,
+		base+"/api/admin/groups/"+created.ID+"/members/carol", "", csrf())
+	env = decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("remove: %d %+v", res.StatusCode, env)
+	}
+
+	res = doReq(t, client, http.MethodDelete, base+"/api/admin/groups/"+created.ID, "", csrf())
+	env = decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("dissolve: %d %+v", res.StatusCode, env)
+	}
+}
+
+func TestOverviewAndSettings(t *testing.T) {
+	db, srv, client := setupA3(t)
+	base := srv.URL
+	insertEndpoint(t, db, "a1", "admin", true, true)
+	insertEndpoint(t, db, "s1", "self", true, false)
+	insertEndpoint(t, db, "d1", "admin", false, false)
+
+	res := doReq(t, client, http.MethodGet, base+"/api/admin/overview", "", nil)
+	env := decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("overview: %d %+v", res.StatusCode, env)
+	}
+	var ov map[string]any
+	_ = json.Unmarshal(env.Data, &ov)
+	if ov["version"] != "0.1.0-test" {
+		t.Fatalf("version=%v", ov["version"])
+	}
+	if ov["endpoints_total"].(float64) != 3 {
+		t.Fatalf("total=%v", ov["endpoints_total"])
+	}
+	if ov["endpoints_self"].(float64) != 1 {
+		t.Fatalf("self=%v", ov["endpoints_self"])
+	}
+	if ov["endpoints_online"].(float64) != 1 {
+		t.Fatalf("online=%v", ov["endpoints_online"])
+	}
+	if ov["endpoints_disabled"].(float64) != 1 {
+		t.Fatalf("disabled=%v", ov["endpoints_disabled"])
+	}
+
+	res = doReq(t, client, http.MethodGet, base+"/api/admin/settings", "", nil)
+	env = decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("settings: %d %+v", res.StatusCode, env)
+	}
+	raw := string(env.Data)
+	if strings.Contains(raw, "token") || strings.Contains(raw, "password") {
+		t.Fatalf("settings leaked secret: %s", raw)
+	}
+	var st map[string]any
+	_ = json.Unmarshal(env.Data, &st)
+	if st["listen"] == nil || st["limits"] == nil {
+		t.Fatalf("settings=%v", st)
+	}
+}

internal/admin/admin_test.go

@@ -214,11 +214,11 @@ func TestAPITokenAuthAndRestrictions(t *testing.T) {
 		t.Fatalf("auth=%v", me["auth"])
 	}
 
-	// 普通管理接口鉴权通过（业务 501）
+	// 普通管理接口鉴权通过（A3 overview）
 	res = doReq(t, tokClient, http.MethodGet, base+"/api/admin/overview", "", hdr)
 	env = decodeEnv(t, res)
-	if res.StatusCode != http.StatusNotImplemented {
-		t.Fatalf("overview want 501 got %d %+v", res.StatusCode, env)
+	if res.StatusCode != http.StatusOK || !env.OK {
+		t.Fatalf("overview want 200 got %d %+v", res.StatusCode, env)
 	}
 
 	// 禁止 password / tokens

internal/admin/endpoints.go

@@ -0,0 +1,619 @@
+package admin
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"net/http"
+	"strconv"
+	"strings"
+	"unicode/utf8"
+
+	"git.asio.asia/nixevol/NixMsg/internal/auth"
+	"git.asio.asia/nixevol/NixMsg/internal/httpx"
+	"git.asio.asia/nixevol/NixMsg/internal/protocol"
+)
+
+const (
+	maxRemarkChars       = 200
+	defaultListLimit     = 50
+	maxListLimit         = 200
+	maxImportRows        = 1000
+	sourceAdmin          = "admin"
+	loginPasswordOnceKey = "login_password"
+)
+
+// EndpointKickFunc 只断开端当前连接；是否作废令牌由调用方在踢线前决定。
+// 返回 kicked=true 表示当时有连接被断开。nil 表示未接线（踢线为 no-op）。
+type EndpointKickFunc func(ctx context.Context, endpointID string) (kicked bool, err error)
+
+type endpointRow struct {
+	ID              string
+	Name            string
+	Remark          string
+	Source          string
+	Enabled         bool
+	TalkHash        sql.NullString
+	DefaultDelayMs  int64
+	CreatedAtMs     int64
+	OnlineSinceMs   sql.NullInt64
+	OfflineSinceMs  sql.NullInt64
+	SessionIssuedAt sql.NullInt64
+	SessionUsedAt   sql.NullInt64
+}
+
+func (e endpointRow) toAPI(loginLocked bool, detail bool) map[string]any {
+	online, onlineSince, offlineSince := presenceFields(e.OnlineSinceMs, e.OfflineSinceMs)
+	out := map[string]any{
+		"id":                e.ID,
+		"name":              e.Name,
+		"remark":            e.Remark,
+		"source":            e.Source,
+		"enabled":           e.Enabled,
+		"online":            online,
+		"online_since_ms":   onlineSince,
+		"offline_since_ms":  offlineSince,
+		"talk_password_set": e.TalkHash.Valid && e.TalkHash.String != "",
+		"default_delay_ms":  e.DefaultDelayMs,
+		"created_at_ms":     e.CreatedAtMs,
+		"login_locked":      loginLocked,
+	}
+	if detail {
+		var issued, used any
+		if e.SessionIssuedAt.Valid {
+			issued = e.SessionIssuedAt.Int64
+		}
+		if e.SessionUsedAt.Valid {
+			used = e.SessionUsedAt.Int64
+		}
+		out["session_issued_at_ms"] = issued
+		out["session_used_at_ms"] = used
+	}
+	return out
+}
+
+func presenceFields(onlineSince, offlineSince sql.NullInt64) (online bool, onlineMs, offlineMs any) {
+	if onlineSince.Valid && (!offlineSince.Valid || onlineSince.Int64 > offlineSince.Int64) {
+		online = true
+		onlineMs = onlineSince.Int64
+		if offlineSince.Valid {
+			offlineMs = offlineSince.Int64
+		}
+		return online, onlineMs, offlineMs
+	}
+	if onlineSince.Valid {
+		onlineMs = onlineSince.Int64
+	}
+	if offlineSince.Valid {
+		offlineMs = offlineSince.Int64
+	}
+	return false, onlineMs, offlineMs
+}
+
+func (h *Handler) endpointLoginLocked(id string) bool {
+	locked, _ := h.locks.Check(auth.LockKey{Kind: auth.LockLoginEndpoint, EndpointID: id})
+	return locked
+}
+
+func (h *Handler) kickEndpoint(ctx context.Context, id string) (bool, error) {
+	if h.kick == nil {
+		return false, nil
+	}
+	return h.kick(ctx, id)
+}
+
+func (h *Handler) handleEndpointList(w http.ResponseWriter, r *http.Request) {
+	q := r.URL.Query()
+	limit := defaultListLimit
+	if s := q.Get("limit"); s != "" {
+		n, err := strconv.Atoi(s)
+		if err != nil || n < 1 {
+			httpx.WriteError(w, http.StatusBadRequest, "bad_request", "limit 无效")
+			return
+		}
+		if n > maxListLimit {
+			n = maxListLimit
+		}
+		limit = n
+	}
+	offset := 0
+	if c := q.Get("cursor"); c != "" {
+		n, err := strconv.Atoi(c)
+		if err != nil || n < 0 {
+			httpx.WriteError(w, http.StatusBadRequest, "bad_request", "cursor 无效")
+			return
+		}
+		offset = n
+	}
+	source := q.Get("source")
+	if source != "" && source != "admin" && source != "self" {
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "source 无效")
+		return
+	}
+	var onlineFilter *bool
+	if s := q.Get("online"); s != "" {
+		switch s {
+		case "true":
+			v := true
+			onlineFilter = &v
+		case "false":
+			v := false
+			onlineFilter = &v
+		default:
+			httpx.WriteError(w, http.StatusBadRequest, "bad_request", "online 无效")
+			return
+		}
+	}
+	var enabledFilter *bool
+	if s := q.Get("enabled"); s != "" {
+		switch s {
+		case "true":
+			v := true
+			enabledFilter = &v
+		case "false":
+			v := false
+			enabledFilter = &v
+		default:
+			httpx.WriteError(w, http.StatusBadRequest, "bad_request", "enabled 无效")
+			return
+		}
+	}
+	query := strings.TrimSpace(q.Get("query"))
+
+	total, err := h.countEndpoints(r.Context(), source, onlineFilter, enabledFilter, query)
+	if err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	rows, err := h.listEndpoints(r.Context(), source, onlineFilter, enabledFilter, query, limit, offset)
+	if err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	items := make([]map[string]any, 0, len(rows))
+	for _, row := range rows {
+		items = append(items, row.toAPI(h.endpointLoginLocked(row.ID), false))
+	}
+	next := ""
+	if offset+len(rows) < total {
+		next = strconv.Itoa(offset + len(rows))
+	}
+	httpx.WriteOK(w, map[string]any{
+		"items":       items,
+		"next_cursor": next,
+		"total":       total,
+	})
+}
+
+func (h *Handler) handleEndpointCreate(w http.ResponseWriter, r *http.Request) {
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+
+	var req struct {
+		ID                  string `json:"id"`
+		Name                string `json:"name"`
+		Remark              string `json:"remark"`
+		LoginPassword       string `json:"login_password"`
+		TalkPassword        string `json:"talk_password"`
+		DefaultDelaySeconds *int64 `json:"default_delay_seconds"`
+	}
+	if err := httpx.DecodeJSON(r, &req); err != nil {
+		h.audit(actorString(p), "endpoint_create", "", "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "请求体无效")
+		return
+	}
+	delaySec := int64(0)
+	if req.DefaultDelaySeconds != nil {
+		delaySec = *req.DefaultDelaySeconds
+	}
+	if errMsg := validateEndpointFields(req.ID, req.Name, req.Remark, req.LoginPassword, req.TalkPassword, delaySec); errMsg != "" {
+		h.audit(actorString(p), "endpoint_create", req.ID, "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", errMsg)
+		return
+	}
+
+	id := req.ID
+	loginPW := req.LoginPassword
+	pwGenerated := false
+	if loginPW == "" {
+		pw, err := generateLoginPassword()
+		if err != nil {
+			h.audit(actorString(p), "endpoint_create", id, "error", ip)
+			httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+			return
+		}
+		loginPW = pw
+		pwGenerated = true
+	}
+	loginHash, err := h.hash.Hash(r.Context(), auth.PasswordLogin, loginPW)
+	if err != nil {
+		h.audit(actorString(p), "endpoint_create", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	var talkHash sql.NullString
+	if req.TalkPassword != "" {
+		th, hashErr := h.hash.Hash(r.Context(), auth.PasswordTalk, req.TalkPassword)
+		if hashErr != nil {
+			h.audit(actorString(p), "endpoint_create", id, "error", ip)
+			httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+			return
+		}
+		talkHash = sql.NullString{String: th, Valid: true}
+	}
+
+	createdID, err := h.insertEndpoint(r.Context(), endpointInsert{
+		ID:             id,
+		Name:           req.Name,
+		Remark:         req.Remark,
+		Source:         sourceAdmin,
+		LoginHash:      loginHash,
+		TalkHash:       talkHash,
+		DefaultDelayMs: delaySec * 1000,
+	})
+	if err != nil {
+		if isUniqueConstraint(err) {
+			h.audit(actorString(p), "endpoint_create", id, "id_taken", ip)
+			httpx.WriteError(w, http.StatusConflict, "id_taken", "编号已占用")
+			return
+		}
+		h.audit(actorString(p), "endpoint_create", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	h.audit(actorString(p), "endpoint_create", createdID, "ok", ip)
+	data := map[string]any{"id": createdID}
+	if pwGenerated {
+		data[loginPasswordOnceKey] = loginPW
+	}
+	httpx.WriteOK(w, data)
+}
+
+func (h *Handler) handleEndpointGet(w http.ResponseWriter, r *http.Request) {
+	id := r.PathValue("id")
+	row, err := h.getEndpoint(r.Context(), id)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			httpx.WriteError(w, http.StatusNotFound, "not_found", "端不存在")
+			return
+		}
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	httpx.WriteOK(w, row.toAPI(h.endpointLoginLocked(row.ID), true))
+}
+
+func (h *Handler) handleEndpointPatch(w http.ResponseWriter, r *http.Request) {
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+	id := r.PathValue("id")
+
+	var req struct {
+		Name                *string `json:"name"`
+		Remark              *string `json:"remark"`
+		DefaultDelaySeconds *int64  `json:"default_delay_seconds"`
+		Enabled             *bool   `json:"enabled"`
+	}
+	if err := httpx.DecodeJSON(r, &req); err != nil {
+		h.audit(actorString(p), "endpoint_patch", id, "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "请求体无效")
+		return
+	}
+	if req.Name == nil && req.Remark == nil && req.DefaultDelaySeconds == nil && req.Enabled == nil {
+		h.audit(actorString(p), "endpoint_patch", id, "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "无更新字段")
+		return
+	}
+	if req.Name != nil && !protocol.ValidName(*req.Name) {
+		h.audit(actorString(p), "endpoint_patch", id, "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "名称不合法")
+		return
+	}
+	if req.Remark != nil && utf8.RuneCountInString(*req.Remark) > maxRemarkChars {
+		h.audit(actorString(p), "endpoint_patch", id, "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "备注过长")
+		return
+	}
+	if req.DefaultDelaySeconds != nil && *req.DefaultDelaySeconds < 0 {
+		h.audit(actorString(p), "endpoint_patch", id, "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "默认延迟无效")
+		return
+	}
+
+	hasMeta := req.Name != nil || req.Remark != nil || req.DefaultDelaySeconds != nil
+	var wasEnabled bool
+	var err error
+	if hasMeta || (req.Enabled != nil && h.identity == nil) {
+		wasEnabled, err = h.patchEndpoint(r.Context(), id, req.Name, req.Remark, req.DefaultDelaySeconds, req.Enabled)
+		if err != nil {
+			if errors.Is(err, sql.ErrNoRows) {
+				h.audit(actorString(p), "endpoint_patch", id, "not_found", ip)
+				httpx.WriteError(w, http.StatusNotFound, "not_found", "端不存在")
+				return
+			}
+			h.audit(actorString(p), "endpoint_patch", id, "error", ip)
+			httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+			return
+		}
+	}
+	if req.Enabled != nil && h.identity != nil {
+		var found bool
+		found, err = h.setEndpointEnabled(r.Context(), id, *req.Enabled)
+		if err != nil {
+			h.audit(actorString(p), "endpoint_patch", id, "error", ip)
+			httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+			return
+		}
+		if !found {
+			h.audit(actorString(p), "endpoint_patch", id, "not_found", ip)
+			httpx.WriteError(w, http.StatusNotFound, "not_found", "端不存在")
+			return
+		}
+		if !*req.Enabled {
+			_, _ = h.kickEndpoint(r.Context(), id)
+		}
+	} else if req.Enabled != nil && !*req.Enabled && wasEnabled {
+		_, _ = h.kickEndpoint(r.Context(), id)
+	}
+	row, err := h.getEndpoint(r.Context(), id)
+	if err != nil {
+		h.audit(actorString(p), "endpoint_patch", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	h.audit(actorString(p), "endpoint_patch", id, "ok", ip)
+	httpx.WriteOK(w, row.toAPI(h.endpointLoginLocked(row.ID), true))
+}
+
+func (h *Handler) handleEndpointDelete(w http.ResponseWriter, r *http.Request) {
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+	id := r.PathValue("id")
+
+	ok, err := h.deleteEndpointBasic(r.Context(), id)
+	if err != nil {
+		h.audit(actorString(p), "endpoint_delete", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	if !ok {
+		h.audit(actorString(p), "endpoint_delete", id, "not_found", ip)
+		httpx.WriteError(w, http.StatusNotFound, "not_found", "端不存在")
+		return
+	}
+	_, _ = h.kickEndpoint(r.Context(), id)
+	h.audit(actorString(p), "endpoint_delete", id, "ok", ip)
+	httpx.WriteOK(w, map[string]any{})
+}
+
+func (h *Handler) handleEndpointBatch(w http.ResponseWriter, r *http.Request) {
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+
+	var req struct {
+		IDs    []string `json:"ids"`
+		Action string   `json:"action"`
+	}
+	if err := httpx.DecodeJSON(r, &req); err != nil || len(req.IDs) == 0 {
+		h.audit(actorString(p), "endpoint_batch", "", "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "请求体无效")
+		return
+	}
+	switch req.Action {
+	case "disable", "enable", "delete":
+	default:
+		h.audit(actorString(p), "endpoint_batch", "", "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "action 无效")
+		return
+	}
+
+	okIDs := make([]string, 0, len(req.IDs))
+	failed := make([]map[string]string, 0)
+	for _, id := range req.IDs {
+		var opErr error
+		var found bool
+		switch req.Action {
+		case "disable":
+			found, opErr = h.setEndpointEnabled(r.Context(), id, false)
+			if found && opErr == nil {
+				_, _ = h.kickEndpoint(r.Context(), id)
+			}
+		case "enable":
+			found, opErr = h.setEndpointEnabled(r.Context(), id, true)
+		case "delete":
+			found, opErr = h.deleteEndpointBasic(r.Context(), id)
+			if found && opErr == nil {
+				_, _ = h.kickEndpoint(r.Context(), id)
+			}
+		}
+		if opErr != nil {
+			failed = append(failed, map[string]string{"id": id, "code": "internal"})
+			continue
+		}
+		if !found {
+			failed = append(failed, map[string]string{"id": id, "code": "not_found"})
+			continue
+		}
+		okIDs = append(okIDs, id)
+	}
+	h.audit(actorString(p), "endpoint_batch_"+req.Action, strings.Join(okIDs, ","), "ok", ip)
+	httpx.WriteOK(w, map[string]any{"ok_ids": okIDs, "failed": failed})
+}
+
+func (h *Handler) handleEndpointKick(w http.ResponseWriter, r *http.Request) {
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+	id := r.PathValue("id")
+
+	exists, err := h.endpointExists(r.Context(), id)
+	if err != nil {
+		h.audit(actorString(p), "endpoint_kick", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	if !exists {
+		h.audit(actorString(p), "endpoint_kick", id, "not_found", ip)
+		httpx.WriteError(w, http.StatusNotFound, "not_found", "端不存在")
+		return
+	}
+	kicked, err := h.kickEndpoint(r.Context(), id)
+	if err != nil {
+		h.audit(actorString(p), "endpoint_kick", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	h.audit(actorString(p), "endpoint_kick", id, "ok", ip)
+	httpx.WriteOK(w, map[string]any{"kicked": kicked})
+}
+
+func (h *Handler) handleEndpointResetLoginPassword(w http.ResponseWriter, r *http.Request) {
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+	id := r.PathValue("id")
+
+	var req struct {
+		LoginPassword string `json:"login_password"`
+	}
+	if err := httpx.DecodeJSON(r, &req); err != nil {
+		h.audit(actorString(p), "endpoint_reset_login_password", id, "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "请求体无效")
+		return
+	}
+	if !protocol.ValidLoginPassword(req.LoginPassword) {
+		h.audit(actorString(p), "endpoint_reset_login_password", id, "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "登录密码不合法")
+		return
+	}
+
+	pw := req.LoginPassword
+	if pw == "" {
+		gen, err := generateLoginPassword()
+		if err != nil {
+			h.audit(actorString(p), "endpoint_reset_login_password", id, "error", ip)
+			httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+			return
+		}
+		pw = gen
+	}
+	hash, err := h.hash.Hash(r.Context(), auth.PasswordLogin, pw)
+	if err != nil {
+		h.audit(actorString(p), "endpoint_reset_login_password", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	ok, err := h.resetLoginPassword(r.Context(), id, hash)
+	if err != nil {
+		h.audit(actorString(p), "endpoint_reset_login_password", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	if !ok {
+		h.audit(actorString(p), "endpoint_reset_login_password", id, "not_found", ip)
+		httpx.WriteError(w, http.StatusNotFound, "not_found", "端不存在")
+		return
+	}
+	_, _ = h.kickEndpoint(r.Context(), id)
+	h.audit(actorString(p), "endpoint_reset_login_password", id, "ok", ip)
+	httpx.WriteOK(w, map[string]any{loginPasswordOnceKey: pw})
+}
+
+func (h *Handler) handleEndpointTalkPassword(w http.ResponseWriter, r *http.Request) {
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+	id := r.PathValue("id")
+
+	var req struct {
+		TalkPassword string `json:"talk_password"`
+	}
+	if err := httpx.DecodeJSON(r, &req); err != nil {
+		h.audit(actorString(p), "endpoint_talk_password", id, "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "请求体无效")
+		return
+	}
+	if !protocol.ValidTalkPassword(req.TalkPassword) {
+		h.audit(actorString(p), "endpoint_talk_password", id, "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "对话密码不合法")
+		return
+	}
+
+	var talkHash sql.NullString
+	if req.TalkPassword != "" {
+		th, err := h.hash.Hash(r.Context(), auth.PasswordTalk, req.TalkPassword)
+		if err != nil {
+			h.audit(actorString(p), "endpoint_talk_password", id, "error", ip)
+			httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+			return
+		}
+		talkHash = sql.NullString{String: th, Valid: true}
+	}
+	ok, err := h.setTalkPassword(r.Context(), id, talkHash)
+	if err != nil {
+		h.audit(actorString(p), "endpoint_talk_password", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	if !ok {
+		h.audit(actorString(p), "endpoint_talk_password", id, "not_found", ip)
+		httpx.WriteError(w, http.StatusNotFound, "not_found", "端不存在")
+		return
+	}
+	h.audit(actorString(p), "endpoint_talk_password", id, "ok", ip)
+	httpx.WriteOK(w, map[string]any{"talk_password_set": talkHash.Valid})
+}
+
+func (h *Handler) handleEndpointUnlock(w http.ResponseWriter, r *http.Request) {
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+	id := r.PathValue("id")
+
+	exists, err := h.endpointExists(r.Context(), id)
+	if err != nil {
+		h.audit(actorString(p), "endpoint_unlock", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	if !exists {
+		h.audit(actorString(p), "endpoint_unlock", id, "not_found", ip)
+		httpx.WriteError(w, http.StatusNotFound, "not_found", "端不存在")
+		return
+	}
+	h.locks.ClearEndpoint(id)
+	h.audit(actorString(p), "endpoint_unlock", id, "ok", ip)
+	httpx.WriteOK(w, map[string]any{})
+}
+
+func validateEndpointFields(id, name, remark, loginPW, talkPW string, delaySec int64) string {
+	if id != "" && !protocol.ValidEndpointID(id) {
+		return "编号不合法"
+	}
+	if !protocol.ValidName(name) {
+		return "名称不合法"
+	}
+	if utf8.RuneCountInString(remark) > maxRemarkChars {
+		return "备注过长"
+	}
+	if protocol.LoginPasswordForbiddenPrefix(loginPW) {
+		return "登录密码不能以 nst_ 开头"
+	}
+	if !protocol.ValidLoginPassword(loginPW) {
+		return "登录密码不合法"
+	}
+	if !protocol.ValidTalkPassword(talkPW) {
+		return "对话密码不合法"
+	}
+	if delaySec < 0 {
+		return "默认延迟无效"
+	}
+	return ""
+}
+
+func writeCSVValidationError(w http.ResponseWriter, errs []csvLineError) {
+	httpx.WriteJSON(w, http.StatusBadRequest, httpx.Envelope{
+		OK: false,
+		Error: &httpx.ErrorBody{
+			Code:    "bad_request",
+			Message: "CSV 校验失败",
+		},
+		Data: map[string]any{"errors": errs},
+	})
+}

internal/admin/endpoints_csv.go

@@ -0,0 +1,310 @@
+package admin
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"encoding/csv"
+	"io"
+	"mime"
+	"mime/multipart"
+	"net/http"
+	"strconv"
+	"strings"
+	"unicode/utf8"
+
+	"git.asio.asia/nixevol/NixMsg/internal/auth"
+	"git.asio.asia/nixevol/NixMsg/internal/httpx"
+	"git.asio.asia/nixevol/NixMsg/internal/protocol"
+)
+
+type csvLineError struct {
+	Line   int    `json:"line"`
+	Reason string `json:"reason"`
+}
+
+type importPrepared struct {
+	Insert endpointInsert
+	// PlainLogin 始终填入响应（生成或原文，仅此一次）。
+	PlainLogin string
+	Name       string
+	Line       int
+}
+
+func (h *Handler) handleEndpointImport(w http.ResponseWriter, r *http.Request) {
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+
+	raw, err := readImportCSV(r)
+	if err != nil {
+		h.audit(actorString(p), "endpoint_import", "", "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", err.Error())
+		return
+	}
+
+	prepared, errs := h.validateImportCSV(r.Context(), raw)
+	if len(errs) > 0 {
+		h.audit(actorString(p), "endpoint_import", "", "bad_request", ip)
+		writeCSVValidationError(w, errs)
+		return
+	}
+
+	rows := make([]endpointInsert, len(prepared))
+	items := make([]map[string]any, 0, len(prepared))
+	for i, pRow := range prepared {
+		rows[i] = pRow.Insert
+		items = append(items, map[string]any{
+			"id":             pRow.Insert.ID,
+			"login_password": pRow.PlainLogin,
+			"name":           pRow.Name,
+		})
+	}
+	if err := h.insertEndpointsBatch(r.Context(), rows); err != nil {
+		h.audit(actorString(p), "endpoint_import", "", "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	h.audit(actorString(p), "endpoint_import", strconv.Itoa(len(items)), "ok", ip)
+	httpx.WriteOK(w, map[string]any{"items": items})
+}
+
+func readImportCSV(r *http.Request) ([]byte, error) {
+	ct := r.Header.Get("Content-Type")
+	mediaType, params, err := mime.ParseMediaType(ct)
+	if err != nil {
+		mediaType = strings.TrimSpace(strings.Split(ct, ";")[0])
+	}
+	switch {
+	case strings.HasPrefix(mediaType, "multipart/"):
+		boundary := params["boundary"]
+		if boundary == "" {
+			return nil, errBadRequest("multipart 缺少 boundary")
+		}
+		mr := multipart.NewReader(r.Body, boundary)
+		for {
+			part, pErr := mr.NextPart()
+			if pErr == io.EOF {
+				break
+			}
+			if pErr != nil {
+				return nil, errBadRequest("读取 multipart 失败")
+			}
+			name := part.FormName()
+			if name == "file" || name == "" {
+				b, readErr := io.ReadAll(io.LimitReader(part, 8<<20))
+				_ = part.Close()
+				if readErr != nil {
+					return nil, errBadRequest("读取文件失败")
+				}
+				return b, nil
+			}
+			_ = part.Close()
+		}
+		return nil, errBadRequest("缺少 file 字段")
+	default:
+		// text/csv 或未标明时按原始体
+		b, readErr := io.ReadAll(io.LimitReader(r.Body, 8<<20))
+		if readErr != nil {
+			return nil, errBadRequest("读取 CSV 失败")
+		}
+		return b, nil
+	}
+}
+
+type badRequestError string
+
+func (e badRequestError) Error() string { return string(e) }
+
+func errBadRequest(msg string) error { return badRequestError(msg) }
+
+func (h *Handler) validateImportCSV(ctx context.Context, raw []byte) ([]importPrepared, []csvLineError) {
+	raw = bytes.TrimPrefix(raw, []byte{0xEF, 0xBB, 0xBF})
+	reader := csv.NewReader(bytes.NewReader(raw))
+	reader.FieldsPerRecord = -1
+	reader.TrimLeadingSpace = true
+
+	records, err := reader.ReadAll()
+	if err != nil {
+		return nil, []csvLineError{{Line: 1, Reason: "CSV 解析失败"}}
+	}
+	if len(records) < 1 {
+		return nil, []csvLineError{{Line: 1, Reason: "表头不正确"}}
+	}
+
+	header := normalizeCSVHeader(records[0])
+	expected := []string{"id", "name", "login_password", "talk_password", "default_delay_seconds", "remark"}
+	if len(header) < len(expected) {
+		return nil, []csvLineError{{Line: 1, Reason: "表头不正确"}}
+	}
+	for i, want := range expected {
+		if header[i] != want {
+			return nil, []csvLineError{{Line: 1, Reason: "表头不正确"}}
+		}
+	}
+
+	dataRows := records[1:]
+	if len(dataRows) == 0 {
+		return nil, []csvLineError{{Line: 2, Reason: "没有数据行"}}
+	}
+	if len(dataRows) > maxImportRows {
+		return nil, []csvLineError{{Line: 1, Reason: "最多 1000 行"}}
+	}
+
+	errs := make([]csvLineError, 0)
+	prepared := make([]importPrepared, 0, len(dataRows))
+	seen := make(map[string]int) // id -> first line
+	checkIDs := make([]string, 0, len(dataRows))
+
+	type pending struct {
+		line              int
+		id                string
+		name              string
+		remark            string
+		loginPW           string
+		talkPW            string
+		delaySec          int64
+		needGenerateID    bool
+		needGenerateLogin bool
+	}
+	pendings := make([]pending, 0, len(dataRows))
+
+	for i, cols := range dataRows {
+		line := i + 2 // 表头为 1
+		for len(cols) < 6 {
+			cols = append(cols, "")
+		}
+		id := strings.TrimSpace(cols[0])
+		name := cols[1]
+		loginPW := cols[2]
+		talkPW := cols[3]
+		delayRaw := strings.TrimSpace(cols[4])
+		remark := cols[5]
+
+		delaySec := int64(0)
+		if delayRaw != "" {
+			n, parseErr := strconv.ParseInt(delayRaw, 10, 64)
+			if parseErr != nil || n < 0 {
+				errs = append(errs, csvLineError{Line: line, Reason: "默认延迟无效"})
+				continue
+			}
+			delaySec = n
+		}
+		if msg := validateEndpointFields(id, name, remark, loginPW, talkPW, delaySec); msg != "" {
+			errs = append(errs, csvLineError{Line: line, Reason: msg})
+			continue
+		}
+		if utf8.RuneCountInString(name) > protocol.MaxNameChars {
+			errs = append(errs, csvLineError{Line: line, Reason: "名称不合法"})
+			continue
+		}
+
+		p := pending{
+			line:              line,
+			id:                id,
+			name:              name,
+			remark:            remark,
+			loginPW:           loginPW,
+			talkPW:            talkPW,
+			delaySec:          delaySec,
+			needGenerateID:    id == "",
+			needGenerateLogin: loginPW == "",
+		}
+		if !p.needGenerateID {
+			if first, ok := seen[id]; ok {
+				errs = append(errs, csvLineError{Line: line, Reason: "编号与第 " + strconv.Itoa(first) + " 行重复"})
+				continue
+			}
+			seen[id] = line
+			checkIDs = append(checkIDs, id)
+		}
+		pendings = append(pendings, p)
+	}
+
+	if len(errs) > 0 {
+		return nil, errs
+	}
+
+	existing, err := h.existingEndpointIDs(ctx, checkIDs)
+	if err != nil {
+		return nil, []csvLineError{{Line: 1, Reason: "校验失败"}}
+	}
+	for _, p := range pendings {
+		if p.needGenerateID {
+			continue
+		}
+		if _, ok := existing[p.id]; ok {
+			errs = append(errs, csvLineError{Line: p.line, Reason: "编号已占用"})
+		}
+	}
+	if len(errs) > 0 {
+		return nil, errs
+	}
+
+	for _, p := range pendings {
+		useID := p.id
+		if p.needGenerateID {
+			for attempt := 0; attempt < 16; attempt++ {
+				genID, genErr := generateEndpointID()
+				if genErr != nil {
+					return nil, []csvLineError{{Line: p.line, Reason: "生成编号失败"}}
+				}
+				if _, clash := seen[genID]; clash {
+					continue
+				}
+				if _, clash := existing[genID]; clash {
+					continue
+				}
+				useID = genID
+				seen[useID] = p.line
+				break
+			}
+			if useID == "" {
+				return nil, []csvLineError{{Line: p.line, Reason: "生成编号失败"}}
+			}
+		}
+
+		loginPW := p.loginPW
+		if p.needGenerateLogin {
+			pw, genErr := generateLoginPassword()
+			if genErr != nil {
+				return nil, []csvLineError{{Line: p.line, Reason: "生成密码失败"}}
+			}
+			loginPW = pw
+		}
+		loginHash, hashErr := h.hash.Hash(ctx, auth.PasswordLogin, loginPW)
+		if hashErr != nil {
+			return nil, []csvLineError{{Line: p.line, Reason: "哈希失败"}}
+		}
+		var talkHash sql.NullString
+		if p.talkPW != "" {
+			th, thErr := h.hash.Hash(ctx, auth.PasswordTalk, p.talkPW)
+			if thErr != nil {
+				return nil, []csvLineError{{Line: p.line, Reason: "哈希失败"}}
+			}
+			talkHash = sql.NullString{String: th, Valid: true}
+		}
+		prepared = append(prepared, importPrepared{
+			Insert: endpointInsert{
+				ID:             useID,
+				Name:           p.name,
+				Remark:         p.remark,
+				Source:         sourceAdmin,
+				LoginHash:      loginHash,
+				TalkHash:       talkHash,
+				DefaultDelayMs: p.delaySec * 1000,
+			},
+			PlainLogin: loginPW,
+			Name:       p.name,
+			Line:       p.line,
+		})
+	}
+	return prepared, nil
+}
+
+func normalizeCSVHeader(cols []string) []string {
+	out := make([]string, len(cols))
+	for i, c := range cols {
+		out[i] = strings.TrimSpace(strings.TrimPrefix(c, "\ufeff"))
+	}
+	return out
+}

internal/admin/endpoints_db.go

@@ -0,0 +1,413 @@
+package admin
+
+import (
+	"context"
+	"crypto/rand"
+	"database/sql"
+	"errors"
+	"strings"
+	"time"
+	"unicode/utf8"
+
+	"git.asio.asia/nixevol/NixMsg/internal/protocol"
+)
+
+const idAlphabet = "abcdefghijklmnopqrstuvwxyz0123456789"
+
+type endpointInsert struct {
+	ID             string
+	Name           string
+	Remark         string
+	Source         string
+	LoginHash      string
+	TalkHash       sql.NullString
+	DefaultDelayMs int64
+	CreatedAtMs    int64
+}
+
+func generateEndpointID() (string, error) {
+	b := make([]byte, 8)
+	if _, err := rand.Read(b); err != nil {
+		return "", err
+	}
+	out := make([]byte, 8)
+	for i := range b {
+		out[i] = idAlphabet[int(b[i])%len(idAlphabet)]
+	}
+	return "e_" + string(out), nil
+}
+
+func generateLoginPassword() (string, error) {
+	const alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
+	for range 8 {
+		b := make([]byte, 20)
+		if _, err := rand.Read(b); err != nil {
+			return "", err
+		}
+		out := make([]byte, 20)
+		for i := range b {
+			out[i] = alphabet[int(b[i])%len(alphabet)]
+		}
+		pw := string(out)
+		if !strings.HasPrefix(pw, protocol.SessionTokenPrefix) && utf8.RuneCountInString(pw) >= protocol.MinLoginPasswordLen {
+			return pw, nil
+		}
+	}
+	return "", errors.New("admin: generate login password failed")
+}
+
+func isUniqueConstraint(err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := strings.ToLower(err.Error())
+	return strings.Contains(msg, "unique constraint") || strings.Contains(msg, "constraint failed")
+}
+
+func (h *Handler) insertEndpoint(ctx context.Context, in endpointInsert) (string, error) {
+	nowMs := in.CreatedAtMs
+	if nowMs == 0 {
+		nowMs = time.Now().UnixMilli()
+	}
+	const maxAttempts = 8
+	requestedID := in.ID
+	for attempt := 0; attempt < maxAttempts; attempt++ {
+		useID := requestedID
+		if useID == "" {
+			genID, err := generateEndpointID()
+			if err != nil {
+				return "", err
+			}
+			useID = genID
+		}
+		err := h.db.Queue.Do(ctx, func(tx *sql.Tx) error {
+			_, execErr := tx.ExecContext(ctx, `
+INSERT INTO endpoints(
+  id, name, remark, source, login_hash, talk_hash, talk_version,
+  default_delay_ms, enabled, created_at
+) VALUES (?, ?, ?, ?, ?, ?, 0, ?, 1, ?)`,
+				useID, in.Name, in.Remark, in.Source, in.LoginHash, in.TalkHash, in.DefaultDelayMs, nowMs,
+			)
+			return execErr
+		})
+		if err == nil {
+			return useID, nil
+		}
+		if isUniqueConstraint(err) {
+			if requestedID != "" {
+				return "", err
+			}
+			continue
+		}
+		return "", err
+	}
+	return "", errors.New("admin: generate endpoint id exhausted")
+}
+
+func (h *Handler) insertEndpointsBatch(ctx context.Context, rows []endpointInsert) error {
+	nowMs := time.Now().UnixMilli()
+	return h.db.Queue.Do(ctx, func(tx *sql.Tx) error {
+		for i := range rows {
+			in := rows[i]
+			created := in.CreatedAtMs
+			if created == 0 {
+				created = nowMs
+			}
+			if _, err := tx.ExecContext(ctx, `
+INSERT INTO endpoints(
+  id, name, remark, source, login_hash, talk_hash, talk_version,
+  default_delay_ms, enabled, created_at
+) VALUES (?, ?, ?, ?, ?, ?, 0, ?, 1, ?)`,
+				in.ID, in.Name, in.Remark, in.Source, in.LoginHash, in.TalkHash, in.DefaultDelayMs, created,
+			); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}
+
+func onlineSQLExpr() string {
+	return `(online_since IS NOT NULL AND (offline_since IS NULL OR online_since > offline_since))`
+}
+
+func (h *Handler) buildEndpointFilter(source string, online, enabled *bool, query string) (where string, args []any) {
+	conds := []string{"1=1"}
+	if source != "" {
+		conds = append(conds, "source = ?")
+		args = append(args, source)
+	}
+	if enabled != nil {
+		if *enabled {
+			conds = append(conds, "enabled = 1")
+		} else {
+			conds = append(conds, "enabled = 0")
+		}
+	}
+	if online != nil {
+		if *online {
+			conds = append(conds, onlineSQLExpr())
+		} else {
+			conds = append(conds, "NOT "+onlineSQLExpr())
+		}
+	}
+	if query != "" {
+		conds = append(conds, "(id LIKE ? COLLATE NOCASE OR instr(lower(name), lower(?)) > 0)")
+		args = append(args, query+"%", query)
+	}
+	return strings.Join(conds, " AND "), args
+}
+
+func (h *Handler) countEndpoints(ctx context.Context, source string, online, enabled *bool, query string) (int, error) {
+	where, args := h.buildEndpointFilter(source, online, enabled, query)
+	var n int
+	err := h.db.Read.QueryRowContext(ctx, `SELECT COUNT(*) FROM endpoints WHERE `+where, args...).Scan(&n)
+	return n, err
+}
+
+func (h *Handler) listEndpoints(ctx context.Context, source string, online, enabled *bool, query string, limit, offset int) ([]endpointRow, error) {
+	where, args := h.buildEndpointFilter(source, online, enabled, query)
+	args = append(args, limit, offset)
+	rows, err := h.db.Read.QueryContext(ctx, `
+SELECT id, name, remark, source, enabled, talk_hash, default_delay_ms, created_at,
+       online_since, offline_since, session_issued_at, session_used_at
+FROM endpoints
+WHERE `+where+`
+ORDER BY id ASC
+LIMIT ? OFFSET ?`, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = rows.Close() }()
+
+	out := make([]endpointRow, 0)
+	for rows.Next() {
+		var e endpointRow
+		var enabledInt int
+		if scanErr := rows.Scan(
+			&e.ID, &e.Name, &e.Remark, &e.Source, &enabledInt, &e.TalkHash, &e.DefaultDelayMs, &e.CreatedAtMs,
+			&e.OnlineSinceMs, &e.OfflineSinceMs, &e.SessionIssuedAt, &e.SessionUsedAt,
+		); scanErr != nil {
+			return nil, scanErr
+		}
+		e.Enabled = enabledInt != 0
+		out = append(out, e)
+	}
+	return out, rows.Err()
+}
+
+func (h *Handler) getEndpoint(ctx context.Context, id string) (endpointRow, error) {
+	var e endpointRow
+	var enabledInt int
+	err := h.db.Read.QueryRowContext(ctx, `
+SELECT id, name, remark, source, enabled, talk_hash, default_delay_ms, created_at,
+       online_since, offline_since, session_issued_at, session_used_at
+FROM endpoints WHERE id = ?`, id).Scan(
+		&e.ID, &e.Name, &e.Remark, &e.Source, &enabledInt, &e.TalkHash, &e.DefaultDelayMs, &e.CreatedAtMs,
+		&e.OnlineSinceMs, &e.OfflineSinceMs, &e.SessionIssuedAt, &e.SessionUsedAt,
+	)
+	if err != nil {
+		return endpointRow{}, err
+	}
+	e.Enabled = enabledInt != 0
+	return e, nil
+}
+
+func (h *Handler) endpointExists(ctx context.Context, id string) (bool, error) {
+	var n int
+	err := h.db.Read.QueryRowContext(ctx, `SELECT COUNT(1) FROM endpoints WHERE id = ?`, id).Scan(&n)
+	return n > 0, err
+}
+
+func (h *Handler) existingEndpointIDs(ctx context.Context, ids []string) (map[string]struct{}, error) {
+	out := make(map[string]struct{})
+	if len(ids) == 0 {
+		return out, nil
+	}
+	// 分批 IN，避免超长 SQL；导入最多 1000。
+	const chunk = 200
+	for i := 0; i < len(ids); i += chunk {
+		end := i + chunk
+		if end > len(ids) {
+			end = len(ids)
+		}
+		part := ids[i:end]
+		placeholders := make([]string, len(part))
+		args := make([]any, len(part))
+		for j, id := range part {
+			placeholders[j] = "?"
+			args[j] = id
+		}
+		rows, err := h.db.Read.QueryContext(ctx,
+			`SELECT id FROM endpoints WHERE id IN (`+strings.Join(placeholders, ",")+`)`, args...)
+		if err != nil {
+			return nil, err
+		}
+		for rows.Next() {
+			var id string
+			if scanErr := rows.Scan(&id); scanErr != nil {
+				_ = rows.Close()
+				return nil, scanErr
+			}
+			out[id] = struct{}{}
+		}
+		err = rows.Err()
+		_ = rows.Close()
+		if err != nil {
+			return nil, err
+		}
+	}
+	return out, nil
+}
+
+// patchEndpoint 更新字段。返回更新前是否 enabled，便于停用时踢线。
+func (h *Handler) patchEndpoint(ctx context.Context, id string, name, remark *string, delaySec *int64, enabled *bool) (wasEnabled bool, err error) {
+	err = h.db.Queue.Do(ctx, func(tx *sql.Tx) error {
+		var enabledInt int
+		if qErr := tx.QueryRowContext(ctx, `SELECT enabled FROM endpoints WHERE id = ?`, id).Scan(&enabledInt); qErr != nil {
+			return qErr
+		}
+		wasEnabled = enabledInt != 0
+		sets := make([]string, 0, 4)
+		args := make([]any, 0, 5)
+		if name != nil {
+			sets = append(sets, "name = ?")
+			args = append(args, *name)
+		}
+		if remark != nil {
+			sets = append(sets, "remark = ?")
+			args = append(args, *remark)
+		}
+		if delaySec != nil {
+			sets = append(sets, "default_delay_ms = ?")
+			args = append(args, (*delaySec)*1000)
+		}
+		if enabled != nil {
+			v := 0
+			if *enabled {
+				v = 1
+			}
+			sets = append(sets, "enabled = ?")
+			args = append(args, v)
+			if !*enabled {
+				sets = append(sets, "session_hash = NULL", "session_issued_at = NULL", "session_used_at = NULL")
+			}
+		}
+		args = append(args, id)
+		res, execErr := tx.ExecContext(ctx, `UPDATE endpoints SET `+strings.Join(sets, ", ")+` WHERE id = ?`, args...)
+		if execErr != nil {
+			return execErr
+		}
+		n, _ := res.RowsAffected()
+		if n == 0 {
+			return sql.ErrNoRows
+		}
+		return nil
+	})
+	return wasEnabled, err
+}
+
+func (h *Handler) setEndpointEnabled(ctx context.Context, id string, enabled bool) (found bool, err error) {
+	if h.identity != nil {
+		var opErr error
+		if enabled {
+			opErr = h.identity.Enable(ctx, id)
+		} else {
+			opErr = h.identity.Disable(ctx, id)
+		}
+		if opErr != nil {
+			if isEndpointNotFound(opErr) {
+				return false, nil
+			}
+			return false, opErr
+		}
+		return true, nil
+	}
+	err = h.db.Queue.Do(ctx, func(tx *sql.Tx) error {
+		v := 0
+		if enabled {
+			v = 1
+		}
+		query := `UPDATE endpoints SET enabled = ?`
+		args := []any{v}
+		if !enabled {
+			query += `, session_hash = NULL, session_issued_at = NULL, session_used_at = NULL`
+		}
+		query += ` WHERE id = ?`
+		args = append(args, id)
+		res, execErr := tx.ExecContext(ctx, query, args...)
+		if execErr != nil {
+			return execErr
+		}
+		n, _ := res.RowsAffected()
+		found = n > 0
+		return nil
+	})
+	return found, err
+}
+
+// deleteEndpointBasic 删除端；注入 Identity 时走 I5 完整级联。
+func (h *Handler) deleteEndpointBasic(ctx context.Context, id string) (found bool, err error) {
+	if h.identity != nil {
+		opErr := h.identity.Delete(ctx, id)
+		if opErr != nil {
+			if isEndpointNotFound(opErr) {
+				return false, nil
+			}
+			return false, opErr
+		}
+		return true, nil
+	}
+	err = h.db.Queue.Do(ctx, func(tx *sql.Tx) error {
+		if _, e := tx.ExecContext(ctx, `DELETE FROM talk_grants WHERE sender_id = ? OR target_id = ?`, id, id); e != nil {
+			return e
+		}
+		res, e := tx.ExecContext(ctx, `DELETE FROM endpoints WHERE id = ?`, id)
+		if e != nil {
+			return e
+		}
+		n, _ := res.RowsAffected()
+		found = n > 0
+		return nil
+	})
+	return found, err
+}
+
+func isEndpointNotFound(err error) bool {
+	var pe *protocol.Error
+	return errors.As(err, &pe) && pe.Code == protocol.CodeNotFound
+}
+
+func (h *Handler) resetLoginPassword(ctx context.Context, id, loginHash string) (bool, error) {
+	var found bool
+	err := h.db.Queue.Do(ctx, func(tx *sql.Tx) error {
+		res, e := tx.ExecContext(ctx, `
+UPDATE endpoints
+SET login_hash = ?, session_hash = NULL, session_issued_at = NULL, session_used_at = NULL
+WHERE id = ?`, loginHash, id)
+		if e != nil {
+			return e
+		}
+		n, _ := res.RowsAffected()
+		found = n > 0
+		return nil
+	})
+	return found, err
+}
+
+func (h *Handler) setTalkPassword(ctx context.Context, id string, talkHash sql.NullString) (bool, error) {
+	var found bool
+	err := h.db.Queue.Do(ctx, func(tx *sql.Tx) error {
+		res, e := tx.ExecContext(ctx, `
+UPDATE endpoints
+SET talk_hash = ?, talk_version = talk_version + 1
+WHERE id = ?`, talkHash, id)
+		if e != nil {
+			return e
+		}
+		n, _ := res.RowsAffected()
+		found = n > 0
+		return nil
+	})
+	return found, err
+}

internal/admin/endpoints_test.go

@@ -0,0 +1,364 @@
+package admin_test
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/cookiejar"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+
+	"git.asio.asia/nixevol/NixMsg/internal/admin"
+	"git.asio.asia/nixevol/NixMsg/internal/auth"
+	"git.asio.asia/nixevol/NixMsg/internal/store"
+)
+
+type kickRecorder struct {
+	mu    sync.Mutex
+	Calls []string
+}
+
+func (k *kickRecorder) Kick(_ context.Context, endpointID string) (bool, error) {
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	k.Calls = append(k.Calls, endpointID)
+	return true, nil
+}
+
+func (k *kickRecorder) count() int {
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	return len(k.Calls)
+}
+
+func setupEndpoints(t *testing.T) (*store.DB, *httptest.Server, *http.Client, *kickRecorder, *admin.MemoryLoginLocks) {
+	t.Helper()
+	dir := t.TempDir()
+	db, err := store.Open(filepath.Join(dir, "data"), "FULL")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+
+	hash := auth.NewStubHashPool()
+	if seedErr := admin.SeedAdminPassword(context.Background(), db, hash, testPassword); seedErr != nil {
+		t.Fatal(seedErr)
+	}
+	kick := &kickRecorder{}
+	locks := admin.NewMemoryLoginLocks()
+	h := admin.New(admin.Deps{
+		DB:           db,
+		Hash:         hash,
+		Tokens:       admin.NewRandomAPITokens(),
+		Locks:        locks,
+		KickEndpoint: kick.Kick,
+	})
+	srv := httptest.NewServer(h)
+	t.Cleanup(srv.Close)
+
+	jar, err := cookiejar.New(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	client := &http.Client{Jar: jar}
+	res := postJSON(t, client, srv.URL+"/api/admin/login",
+		`{"username":"admin","password":"`+testPassword+`"}`, nil)
+	env := decodeEnv(t, res)
+	if res.StatusCode != http.StatusOK || !env.OK {
+		t.Fatalf("login failed: %d %+v", res.StatusCode, env)
+	}
+	return db, srv, client, kick, locks
+}
+
+func TestEndpointImportDuplicateRejectsAll(t *testing.T) {
+	db, srv, client, _, _ := setupEndpoints(t)
+	base := srv.URL
+
+	csvBody := "" +
+		"id,name,login_password,talk_password,default_delay_seconds,remark\n" +
+		"ep-a,甲,password1,,0,\n" +
+		"ep-b,乙,password2,,0,\n" +
+		"ep-a,丙,password3,,0,\n"
+
+	req, err := http.NewRequest(http.MethodPost, base+"/api/admin/endpoints/import", strings.NewReader(csvBody))
+	if err != nil {
+		t.Fatal(err)
+	}
+	req.Header.Set("Content-Type", "text/csv")
+	req.Header.Set("X-Nixmsg-Request", "1")
+	res, err := client.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	raw, _ := io.ReadAll(res.Body)
+	_ = res.Body.Close()
+	if res.StatusCode != http.StatusBadRequest {
+		t.Fatalf("want 400 got %d body=%s", res.StatusCode, raw)
+	}
+	var env struct {
+		OK    bool `json:"ok"`
+		Error *struct {
+			Code    string `json:"code"`
+			Message string `json:"message"`
+		} `json:"error"`
+		Data *struct {
+			Errors []struct {
+				Line   int    `json:"line"`
+				Reason string `json:"reason"`
+			} `json:"errors"`
+		} `json:"data"`
+	}
+	if err := json.Unmarshal(raw, &env); err != nil {
+		t.Fatal(err)
+	}
+	if env.OK || env.Error == nil || env.Error.Code != "bad_request" {
+		t.Fatalf("env=%+v", env)
+	}
+	if env.Data == nil || len(env.Data.Errors) == 0 {
+		t.Fatalf("missing errors: %s", raw)
+	}
+	foundLine := false
+	for _, e := range env.Data.Errors {
+		if e.Line == 4 {
+			foundLine = true
+			break
+		}
+	}
+	if !foundLine {
+		t.Fatalf("want line 4 in errors: %+v", env.Data.Errors)
+	}
+
+	var n int
+	if err := db.Read.QueryRow(`SELECT COUNT(*) FROM endpoints`).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != 0 {
+		t.Fatalf("want 0 endpoints created, got %d", n)
+	}
+}
+
+func TestEndpointResetPasswordAppearsOnce(t *testing.T) {
+	db, srv, client, kick, _ := setupEndpoints(t)
+	base := srv.URL
+
+	res := postJSON(t, client, base+"/api/admin/endpoints",
+		`{"id":"dev-1","name":"门口","login_password":"oldpass12"}`,
+		csrfHeaders())
+	env := decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("create: %d %+v", res.StatusCode, env)
+	}
+
+	// 写入假会话令牌，确认重置会清空
+	err := db.Queue.Do(context.Background(), func(tx *sql.Tx) error {
+		_, e := tx.Exec(`UPDATE endpoints SET session_hash='abc', session_issued_at=1, session_used_at=1 WHERE id='dev-1'`)
+		return e
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	res = postJSON(t, client, base+"/api/admin/endpoints/dev-1/reset-login-password",
+		`{}`, csrfHeaders())
+	body, _ := io.ReadAll(res.Body)
+	_ = res.Body.Close()
+	if res.StatusCode != 200 {
+		t.Fatalf("reset status=%d body=%s", res.StatusCode, body)
+	}
+	var resetEnv struct {
+		OK   bool `json:"ok"`
+		Data struct {
+			LoginPassword string `json:"login_password"`
+		} `json:"data"`
+	}
+	if err := json.Unmarshal(body, &resetEnv); err != nil {
+		t.Fatal(err)
+	}
+	if !resetEnv.OK || resetEnv.Data.LoginPassword == "" {
+		t.Fatalf("want one-time password, got %s", body)
+	}
+	pw := resetEnv.Data.LoginPassword
+	if strings.Count(string(body), pw) != 1 {
+		t.Fatalf("password should appear exactly once in response: %s", body)
+	}
+	if strings.HasPrefix(pw, "nst_") {
+		t.Fatalf("password must not start with nst_: %q", pw)
+	}
+
+	var session sql.NullString
+	if err := db.Read.QueryRow(`SELECT session_hash FROM endpoints WHERE id='dev-1'`).Scan(&session); err != nil {
+		t.Fatal(err)
+	}
+	if session.Valid {
+		t.Fatal("session_hash should be cleared")
+	}
+	if kick.count() < 1 {
+		t.Fatal("expected kick hook after reset")
+	}
+
+	// 详情中不得再出现明文密码
+	res = doReq(t, client, http.MethodGet, base+"/api/admin/endpoints/dev-1", "", nil)
+	detailBody, _ := io.ReadAll(res.Body)
+	_ = res.Body.Close()
+	if strings.Contains(string(detailBody), pw) {
+		t.Fatalf("password leaked in detail: %s", detailBody)
+	}
+}
+
+func TestEndpointMutatingRequiresCSRF(t *testing.T) {
+	_, srv, client, _, _ := setupEndpoints(t)
+	base := srv.URL
+
+	res := postJSON(t, client, base+"/api/admin/endpoints",
+		`{"id":"no-csrf","name":"x","login_password":"password1"}`,
+		nil) // 无 CSRF
+	env := decodeEnv(t, res)
+	if res.StatusCode != http.StatusForbidden || env.Error == nil || env.Error.Code != "forbidden" {
+		t.Fatalf("want 403 forbidden got %d %+v", res.StatusCode, env)
+	}
+}
+
+func TestEndpointDisableKickAndUnlock(t *testing.T) {
+	db, srv, client, kick, locks := setupEndpoints(t)
+	base := srv.URL
+
+	res := postJSON(t, client, base+"/api/admin/endpoints",
+		`{"id":"lock-1","name":"锁","login_password":"password1"}`,
+		csrfHeaders())
+	env := decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("create: %d %+v", res.StatusCode, env)
+	}
+
+	res = postJSON(t, client, base+"/api/admin/endpoints/batch",
+		`{"ids":["lock-1"],"action":"disable"}`, csrfHeaders())
+	env = decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("disable: %d %+v", res.StatusCode, env)
+	}
+	var enabled int
+	if err := db.Read.QueryRow(`SELECT enabled FROM endpoints WHERE id='lock-1'`).Scan(&enabled); err != nil {
+		t.Fatal(err)
+	}
+	if enabled != 0 {
+		t.Fatalf("want enabled=0 got %d", enabled)
+	}
+	if kick.count() < 1 {
+		t.Fatal("disable should kick")
+	}
+
+	for i := 0; i < 50; i++ {
+		locks.Fail(auth.LockKey{Kind: auth.LockLoginEndpoint, EndpointID: "lock-1"})
+	}
+	if locked, _ := locks.Check(auth.LockKey{Kind: auth.LockLoginEndpoint, EndpointID: "lock-1"}); !locked {
+		t.Fatal("expected endpoint locked before unlock")
+	}
+
+	res = postJSON(t, client, base+"/api/admin/endpoints/lock-1/unlock", `{}`, csrfHeaders())
+	env = decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("unlock: %d %+v", res.StatusCode, env)
+	}
+	if locked, _ := locks.Check(auth.LockKey{Kind: auth.LockLoginEndpoint, EndpointID: "lock-1"}); locked {
+		t.Fatal("expected unlocked")
+	}
+
+	res = doReq(t, client, http.MethodPut, base+"/api/admin/endpoints/lock-1/talk-password",
+		`{"talk_password":"talk"}`,
+		map[string]string{"X-Nixmsg-Request": "1", "Content-Type": "application/json"})
+	env = decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("talk-password: %d %+v", res.StatusCode, env)
+	}
+	var talkSet struct {
+		TalkPasswordSet bool `json:"talk_password_set"`
+	}
+	_ = json.Unmarshal(env.Data, &talkSet)
+	if !talkSet.TalkPasswordSet {
+		t.Fatal("want talk_password_set true")
+	}
+
+	var ver int
+	if err := db.Read.QueryRow(`SELECT talk_version FROM endpoints WHERE id='lock-1'`).Scan(&ver); err != nil {
+		t.Fatal(err)
+	}
+	if ver != 1 {
+		t.Fatalf("talk_version want 1 got %d", ver)
+	}
+
+	res = doReq(t, client, http.MethodPut, base+"/api/admin/endpoints/lock-1/talk-password",
+		`{"talk_password":""}`,
+		map[string]string{"X-Nixmsg-Request": "1", "Content-Type": "application/json"})
+	env = decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("clear talk: %d %+v", res.StatusCode, env)
+	}
+	if err := db.Read.QueryRow(`SELECT talk_version FROM endpoints WHERE id='lock-1'`).Scan(&ver); err != nil {
+		t.Fatal(err)
+	}
+	if ver != 2 {
+		t.Fatalf("talk_version want 2 got %d", ver)
+	}
+}
+
+func TestEndpointImportBOMAndCreate(t *testing.T) {
+	db, srv, client, _, _ := setupEndpoints(t)
+	base := srv.URL
+
+	var buf bytes.Buffer
+	buf.Write([]byte{0xEF, 0xBB, 0xBF})
+	buf.WriteString("id,name,login_password,talk_password,default_delay_seconds,remark\n")
+	buf.WriteString("bom-1,门,,secret,5,备注\n")
+
+	req, err := http.NewRequest(http.MethodPost, base+"/api/admin/endpoints/import", &buf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	req.Header.Set("Content-Type", "text/csv; charset=utf-8")
+	req.Header.Set("X-Nixmsg-Request", "1")
+	res, err := client.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	env := decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("import: %d %+v", res.StatusCode, env)
+	}
+	var data struct {
+		Items []struct {
+			ID            string `json:"id"`
+			LoginPassword string `json:"login_password"`
+			Name          string `json:"name"`
+		} `json:"items"`
+	}
+	if err := json.Unmarshal(env.Data, &data); err != nil {
+		t.Fatal(err)
+	}
+	if len(data.Items) != 1 || data.Items[0].ID != "bom-1" || data.Items[0].LoginPassword == "" {
+		t.Fatalf("items=%+v", data.Items)
+	}
+
+	var n int
+	if err := db.Read.QueryRow(`SELECT COUNT(*) FROM endpoints WHERE id='bom-1'`).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != 1 {
+		t.Fatalf("want 1 row got %d", n)
+	}
+
+	res = doReq(t, client, http.MethodGet, base+"/api/admin/endpoints?source=admin", "", nil)
+	env = decodeEnv(t, res)
+	if res.StatusCode != 200 || !env.OK {
+		t.Fatalf("list: %d %+v", res.StatusCode, env)
+	}
+}
+
+func csrfHeaders() map[string]string {
+	return map[string]string{"X-Nixmsg-Request": "1"}
+}

internal/admin/groups.go

@@ -0,0 +1,472 @@
+package admin
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"net/http"
+	"strconv"
+	"strings"
+
+	"git.asio.asia/nixevol/NixMsg/internal/app/group"
+	"git.asio.asia/nixevol/NixMsg/internal/httpx"
+	"git.asio.asia/nixevol/NixMsg/internal/protocol"
+)
+
+func (h *Handler) requireGroups(w http.ResponseWriter) bool {
+	if h.groups == nil {
+		httpx.WriteError(w, http.StatusServiceUnavailable, "busy", "群服务未注入")
+		return false
+	}
+	return true
+}
+
+func (h *Handler) handleGroupList(w http.ResponseWriter, r *http.Request) {
+	q := r.URL.Query()
+	limit, offset, ok := parsePage(w, q.Get("limit"), q.Get("cursor"))
+	if !ok {
+		return
+	}
+	query := strings.TrimSpace(q.Get("query"))
+
+	where := "1=1"
+	args := make([]any, 0, 4)
+	if query != "" {
+		where += " AND LOWER(g.name) LIKE ?"
+		args = append(args, "%"+strings.ToLower(query)+"%")
+	}
+
+	var total int
+	countSQL := `SELECT COUNT(*) FROM groups g WHERE ` + where
+	if err := h.db.Read.QueryRowContext(r.Context(), countSQL, args...).Scan(&total); err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+
+	listArgs := append(append([]any{}, args...), limit, offset)
+	rows, err := h.db.Read.QueryContext(r.Context(), `
+SELECT g.id, g.name, g.owner_id, g.created_at,
+  (SELECT COUNT(*) FROM group_members gm WHERE gm.group_id = g.id) AS cnt
+FROM groups g
+WHERE `+where+`
+ORDER BY g.id ASC
+LIMIT ? OFFSET ?`, listArgs...)
+	if err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	defer func() { _ = rows.Close() }()
+
+	items := make([]map[string]any, 0)
+	for rows.Next() {
+		var id, name, owner string
+		var created int64
+		var cnt int
+		if scanErr := rows.Scan(&id, &name, &owner, &created, &cnt); scanErr != nil {
+			httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+			return
+		}
+		items = append(items, map[string]any{
+			"id":            id,
+			"name":          name,
+			"owner_id":      owner,
+			"member_count":  cnt,
+			"created_at_ms": created,
+		})
+	}
+	if err := rows.Err(); err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+
+	next := ""
+	if offset+len(items) < total {
+		next = strconv.Itoa(offset + len(items))
+	}
+	httpx.WriteOK(w, map[string]any{"items": items, "next_cursor": next, "total": total})
+}
+
+func (h *Handler) handleGroupCreate(w http.ResponseWriter, r *http.Request) {
+	if !h.requireGroups(w) {
+		return
+	}
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+
+	var req struct {
+		ID        string   `json:"id"`
+		Name      string   `json:"name"`
+		OwnerID   string   `json:"owner_id"`
+		MemberIDs []string `json:"member_ids"`
+	}
+	if err := httpx.DecodeJSON(r, &req); err != nil {
+		h.audit(actorString(p), "group_create", "", "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "请求体无效")
+		return
+	}
+	if strings.TrimSpace(req.ID) != "" {
+		// AdminCreate 不接受自定义 id；与契约「留空则生成」一致时忽略非空会误导，故拒绝。
+		h.audit(actorString(p), "group_create", req.ID, "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "后台创建群请留空 id，由服务器生成")
+		return
+	}
+	if !protocol.ValidName(req.Name) || req.Name == "" {
+		h.audit(actorString(p), "group_create", "", "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "名称不合法")
+		return
+	}
+	if req.OwnerID == "" {
+		h.audit(actorString(p), "group_create", "", "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "缺少 owner_id")
+		return
+	}
+
+	res, err := h.groups.AdminCreate(r.Context(), req.Name, req.OwnerID, req.MemberIDs)
+	if err != nil {
+		h.writeGroupErr(w, p, "group_create", req.OwnerID, ip, err)
+		return
+	}
+	h.audit(actorString(p), "group_create", res.ID, "ok", ip)
+	failed := res.Failed
+	if failed == nil {
+		failed = []group.MemberFail{}
+	}
+	httpx.WriteOK(w, map[string]any{
+		"id":       res.ID,
+		"name":     res.Name,
+		"owner_id": res.OwnerID,
+		"failed":   failed,
+	})
+}
+
+func (h *Handler) handleGroupGet(w http.ResponseWriter, r *http.Request) {
+	id := r.PathValue("id")
+	q := r.URL.Query()
+	limit, offset, ok := parsePage(w, q.Get("limit"), q.Get("cursor"))
+	if !ok {
+		return
+	}
+
+	var name, owner string
+	var created int64
+	err := h.db.Read.QueryRowContext(r.Context(),
+		`SELECT name, owner_id, created_at FROM groups WHERE id = ?`, id,
+	).Scan(&name, &owner, &created)
+	if isNoRows(err) {
+		httpx.WriteError(w, http.StatusNotFound, "not_found", "群不存在")
+		return
+	}
+	if err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+
+	rows, err := h.db.Read.QueryContext(r.Context(), `
+SELECT gm.endpoint_id, COALESCE(e.name, ''), gm.joined_at,
+  e.online_since, e.offline_since
+FROM group_members gm
+LEFT JOIN endpoints e ON e.id = gm.endpoint_id
+WHERE gm.group_id = ?
+ORDER BY gm.endpoint_id ASC
+LIMIT ? OFFSET ?`, id, limit, offset)
+	if err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	defer func() { _ = rows.Close() }()
+
+	members := make([]map[string]any, 0)
+	for rows.Next() {
+		var eid, ename string
+		var joined int64
+		var onlineSince, offlineSince sql.NullInt64
+		if scanErr := rows.Scan(&eid, &ename, &joined, &onlineSince, &offlineSince); scanErr != nil {
+			httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+			return
+		}
+		online, _, _ := presenceFields(onlineSince, offlineSince)
+		members = append(members, map[string]any{
+			"id":           eid,
+			"name":         ename,
+			"online":       online,
+			"joined_at_ms": joined,
+		})
+	}
+	if err := rows.Err(); err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+
+	var memberTotal int
+	_ = h.db.Read.QueryRowContext(r.Context(),
+		`SELECT COUNT(*) FROM group_members WHERE group_id = ?`, id).Scan(&memberTotal)
+	next := ""
+	if offset+len(members) < memberTotal {
+		next = strconv.Itoa(offset + len(members))
+	}
+	httpx.WriteOK(w, map[string]any{
+		"id":            id,
+		"name":          name,
+		"owner_id":      owner,
+		"created_at_ms": created,
+		"members":       members,
+		"next_cursor":   next,
+	})
+}
+
+func (h *Handler) handleGroupRename(w http.ResponseWriter, r *http.Request) {
+	if !h.requireGroups(w) {
+		return
+	}
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+	id := r.PathValue("id")
+
+	var req struct {
+		Name string `json:"name"`
+	}
+	if err := httpx.DecodeJSON(r, &req); err != nil {
+		h.audit(actorString(p), "group_rename", id, "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "请求体无效")
+		return
+	}
+	owner, err := h.groupOwner(r.Context(), id)
+	if isNoRows(err) {
+		h.audit(actorString(p), "group_rename", id, "not_found", ip)
+		httpx.WriteError(w, http.StatusNotFound, "not_found", "群不存在")
+		return
+	}
+	if err != nil {
+		h.audit(actorString(p), "group_rename", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	err = h.groups.Rename(r.Context(), owner, &protocol.GroupRename{
+		V: protocol.Version, Type: protocol.TypeGroupRename, RID: "admin",
+		GroupID: id, Name: req.Name,
+	})
+	if err != nil {
+		h.writeGroupErr(w, p, "group_rename", id, ip, err)
+		return
+	}
+	summary, err := h.groupSummary(r.Context(), id)
+	if err != nil {
+		h.audit(actorString(p), "group_rename", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	h.audit(actorString(p), "group_rename", id, "ok", ip)
+	httpx.WriteOK(w, summary)
+}
+
+func (h *Handler) handleGroupDissolve(w http.ResponseWriter, r *http.Request) {
+	if !h.requireGroups(w) {
+		return
+	}
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+	id := r.PathValue("id")
+
+	owner, err := h.groupOwner(r.Context(), id)
+	if isNoRows(err) {
+		h.audit(actorString(p), "group_dissolve", id, "not_found", ip)
+		httpx.WriteError(w, http.StatusNotFound, "not_found", "群不存在")
+		return
+	}
+	if err != nil {
+		h.audit(actorString(p), "group_dissolve", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	err = h.groups.Dissolve(r.Context(), owner, &protocol.GroupDissolve{
+		V: protocol.Version, Type: protocol.TypeGroupDissolve, RID: "admin", GroupID: id,
+	})
+	if err != nil {
+		h.writeGroupErr(w, p, "group_dissolve", id, ip, err)
+		return
+	}
+	h.audit(actorString(p), "group_dissolve", id, "ok", ip)
+	httpx.WriteOK(w, map[string]any{})
+}
+
+func (h *Handler) handleGroupAddMembers(w http.ResponseWriter, r *http.Request) {
+	if !h.requireGroups(w) {
+		return
+	}
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+	id := r.PathValue("id")
+
+	var req struct {
+		MemberIDs []string `json:"member_ids"`
+	}
+	if err := httpx.DecodeJSON(r, &req); err != nil {
+		h.audit(actorString(p), "group_add_members", id, "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "请求体无效")
+		return
+	}
+	if _, err := h.groupOwner(r.Context(), id); isNoRows(err) {
+		h.audit(actorString(p), "group_add_members", id, "not_found", ip)
+		httpx.WriteError(w, http.StatusNotFound, "not_found", "群不存在")
+		return
+	} else if err != nil {
+		h.audit(actorString(p), "group_add_members", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+
+	res, err := h.groups.AdminAddMembers(r.Context(), id, req.MemberIDs)
+	if err != nil {
+		h.writeGroupErr(w, p, "group_add_members", id, ip, err)
+		return
+	}
+	failed := res.Failed
+	if failed == nil {
+		failed = []group.MemberFail{}
+	}
+	h.audit(actorString(p), "group_add_members", id, "ok", ip)
+	httpx.WriteOK(w, map[string]any{"failed": failed})
+}
+
+func (h *Handler) handleGroupRemoveMember(w http.ResponseWriter, r *http.Request) {
+	if !h.requireGroups(w) {
+		return
+	}
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+	id := r.PathValue("id")
+	endpointID := r.PathValue("endpointId")
+
+	owner, err := h.groupOwner(r.Context(), id)
+	if isNoRows(err) {
+		h.audit(actorString(p), "group_remove_member", id, "not_found", ip)
+		httpx.WriteError(w, http.StatusNotFound, "not_found", "群不存在")
+		return
+	}
+	if err != nil {
+		h.audit(actorString(p), "group_remove_member", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	err = h.groups.Remove(r.Context(), owner, &protocol.GroupRemove{
+		V: protocol.Version, Type: protocol.TypeGroupRemove, RID: "admin",
+		GroupID: id, EndpointID: endpointID,
+	})
+	if err != nil {
+		h.writeGroupErr(w, p, "group_remove_member", id, ip, err)
+		return
+	}
+	h.audit(actorString(p), "group_remove_member", id+"/"+endpointID, "ok", ip)
+	httpx.WriteOK(w, map[string]any{})
+}
+
+func (h *Handler) handleGroupTransfer(w http.ResponseWriter, r *http.Request) {
+	if !h.requireGroups(w) {
+		return
+	}
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+	id := r.PathValue("id")
+
+	var req struct {
+		EndpointID string `json:"endpoint_id"`
+	}
+	if err := httpx.DecodeJSON(r, &req); err != nil {
+		h.audit(actorString(p), "group_transfer", id, "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "请求体无效")
+		return
+	}
+	owner, err := h.groupOwner(r.Context(), id)
+	if isNoRows(err) {
+		h.audit(actorString(p), "group_transfer", id, "not_found", ip)
+		httpx.WriteError(w, http.StatusNotFound, "not_found", "群不存在")
+		return
+	}
+	if err != nil {
+		h.audit(actorString(p), "group_transfer", id, "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	err = h.groups.Transfer(r.Context(), owner, &protocol.GroupTransfer{
+		V: protocol.Version, Type: protocol.TypeGroupTransfer, RID: "admin",
+		GroupID: id, EndpointID: req.EndpointID,
+	})
+	if err != nil {
+		h.writeGroupErr(w, p, "group_transfer", id, ip, err)
+		return
+	}
+	h.audit(actorString(p), "group_transfer", id, "ok", ip)
+	httpx.WriteOK(w, map[string]any{"owner_id": req.EndpointID})
+}
+
+func (h *Handler) groupOwner(ctx context.Context, groupID string) (string, error) {
+	var owner string
+	err := h.db.Read.QueryRowContext(ctx, `SELECT owner_id FROM groups WHERE id = ?`, groupID).Scan(&owner)
+	return owner, err
+}
+
+func (h *Handler) groupSummary(ctx context.Context, groupID string) (map[string]any, error) {
+	var name, owner string
+	var created int64
+	var cnt int
+	err := h.db.Read.QueryRowContext(ctx, `
+SELECT g.name, g.owner_id, g.created_at,
+  (SELECT COUNT(*) FROM group_members gm WHERE gm.group_id = g.id)
+FROM groups g WHERE g.id = ?`, groupID).Scan(&name, &owner, &created, &cnt)
+	if err != nil {
+		return nil, err
+	}
+	return map[string]any{
+		"id":            groupID,
+		"name":          name,
+		"owner_id":      owner,
+		"member_count":  cnt,
+		"created_at_ms": created,
+	}, nil
+}
+
+func (h *Handler) writeGroupErr(w http.ResponseWriter, p principal, action, object, ip string, err error) {
+	var pe *protocol.Error
+	if errors.As(err, &pe) {
+		status := http.StatusBadRequest
+		switch pe.Code {
+		case protocol.CodeNotFound, protocol.CodeInvalidTarget:
+			status = http.StatusNotFound
+		case protocol.CodeForbidden, protocol.CodeNotMember, protocol.CodeOwnerCannotLeave:
+			status = http.StatusForbidden
+		case protocol.CodeIDTaken:
+			status = http.StatusConflict
+		case protocol.CodeBusy:
+			status = http.StatusServiceUnavailable
+		}
+		h.audit(actorString(p), action, object, pe.Code, ip)
+		httpx.WriteError(w, status, pe.Code, pe.Message)
+		return
+	}
+	h.audit(actorString(p), action, object, "error", ip)
+	httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+}
+
+func parsePage(w http.ResponseWriter, limitStr, cursorStr string) (limit, offset int, ok bool) {
+	limit = defaultListLimit
+	if limitStr != "" {
+		n, err := strconv.Atoi(limitStr)
+		if err != nil || n < 1 {
+			httpx.WriteError(w, http.StatusBadRequest, "bad_request", "limit 无效")
+			return 0, 0, false
+		}
+		limit = n
+	}
+	if limit > maxListLimit {
+		limit = maxListLimit
+	}
+	if cursorStr != "" {
+		n, err := strconv.Atoi(cursorStr)
+		if err != nil || n < 0 {
+			httpx.WriteError(w, http.StatusBadRequest, "bad_request", "cursor 无效")
+			return 0, 0, false
+		}
+		offset = n
+	}
+	return limit, offset, true
+}

internal/admin/handler.go

@@ -7,7 +7,10 @@ import (
 	"sync"
 	"time"
 
+	"git.asio.asia/nixevol/NixMsg/internal/app/group"
+	"git.asio.asia/nixevol/NixMsg/internal/app/identity"
 	"git.asio.asia/nixevol/NixMsg/internal/auth"
+	"git.asio.asia/nixevol/NixMsg/internal/config"
 	"git.asio.asia/nixevol/NixMsg/internal/store"
 )
 
@@ -36,18 +39,35 @@ type Deps struct {
 	SessionTTL time.Duration
 	// SecureCookies 为 true 时 Cookie 始终带 Secure；否则按请求是否 HTTPS 决定。
 	SecureCookies bool
+	// KickEndpoint 踢下线钩子（只断开连接）；nil 时踢线为 no-op。
+	KickEndpoint EndpointKickFunc
+	// Identity 端停用/启用/删除级联（I5）；nil 时回退为仅改 enabled/删行。
+	Identity identity.Service
+
+	// Groups 群服务；nil 时群写操作返回 busy。列表/详情可只读库。
+	Groups group.Service
+	// Config 只读运行参数来源；零值用 config.Default()。
+	Config config.Config
+	// Version 概览里的版本号；空则 "dev"。
+	Version string
 }
 
 // Handler 是可挂载的管理接口（路由前缀 /api/admin/）。
 type Handler struct {
-	db       *store.DB
-	hash     auth.HashPool
-	tokens   auth.APITokens
-	locks    auth.LoginLocks
-	log      *slog.Logger
-	trusted  []*net.IPNet
-	ttl      time.Duration
-	forceSec bool
+	db        *store.DB
+	hash      auth.HashPool
+	tokens    auth.APITokens
+	locks     auth.LoginLocks
+	log       *slog.Logger
+	trusted   []*net.IPNet
+	ttl       time.Duration
+	forceSec  bool
+	kick      EndpointKickFunc
+	identity  identity.Service
+	groups    group.Service
+	cfg       config.Config
+	version   string
+	startedAt time.Time
 
 	mux *http.ServeMux
 
@@ -67,17 +87,31 @@ func New(d Deps) *Handler {
 	if ttl <= 0 {
 		ttl = defaultSessionTTL
 	}
+	cfg := d.Config
+	if cfg.Listen == "" {
+		cfg = config.Default()
+	}
+	ver := d.Version
+	if ver == "" {
+		ver = "dev"
+	}
 	h := &Handler{
-		db:       d.DB,
-		hash:     d.Hash,
-		tokens:   d.Tokens,
-		locks:    d.Locks,
-		log:      d.Logger,
-		trusted:  d.TrustedProxies,
-		ttl:      ttl,
-		forceSec: d.SecureCookies,
-		mux:      http.NewServeMux(),
-		lastUsed: make(map[string]time.Time),
+		db:        d.DB,
+		hash:      d.Hash,
+		tokens:    d.Tokens,
+		locks:     d.Locks,
+		log:       d.Logger,
+		trusted:   d.TrustedProxies,
+		ttl:       ttl,
+		forceSec:  d.SecureCookies,
+		kick:      d.KickEndpoint,
+		identity:  d.Identity,
+		groups:    d.Groups,
+		cfg:       cfg,
+		version:   ver,
+		startedAt: time.Now(),
+		mux:       http.NewServeMux(),
+		lastUsed:  make(map[string]time.Time),
 	}
 	h.routes()
 	return h
@@ -103,36 +137,32 @@ func (h *Handler) routes() {
 	h.mux.Handle("PATCH /api/admin/tokens/{id}", h.auth(h.handleTokenPatch))
 	h.mux.Handle("DELETE /api/admin/tokens/{id}", h.auth(h.handleTokenDelete))
 
-	// 其余管理路由：鉴权生效，业务暂 501
-	for _, p := range stubRoutes {
-		h.mux.Handle(p, h.auth(h.handleNotImplemented))
-	}
-}
+	// A2 端管理
+	h.mux.Handle("GET /api/admin/endpoints", h.auth(h.handleEndpointList))
+	h.mux.Handle("POST /api/admin/endpoints", h.auth(h.handleEndpointCreate))
+	h.mux.Handle("POST /api/admin/endpoints/import", h.auth(h.handleEndpointImport))
+	h.mux.Handle("POST /api/admin/endpoints/batch", h.auth(h.handleEndpointBatch))
+	h.mux.Handle("GET /api/admin/endpoints/{id}", h.auth(h.handleEndpointGet))
+	h.mux.Handle("PATCH /api/admin/endpoints/{id}", h.auth(h.handleEndpointPatch))
+	h.mux.Handle("DELETE /api/admin/endpoints/{id}", h.auth(h.handleEndpointDelete))
+	h.mux.Handle("POST /api/admin/endpoints/{id}/kick", h.auth(h.handleEndpointKick))
+	h.mux.Handle("POST /api/admin/endpoints/{id}/reset-login-password", h.auth(h.handleEndpointResetLoginPassword))
+	h.mux.Handle("PUT /api/admin/endpoints/{id}/talk-password", h.auth(h.handleEndpointTalkPassword))
+	h.mux.Handle("POST /api/admin/endpoints/{id}/unlock", h.auth(h.handleEndpointUnlock))
 
-var stubRoutes = []string{
-	"GET /api/admin/overview",
-	"GET /api/admin/endpoints",
-	"POST /api/admin/endpoints",
-	"POST /api/admin/endpoints/import",
-	"POST /api/admin/endpoints/batch",
-	"GET /api/admin/endpoints/{id}",
-	"PATCH /api/admin/endpoints/{id}",
-	"DELETE /api/admin/endpoints/{id}",
-	"POST /api/admin/endpoints/{id}/kick",
-	"POST /api/admin/endpoints/{id}/reset-login-password",
-	"PUT /api/admin/endpoints/{id}/talk-password",
-	"POST /api/admin/endpoints/{id}/unlock",
-	"GET /api/admin/registration",
-	"PUT /api/admin/registration",
-	"GET /api/admin/groups",
-	"POST /api/admin/groups",
-	"GET /api/admin/groups/{id}",
-	"PATCH /api/admin/groups/{id}",
-	"DELETE /api/admin/groups/{id}",
-	"POST /api/admin/groups/{id}/members",
-	"DELETE /api/admin/groups/{id}/members/{endpointId}",
-	"POST /api/admin/groups/{id}/transfer",
-	"GET /api/admin/messages",
-	"GET /api/admin/messages/{seq}",
-	"GET /api/admin/settings",
+	// A3
+	h.mux.Handle("GET /api/admin/overview", h.auth(h.handleOverview))
+	h.mux.Handle("GET /api/admin/registration", h.auth(h.handleRegistrationGet))
+	h.mux.Handle("PUT /api/admin/registration", h.auth(h.handleRegistrationPut))
+	h.mux.Handle("GET /api/admin/groups", h.auth(h.handleGroupList))
+	h.mux.Handle("POST /api/admin/groups", h.auth(h.handleGroupCreate))
+	h.mux.Handle("GET /api/admin/groups/{id}", h.auth(h.handleGroupGet))
+	h.mux.Handle("PATCH /api/admin/groups/{id}", h.auth(h.handleGroupRename))
+	h.mux.Handle("DELETE /api/admin/groups/{id}", h.auth(h.handleGroupDissolve))
+	h.mux.Handle("POST /api/admin/groups/{id}/members", h.auth(h.handleGroupAddMembers))
+	h.mux.Handle("DELETE /api/admin/groups/{id}/members/{endpointId}", h.auth(h.handleGroupRemoveMember))
+	h.mux.Handle("POST /api/admin/groups/{id}/transfer", h.auth(h.handleGroupTransfer))
+	h.mux.Handle("GET /api/admin/messages", h.auth(h.handleMessageList))
+	h.mux.Handle("GET /api/admin/messages/{seq}", h.auth(h.handleMessageGet))
+	h.mux.Handle("GET /api/admin/settings", h.auth(h.handleSettingsGet))
 }

internal/admin/login.go

@@ -4,7 +4,6 @@ import (
 	"database/sql"
 	"errors"
 	"net/http"
-	"strings"
 
 	"git.asio.asia/nixevol/NixMsg/internal/auth"
 	"git.asio.asia/nixevol/NixMsg/internal/httpx"
@@ -157,13 +156,3 @@ func (h *Handler) handlePassword(w http.ResponseWriter, r *http.Request) {
 	h.audit(actorString(p), "password_change", "", "ok", ip)
 	httpx.WriteOK(w, map[string]any{})
 }
-
-func (h *Handler) handleNotImplemented(w http.ResponseWriter, r *http.Request) {
-	p, _ := principalFrom(r.Context())
-	ip := httpx.ClientIP(r, h.trusted)
-	action := strings.ToLower(r.Method) + " " + r.URL.Path
-	if isMutating(r.Method) {
-		h.audit(actorString(p), action, "", "not_implemented", ip)
-	}
-	httpx.WriteError(w, http.StatusNotImplemented, "not_implemented", "接口尚未实现")
-}

internal/admin/messages.go

@@ -0,0 +1,301 @@
+package admin
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"net/http"
+	"strconv"
+	"strings"
+
+	"git.asio.asia/nixevol/NixMsg/internal/httpx"
+)
+
+// 消息查询只读 messages / deliveries，不读 message_bodies，响应不含 body/正文。
+
+func (h *Handler) handleMessageList(w http.ResponseWriter, r *http.Request) {
+	q := r.URL.Query()
+	limit, offset, ok := parsePage(w, q.Get("limit"), q.Get("cursor"))
+	if !ok {
+		return
+	}
+
+	where, args, ok := buildMessageFilter(w, q)
+	if !ok {
+		return
+	}
+
+	var total int
+	if err := h.db.Read.QueryRowContext(r.Context(),
+		`SELECT COUNT(*) FROM messages m WHERE `+where, args...).Scan(&total); err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+
+	listArgs := append(append([]any{}, args...), limit, offset)
+	rows, err := h.db.Read.QueryContext(r.Context(), `
+SELECT m.seq, m.id, m.sender_id, m.dest_kind, m.dest_id, m.state, m.reason,
+  m.send_at, m.created_at, m.keep, m.receipt, m.content_type
+FROM messages m
+WHERE `+where+`
+ORDER BY m.seq DESC
+LIMIT ? OFFSET ?`, listArgs...)
+	if err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	defer func() { _ = rows.Close() }()
+
+	items := make([]map[string]any, 0)
+	seqs := make([]int64, 0)
+	for rows.Next() {
+		var seq int64
+		var id, sender, destKind, destID, state, reason, contentType string
+		var sendAt, created int64
+		var keep, receipt int
+		if scanErr := rows.Scan(&seq, &id, &sender, &destKind, &destID, &state, &reason,
+			&sendAt, &created, &keep, &receipt, &contentType); scanErr != nil {
+			httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+			return
+		}
+		seqs = append(seqs, seq)
+		items = append(items, map[string]any{
+			"seq":           seq,
+			"id":            id,
+			"sender_id":     sender,
+			"dest_kind":     destKind,
+			"dest_id":       destID,
+			"state":         state,
+			"reason":        reason,
+			"send_at_ms":    sendAt,
+			"created_at_ms": created,
+			"keep":          keep != 0,
+			"receipt":       receipt != 0,
+			"content_type":  contentType,
+		})
+	}
+	if rowsErr := rows.Err(); rowsErr != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+
+	counts, err := h.deliveryCounts(r.Context(), seqs)
+	if err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	for i, seq := range seqs {
+		c := counts[seq]
+		if c == nil {
+			c = emptyDeliveryCounts()
+		}
+		items[i]["delivery_counts"] = c
+	}
+
+	next := ""
+	if offset+len(items) < total {
+		next = strconv.Itoa(offset + len(items))
+	}
+	httpx.WriteOK(w, map[string]any{"items": items, "next_cursor": next, "total": total})
+}
+
+func (h *Handler) handleMessageGet(w http.ResponseWriter, r *http.Request) {
+	seqStr := r.PathValue("seq")
+	seq, err := strconv.ParseInt(seqStr, 10, 64)
+	if err != nil || seq < 1 {
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "seq 无效")
+		return
+	}
+	q := r.URL.Query()
+	limit, offset, ok := parsePage(w, q.Get("limit"), q.Get("cursor"))
+	if !ok {
+		return
+	}
+
+	var id, sender, destKind, destID, state, reason, contentType, meta string
+	var sendAt, created int64
+	err = h.db.Read.QueryRowContext(r.Context(), `
+SELECT id, sender_id, dest_kind, dest_id, state, reason, send_at, created_at,
+  content_type, meta
+FROM messages WHERE seq = ?`, seq).Scan(
+		&id, &sender, &destKind, &destID, &state, &reason, &sendAt, &created,
+		&contentType, &meta,
+	)
+	if isNoRows(err) {
+		httpx.WriteError(w, http.StatusNotFound, "not_found", "消息不存在")
+		return
+	}
+	if err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+
+	var total int
+	_ = h.db.Read.QueryRowContext(r.Context(),
+		`SELECT COUNT(*) FROM deliveries WHERE seq = ?`, seq).Scan(&total)
+
+	rows, err := h.db.Read.QueryContext(r.Context(), `
+SELECT endpoint_id, state, reason, attempts, pushed_at, updated_at
+FROM deliveries WHERE seq = ?
+ORDER BY endpoint_id ASC
+LIMIT ? OFFSET ?`, seq, limit, offset)
+	if err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	defer func() { _ = rows.Close() }()
+
+	deliveries := make([]map[string]any, 0)
+	for rows.Next() {
+		var endpointID, dState, dReason string
+		var attempts int
+		var pushedAt, updatedAt sql.NullInt64
+		if scanErr := rows.Scan(&endpointID, &dState, &dReason, &attempts, &pushedAt, &updatedAt); scanErr != nil {
+			httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+			return
+		}
+		deliveries = append(deliveries, map[string]any{
+			"endpoint_id":   endpointID,
+			"state":         dState,
+			"reason":        dReason,
+			"attempts":      attempts,
+			"pushed_at_ms":  nullInt64API(pushedAt),
+			"updated_at_ms": updatedAt.Int64,
+		})
+	}
+	if err := rows.Err(); err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+
+	next := ""
+	if offset+len(deliveries) < total {
+		next = strconv.Itoa(offset + len(deliveries))
+	}
+
+	metaObj := any(map[string]any{})
+	if strings.TrimSpace(meta) != "" && meta != "{}" {
+		metaObj = jsonRawOrObject(meta)
+	}
+
+	httpx.WriteOK(w, map[string]any{
+		"seq":           seq,
+		"id":            id,
+		"sender_id":     sender,
+		"dest_kind":     destKind,
+		"dest_id":       destID,
+		"state":         state,
+		"reason":        reason,
+		"send_at_ms":    sendAt,
+		"created_at_ms": created,
+		"meta":          metaObj,
+		"content_type":  contentType,
+		"deliveries":    deliveries,
+		"next_cursor":   next,
+	})
+}
+
+func buildMessageFilter(w http.ResponseWriter, q interface{ Get(string) string }) (where string, args []any, ok bool) {
+	conds := []string{"1=1"}
+	args = make([]any, 0, 8)
+
+	if v := strings.TrimSpace(q.Get("sender_id")); v != "" {
+		conds = append(conds, "m.sender_id = ?")
+		args = append(args, v)
+	}
+	if v := strings.TrimSpace(q.Get("group_id")); v != "" {
+		conds = append(conds, "m.dest_kind = 'group' AND m.dest_id = ?")
+		args = append(args, v)
+	}
+	if v := strings.TrimSpace(q.Get("state")); v != "" {
+		switch v {
+		case "scheduled", "dispatched", "completed":
+			conds = append(conds, "m.state = ?")
+			args = append(args, v)
+		default:
+			httpx.WriteError(w, http.StatusBadRequest, "bad_request", "state 无效")
+			return "", nil, false
+		}
+	}
+	if v := strings.TrimSpace(q.Get("from_ms")); v != "" {
+		n, okParse := parseInt64Query(v)
+		if !okParse {
+			httpx.WriteError(w, http.StatusBadRequest, "bad_request", "from_ms 无效")
+			return "", nil, false
+		}
+		conds = append(conds, "m.created_at >= ?")
+		args = append(args, n)
+	}
+	if v := strings.TrimSpace(q.Get("to_ms")); v != "" {
+		n, okParse := parseInt64Query(v)
+		if !okParse {
+			httpx.WriteError(w, http.StatusBadRequest, "bad_request", "to_ms 无效")
+			return "", nil, false
+		}
+		conds = append(conds, "m.created_at <= ?")
+		args = append(args, n)
+	}
+	if v := strings.TrimSpace(q.Get("endpoint_id")); v != "" {
+		conds = append(conds, `EXISTS (
+SELECT 1 FROM deliveries d WHERE d.seq = m.seq AND d.endpoint_id = ?)`)
+		args = append(args, v)
+	}
+
+	return strings.Join(conds, " AND "), args, true
+}
+
+func (h *Handler) deliveryCounts(ctx context.Context, seqs []int64) (map[int64]map[string]int, error) {
+	out := make(map[int64]map[string]int, len(seqs))
+	if len(seqs) == 0 {
+		return out, nil
+	}
+	placeholders := make([]string, len(seqs))
+	args := make([]any, len(seqs))
+	for i, s := range seqs {
+		placeholders[i] = "?"
+		args[i] = s
+		out[s] = emptyDeliveryCounts()
+	}
+	rows, err := h.db.Read.QueryContext(ctx, `
+SELECT seq, state, COUNT(*) FROM deliveries
+WHERE seq IN (`+strings.Join(placeholders, ",")+`)
+GROUP BY seq, state`, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = rows.Close() }()
+	for rows.Next() {
+		var seq int64
+		var state string
+		var n int
+		if scanErr := rows.Scan(&seq, &state, &n); scanErr != nil {
+			return nil, scanErr
+		}
+		c := out[seq]
+		switch state {
+		case "pending", "accepted", "recalled", "expired", "dropped", "rejected":
+			c[state] = n
+		}
+	}
+	return out, rows.Err()
+}
+
+func emptyDeliveryCounts() map[string]int {
+	return map[string]int{
+		"pending":  0,
+		"accepted": 0,
+		"recalled": 0,
+		"expired":  0,
+		"dropped":  0,
+		"rejected": 0,
+	}
+}
+
+func jsonRawOrObject(s string) any {
+	// meta 存的是 JSON 文本；解析失败时返回空对象，避免把原始串当正文泄漏路径
+	var v any
+	if err := json.Unmarshal([]byte(s), &v); err != nil || v == nil {
+		return map[string]any{}
+	}
+	return v
+}

internal/admin/overview.go

@@ -0,0 +1,113 @@
+package admin
+
+import (
+	"database/sql"
+	"errors"
+	"net/http"
+	"strconv"
+	"time"
+
+	"git.asio.asia/nixevol/NixMsg/internal/httpx"
+)
+
+func (h *Handler) handleOverview(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+
+	var total, selfCount, disabled, online int
+	err := h.db.Read.QueryRowContext(ctx, `
+SELECT
+  COUNT(*),
+  COALESCE(SUM(CASE WHEN source = 'self' THEN 1 ELSE 0 END), 0),
+  COALESCE(SUM(CASE WHEN enabled = 0 THEN 1 ELSE 0 END), 0),
+  COALESCE(SUM(CASE
+    WHEN online_since IS NOT NULL
+     AND (offline_since IS NULL OR online_since > offline_since) THEN 1
+    ELSE 0 END), 0)
+FROM endpoints`).Scan(&total, &selfCount, &disabled, &online)
+	if err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+
+	var groupsTotal int
+	if err := h.db.Read.QueryRowContext(ctx, `SELECT COUNT(*) FROM groups`).Scan(&groupsTotal); err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+
+	var pending int
+	if err := h.db.Read.QueryRowContext(ctx,
+		`SELECT COUNT(*) FROM deliveries WHERE state = 'pending'`).Scan(&pending); err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+
+	var scheduled int
+	if err := h.db.Read.QueryRowContext(ctx,
+		`SELECT COUNT(*) FROM messages WHERE state = 'scheduled'`).Scan(&scheduled); err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+
+	httpx.WriteOK(w, map[string]any{
+		"version":            h.version,
+		"endpoints_total":    total,
+		"endpoints_self":     selfCount,
+		"endpoints_online":   online,
+		"endpoints_disabled": disabled,
+		"groups_total":       groupsTotal,
+		"messages_pending":   pending,
+		"messages_scheduled": scheduled,
+		"uptime_ms":          time.Since(h.startedAt).Milliseconds(),
+	})
+}
+
+func (h *Handler) handleSettingsGet(w http.ResponseWriter, _ *http.Request) {
+	cfg := h.cfg
+	httpx.WriteOK(w, map[string]any{
+		"listen":                 cfg.Listen,
+		"admin_listen":           cfg.AdminListen,
+		"session_idle_days":      cfg.SessionIdleDays,
+		"record_retention_days":  cfg.RecordRetentionDays,
+		"idempotency_hours":      cfg.IdempotencyHours,
+		"receipt_retention_days": cfg.ReceiptRetentionDays,
+		"sqlite_synchronous":     cfg.SQLiteSynchronous,
+		"limits": map[string]any{
+			"max_body_bytes":           cfg.Limits.MaxBodyBytes,
+			"max_meta_bytes":           cfg.Limits.MaxMetaBytes,
+			"max_frame_bytes":          cfg.Limits.MaxFrameBytes,
+			"max_ttl_seconds":          cfg.Limits.MaxTTLSeconds,
+			"max_schedule_seconds":     cfg.Limits.MaxScheduleSeconds,
+			"max_group_members":        cfg.Limits.MaxGroupMembers,
+			"grace_seconds":            cfg.Limits.GraceSeconds,
+			"ack_timeout_seconds":      cfg.Limits.AckTimeoutSeconds,
+			"delivery_window":          cfg.Limits.DeliveryWindow,
+			"receipt_window":           cfg.Limits.ReceiptWindow,
+			"requests_per_second":      cfg.Limits.RequestsPerSecond,
+			"max_pending_per_sender":   cfg.Limits.MaxPendingPerSender,
+			"max_pending_per_receiver": cfg.Limits.MaxPendingPerReceiver,
+		},
+	})
+}
+
+func nullInt64API(v sql.NullInt64) any {
+	if !v.Valid {
+		return nil
+	}
+	return v.Int64
+}
+
+func parseInt64Query(s string) (int64, bool) {
+	if s == "" {
+		return 0, false
+	}
+	n, err := strconv.ParseInt(s, 10, 64)
+	if err != nil {
+		return 0, false
+	}
+	return n, true
+}
+
+func isNoRows(err error) bool {
+	return errors.Is(err, sql.ErrNoRows)
+}

internal/admin/registration.go

@@ -0,0 +1,173 @@
+package admin
+
+import (
+	"context"
+	"crypto/rand"
+	"database/sql"
+	"errors"
+	"net/http"
+	"strings"
+	"time"
+	"unicode/utf8"
+
+	"git.asio.asia/nixevol/NixMsg/internal/httpx"
+)
+
+const (
+	settingRegistrationEnabled = "registration_enabled"
+	settingRegistrationCode    = "registration_code"
+	minRegistrationCodeLen     = 8
+	maxRegistrationCodeLen     = 64
+	generatedRegistrationLen   = 16
+)
+
+func (h *Handler) handleRegistrationGet(w http.ResponseWriter, r *http.Request) {
+	data, err := h.loadRegistration(r.Context())
+	if err != nil {
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	httpx.WriteOK(w, data)
+}
+
+func (h *Handler) handleRegistrationPut(w http.ResponseWriter, r *http.Request) {
+	p, _ := principalFrom(r.Context())
+	ip := httpx.ClientIP(r, h.trusted)
+
+	var req struct {
+		Enabled  *bool   `json:"enabled"`
+		Code     *string `json:"code"`
+		Generate bool    `json:"generate"`
+	}
+	if err := httpx.DecodeJSON(r, &req); err != nil {
+		h.audit(actorString(p), "registration_update", "", "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "请求体无效")
+		return
+	}
+	if req.Enabled == nil && req.Code == nil && !req.Generate {
+		h.audit(actorString(p), "registration_update", "", "bad_request", ip)
+		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "至少提供一项更新")
+		return
+	}
+
+	nowMs := time.Now().UnixMilli()
+	err := h.db.Queue.Do(r.Context(), func(tx *sql.Tx) error {
+		if req.Enabled != nil {
+			val := "0"
+			if *req.Enabled {
+				val = "1"
+			}
+			if _, e := tx.Exec(
+				`INSERT INTO settings(key, value, updated_at) VALUES(?, ?, ?)
+				 ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at`,
+				settingRegistrationEnabled, val, nowMs,
+			); e != nil {
+				return e
+			}
+		}
+		if req.Generate {
+			code, genErr := generateRegistrationCode()
+			if genErr != nil {
+				return genErr
+			}
+			if _, e := tx.Exec(
+				`INSERT INTO settings(key, value, updated_at) VALUES(?, ?, ?)
+				 ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at`,
+				settingRegistrationCode, code, nowMs,
+			); e != nil {
+				return e
+			}
+			return nil
+		}
+		if req.Code != nil {
+			code := *req.Code
+			n := utf8.RuneCountInString(code)
+			if n < minRegistrationCodeLen || n > maxRegistrationCodeLen {
+				return errBadRequest("安全码须为 8–64 字符")
+			}
+			if _, e := tx.Exec(
+				`INSERT INTO settings(key, value, updated_at) VALUES(?, ?, ?)
+				 ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at`,
+				settingRegistrationCode, code, nowMs,
+			); e != nil {
+				return e
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		var br badRequestError
+		if errors.As(err, &br) {
+			h.audit(actorString(p), "registration_update", "", "bad_request", ip)
+			httpx.WriteError(w, http.StatusBadRequest, "bad_request", string(br))
+			return
+		}
+		h.audit(actorString(p), "registration_update", "", "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+
+	data, err := h.loadRegistration(r.Context())
+	if err != nil {
+		h.audit(actorString(p), "registration_update", "", "error", ip)
+		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
+		return
+	}
+	// 审计不写安全码明文
+	h.audit(actorString(p), "registration_update", "", "ok", ip)
+	httpx.WriteOK(w, data)
+}
+
+func (h *Handler) loadRegistration(ctx context.Context) (map[string]any, error) {
+	var enabledVal, codeVal sql.NullString
+	var enabledAt, codeAt sql.NullInt64
+
+	err := h.db.Read.QueryRowContext(ctx,
+		`SELECT value, updated_at FROM settings WHERE key = ?`, settingRegistrationEnabled,
+	).Scan(&enabledVal, &enabledAt)
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return nil, err
+	}
+	err = h.db.Read.QueryRowContext(ctx,
+		`SELECT value, updated_at FROM settings WHERE key = ?`, settingRegistrationCode,
+	).Scan(&codeVal, &codeAt)
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return nil, err
+	}
+
+	updated := int64(0)
+	if enabledAt.Valid && enabledAt.Int64 > updated {
+		updated = enabledAt.Int64
+	}
+	if codeAt.Valid && codeAt.Int64 > updated {
+		updated = codeAt.Int64
+	}
+
+	return map[string]any{
+		"enabled":       registrationTruthy(enabledVal.String),
+		"code":          codeVal.String,
+		"updated_at_ms": updated,
+	}, nil
+}
+
+func registrationTruthy(v string) bool {
+	switch strings.TrimSpace(strings.ToLower(v)) {
+	case "1", "true", "yes", "on":
+		return true
+	default:
+		return false
+	}
+}
+
+func generateRegistrationCode() (string, error) {
+	const alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
+	b := make([]byte, generatedRegistrationLen)
+	if _, err := rand.Read(b); err != nil {
+		return "", err
+	}
+	out := make([]byte, generatedRegistrationLen)
+	for i := range b {
+		out[i] = alphabet[int(b[i])%len(alphabet)]
+	}
+	return string(out), nil
+}

internal/app/identity/app.go

@@ -32,8 +32,10 @@ type Config struct {
 	Sessions auth.SessionTokens
 	// MaxScheduleSeconds 限制 self.update 的 default_delay_ms。
 	MaxScheduleSeconds int64
-	// ConnControl 可选：logout 后踢线；未接线时为 nil。
+	// ConnControl 可选：logout / 停用 / 删除后踢线；未接线时为 nil。
 	ConnControl port.ConnControl
+	// Downlink 可选：停用/删除时发 revoked 与群事件；未接线时为 nil。
+	Downlink port.Downlink
 }
 
 // App 实现 identity.Service（含 I1 注册与 I2 self/对话密码）。
@@ -45,6 +47,7 @@ type App struct {
 	sessions           auth.SessionTokens
 	maxScheduleSeconds int64
 	connCtrl           port.ConnControl
+	down               port.Downlink
 	nowFn              func() time.Time
 }
 
@@ -81,6 +84,7 @@ func New(cfg Config) *App {
 		sessions:           cfg.Sessions,
 		maxScheduleSeconds: cfg.MaxScheduleSeconds,
 		connCtrl:           cfg.ConnControl,
+		down:               cfg.Downlink,
 		nowFn:              cfg.Now,
 	}
 }

internal/app/identity/lifecycle.go

@@ -0,0 +1,664 @@
+package identity
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"errors"
+
+	"git.asio.asia/nixevol/NixMsg/internal/app/port"
+	"git.asio.asia/nixevol/NixMsg/internal/protocol"
+)
+
+const (
+	reasonEndpointDisabled = "endpoint_disabled"
+	reasonEndpointDeleted  = "endpoint_deleted"
+	reasonSenderDisabled   = "sender_disabled"
+	reasonSenderDeleted    = "sender_deleted"
+	reasonLeftGroup        = "left_group"
+	reasonGroupDissolved   = "group_dissolved"
+
+	eventOwnerChanged  = "owner_changed"
+	eventLeft          = "left"
+	eventMemberRemoved = "member_removed"
+	eventDissolved     = "dissolved"
+)
+
+type revokeItem struct {
+	endpointID string
+	msgID      string
+	fromID     string
+	reason     string
+}
+
+type groupNotify struct {
+	recipients []string
+	groupID    string
+	event      string
+	endpointID string
+	atMs       int64
+}
+
+// Disable 停用端：清令牌、作废相关消息/投递，并踢连接（DEVELOPMENT 7.6）。
+func (a *App) Disable(ctx context.Context, endpointID string) error {
+	return a.disableOrDelete(ctx, endpointID, false)
+}
+
+// Enable 仅恢复 enabled=1；已作废消息不恢复。
+func (a *App) Enable(ctx context.Context, endpointID string) error {
+	err := a.db.Queue.Do(ctx, func(tx *sql.Tx) error {
+		res, e := tx.ExecContext(ctx, `UPDATE endpoints SET enabled = 1 WHERE id = ?`, endpointID)
+		if e != nil {
+			return e
+		}
+		n, _ := res.RowsAffected()
+		if n == 0 {
+			return errCode(protocol.CodeNotFound, "endpoint not found")
+		}
+		return nil
+	})
+	return err
+}
+
+// Delete 删除端：停用效果（原因改为 deleted）+ 退群/转让群主 + 清授权与发出记录。
+func (a *App) Delete(ctx context.Context, endpointID string) error {
+	return a.disableOrDelete(ctx, endpointID, true)
+}
+
+func (a *App) disableOrDelete(ctx context.Context, endpointID string, hardDelete bool) error {
+	nowMs := a.now().UnixMilli()
+	recvReason := reasonEndpointDisabled
+	sendReason := reasonSenderDisabled
+	if hardDelete {
+		recvReason = reasonEndpointDeleted
+		sendReason = reasonSenderDeleted
+	}
+
+	var revokes []revokeItem
+	var notifies []groupNotify
+	err := a.db.Queue.Do(ctx, func(tx *sql.Tx) error {
+		var one int
+		if e := tx.QueryRowContext(ctx, `SELECT 1 FROM endpoints WHERE id = ?`, endpointID).Scan(&one); e != nil {
+			if errors.Is(e, sql.ErrNoRows) {
+				return errCode(protocol.CodeNotFound, "endpoint not found")
+			}
+			return e
+		}
+
+		if _, e := tx.ExecContext(ctx, `
+UPDATE endpoints SET enabled = 0,
+  session_hash = NULL, session_issued_at = NULL, session_used_at = NULL
+WHERE id = ?`, endpointID); e != nil {
+			return e
+		}
+
+		if e := voidEndpointMessagesTx(tx, endpointID, recvReason, sendReason, nowMs, &revokes); e != nil {
+			return e
+		}
+
+		if !hardDelete {
+			return nil
+		}
+
+		if e := leaveAllGroupsTx(tx, endpointID, nowMs, &revokes, &notifies); e != nil {
+			return e
+		}
+		if _, e := tx.ExecContext(ctx, `DELETE FROM talk_grants WHERE sender_id = ? OR target_id = ?`, endpointID, endpointID); e != nil {
+			return e
+		}
+		if _, e := tx.ExecContext(ctx, `DELETE FROM receipts WHERE sender_id = ?`, endpointID); e != nil {
+			return e
+		}
+		if _, e := tx.ExecContext(ctx, `DELETE FROM send_keys WHERE sender_id = ?`, endpointID); e != nil {
+			return e
+		}
+		if _, e := tx.ExecContext(ctx, `DELETE FROM messages WHERE sender_id = ?`, endpointID); e != nil {
+			return e
+		}
+		_, e := tx.ExecContext(ctx, `DELETE FROM endpoints WHERE id = ?`, endpointID)
+		return e
+	})
+	if err != nil {
+		return err
+	}
+
+	a.publishRevokes(ctx, revokes)
+	a.publishGroupEvents(ctx, notifies)
+	if a.connCtrl != nil {
+		_ = a.connCtrl.Disconnect(ctx, endpointID, "", port.DisconnectFatal)
+	}
+	return nil
+}
+
+func voidEndpointMessagesTx(tx *sql.Tx, endpointID, recvReason, sendReason string, nowMs int64, revokes *[]revokeItem) error {
+	// 发给 X 的 pending → rejected
+	rows, err := tx.Query(`
+SELECT d.seq, d.pushed_at, m.id, m.sender_id, m.receipt
+FROM deliveries d
+JOIN messages m ON m.seq = d.seq
+WHERE d.endpoint_id = ? AND d.state = 'pending'`, endpointID)
+	if err != nil {
+		return err
+	}
+	type pendRow struct {
+		seq      int64
+		pushed   sql.NullInt64
+		msgID    string
+		senderID string
+		receipt  int
+	}
+	var pending []pendRow
+	for rows.Next() {
+		var r pendRow
+		if scanErr := rows.Scan(&r.seq, &r.pushed, &r.msgID, &r.senderID, &r.receipt); scanErr != nil {
+			_ = rows.Close()
+			return scanErr
+		}
+		pending = append(pending, r)
+	}
+	_ = rows.Close()
+	if err = rows.Err(); err != nil {
+		return err
+	}
+	finalSeqs := map[int64]struct{}{}
+	for _, r := range pending {
+		if _, execErr := tx.Exec(`
+UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ?
+WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`,
+			recvReason, nowMs, r.seq, endpointID); execErr != nil {
+			return execErr
+		}
+		if r.receipt != 0 {
+			if e := insertReceiptIfWantedTx(tx, r.senderID, r.msgID, endpointID, "rejected", recvReason, nowMs, true); e != nil {
+				return e
+			}
+		}
+		if r.pushed.Valid && revokes != nil {
+			*revokes = append(*revokes, revokeItem{
+				endpointID: endpointID, msgID: r.msgID, fromID: r.senderID, reason: recvReason,
+			})
+		}
+		finalSeqs[r.seq] = struct{}{}
+	}
+
+	// 发给 X 的 scheduled 单聊 → completed，要回执则写
+	srows, err := tx.Query(`
+SELECT seq, id, sender_id, receipt FROM messages
+WHERE dest_kind = 'endpoint' AND dest_id = ? AND state = 'scheduled'`, endpointID)
+	if err != nil {
+		return err
+	}
+	type schedRow struct {
+		seq      int64
+		msgID    string
+		senderID string
+		receipt  int
+	}
+	var scheduledTo []schedRow
+	for srows.Next() {
+		var r schedRow
+		if scanErr := srows.Scan(&r.seq, &r.msgID, &r.senderID, &r.receipt); scanErr != nil {
+			_ = srows.Close()
+			return scanErr
+		}
+		scheduledTo = append(scheduledTo, r)
+	}
+	_ = srows.Close()
+	if err = srows.Err(); err != nil {
+		return err
+	}
+	for _, r := range scheduledTo {
+		if _, execErr := tx.Exec(`
+UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 'scheduled'`,
+			recvReason, r.seq); execErr != nil {
+			return execErr
+		}
+		if _, execErr := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, r.seq); execErr != nil {
+			return execErr
+		}
+		if r.receipt != 0 {
+			// 消息级作废：endpoint_id 空，state=rejected（DEVELOPMENT 6.4）
+			if e := insertReceiptIfWantedTx(tx, r.senderID, r.msgID, "", "rejected", recvReason, nowMs, true); e != nil {
+				return e
+			}
+		}
+	}
+
+	// X 发出的 scheduled → completed（sender_*），不写回执
+	outSched, err := tx.Query(`SELECT seq FROM messages WHERE sender_id = ? AND state = 'scheduled'`, endpointID)
+	if err != nil {
+		return err
+	}
+	var outSeqs []int64
+	for outSched.Next() {
+		var seq int64
+		if scanErr := outSched.Scan(&seq); scanErr != nil {
+			_ = outSched.Close()
+			return scanErr
+		}
+		outSeqs = append(outSeqs, seq)
+	}
+	_ = outSched.Close()
+	if err = outSched.Err(); err != nil {
+		return err
+	}
+	for _, seq := range outSeqs {
+		if _, execErr := tx.Exec(`
+UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 'scheduled'`,
+			sendReason, seq); execErr != nil {
+			return execErr
+		}
+		if _, execErr := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, seq); execErr != nil {
+			return execErr
+		}
+	}
+
+	// X 发出的消息的 pending 投递 → rejected（sender_*），不写回执
+	drows, err := tx.Query(`
+SELECT d.seq, d.endpoint_id, d.pushed_at, m.id, m.sender_id
+FROM deliveries d
+JOIN messages m ON m.seq = d.seq
+WHERE m.sender_id = ? AND d.state = 'pending'`, endpointID)
+	if err != nil {
+		return err
+	}
+	type outPend struct {
+		seq        int64
+		endpointID string
+		pushed     sql.NullInt64
+		msgID      string
+		senderID   string
+	}
+	var outPending []outPend
+	for drows.Next() {
+		var r outPend
+		if scanErr := drows.Scan(&r.seq, &r.endpointID, &r.pushed, &r.msgID, &r.senderID); scanErr != nil {
+			_ = drows.Close()
+			return scanErr
+		}
+		outPending = append(outPending, r)
+	}
+	_ = drows.Close()
+	if err = drows.Err(); err != nil {
+		return err
+	}
+	for _, r := range outPending {
+		if _, execErr := tx.Exec(`
+UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ?
+WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`,
+			sendReason, nowMs, r.seq, r.endpointID); execErr != nil {
+			return execErr
+		}
+		if r.pushed.Valid && revokes != nil {
+			*revokes = append(*revokes, revokeItem{
+				endpointID: r.endpointID, msgID: r.msgID, fromID: r.senderID, reason: sendReason,
+			})
+		}
+		finalSeqs[r.seq] = struct{}{}
+	}
+
+	for seq := range finalSeqs {
+		if e := tryFinalizeTx(tx, seq); e != nil {
+			return e
+		}
+	}
+	return nil
+}
+
+func leaveAllGroupsTx(tx *sql.Tx, endpointID string, nowMs int64, revokes *[]revokeItem, notifies *[]groupNotify) error {
+	grows, err := tx.Query(`
+SELECT g.id, g.owner_id
+FROM groups g
+JOIN group_members gm ON gm.group_id = g.id
+WHERE gm.endpoint_id = ?`, endpointID)
+	if err != nil {
+		return err
+	}
+	type grow struct {
+		id    string
+		owner string
+	}
+	var groups []grow
+	for grows.Next() {
+		var g grow
+		if scanErr := grows.Scan(&g.id, &g.owner); scanErr != nil {
+			_ = grows.Close()
+			return scanErr
+		}
+		groups = append(groups, g)
+	}
+	_ = grows.Close()
+	if err = grows.Err(); err != nil {
+		return err
+	}
+
+	for _, g := range groups {
+		members, memErr := loadMembersTx(tx, g.id)
+		if memErr != nil {
+			return memErr
+		}
+		if g.owner == endpointID {
+			others := withoutMember(members, endpointID)
+			if len(others) == 0 {
+				if e := voidGroupAllTx(tx, g.id, nowMs, revokes); e != nil {
+					return e
+				}
+				if _, e := tx.Exec(`DELETE FROM group_members WHERE group_id = ?`, g.id); e != nil {
+					return e
+				}
+				if _, e := tx.Exec(`DELETE FROM groups WHERE id = ?`, g.id); e != nil {
+					return e
+				}
+				if notifies != nil {
+					*notifies = append(*notifies, groupNotify{
+						recipients: members, groupID: g.id, event: eventDissolved, atMs: nowMs,
+					})
+				}
+				continue
+			}
+			newOwner, ownErr := earliestOtherMemberTx(tx, g.id, endpointID)
+			if ownErr != nil {
+				return ownErr
+			}
+			if _, e := tx.Exec(`UPDATE groups SET owner_id = ? WHERE id = ?`, newOwner, g.id); e != nil {
+				return e
+			}
+			if _, e := tx.Exec(`DELETE FROM group_members WHERE group_id = ? AND endpoint_id = ?`, g.id, endpointID); e != nil {
+				return e
+			}
+			if e := voidMemberDeliveriesTx(tx, g.id, endpointID, reasonLeftGroup, nowMs, revokes); e != nil {
+				return e
+			}
+			left := withoutMember(members, endpointID)
+			if notifies != nil {
+				*notifies = append(*notifies,
+					groupNotify{recipients: left, groupID: g.id, event: eventOwnerChanged, endpointID: newOwner, atMs: nowMs},
+					groupNotify{recipients: append(append([]string{}, left...), endpointID), groupID: g.id, event: eventMemberRemoved, endpointID: endpointID, atMs: nowMs},
+				)
+			}
+			continue
+		}
+
+		if _, e := tx.Exec(`DELETE FROM group_members WHERE group_id = ? AND endpoint_id = ?`, g.id, endpointID); e != nil {
+			return e
+		}
+		if e := voidMemberDeliveriesTx(tx, g.id, endpointID, reasonLeftGroup, nowMs, revokes); e != nil {
+			return e
+		}
+		left := withoutMember(members, endpointID)
+		if notifies != nil {
+			*notifies = append(*notifies, groupNotify{
+				recipients: append(append([]string{}, left...), endpointID),
+				groupID:    g.id, event: eventLeft, endpointID: endpointID, atMs: nowMs,
+			})
+		}
+	}
+	return nil
+}
+
+func loadMembersTx(tx *sql.Tx, groupID string) ([]string, error) {
+	rows, err := tx.Query(`SELECT endpoint_id FROM group_members WHERE group_id = ? ORDER BY joined_at ASC, endpoint_id ASC`, groupID)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = rows.Close() }()
+	var out []string
+	for rows.Next() {
+		var id string
+		if e := rows.Scan(&id); e != nil {
+			return nil, e
+		}
+		out = append(out, id)
+	}
+	return out, rows.Err()
+}
+
+func earliestOtherMemberTx(tx *sql.Tx, groupID, exceptID string) (string, error) {
+	var id string
+	err := tx.QueryRow(`
+SELECT endpoint_id FROM group_members
+WHERE group_id = ? AND endpoint_id != ?
+ORDER BY joined_at ASC, endpoint_id ASC
+LIMIT 1`, groupID, exceptID).Scan(&id)
+	return id, err
+}
+
+func withoutMember(ids []string, drop string) []string {
+	out := make([]string, 0, len(ids))
+	for _, id := range ids {
+		if id != drop {
+			out = append(out, id)
+		}
+	}
+	return out
+}
+
+// voidMemberDeliveriesTx 与 group 包同语义：退群成员的 pending 群投递改 rejected。
+func voidMemberDeliveriesTx(tx *sql.Tx, groupID, endpointID, reason string, nowMs int64, revokes *[]revokeItem) error {
+	rows, err := tx.Query(`
+SELECT d.seq, d.pushed_at, m.id, m.sender_id
+FROM deliveries d
+JOIN messages m ON m.seq = d.seq
+WHERE d.endpoint_id = ? AND d.state = 'pending'
+  AND m.dest_kind = 'group' AND m.dest_id = ?`, endpointID, groupID)
+	if err != nil {
+		return err
+	}
+	type row struct {
+		seq      int64
+		pushed   sql.NullInt64
+		msgID    string
+		senderID string
+	}
+	var list []row
+	for rows.Next() {
+		var r row
+		if scanErr := rows.Scan(&r.seq, &r.pushed, &r.msgID, &r.senderID); scanErr != nil {
+			_ = rows.Close()
+			return scanErr
+		}
+		list = append(list, r)
+	}
+	_ = rows.Close()
+	if err = rows.Err(); err != nil {
+		return err
+	}
+	for _, r := range list {
+		if _, execErr := tx.Exec(`
+UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ?
+WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`,
+			reason, nowMs, r.seq, endpointID); execErr != nil {
+			return execErr
+		}
+		if r.pushed.Valid && revokes != nil {
+			*revokes = append(*revokes, revokeItem{
+				endpointID: endpointID, msgID: r.msgID, fromID: r.senderID, reason: reason,
+			})
+		}
+		if e := tryFinalizeTx(tx, r.seq); e != nil {
+			return e
+		}
+	}
+	return nil
+}
+
+func voidGroupAllTx(tx *sql.Tx, groupID string, nowMs int64, revokes *[]revokeItem) error {
+	rows, err := tx.Query(`
+SELECT d.seq, d.endpoint_id, d.pushed_at, m.id, m.sender_id
+FROM deliveries d
+JOIN messages m ON m.seq = d.seq
+WHERE d.state = 'pending' AND m.dest_kind = 'group' AND m.dest_id = ?`, groupID)
+	if err != nil {
+		return err
+	}
+	type drow struct {
+		seq        int64
+		endpointID string
+		pushed     sql.NullInt64
+		msgID      string
+		senderID   string
+	}
+	var dlist []drow
+	for rows.Next() {
+		var r drow
+		if scanErr := rows.Scan(&r.seq, &r.endpointID, &r.pushed, &r.msgID, &r.senderID); scanErr != nil {
+			_ = rows.Close()
+			return scanErr
+		}
+		dlist = append(dlist, r)
+	}
+	_ = rows.Close()
+	if err = rows.Err(); err != nil {
+		return err
+	}
+	for _, r := range dlist {
+		if _, execErr := tx.Exec(`
+UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ?
+WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`,
+			reasonGroupDissolved, nowMs, r.seq, r.endpointID); execErr != nil {
+			return execErr
+		}
+		if r.pushed.Valid && revokes != nil {
+			*revokes = append(*revokes, revokeItem{
+				endpointID: r.endpointID, msgID: r.msgID, fromID: r.senderID, reason: reasonGroupDissolved,
+			})
+		}
+		if e := tryFinalizeTx(tx, r.seq); e != nil {
+			return e
+		}
+	}
+
+	srows, err := tx.Query(`
+SELECT seq, id, sender_id, receipt FROM messages
+WHERE dest_kind = 'group' AND dest_id = ? AND state = 'scheduled'`, groupID)
+	if err != nil {
+		return err
+	}
+	type srow struct {
+		seq      int64
+		msgID    string
+		senderID string
+		receipt  int
+	}
+	var slist []srow
+	for srows.Next() {
+		var r srow
+		if scanErr := srows.Scan(&r.seq, &r.msgID, &r.senderID, &r.receipt); scanErr != nil {
+			_ = srows.Close()
+			return scanErr
+		}
+		slist = append(slist, r)
+	}
+	_ = srows.Close()
+	if err = srows.Err(); err != nil {
+		return err
+	}
+	for _, r := range slist {
+		if _, execErr := tx.Exec(`
+UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 'scheduled'`,
+			reasonGroupDissolved, r.seq); execErr != nil {
+			return execErr
+		}
+		if _, execErr := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, r.seq); execErr != nil {
+			return execErr
+		}
+		if r.receipt != 0 {
+			if e := insertReceiptIfWantedTx(tx, r.senderID, r.msgID, "", "rejected", reasonGroupDissolved, nowMs, true); e != nil {
+				return e
+			}
+		}
+	}
+	return nil
+}
+
+func insertReceiptIfWantedTx(tx *sql.Tx, senderID, msgID, endpointID, state, reason string, nowMs int64, alreadyWanted bool) error {
+	if !alreadyWanted {
+		return nil
+	}
+	var one int
+	err := tx.QueryRow(`SELECT 1 FROM endpoints WHERE id = ?`, senderID).Scan(&one)
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	_, err = tx.Exec(`
+INSERT INTO receipts(sender_id, msg_id, endpoint_id, state, reason, created_at, acked)
+VALUES(?,?,?,?,?,?,0)`, senderID, msgID, endpointID, state, reason, nowMs)
+	return err
+}
+
+func tryFinalizeTx(tx *sql.Tx, seq int64) error {
+	var n int
+	if err := tx.QueryRow(`SELECT COUNT(*) FROM deliveries WHERE seq = ? AND state = 'pending'`, seq).Scan(&n); err != nil {
+		return err
+	}
+	if n > 0 {
+		return nil
+	}
+	var state string
+	if err := tx.QueryRow(`SELECT state FROM messages WHERE seq = ?`, seq).Scan(&state); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil
+		}
+		return err
+	}
+	if state == "completed" {
+		return nil
+	}
+	if _, err := tx.Exec(`UPDATE messages SET state = 'completed' WHERE seq = ?`, seq); err != nil {
+		return err
+	}
+	_, err := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, seq)
+	return err
+}
+
+func (a *App) publishRevokes(ctx context.Context, items []revokeItem) {
+	if a.down == nil || len(items) == 0 {
+		return
+	}
+	for _, it := range items {
+		frame := protocol.Revoked{
+			V: protocol.Version, Type: protocol.TypeRevoked,
+			ID: it.msgID, From: it.fromID, Reason: it.reason,
+		}
+		payload, encErr := encodeFrame(frame)
+		if encErr != nil {
+			continue
+		}
+		_ = a.down.PublishDown(ctx, it.endpointID, "", payload, port.PublishOpts{QoS: 1})
+	}
+}
+
+func (a *App) publishGroupEvents(ctx context.Context, items []groupNotify) {
+	if a.down == nil || len(items) == 0 {
+		return
+	}
+	for _, it := range items {
+		frame := protocol.GroupEvent{
+			V: protocol.Version, Type: protocol.TypeGroupEvent,
+			GroupID: it.groupID, Event: it.event, EndpointID: it.endpointID, AtMs: it.atMs,
+		}
+		payload, encErr := encodeFrame(frame)
+		if encErr != nil {
+			continue
+		}
+		seen := map[string]struct{}{}
+		for _, id := range it.recipients {
+			if _, ok := seen[id]; ok {
+				continue
+			}
+			seen[id] = struct{}{}
+			_ = a.down.PublishDown(ctx, id, "", payload, port.PublishOpts{QoS: 0})
+		}
+	}
+}
+
+func encodeFrame(v any) ([]byte, error) {
+	var buf bytes.Buffer
+	if err := protocol.Encode(&buf, v); err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}

internal/app/identity/lifecycle_test.go

@@ -0,0 +1,386 @@
+package identity_test
+
+import (
+	"context"
+	"database/sql"
+	"io"
+	"net/http"
+	"net/http/cookiejar"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"git.asio.asia/nixevol/NixMsg/internal/admin"
+	"git.asio.asia/nixevol/NixMsg/internal/app/identity"
+	"git.asio.asia/nixevol/NixMsg/internal/app/message"
+	"git.asio.asia/nixevol/NixMsg/internal/app/port"
+	"git.asio.asia/nixevol/NixMsg/internal/auth"
+	"git.asio.asia/nixevol/NixMsg/internal/config"
+	"git.asio.asia/nixevol/NixMsg/internal/protocol"
+	"git.asio.asia/nixevol/NixMsg/internal/store"
+)
+
+func openLifecycle(t *testing.T) (*identity.App, *message.App, *store.DB) {
+	t.Helper()
+	dir := t.TempDir()
+	db, err := store.Open(filepath.Join(dir, "data"), "FULL")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	fixed := time.UnixMilli(1_700_000_000_000)
+	idApp := identity.New(identity.Config{
+		DB:                 db,
+		Hash:               auth.NewStubHashPool(),
+		Locks:              auth.NewStubLoginLocks(),
+		Sessions:           auth.NewSessionTokens(),
+		MaxScheduleSeconds: int64(config.Default().Limits.MaxScheduleSeconds),
+		Now:                func() time.Time { return fixed },
+		ConnControl:        &port.StubConnControl{},
+		Downlink:           &port.StubDownlink{},
+	})
+	lim := message.LimitsFromConfig(config.Default().Limits)
+	lim.RequestsPerSecond = 0
+	lim.RecordRetentionDays = 7
+	msgApp := message.New(db, lim, auth.NewStubHashPool(),
+		message.WithNow(func() time.Time { return fixed }),
+		message.WithLocks(auth.NewStubLoginLocks()),
+	)
+	return idApp, msgApp, db
+}
+
+func insertEPFull(t *testing.T, db *store.DB, id string) {
+	t.Helper()
+	err := db.Queue.Do(context.Background(), func(tx *sql.Tx) error {
+		_, e := tx.Exec(`
+INSERT INTO endpoints(id, name, login_hash, talk_hash, talk_version, default_delay_ms, enabled, created_at, offline_since)
+VALUES(?,?,?,?,0,0,1,?,?)`, id, id, "stub$login", nil, 1_700_000_000_000, 1_700_000_000_000)
+		return e
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestF01DisableVoidsScheduledAndRejectsNew(t *testing.T) {
+	t.Parallel()
+	idApp, msgApp, db := openLifecycle(t)
+	ctx := context.Background()
+	insertEPFull(t, db, "alice")
+	insertEPFull(t, db, "bob")
+
+	future := int64(1_700_000_000_000 + 3600_000)
+	receipt := true
+	toBob := &protocol.Send{
+		V: protocol.Version, Type: protocol.TypeSend, RID: "1", ID: "m-to-bob",
+		To:       protocol.Target{Kind: protocol.TargetEndpoint, ID: "bob"},
+		Body:     protocol.Body{Enc: protocol.EncUTF8, Data: "hi"},
+		SendAtMs: &future,
+		Receipt:  &receipt,
+	}
+	if _, err := msgApp.Submit(ctx, "alice", port.ConnInfo{EndpointID: "alice"}, toBob); err != nil {
+		t.Fatal(err)
+	}
+	fromBob := &protocol.Send{
+		V: protocol.Version, Type: protocol.TypeSend, RID: "2", ID: "m-from-bob",
+		To:       protocol.Target{Kind: protocol.TargetEndpoint, ID: "alice"},
+		Body:     protocol.Body{Enc: protocol.EncUTF8, Data: "bye"},
+		SendAtMs: &future,
+	}
+	if _, err := msgApp.Submit(ctx, "bob", port.ConnInfo{EndpointID: "bob"}, fromBob); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := idApp.Disable(ctx, "bob"); err != nil {
+		t.Fatal(err)
+	}
+
+	var enabled int
+	if err := db.Read.QueryRow(`SELECT enabled FROM endpoints WHERE id='bob'`).Scan(&enabled); err != nil {
+		t.Fatal(err)
+	}
+	if enabled != 0 {
+		t.Fatalf("enabled=%d", enabled)
+	}
+
+	var toState, toReason string
+	if err := db.Read.QueryRow(`SELECT state, reason FROM messages WHERE sender_id='alice' AND id='m-to-bob'`).
+		Scan(&toState, &toReason); err != nil {
+		t.Fatal(err)
+	}
+	if toState != "completed" || toReason != "endpoint_disabled" {
+		t.Fatalf("to bob: state=%s reason=%s", toState, toReason)
+	}
+	var fromState, fromReason string
+	if err := db.Read.QueryRow(`SELECT state, reason FROM messages WHERE sender_id='bob' AND id='m-from-bob'`).
+		Scan(&fromState, &fromReason); err != nil {
+		t.Fatal(err)
+	}
+	if fromState != "completed" || fromReason != "sender_disabled" {
+		t.Fatalf("from bob: state=%s reason=%s", fromState, fromReason)
+	}
+
+	newSend := &protocol.Send{
+		V: protocol.Version, Type: protocol.TypeSend, RID: "3", ID: "m-new",
+		To:   protocol.Target{Kind: protocol.TargetEndpoint, ID: "bob"},
+		Body: protocol.Body{Enc: protocol.EncUTF8, Data: "x"},
+	}
+	_, err := msgApp.Submit(ctx, "alice", port.ConnInfo{EndpointID: "alice"}, newSend)
+	if protoCode(err) != protocol.CodeEndpointDisabled {
+		t.Fatalf("want endpoint_disabled got %v", err)
+	}
+
+	if err := idApp.Enable(ctx, "bob"); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Read.QueryRow(`SELECT state FROM messages WHERE id='m-to-bob'`).Scan(&toState); err != nil {
+		t.Fatal(err)
+	}
+	if toState != "completed" {
+		t.Fatalf("voided message restored? %s", toState)
+	}
+	okSend := &protocol.Send{
+		V: protocol.Version, Type: protocol.TypeSend, RID: "4", ID: "m-after",
+		To:   protocol.Target{Kind: protocol.TargetEndpoint, ID: "bob"},
+		Body: protocol.Body{Enc: protocol.EncUTF8, Data: "ok"},
+	}
+	if _, err := msgApp.Submit(ctx, "alice", port.ConnInfo{EndpointID: "alice"}, okSend); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestF01DeleteOwnerTransfersEarliest(t *testing.T) {
+	t.Parallel()
+	idApp, _, db := openLifecycle(t)
+	ctx := context.Background()
+	insertEPFull(t, db, "owner")
+	insertEPFull(t, db, "early")
+	insertEPFull(t, db, "late")
+
+	err := db.Queue.Do(ctx, func(tx *sql.Tx) error {
+		if _, e := tx.Exec(`INSERT INTO groups(id, name, owner_id, created_at) VALUES('g1','群','owner',?)`, 1_700_000_000_000); e != nil {
+			return e
+		}
+		_, e := tx.Exec(`INSERT INTO group_members(group_id, endpoint_id, joined_at) VALUES
+('g1','owner',100),('g1','early',200),('g1','late',300)`)
+		return e
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := idApp.Delete(ctx, "owner"); err != nil {
+		t.Fatal(err)
+	}
+
+	var owner string
+	if err := db.Read.QueryRow(`SELECT owner_id FROM groups WHERE id='g1'`).Scan(&owner); err != nil {
+		t.Fatal(err)
+	}
+	if owner != "early" {
+		t.Fatalf("want earliest other member early, got %s", owner)
+	}
+	var n int
+	if err := db.Read.QueryRow(`SELECT COUNT(*) FROM group_members WHERE group_id='g1' AND endpoint_id='owner'`).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != 0 {
+		t.Fatal("owner should have left the group")
+	}
+}
+
+func TestF01DeleteReopenNoOldReceipts(t *testing.T) {
+	t.Parallel()
+	idApp, msgApp, db := openLifecycle(t)
+	ctx := context.Background()
+	insertEPFull(t, db, "alice")
+	insertEPFull(t, db, "bob")
+
+	receipt := true
+	send := &protocol.Send{
+		V: protocol.Version, Type: protocol.TypeSend, RID: "1", ID: "m1",
+		To:      protocol.Target{Kind: protocol.TargetEndpoint, ID: "alice"},
+		Body:    protocol.Body{Enc: protocol.EncUTF8, Data: "hi"},
+		Receipt: &receipt,
+	}
+	if _, err := msgApp.Submit(ctx, "bob", port.ConnInfo{EndpointID: "bob"}, send); err != nil {
+		t.Fatal(err)
+	}
+	err := db.Queue.Do(ctx, func(tx *sql.Tx) error {
+		_, e := tx.Exec(`
+INSERT INTO receipts(sender_id, msg_id, endpoint_id, state, reason, created_at, acked)
+VALUES('bob','m1','alice','accepted','',?,0)`, 1_700_000_000_000)
+		return e
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := idApp.Delete(ctx, "bob"); err != nil {
+		t.Fatal(err)
+	}
+
+	var n int
+	if err := db.Read.QueryRow(`SELECT COUNT(*) FROM receipts WHERE sender_id='bob'`).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != 0 {
+		t.Fatalf("old receipts should be gone, got %d", n)
+	}
+
+	insertEPFull(t, db, "bob")
+	if err := db.Read.QueryRow(`SELECT COUNT(*) FROM receipts WHERE sender_id='bob'`).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != 0 {
+		t.Fatalf("reopened endpoint must not inherit receipts, got %d", n)
+	}
+	if err := db.Read.QueryRow(`SELECT COUNT(*) FROM messages WHERE sender_id='bob'`).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != 0 {
+		t.Fatalf("reopened endpoint must not inherit messages, got %d", n)
+	}
+}
+
+func TestAdminDisableDeleteHTTP(t *testing.T) {
+	t.Parallel()
+	dir := t.TempDir()
+	db, err := store.Open(filepath.Join(dir, "data"), "FULL")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+
+	fixed := time.UnixMilli(1_700_000_000_000)
+	hash := auth.NewStubHashPool()
+	if seedErr := admin.SeedAdminPassword(context.Background(), db, hash, "adminpassword1"); seedErr != nil {
+		t.Fatal(seedErr)
+	}
+	idApp := identity.New(identity.Config{
+		DB: db, Hash: hash, Locks: auth.NewStubLoginLocks(), Sessions: auth.NewSessionTokens(),
+		MaxScheduleSeconds: 86400, Now: func() time.Time { return fixed },
+	})
+	lim := message.LimitsFromConfig(config.Default().Limits)
+	lim.RequestsPerSecond = 0
+	msgApp := message.New(db, lim, hash,
+		message.WithNow(func() time.Time { return fixed }),
+		message.WithLocks(auth.NewStubLoginLocks()),
+	)
+
+	kick := &lifecycleKick{}
+	h := admin.New(admin.Deps{
+		DB: db, Hash: hash, Tokens: admin.NewRandomAPITokens(),
+		Locks: admin.NewMemoryLoginLocks(), KickEndpoint: kick.Kick, Identity: idApp,
+	})
+	srv := httptest.NewServer(h)
+	t.Cleanup(srv.Close)
+
+	jar, _ := cookiejar.New(nil)
+	client := &http.Client{Jar: jar}
+	loginRes, err := client.Post(srv.URL+"/api/admin/login", "application/json",
+		strings.NewReader(`{"username":"admin","password":"adminpassword1"}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = loginRes.Body.Close()
+	if loginRes.StatusCode != 200 {
+		t.Fatalf("login %d", loginRes.StatusCode)
+	}
+
+	createEP := func(id string) {
+		t.Helper()
+		req, _ := http.NewRequest(http.MethodPost, srv.URL+"/api/admin/endpoints",
+			strings.NewReader(`{"id":"`+id+`","name":"`+id+`","login_password":"password12"}`))
+		req.Header.Set("Content-Type", "application/json")
+		req.Header.Set("X-Nixmsg-Request", "1")
+		res, e := client.Do(req)
+		if e != nil {
+			t.Fatal(e)
+		}
+		raw, _ := io.ReadAll(res.Body)
+		_ = res.Body.Close()
+		if res.StatusCode != 200 {
+			t.Fatalf("create %s: %d %s", id, res.StatusCode, raw)
+		}
+	}
+	createEP("alice")
+	createEP("bob")
+	createEP("carol")
+
+	ctx := context.Background()
+	future := fixed.UnixMilli() + 3600_000
+	toBob := &protocol.Send{
+		V: protocol.Version, Type: protocol.TypeSend, RID: "1", ID: "sched-bob",
+		To:       protocol.Target{Kind: protocol.TargetEndpoint, ID: "bob"},
+		Body:     protocol.Body{Enc: protocol.EncUTF8, Data: "x"},
+		SendAtMs: &future,
+	}
+	if _, subErr := msgApp.Submit(ctx, "alice", port.ConnInfo{EndpointID: "alice"}, toBob); subErr != nil {
+		t.Fatal(subErr)
+	}
+
+	req, _ := http.NewRequest(http.MethodPost, srv.URL+"/api/admin/endpoints/batch",
+		strings.NewReader(`{"ids":["bob"],"action":"disable"}`))
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Nixmsg-Request", "1")
+	res, err := client.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	body, _ := io.ReadAll(res.Body)
+	_ = res.Body.Close()
+	if res.StatusCode != 200 {
+		t.Fatalf("disable: %d %s", res.StatusCode, body)
+	}
+	var st string
+	if scanErr := db.Read.QueryRow(`SELECT state FROM messages WHERE id='sched-bob'`).Scan(&st); scanErr != nil {
+		t.Fatal(scanErr)
+	}
+	if st != "completed" {
+		t.Fatalf("scheduled should be voided, got %s", st)
+	}
+
+	err = db.Queue.Do(ctx, func(tx *sql.Tx) error {
+		if _, e := tx.Exec(`INSERT INTO groups(id, name, owner_id, created_at) VALUES('ghttp','G','bob',?)`, fixed.UnixMilli()); e != nil {
+			return e
+		}
+		_, e := tx.Exec(`INSERT INTO group_members(group_id, endpoint_id, joined_at) VALUES
+('ghttp','bob',1),('ghttp','alice',2),('ghttp','carol',3)`)
+		return e
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// bob 已停用，需先启用才能作为「仍存在的群主」再删除？删除不要求 enabled。
+	req, _ = http.NewRequest(http.MethodDelete, srv.URL+"/api/admin/endpoints/bob", nil)
+	req.Header.Set("X-Nixmsg-Request", "1")
+	res, err = client.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	raw, _ := io.ReadAll(res.Body)
+	_ = res.Body.Close()
+	if res.StatusCode != 200 {
+		t.Fatalf("delete: %d %s", res.StatusCode, raw)
+	}
+	var owner string
+	if err := db.Read.QueryRow(`SELECT owner_id FROM groups WHERE id='ghttp'`).Scan(&owner); err != nil {
+		t.Fatal(err)
+	}
+	if owner != "alice" {
+		t.Fatalf("want alice as new owner, got %s", owner)
+	}
+}
+
+type lifecycleKick struct {
+	Calls []string
+}
+
+func (k *lifecycleKick) Kick(_ context.Context, endpointID string) (bool, error) {
+	k.Calls = append(k.Calls, endpointID)
+	return true, nil
+}

internal/app/identity/self.go

@@ -215,8 +215,3 @@ WHERE id = ?`, endpointID)
 	}
 	return nil
 }
-
-// Disable / Enable / Delete 属 I5，此处保留未实现。
-func (a *App) Disable(context.Context, string) error { return ErrNotImplemented }
-func (a *App) Enable(context.Context, string) error  { return ErrNotImplemented }
-func (a *App) Delete(context.Context, string) error  { return ErrNotImplemented }

internal/app/message/conn.go

@@ -67,6 +67,17 @@ func (c *MemoryConns) Current(endpointID string) (LiveConn, bool) {
 	return v, ok
 }
 
+// Snapshot 返回当前连接表副本（供推送循环遍历）。
+func (c *MemoryConns) Snapshot() map[string]LiveConn {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	out := make(map[string]LiveConn, len(c.m))
+	for k, v := range c.m {
+		out[k] = v
+	}
+	return out
+}
+
 // RecordingDownlink 记录下行发布，供测试断言。
 type RecordingDownlink struct {
 	mu        sync.Mutex

internal/app/presence/encode.go

@@ -0,0 +1,22 @@
+package presence
+
+// EncodeGetData 把 Get 结果编成 presence.get 的 resp data（items 数组）。
+// 未知编号项为 {"id":"...","not_found":true}，其余含 id/online/since_ms。
+func EncodeGetData(items []StatusItem) map[string]any {
+	out := make([]map[string]any, 0, len(items))
+	for _, it := range items {
+		if it.NotFound {
+			out = append(out, map[string]any{
+				"id":        it.ID,
+				"not_found": true,
+			})
+			continue
+		}
+		out = append(out, map[string]any{
+			"id":       it.ID,
+			"online":   it.Online,
+			"since_ms": it.SinceMs,
+		})
+	}
+	return map[string]any{"items": out}
+}

internal/broker/broker.go

@@ -58,11 +58,16 @@ func (AllowAuthenticator) Authenticate(context.Context, string, []byte, string)
 	return AuthResult{OK: true}, nil
 }
 
+// PublishDroppedFunc 下行未写入发送队列时回调（对接消息线 OnPublishDropped）。
+type PublishDroppedFunc func(ctx context.Context, endpointID string, connID port.ConnID, payload []byte)
+
 // Options 装配 broker。
 type Options struct {
 	Authenticator Authenticator
 	Uplink        port.UplinkHandler
 	Logger        *slog.Logger
+	// OnPublishDropped 可选；nil 时仅打 debug 日志。
+	OnPublishDropped PublishDroppedFunc
 }
 
 // Broker 内置 mochi，不自带监听端口。
@@ -71,6 +76,7 @@ type Broker struct {
 	auth   Authenticator
 	uplink port.UplinkHandler
 	log    *slog.Logger
+	onDrop PublishDroppedFunc
 
 	hook *nixHook
 
@@ -144,6 +150,7 @@ func New(opts Options) (*Broker, error) {
 		auth:     auth,
 		uplink:   uplink,
 		log:      log,
+		onDrop:   opts.OnPublishDropped,
 		current:  make(map[string]*connState),
 		byClient: make(map[*mqtt.Client]*connState),
 		queues:   make(map[string]*uplinkQueue),

internal/broker/hooks.go

@@ -159,6 +159,16 @@ func (h *nixHook) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes [
 
 func (h *nixHook) OnPublishDropped(cl *mqtt.Client, pk packets.Packet) {
 	h.b.log.Debug("publish dropped", "client", cl.ID, "topic", pk.TopicName, "size", len(pk.Payload))
+	if h.b.onDrop == nil {
+		return
+	}
+	h.b.connsMu.RLock()
+	st := h.b.byClient[cl]
+	h.b.connsMu.RUnlock()
+	if st == nil {
+		return
+	}
+	h.b.onDrop(context.Background(), st.endpointID, st.connID, append([]byte(nil), pk.Payload...))
 }
 
 func (h *nixHook) OnSessionEstablished(cl *mqtt.Client, _ packets.Packet) {

internal/broker/session.go

@@ -101,6 +101,11 @@ func (s *Session) Attach(b *Broker) {
 	s.b = b
 }
 
+// SetPresence 接线时在创建最终 presence 实现后注入（可替换占位）。
+func (s *Session) SetPresence(p PresenceSink) {
+	s.presence = p
+}
+
 func (s *Session) OnSessionEstablished(ctx context.Context, conn port.ConnInfo) error {
 	if s.b != nil {
 		s.b.startHandshakeDeadline(conn.EndpointID, conn.ConnID, handshakeTimeout)

internal/httpx/metrics.go

@@ -0,0 +1,29 @@
+package httpx
+
+import (
+	"net/http"
+	"strings"
+)
+
+// MetricsGate 实现 DEVELOPMENT 4.3 节 /metrics 访问规则（共用端口）。
+// token 为空返回 404；Authorization Bearer 不匹配返回 401；匹配则交给 next。
+// 后台单独监听时不要包这层，直接挂 metrics Handler。
+func MetricsGate(token string, next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if token == "" {
+			http.NotFound(w, r)
+			return
+		}
+		auth := r.Header.Get("Authorization")
+		const prefix = "Bearer "
+		if !strings.HasPrefix(auth, prefix) || auth[len(prefix):] != token {
+			w.WriteHeader(http.StatusUnauthorized)
+			return
+		}
+		if next == nil {
+			http.NotFound(w, r)
+			return
+		}
+		next.ServeHTTP(w, r)
+	})
+}

internal/httpx/metrics_test.go

@@ -0,0 +1,98 @@
+package httpx_test
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"git.asio.asia/nixevol/NixMsg/internal/httpx"
+	"git.asio.asia/nixevol/NixMsg/internal/listener"
+	"git.asio.asia/nixevol/NixMsg/internal/metrics"
+)
+
+func TestMetricsGateSharedPort(t *testing.T) {
+	reg := metrics.New()
+	okHandler := reg.Handler()
+
+	t.Run("empty token 404", func(t *testing.T) {
+		mux := listener.NewMux(listener.RoleShared, listener.Handlers{
+			Metrics:      okHandler,
+			MetricsToken: "",
+		})
+		res := httptest.NewRecorder()
+		mux.ServeHTTP(res, httptest.NewRequest(http.MethodGet, "/metrics", nil))
+		if res.Code != http.StatusNotFound {
+			t.Fatalf("want 404 got %d", res.Code)
+		}
+	})
+
+	t.Run("wrong token 401", func(t *testing.T) {
+		mux := listener.NewMux(listener.RoleShared, listener.Handlers{
+			Metrics:      okHandler,
+			MetricsToken: "secret-token",
+		})
+		req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
+		req.Header.Set("Authorization", "Bearer wrong")
+		res := httptest.NewRecorder()
+		mux.ServeHTTP(res, req)
+		if res.Code != http.StatusUnauthorized {
+			t.Fatalf("want 401 got %d", res.Code)
+		}
+	})
+
+	t.Run("no auth 401", func(t *testing.T) {
+		mux := listener.NewMux(listener.RoleShared, listener.Handlers{
+			Metrics:      okHandler,
+			MetricsToken: "secret-token",
+		})
+		res := httptest.NewRecorder()
+		mux.ServeHTTP(res, httptest.NewRequest(http.MethodGet, "/metrics", nil))
+		if res.Code != http.StatusUnauthorized {
+			t.Fatalf("want 401 got %d", res.Code)
+		}
+	})
+
+	t.Run("good token 200", func(t *testing.T) {
+		mux := listener.NewMux(listener.RoleShared, listener.Handlers{
+			Metrics:      okHandler,
+			MetricsToken: "secret-token",
+		})
+		req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
+		req.Header.Set("Authorization", "Bearer secret-token")
+		res := httptest.NewRecorder()
+		mux.ServeHTTP(res, req)
+		if res.Code != http.StatusOK {
+			t.Fatalf("want 200 got %d", res.Code)
+		}
+		body, _ := io.ReadAll(res.Body)
+		if len(body) == 0 {
+			t.Fatal("empty metrics body")
+		}
+	})
+
+	t.Run("admin role no auth", func(t *testing.T) {
+		mux := listener.NewMux(listener.RoleAdmin, listener.Handlers{
+			Metrics: okHandler,
+		})
+		res := httptest.NewRecorder()
+		mux.ServeHTTP(res, httptest.NewRequest(http.MethodGet, "/metrics", nil))
+		if res.Code != http.StatusOK {
+			t.Fatalf("admin metrics want 200 got %d", res.Code)
+		}
+	})
+}
+
+func TestMetricsGateDirect(t *testing.T) {
+	h := httpx.MetricsGate("tok", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte("ok"))
+	}))
+	res := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
+	req.Header.Set("Authorization", "Bearer tok")
+	h.ServeHTTP(res, req)
+	if res.Code != 200 || res.Body.String() != "ok" {
+		t.Fatalf("got %d %q", res.Code, res.Body.String())
+	}
+}

internal/listener/routes.go

@@ -2,7 +2,8 @@ package listener
 
 import (
 	"net/http"
-	"strings"
+
+	"git.asio.asia/nixevol/NixMsg/internal/httpx"
 )
 
 // RouteRole 区分监听用途，决定哪些路径可用。
@@ -92,7 +93,7 @@ func NewMux(role RouteRole, h Handlers) http.Handler {
 		if h.AdminAPI != nil {
 			mux.Handle("/api/admin/", h.AdminAPI)
 		}
-		mux.Handle("GET /metrics", metricsGate(h.MetricsToken, h.Metrics))
+		mux.Handle("GET /metrics", httpx.MetricsGate(h.MetricsToken, h.Metrics))
 		if h.Static != nil {
 			mux.Handle("/", h.Static)
 		} else {
@@ -102,23 +103,3 @@ func NewMux(role RouteRole, h Handlers) http.Handler {
 
 	return mux
 }
-
-func metricsGate(token string, next http.Handler) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if token == "" {
-			http.NotFound(w, r)
-			return
-		}
-		auth := r.Header.Get("Authorization")
-		const prefix = "Bearer "
-		if !strings.HasPrefix(auth, prefix) || auth[len(prefix):] != token {
-			w.WriteHeader(http.StatusUnauthorized)
-			return
-		}
-		if next == nil {
-			http.NotFound(w, r)
-			return
-		}
-		next.ServeHTTP(w, r)
-	})
-}

taskfiles/q.yml

@@ -6,6 +6,13 @@ tasks:
     cmds:
       - go test ./test/chaos/ -count=1 -v
 
+  q:accept:
+    desc: 跑 Q2 验收集成测试并写入 F01–F23 对照表
+    cmds:
+      - go test ./test/accept/ -count=1 -v
+    env:
+      NIXMSG_WRITE_ACCEPT_REPORT: "1"
+
   q:report:
     desc: 从结果 JSON 生成 F01–F23 验收对照表（默认空样例）
     cmds:
@@ -16,6 +23,11 @@ tasks:
     cmds:
       - go run ./test/report/cmd/genreport ./test/report/testdata/sample_results.json
 
+  q:report-q2:
+    desc: 用 Q2 验收结果生成对照表到 stdout
+    cmds:
+      - go run ./test/report/cmd/genreport ./test/report/testdata/q2_results.json
+
   q:load-test:
     desc: 压测客户端骨架单元测试
     cmds:

test/accept/ACCEPTANCE.md

@@ -0,0 +1,31 @@
+# NixMsg 验收对照表（PRD 第 10 节）
+
+生成时间：2026-09-29T23:20:07Z
+
+汇总：通过 1，失败 0，未测 22
+
+| 编号 | 一句话 | 结果 | 备注 |
+|---|---|---|---|
+| F01 | 批量开通整批校验、停用、删除群主转让、删除后同编号重开不串数据 | 未测 | 未测：serve 未挂载 POST /api/admin/endpoints（A2 未合入或未接线；当前 A1 对端路由返回 501 亦未挂到进程） |
+| F02 | 新设备登录后旧设备自动退出、换 IP 用令牌重连、两种密码锁定、重置密码后被踢、服务器故障不误报密码错误 | 未测 | 未测：端登录/会话令牌属连接 N3，main 上 serve 未挂 broker |
+| F03 | 断开后状态及时变离线，全表可列出 | 未测 | 未测：在线状态属身份 I3 + 连接 N3，未接线 |
+| F04 | 只通知订阅了的端 | 未测 | 未测：presence.watch 属身份 I3，未接线 |
+| F05 | 崩溃不丢已提交消息，消息号去重和冲突，密码门生效，配额生效 | 未测 | 未测：消息提交属消息 M1，未挂入 broker 上行 |
+| F06 | 群成员收到同一份，入群前不补，发送者不收到自己的 | 未测 | 未测：群消息属消息 M + 身份 I4，未接线 |
+| F07 | 256 KiB 通过，超出拒绝，接收上限生效 | 未测 | 未测：大小限制属消息/连接，未接线 |
+| F08 | 弱网最终送达且应用层不重复，重启后续传 | 未测 | 未测：投递确认属消息 M2，未接线 |
+| F09 | 保留时间从发送时刻起算，超时过期 | 未测 | 未测：保留期属消息 M2，未接线 |
+| F10 | 短断线送到，长断线丢弃，服务器重启后宽限内重连送到 | 未测 | 未测：断线策略属消息 M2，未接线 |
+| F11 | 发送方离线后到点仍发送 | 未测 | 未测：定时发送属消息 M2/M4，未接线 |
+| F12 | 延迟窗口内撤回对方收不到 | 未测 | 未测：延迟撤回属消息 M3，未接线 |
+| F13 | 未推送必撤成功；群部分确认得到部分撤回 | 未测 | 未测：撤回判定属消息 M3，未接线 |
+| F14 | 回执能补送给当时离线的发送方 | 未测 | 未测：回执属消息 M3，未接线 |
+| F15 | 输一次记住、改密失效、回复免密、进群仍要密码、防多账号轮流猜 | 未测 | 未测：对话密码属身份 I2，未接线 |
+| F16 | 群主权限、退出后不再收到、解散后同编号新群不收旧消息 | 未测 | 未测：群权限属身份 I4，未接线 |
+| F17 | 后台管端、管注册、管群、查记录，响应里没有正文；API 令牌可用且不能越权 | 未测 | 未测：serve 未挂载 /api/admin/login（A1 Handler 已实现，缺总控/接线挂到 cmd/nixmsg；DEVIATIONS 后台接口 A §1） |
+| F18 | 送达后正文消失；记录天数 0 时连记录消失；防重仍在 | 未测 | 未测：正文清理属消息 M3，未接线 |
+| F19 | 四种 SDK 通过同一清单 | 未测 | 未测：SDK 接入清单属 S1/S2 任务 4，依赖真实服务接线 |
+| F20 | 裸 MQTT 能登录、收、确认、发 | 未测 | 未测：裸 MQTT 属连接 N，serve 未挂 broker |
+| F21 | 默认一个端口提供后台、WebSocket、TCP、注册；后台可分到单独端口 | 未测 | 仅验证 listen 上 /healthz+/readyz；后台 API、/mqtt、裸 TCP、注册未挂入 serve（缺总控接线 + 连接 N） |
+| F22 | 初始化后单文件或 Docker 启动、备份恢复、升级迁移、证书自动重载、指标可抓取 | 通过 | 已测：空目录 admin init + serve，/healthz 与 /readyz 成功，密码不在 serve 日志；未测：备份恢复、升级迁移、证书重载、Docker、/metrics |
+| F23 | 注册开关、安全码校验、换码不影响已注册、输错锁定 | 未测 | 未测：serve 未挂载 POST /api/client/register（I1 Handler 已实现，缺总控/连接 N 接线；DEVIATIONS 身份 I §1） |

test/accept/accept_test.go

@@ -0,0 +1,496 @@
+package accept_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"git.asio.asia/nixevol/NixMsg/test/accept"
+	"git.asio.asia/nixevol/NixMsg/test/harness"
+	"git.asio.asia/nixevol/NixMsg/test/report"
+)
+
+// TestQ2AcceptAndReport 是 Q2 第一部分：对 main 已有能力做集成探测，并生成 F01–F23 对照表。
+// 未接线的路由记为「未测」并写明缺哪条线；不改业务代码以求变绿。
+func TestQ2AcceptAndReport(t *testing.T) {
+	items := make(map[string]report.Item, len(report.Features))
+	for _, f := range report.Features {
+		items[f.ID] = report.Item{
+			ID:     f.ID,
+			Status: report.StatusUntested,
+			Note:   "本波未覆盖；依赖后续线合入与接线",
+		}
+	}
+	set := func(id string, st report.Status, note string) {
+		items[id] = report.Item{ID: id, Status: st, Note: note}
+	}
+
+	// —— 1) admin init + serve + /healthz + /readyz，密码不在 serve 日志 ——
+	runInitHealthz(t, set)
+
+	// —— 共用 harness 进程，探测管理 / 注册 / 开通 ——
+	srv, err := harness.Start(harness.Options{})
+	if err != nil {
+		t.Fatalf("harness start: %v", err)
+	}
+	defer func() { _ = srv.Stop() }()
+
+	if srv.AdminPassword == "" {
+		t.Fatal("admin init 未返回密码（P1 应已合入）")
+	}
+
+	runAdminAuth(t, srv, set)
+	runRegistration(t, srv, set)
+	runEndpointCreate(t, srv, set)
+
+	// 其余条目写清未测原因（缺哪条线）
+	setDefaultUntested(set)
+
+	out := make([]report.Item, 0, len(report.Features))
+	for _, f := range report.Features {
+		out = append(out, items[f.ID])
+	}
+	results := report.Results{
+		GeneratedAt: time.Now().UTC().Format(time.RFC3339),
+		Items:       out,
+	}
+	normalized, err := report.Normalize(results)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var md bytes.Buffer
+	if werr := report.WriteMarkdown(&md, normalized); werr != nil {
+		t.Fatal(werr)
+	}
+	if !strings.Contains(md.String(), "F01") || !strings.Contains(md.String(), "F23") {
+		t.Fatalf("report missing features:\n%s", md.String())
+	}
+	if !strings.Contains(md.String(), "通过") && !strings.Contains(md.String(), "未测") {
+		t.Fatalf("report missing status words:\n%s", md.String())
+	}
+
+	// 默认写到临时目录验证；设 NIXMSG_WRITE_ACCEPT_REPORT=1 时写入仓库产物供交付。
+	dir := t.TempDir()
+	resultsPath := filepath.Join(dir, "q2_results.json")
+	mdPath := filepath.Join(dir, "ACCEPTANCE.md")
+	if os.Getenv("NIXMSG_WRITE_ACCEPT_REPORT") == "1" {
+		root := findModuleRoot(t)
+		resultsPath = filepath.Join(root, "test", "report", "testdata", "q2_results.json")
+		mdPath = filepath.Join(root, "test", "accept", "ACCEPTANCE.md")
+	}
+	raw, err := json.MarshalIndent(results, "", "  ")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(resultsPath, append(raw, '\n'), 0o644); err != nil {
+		t.Fatalf("write results: %v", err)
+	}
+	if err := os.WriteFile(mdPath, md.Bytes(), 0o644); err != nil {
+		t.Fatalf("write acceptance md: %v", err)
+	}
+	t.Logf("wrote %s and %s", resultsPath, mdPath)
+}
+
+func runInitHealthz(t *testing.T, set func(string, report.Status, string)) {
+	t.Helper()
+	ls, err := accept.StartWithLogCapture()
+	if err != nil {
+		set("F22", report.StatusFail, "admin init/serve 失败（平台 P）: "+err.Error())
+		t.Errorf("init/serve: %v", err)
+		return
+	}
+	defer func() { _ = ls.Stop() }()
+
+	pass := ls.AdminPassword
+	if pass == "" {
+		set("F22", report.StatusFail, "admin init 未打印密码（平台 P）")
+		t.Error("empty admin password")
+		return
+	}
+
+	hz, err := http.Get(ls.HTTPBase + "/healthz")
+	if err != nil {
+		set("F22", report.StatusFail, "/healthz 不可达（平台 P）: "+err.Error())
+		t.Errorf("healthz: %v", err)
+		return
+	}
+	body, _ := io.ReadAll(hz.Body)
+	_ = hz.Body.Close()
+	if hz.StatusCode != http.StatusOK || string(body) != "ok" {
+		set("F22", report.StatusFail, fmt.Sprintf("/healthz status=%d body=%q（平台 P）", hz.StatusCode, body))
+		t.Errorf("healthz status=%d body=%q", hz.StatusCode, body)
+		return
+	}
+
+	rz, err := http.Get(ls.HTTPBase + "/readyz")
+	if err != nil {
+		set("F22", report.StatusFail, "/readyz 不可达（平台 P）: "+err.Error())
+		t.Errorf("readyz: %v", err)
+		return
+	}
+	rbody, _ := io.ReadAll(rz.Body)
+	_ = rz.Body.Close()
+	if rz.StatusCode != http.StatusOK || string(rbody) != "ok" {
+		set("F22", report.StatusFail, fmt.Sprintf("/readyz status=%d body=%q（平台 P）", rz.StatusCode, rbody))
+		t.Errorf("readyz status=%d body=%q", rz.StatusCode, rbody)
+		return
+	}
+
+	logs := ls.LogBuf.String()
+	if strings.Contains(logs, pass) {
+		set("F22", report.StatusFail, "管理员密码出现在 serve 日志（平台 P）")
+		t.Errorf("password leaked into serve logs")
+		return
+	}
+
+	set("F22", report.StatusPass, "已测：空目录 admin init + serve，/healthz 与 /readyz 成功，密码不在 serve 日志；未测：备份恢复、升级迁移、证书重载、Docker、/metrics")
+	// F21：当前进程至少在单一 listen 上提供健康检查；后台 API / MQTT 尚未接线
+	set("F21", report.StatusUntested, "仅验证 listen 上 /healthz+/readyz；后台 API、/mqtt、裸 TCP、注册未挂入 serve（缺总控接线 + 连接 N）")
+}
+
+func runAdminAuth(t *testing.T, srv *harness.Server, set func(string, report.Status, string)) {
+	t.Helper()
+	code, body, err := accept.ProbeMethod(srv.AdminHTTPBase, http.MethodPost, "/api/admin/login",
+		[]byte(`{"username":"admin","password":"not-the-password!!"}`))
+	if err != nil {
+		set("F17", report.StatusFail, "探测管理登录失败: "+err.Error())
+		t.Errorf("probe login: %v", err)
+		return
+	}
+	if accept.ClassifyAdminLogin(code) == accept.RouteMissing {
+		note := "未测：serve 未挂载 /api/admin/login（A1 Handler 已实现，缺总控/接线挂到 cmd/nixmsg；DEVIATIONS 后台接口 A §1）"
+		set("F17", report.StatusUntested, note)
+		t.Log(note)
+		return
+	}
+
+	// 路由已挂上：跑登录、锁定、CSRF
+	client, err := srv.AdminClient()
+	if err != nil {
+		set("F17", report.StatusFail, "AdminClient: "+err.Error())
+		t.Fatal(err)
+	}
+
+	// 正确密码登录
+	loginBody, _ := json.Marshal(map[string]string{
+		"username": "admin",
+		"password": srv.AdminPassword,
+	})
+	resp, err := client.PostJSON("/api/admin/login", loginBody)
+	if err != nil {
+		set("F17", report.StatusFail, "登录请求失败: "+err.Error())
+		t.Fatal(err)
+	}
+	loginRaw, _ := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		set("F17", report.StatusFail, fmt.Sprintf("正确密码登录失败 status=%d body=%s（后台接口 A）", resp.StatusCode, loginRaw))
+		t.Errorf("login want 200 got %d %s", resp.StatusCode, loginRaw)
+		return
+	}
+
+	// 无 CSRF 的改状态请求应被拒
+	req, err := http.NewRequest(http.MethodPost, srv.AdminHTTPBase+"/api/admin/logout", strings.NewReader(`{}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	// 故意不加 X-Nixmsg-Request
+	noCSRF, err := client.HTTP.Do(req)
+	if err != nil {
+		set("F17", report.StatusFail, "无 CSRF 请求失败: "+err.Error())
+		t.Fatal(err)
+	}
+	noBody, _ := io.ReadAll(noCSRF.Body)
+	_ = noCSRF.Body.Close()
+	if noCSRF.StatusCode != http.StatusForbidden {
+		set("F17", report.StatusFail, fmt.Sprintf("无 CSRF 期望 403 得 %d body=%s（后台接口 A）", noCSRF.StatusCode, noBody))
+		t.Errorf("csrf: want 403 got %d %s", noCSRF.StatusCode, noBody)
+		return
+	}
+
+	// 带 CSRF 的 logout 应成功
+	okLogout, err := client.PostJSON("/api/admin/logout", []byte(`{}`))
+	if err != nil {
+		set("F17", report.StatusFail, "带 CSRF logout 失败: "+err.Error())
+		t.Fatal(err)
+	}
+	_ = okLogout.Body.Close()
+	if okLogout.StatusCode != http.StatusOK {
+		set("F17", report.StatusFail, fmt.Sprintf("带 CSRF logout 期望 200 得 %d（后台接口 A）", okLogout.StatusCode))
+		t.Errorf("logout with csrf: want 200 got %d", okLogout.StatusCode)
+		return
+	}
+
+	// 错误密码锁定（新客户端，避免 Cookie 干扰；10 次错）
+	lockClient, err := harness.NewAdminClient(srv.AdminHTTPBase)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var locked bool
+	for i := 0; i < 12; i++ {
+		r, rerr := lockClient.PostJSON("/api/admin/login", []byte(`{"username":"admin","password":"wrong-password!!"}`))
+		if rerr != nil {
+			set("F17", report.StatusFail, "错误密码请求失败: "+rerr.Error())
+			t.Fatal(rerr)
+		}
+		raw, _ := io.ReadAll(r.Body)
+		_ = r.Body.Close()
+		if r.StatusCode == http.StatusTooManyRequests {
+			locked = true
+			break
+		}
+		if r.StatusCode != http.StatusUnauthorized {
+			set("F17", report.StatusFail, fmt.Sprintf("错误密码第 %d 次期望 401/429 得 %d body=%s（后台接口 A）", i+1, r.StatusCode, raw))
+			t.Errorf("bad login %d: %d %s", i, r.StatusCode, raw)
+			return
+		}
+	}
+	if !locked {
+		set("F17", report.StatusFail, "错误密码未触发锁定（后台接口 A / 平台 P3）")
+		t.Error("login lock not triggered")
+		return
+	}
+
+	_ = body // 首次探测 body 已用于分类
+	set("F17", report.StatusPass, "已测：管理登录、错误密码锁定、Cookie 会话下无 CSRF 被拒 / 有 CSRF 可通过；管端/管注册/管群/查记录/令牌越权等未在本波覆盖（A2/A3）")
+}
+
+func runRegistration(t *testing.T, srv *harness.Server, set func(string, report.Status, string)) {
+	t.Helper()
+	code, body, err := accept.ProbeMethod(srv.HTTPBase, http.MethodPost, "/api/client/register",
+		[]byte(`{"code":"x"}`))
+	if err != nil {
+		set("F23", report.StatusFail, "探测注册失败: "+err.Error())
+		t.Errorf("probe register: %v", err)
+		return
+	}
+	if accept.ClassifyRegister(code) == accept.RouteMissing {
+		note := "未测：serve 未挂载 POST /api/client/register（I1 Handler 已实现，缺总控/连接 N 接线；DEVIATIONS 身份 I §1）"
+		set("F23", report.StatusUntested, note)
+		t.Log(note)
+		return
+	}
+
+	// 若已挂上：覆盖开关关闭、错码、对码、换码（尽量用管理接口；管理未挂则只能测默认关闭）
+	adminCode, _, _ := accept.ProbeMethod(srv.AdminHTTPBase, http.MethodGet, "/api/admin/registration", nil)
+	if accept.ClassifyAdminLogin(adminCode) == accept.RouteMissing || adminCode == http.StatusNotFound {
+		// 注册路由在、管理注册设置不在：至少验证默认关闭
+		if code == http.StatusForbidden || code == http.StatusNotFound || code == http.StatusBadRequest || code == http.StatusConflict {
+			set("F23", report.StatusPass, fmt.Sprintf("注册路由已挂；默认关闭或校验拒绝（status=%d）。管理注册设置未挂，换码路径未测（缺 A3 接线） body=%s", code, trim(body)))
+			return
+		}
+		set("F23", report.StatusFail, fmt.Sprintf("注册路由异常 status=%d body=%s（身份 I）", code, trim(body)))
+		t.Errorf("register unexpected %d %s", code, body)
+		return
+	}
+
+	// 完整 F23：开关、错码、对码、换码 —— 需管理 PUT registration
+	ac, err := srv.AdminClient()
+	if err != nil {
+		t.Fatal(err)
+	}
+	loginBody, _ := json.Marshal(map[string]string{"username": "admin", "password": srv.AdminPassword})
+	lr, err := ac.PostJSON("/api/admin/login", loginBody)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = lr.Body.Close()
+	if lr.StatusCode != http.StatusOK {
+		set("F23", report.StatusFail, fmt.Sprintf("F23 前置管理登录失败 %d（后台接口 A）", lr.StatusCode))
+		return
+	}
+
+	code1 := "accept-code-one-aaaa"
+	put1, err := ac.Do(http.MethodPut, "/api/admin/registration",
+		[]byte(fmt.Sprintf(`{"enabled":true,"code":%q}`, code1)), "application/json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	put1Body, _ := io.ReadAll(put1.Body)
+	_ = put1.Body.Close()
+	if put1.StatusCode == http.StatusNotImplemented {
+		set("F23", report.StatusUntested, "注册路由已挂，但 PUT /api/admin/registration 返回 501（缺 A3）")
+		t.Log("registration settings 501")
+		return
+	}
+	if put1.StatusCode != http.StatusOK {
+		set("F23", report.StatusFail, fmt.Sprintf("开启注册失败 status=%d body=%s（后台接口 A / 身份 I）", put1.StatusCode, put1Body))
+		return
+	}
+
+	// 错码
+	bad, _, berr := accept.ProbeMethod(srv.HTTPBase, http.MethodPost, "/api/client/register",
+		[]byte(`{"code":"wrong-code","id":"q2bad001"}`))
+	if berr != nil {
+		t.Fatal(berr)
+	}
+	if bad != http.StatusUnauthorized && bad != http.StatusForbidden {
+		set("F23", report.StatusFail, fmt.Sprintf("错码期望 401/403 得 %d（身份 I）", bad))
+		return
+	}
+
+	// 对码
+	okCode, okBody, oerr := accept.ProbeMethod(srv.HTTPBase, http.MethodPost, "/api/client/register",
+		[]byte(fmt.Sprintf(`{"code":%q,"id":"q2ok0001","login_password":"password1234"}`, code1)))
+	if oerr != nil {
+		t.Fatal(oerr)
+	}
+	if okCode != http.StatusOK {
+		set("F23", report.StatusFail, fmt.Sprintf("对码注册失败 status=%d body=%s（身份 I）", okCode, trim(okBody)))
+		return
+	}
+
+	// 换码后旧码失败、新码成功
+	code2 := "accept-code-two-bbbb"
+	put2, err := ac.Do(http.MethodPut, "/api/admin/registration",
+		[]byte(fmt.Sprintf(`{"enabled":true,"code":%q}`, code2)), "application/json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = put2.Body.Close()
+	if put2.StatusCode != http.StatusOK {
+		set("F23", report.StatusFail, fmt.Sprintf("换码失败 status=%d（后台接口 A）", put2.StatusCode))
+		return
+	}
+	oldBad, _, _ := accept.ProbeMethod(srv.HTTPBase, http.MethodPost, "/api/client/register",
+		[]byte(fmt.Sprintf(`{"code":%q,"id":"q2old001","login_password":"password1234"}`, code1)))
+	if oldBad == http.StatusOK {
+		set("F23", report.StatusFail, "换码后旧码仍可注册（身份 I）")
+		return
+	}
+	newOK, _, _ := accept.ProbeMethod(srv.HTTPBase, http.MethodPost, "/api/client/register",
+		[]byte(fmt.Sprintf(`{"code":%q,"id":"q2new001","login_password":"password1234"}`, code2)))
+	if newOK != http.StatusOK {
+		set("F23", report.StatusFail, fmt.Sprintf("换码后新码注册失败 status=%d（身份 I）", newOK))
+		return
+	}
+
+	set("F23", report.StatusPass, "已测：开关、错码、对码、换码；输错锁定未在本用例穷尽")
+}
+
+func runEndpointCreate(t *testing.T, srv *harness.Server, set func(string, report.Status, string)) {
+	t.Helper()
+	// 先看未登录时路由是否存在
+	code, body, err := accept.ProbeMethod(srv.AdminHTTPBase, http.MethodPost, "/api/admin/endpoints",
+		[]byte(`{"id":"q2ep0001","login_password":"password1234"}`))
+	if err != nil {
+		set("F01", report.StatusFail, "探测开通端失败: "+err.Error())
+		t.Errorf("probe endpoints: %v", err)
+		return
+	}
+	if accept.ClassifyCreateEndpoint(code) == accept.RouteMissing {
+		note := "未测：serve 未挂载 POST /api/admin/endpoints（A2 未合入或未接线；当前 A1 对端路由返回 501 亦未挂到进程）"
+		set("F01", report.StatusUntested, note)
+		t.Log(note)
+		return
+	}
+
+	client, err := srv.AdminClient()
+	if err != nil {
+		t.Fatal(err)
+	}
+	loginBody, _ := json.Marshal(map[string]string{"username": "admin", "password": srv.AdminPassword})
+	lr, err := client.PostJSON("/api/admin/login", loginBody)
+	if err != nil {
+		t.Fatal(err)
+	}
+	loginRaw, _ := io.ReadAll(lr.Body)
+	_ = lr.Body.Close()
+	if lr.StatusCode == http.StatusNotFound {
+		set("F01", report.StatusUntested, "端开通路由有响应但管理登录未挂载，无法完成开通验收（缺接线）")
+		return
+	}
+	if lr.StatusCode != http.StatusOK {
+		set("F01", report.StatusFail, fmt.Sprintf("开通前置登录失败 %d %s（后台接口 A）", lr.StatusCode, loginRaw))
+		return
+	}
+
+	create, err := client.PostJSON("/api/admin/endpoints",
+		[]byte(`{"id":"q2ep0001","login_password":"password1234"}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	craw, _ := io.ReadAll(create.Body)
+	_ = create.Body.Close()
+	if create.StatusCode == http.StatusNotImplemented {
+		set("F01", report.StatusUntested, "POST /api/admin/endpoints 返回 501（缺 A2 业务实现）")
+		t.Log("endpoints 501")
+		return
+	}
+	if create.StatusCode != http.StatusOK && create.StatusCode != http.StatusCreated {
+		set("F01", report.StatusFail, fmt.Sprintf("开通端失败 status=%d body=%s（后台接口 A）", create.StatusCode, trim(string(craw))))
+		return
+	}
+
+	// 错误密码连不上：依赖 MQTT 登录（连接 N3）。若 /mqtt 未挂则记未测。
+	mqttCode, _, _ := accept.ProbeMethod(srv.HTTPBase, http.MethodGet, "/mqtt", nil)
+	if mqttCode == http.StatusNotFound {
+		set("F01", report.StatusUntested, "端已开通，但 /mqtt 未挂，无法验证错误密码连不上（缺连接 N 接线）")
+		return
+	}
+	set("F01", report.StatusPass, "已测：开通一端；错误密码 MQTT 连接拒绝需 N3 联调细节，本波仅确认路由可用。批量/停用/删除转让等未覆盖")
+	_ = body
+	_ = code
+}
+
+func setDefaultUntested(set func(string, report.Status, string)) {
+	// 只填尚未被专项用例写入的条目；F01/F17/F21/F22/F23 由探测结果决定。
+	defaults := map[string]string{
+		"F02": "未测：端登录/会话令牌属连接 N3，main 上 serve 未挂 broker",
+		"F03": "未测：在线状态属身份 I3 + 连接 N3，未接线",
+		"F04": "未测：presence.watch 属身份 I3，未接线",
+		"F05": "未测：消息提交属消息 M1，未挂入 broker 上行",
+		"F06": "未测：群消息属消息 M + 身份 I4，未接线",
+		"F07": "未测：大小限制属消息/连接，未接线",
+		"F08": "未测：投递确认属消息 M2，未接线",
+		"F09": "未测：保留期属消息 M2，未接线",
+		"F10": "未测：断线策略属消息 M2，未接线",
+		"F11": "未测：定时发送属消息 M2/M4，未接线",
+		"F12": "未测：延迟撤回属消息 M3，未接线",
+		"F13": "未测：撤回判定属消息 M3，未接线",
+		"F14": "未测：回执属消息 M3，未接线",
+		"F15": "未测：对话密码属身份 I2，未接线",
+		"F16": "未测：群权限属身份 I4，未接线",
+		"F18": "未测：正文清理属消息 M3，未接线",
+		"F19": "未测：SDK 接入清单属 S1/S2 任务 4，依赖真实服务接线",
+		"F20": "未测：裸 MQTT 属连接 N，serve 未挂 broker",
+	}
+	for id, note := range defaults {
+		set(id, report.StatusUntested, note)
+	}
+}
+
+func findModuleRoot(t *testing.T) string {
+	t.Helper()
+	dir, err := os.Getwd()
+	if err != nil {
+		t.Fatal(err)
+	}
+	for {
+		if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil {
+			return dir
+		}
+		parent := filepath.Dir(dir)
+		if parent == dir {
+			t.Fatal("go.mod not found")
+		}
+		dir = parent
+	}
+}
+
+func trim(s string) string {
+	s = strings.TrimSpace(s)
+	if len(s) > 180 {
+		return s[:180] + "..."
+	}
+	return s
+}

test/accept/routes.go

@@ -0,0 +1,65 @@
+package accept
+
+import (
+	"bytes"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+)
+
+// RouteStatus 探测结果。
+type RouteStatus int
+
+const (
+	RouteMissing RouteStatus = iota // 404 / 无此路由
+	RoutePresent                    // 路由存在（含 4xx/5xx 业务响应）
+)
+
+// ProbeMethod 对运行中的服务发一次 HTTP 请求。
+func ProbeMethod(base, method, path string, body []byte) (statusCode int, bodyText string, err error) {
+	var rdr io.Reader
+	if body != nil {
+		rdr = bytes.NewReader(body)
+	}
+	req, err := http.NewRequest(method, strings.TrimRight(base, "/")+path, rdr)
+	if err != nil {
+		return 0, "", err
+	}
+	if body != nil {
+		req.Header.Set("Content-Type", "application/json")
+	}
+	client := &http.Client{Timeout: 10 * time.Second}
+	resp, err := client.Do(req)
+	if err != nil {
+		return 0, "", err
+	}
+	defer func() { _ = resp.Body.Close() }()
+	b, _ := io.ReadAll(io.LimitReader(resp.Body, 64*1024))
+	return resp.StatusCode, string(b), nil
+}
+
+// ClassifyAdminLogin 判断管理登录是否已挂到进程。
+// 404 → 未挂载；其它（含 400/401/429）视为已挂载。
+func ClassifyAdminLogin(code int) RouteStatus {
+	if code == http.StatusNotFound {
+		return RouteMissing
+	}
+	return RoutePresent
+}
+
+// ClassifyRegister 判断注册路由是否已挂到进程。
+func ClassifyRegister(code int) RouteStatus {
+	if code == http.StatusNotFound {
+		return RouteMissing
+	}
+	return RoutePresent
+}
+
+// ClassifyCreateEndpoint 判断开通端路由是否已挂到进程。
+func ClassifyCreateEndpoint(code int) RouteStatus {
+	if code == http.StatusNotFound {
+		return RouteMissing
+	}
+	return RoutePresent
+}

test/accept/server.go

@@ -0,0 +1,128 @@
+package accept
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"git.asio.asia/nixevol/NixMsg/test/harness"
+)
+
+// LoggedServer 在捕获 stdout/stderr 的情况下启动 serve（用于断言密码不进日志）。
+type LoggedServer struct {
+	*harness.Server
+	LogBuf *bytes.Buffer
+	cmd    *exec.Cmd
+}
+
+// StartWithLogCapture 执行 admin init 后启动 serve，并把进程日志写入 LogBuf。
+func StartWithLogCapture() (*LoggedServer, error) {
+	bin, err := harness.Binary()
+	if err != nil {
+		return nil, err
+	}
+	dataDir, err := os.MkdirTemp("", "nixmsg-accept-*")
+	if err != nil {
+		return nil, err
+	}
+	cfgPath := filepath.Join(dataDir, "config.yaml")
+	cfg := fmt.Sprintf("listen: %q\ndata_dir: %q\n", "127.0.0.1:0", filepath.ToSlash(dataDir))
+	if err = os.WriteFile(cfgPath, []byte(cfg), 0o644); err != nil {
+		_ = os.RemoveAll(dataDir)
+		return nil, err
+	}
+
+	initCmd := exec.Command(bin, "admin", "init")
+	initCmd.Env = append(os.Environ(), "NIXMSG_CONFIG="+cfgPath)
+	initOut, initErr := initCmd.CombinedOutput()
+	if initErr != nil {
+		_ = os.RemoveAll(dataDir)
+		return nil, fmt.Errorf("admin init: %w\n%s", initErr, initOut)
+	}
+	password := parsePassword(string(initOut))
+	if password == "" {
+		_ = os.RemoveAll(dataDir)
+		return nil, fmt.Errorf("admin init password not found:\n%s", initOut)
+	}
+
+	logBuf := &bytes.Buffer{}
+	cmd := exec.Command(bin, "serve")
+	cmd.Env = append(os.Environ(), "NIXMSG_CONFIG="+cfgPath)
+	cmd.Stdout = logBuf
+	cmd.Stderr = logBuf
+	if err = cmd.Start(); err != nil {
+		_ = os.RemoveAll(dataDir)
+		return nil, fmt.Errorf("start serve: %w", err)
+	}
+
+	s := &harness.Server{
+		BinPath:       bin,
+		ConfigPath:    cfgPath,
+		DataDir:       dataDir,
+		AdminPassword: password,
+	}
+	addr, err := waitListenAddr(filepath.Join(dataDir, "listen.addr"), 15*time.Second)
+	if err != nil {
+		_ = cmd.Process.Kill()
+		_, _ = cmd.Process.Wait()
+		_ = os.RemoveAll(dataDir)
+		return nil, fmt.Errorf("wait listen.addr: %w\nlogs:\n%s", err, logBuf.String())
+	}
+	s.Addr = addr
+	s.HTTPBase = "http://" + addr
+	s.AdminHTTPBase = s.HTTPBase
+
+	return &LoggedServer{Server: s, LogBuf: logBuf, cmd: cmd}, nil
+}
+
+// Stop 结束进程并删除临时目录。
+func (s *LoggedServer) Stop() error {
+	if s == nil {
+		return nil
+	}
+	if s.cmd != nil && s.cmd.Process != nil {
+		_ = s.cmd.Process.Kill()
+		_, _ = s.cmd.Process.Wait()
+	}
+	if s.DataDir != "" {
+		return os.RemoveAll(s.DataDir)
+	}
+	return nil
+}
+
+func parsePassword(out string) string {
+	for _, line := range strings.Split(out, "\n") {
+		line = strings.TrimSpace(line)
+		lower := strings.ToLower(line)
+		if strings.HasPrefix(lower, "admin password:") {
+			return strings.TrimSpace(line[len("admin password:"):])
+		}
+		if strings.HasPrefix(lower, "password:") {
+			return strings.TrimSpace(line[len("password:"):])
+		}
+	}
+	return ""
+}
+
+func waitListenAddr(path string, timeout time.Duration) (string, error) {
+	deadline := time.Now().Add(timeout)
+	var lastErr error
+	for time.Now().Before(deadline) {
+		b, err := os.ReadFile(path)
+		if err == nil {
+			addr := strings.TrimSpace(string(b))
+			if addr != "" {
+				return addr, nil
+			}
+			lastErr = fmt.Errorf("empty addr file")
+		} else {
+			lastErr = err
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	return "", lastErr
+}

test/report/testdata/q2_results.json

@@ -0,0 +1,120 @@
+{
+  "generated_at": "2026-09-29T23:20:07Z",
+  "items": [
+    {
+      "id": "F01",
+      "status": "untested",
+      "note": "未测：serve 未挂载 POST /api/admin/endpoints（A2 未合入或未接线；当前 A1 对端路由返回 501 亦未挂到进程）"
+    },
+    {
+      "id": "F02",
+      "status": "untested",
+      "note": "未测：端登录/会话令牌属连接 N3，main 上 serve 未挂 broker"
+    },
+    {
+      "id": "F03",
+      "status": "untested",
+      "note": "未测：在线状态属身份 I3 + 连接 N3，未接线"
+    },
+    {
+      "id": "F04",
+      "status": "untested",
+      "note": "未测：presence.watch 属身份 I3，未接线"
+    },
+    {
+      "id": "F05",
+      "status": "untested",
+      "note": "未测：消息提交属消息 M1，未挂入 broker 上行"
+    },
+    {
+      "id": "F06",
+      "status": "untested",
+      "note": "未测：群消息属消息 M + 身份 I4，未接线"
+    },
+    {
+      "id": "F07",
+      "status": "untested",
+      "note": "未测：大小限制属消息/连接，未接线"
+    },
+    {
+      "id": "F08",
+      "status": "untested",
+      "note": "未测：投递确认属消息 M2，未接线"
+    },
+    {
+      "id": "F09",
+      "status": "untested",
+      "note": "未测：保留期属消息 M2，未接线"
+    },
+    {
+      "id": "F10",
+      "status": "untested",
+      "note": "未测：断线策略属消息 M2，未接线"
+    },
+    {
+      "id": "F11",
+      "status": "untested",
+      "note": "未测：定时发送属消息 M2/M4，未接线"
+    },
+    {
+      "id": "F12",
+      "status": "untested",
+      "note": "未测：延迟撤回属消息 M3，未接线"
+    },
+    {
+      "id": "F13",
+      "status": "untested",
+      "note": "未测：撤回判定属消息 M3，未接线"
+    },
+    {
+      "id": "F14",
+      "status": "untested",
+      "note": "未测：回执属消息 M3，未接线"
+    },
+    {
+      "id": "F15",
+      "status": "untested",
+      "note": "未测：对话密码属身份 I2，未接线"
+    },
+    {
+      "id": "F16",
+      "status": "untested",
+      "note": "未测：群权限属身份 I4，未接线"
+    },
+    {
+      "id": "F17",
+      "status": "untested",
+      "note": "未测：serve 未挂载 /api/admin/login（A1 Handler 已实现，缺总控/接线挂到 cmd/nixmsg；DEVIATIONS 后台接口 A §1）"
+    },
+    {
+      "id": "F18",
+      "status": "untested",
+      "note": "未测：正文清理属消息 M3，未接线"
+    },
+    {
+      "id": "F19",
+      "status": "untested",
+      "note": "未测：SDK 接入清单属 S1/S2 任务 4，依赖真实服务接线"
+    },
+    {
+      "id": "F20",
+      "status": "untested",
+      "note": "未测：裸 MQTT 属连接 N，serve 未挂 broker"
+    },
+    {
+      "id": "F21",
+      "status": "untested",
+      "note": "仅验证 listen 上 /healthz+/readyz；后台 API、/mqtt、裸 TCP、注册未挂入 serve（缺总控接线 + 连接 N）"
+    },
+    {
+      "id": "F22",
+      "status": "pass",
+      "note": "已测：空目录 admin init + serve，/healthz 与 /readyz 成功，密码不在 serve 日志；未测：备份恢复、升级迁移、证书重载、Docker、/metrics"
+    },
+    {
+      "id": "F23",
+      "status": "untested",
+      "note": "未测：serve 未挂载 POST /api/client/register（I1 Handler 已实现，缺总控/连接 N 接线；DEVIATIONS 身份 I §1）"
+    }
+  ]
+}