feat: WebSocket 消息可靠投递（纯内存事件队列）

- agent 数据写入增加 10s 超时，写失败立即断连重连，消除无超时写卡死期间快照被覆盖的静默丢失
- 事件类消息（SSH登录/防篡改/指令响应/监控检测）改走内存 outbox：seq + 服务端累计 ack + 断线重放，上限 1024 条溢出丢最旧，重启清空（可接受）
- 注册请求携带 BootID 会话标识：探针重启后 seq 从 1 重新分配，服务端据此重置确认位点，避免新消息被旧位点误判为重复而丢弃
- server 读循环与处理解耦（per-agent 有序队列 + 串行 worker），慢处理不再阻塞 TCP 读取，队列满断连由探针重放兜底
- 修复探针快速重连时旧连接误注销新连接的竞态（此前会闪断新连接）
- IsAgentEnabled 加 3s TTL 缓存，禁用立即失效；pong 在线状态写库去抖（首次 + 5 分钟心跳刷新）
- VM 写超时默认 30s -> 10s，快速失败配合重放兜底

新旧版本双向兼容：注册响应 Reliable 字段协商，旧 server 下探针自动降级直发

pkg/agent/service/agent.go

@@ -11,6 +11,7 @@ import (
 	"runtime"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/pika-monitor/pika/internal/protocol"
@@ -22,6 +23,7 @@ import (
 	"github.com/pika-monitor/pika/pkg/agent/tamper"
 	"github.com/pika-monitor/pika/pkg/version"
 
+	"github.com/google/uuid"
 	"github.com/gorilla/websocket"
 	"github.com/jpillora/backoff"
 	"github.com/sourcegraph/conc"
@@ -32,6 +34,10 @@ const (
 	agentPongWait       = 30 * time.Second
 	agentWriteWait      = 5 * time.Second
 	agentCollectTimeout = 30 * time.Second
+	// agentDataWriteWait 数据帧写入超时。没有超时的写在对端停止读取
+	// （网络劣化、服务端处理阻塞）时会无限阻塞并持有连接写锁，
+	// 期间采集快照被覆盖造成静默丢数据。
+	agentDataWriteWait = 10 * time.Second
 )
 
 // safeConn 线程安全的 WebSocket 连接包装器
@@ -40,17 +46,23 @@ type safeConn struct {
 	mu   sync.Mutex
 }
 
-// WriteJSON 线程安全地写入 JSON 消息
+// WriteJSON 线程安全地写入 JSON 消息（带写超时）
 func (sc *safeConn) WriteJSON(v interface{}) error {
 	sc.mu.Lock()
 	defer sc.mu.Unlock()
+	if err := sc.conn.SetWriteDeadline(time.Now().Add(agentDataWriteWait)); err != nil {
+		return err
+	}
 	return sc.conn.WriteJSON(v)
 }
 
-// WriteMessage 线程安全地写入消息
+// WriteMessage 线程安全地写入消息（带写超时）
 func (sc *safeConn) WriteMessage(messageType int, data []byte) error {
 	sc.mu.Lock()
 	defer sc.mu.Unlock()
+	if err := sc.conn.SetWriteDeadline(time.Now().Add(agentDataWriteWait)); err != nil {
+		return err
+	}
 	return sc.conn.WriteMessage(messageType, data)
 }
 
@@ -75,12 +87,15 @@ func (sc *safeConn) WriteControl(messageType int, data []byte, deadline time.Tim
 type Agent struct {
 	cfg              *config.Config
 	idMgr            *id.Manager
+	bootID           string
 	cancel           context.CancelFunc
 	connMu           sync.RWMutex
 	activeConn       *safeConn
 	collectorMu      sync.RWMutex
 	collectorManager *collector.Manager
 	metricsStore     *metricsStore
+	outbox           *outbox
+	reliable         atomic.Bool
 	tamperProtector  *tamper.Protector
 	sshMonitor       *sshmonitor.Monitor
 }
@@ -90,8 +105,10 @@ func New(cfg *config.Config) *Agent {
 	return &Agent{
 		cfg:              cfg,
 		idMgr:            id.NewManager(),
+		bootID:           uuid.NewString(),
 		collectorManager: collector.NewManager(cfg),
 		metricsStore:     newMetricsStore(),
+		outbox:           newOutbox(),
 		tamperProtector:  tamper.NewProtector(),
 		sshMonitor:       sshmonitor.NewMonitor(),
 	}
@@ -201,7 +218,8 @@ func (a *Agent) runOnce(ctx context.Context, onRegistered func()) error {
 	})
 
 	// 发送注册消息
-	if err := a.registerAgent(conn); err != nil {
+	registerResp, err := a.registerAgent(conn)
+	if err != nil {
 		return fmt.Errorf("register failed: %w", err)
 	}
 	onRegistered()
@@ -212,6 +230,12 @@ func (a *Agent) runOnce(ctx context.Context, onRegistered func()) error {
 	// 重置发送游标，使下一个 sendLoop tick 立即发送当前全量快照，
 	// 保证重连瞬间服务端就能拿到最新数据
 	a.metricsStore.reset()
+	// 服务端声明支持可靠投递时，事件类消息改走 outbox；注册响应携带的
+	// 累计确认位点先修剪队列，避免重放服务端已处理的消息
+	a.reliable.Store(registerResp.Reliable)
+	if registerResp.Reliable && registerResp.AckSeq > 0 {
+		a.outbox.ack(registerResp.AckSeq)
+	}
 	defer func() {
 		a.setActiveConn(nil)
 	}()
@@ -243,6 +267,13 @@ func (a *Agent) runOnce(ctx context.Context, onRegistered func()) error {
 		}
 	})
 
+	// 事件类消息可靠发送（仅服务端支持时启用）
+	if registerResp.Reliable {
+		wg.Go(func() {
+			a.eventSendLoop(done, conn)
+		})
+	}
+
 	// 启动防篡改事件监控
 	wg.Go(func() {
 		a.tamperEventLoop(ctx, done)
@@ -321,6 +352,8 @@ func (a *Agent) readLoop(conn *websocket.Conn, done chan struct{}) error {
 			go a.handleSSHLoginConfig(msg.Data)
 		case protocol.MessageTypeUninstall:
 			go a.handleUninstall()
+		case protocol.MessageTypeAck:
+			a.handleAck(msg.Data)
 		default:
 			// 忽略其他类型
 		}
@@ -345,12 +378,14 @@ func (a *Agent) pingLoop(ctx context.Context, conn *safeConn, done chan struct{}
 	}
 }
 
-// registerAgent 注册探针
-func (a *Agent) registerAgent(conn *safeConn) error {
+// registerAgent 注册探针，返回服务端的注册响应（含可靠投递协商结果）
+func (a *Agent) registerAgent(conn *safeConn) (protocol.RegisterResponse, error) {
+	var registerResp protocol.RegisterResponse
+
 	// 加载或生成探针 ID
 	agentID, err := a.idMgr.Load()
 	if err != nil {
-		return fmt.Errorf("load agent ID failed: %w", err)
+		return registerResp, fmt.Errorf("load agent ID failed: %w", err)
 	}
 	slog.Info("Agent ID", "id", agentID, "path", a.idMgr.GetPath())
 
@@ -366,7 +401,8 @@ func (a *Agent) registerAgent(conn *safeConn) error {
 		agentName = hostname
 	}
 
-	// 构建注册请求
+	// 构建注册请求（BootID 标识本次探针进程，服务端据此识别重启并
+	// 重置事件流确认位点——纯内存 outbox 重启后 seq 从 1 重新分配）
 	registerReq := protocol.RegisterRequest{
 		AgentInfo: protocol.AgentInfo{
 			ID:       agentID,
@@ -377,41 +413,82 @@ func (a *Agent) registerAgent(conn *safeConn) error {
 			Version:  GetVersion(),
 		},
 		ApiKey: a.cfg.Server.APIKey,
+		BootID: a.bootID,
 	}
 
 	if err := conn.WriteJSON(protocol.OutboundMessage{
 		Type: protocol.MessageTypeRegister,
 		Data: registerReq,
 	}); err != nil {
-		return fmt.Errorf("send register message failed: %w", err)
+		return registerResp, fmt.Errorf("send register message failed: %w", err)
 	}
 
 	// 读取注册响应
 	var response protocol.InputMessage
 	if err := conn.ReadJSON(&response); err != nil {
-		return fmt.Errorf("read register response failed: %w", err)
+		return registerResp, fmt.Errorf("read register response failed: %w", err)
 	}
 
 	// 检查响应类型
 	if response.Type == protocol.MessageTypeRegisterErr {
 		var errResp protocol.RegisterResponse
 		if err := json.Unmarshal(response.Data, &errResp); err == nil {
-			return fmt.Errorf("register failed: %s", errResp.Message)
+			return registerResp, fmt.Errorf("register failed: %s", errResp.Message)
 		}
-		return fmt.Errorf("register failed: unknown error")
+		return registerResp, fmt.Errorf("register failed: unknown error")
 	}
 
 	if response.Type != protocol.MessageTypeRegisterAck {
-		return fmt.Errorf("register failed: unexpected response type %s", response.Type)
+		return registerResp, fmt.Errorf("register failed: unexpected response type %s", response.Type)
 	}
 
-	var registerResp protocol.RegisterResponse
 	if err := json.Unmarshal(response.Data, &registerResp); err != nil {
-		return fmt.Errorf("parse register response failed: %w", err)
+		return registerResp, fmt.Errorf("parse register response failed: %w", err)
 	}
 
-	slog.Info("注册成功", "agentId", registerResp.AgentID, "status", registerResp.Status)
-	return nil
+	slog.Info("注册成功", "agentId", registerResp.AgentID, "status", registerResp.Status,
+		"reliable", registerResp.Reliable, "serverAckSeq", registerResp.AckSeq)
+	return registerResp, nil
+}
+
+// handleAck 处理服务端的累计确认，修剪已投递的事件消息
+func (a *Agent) handleAck(data json.RawMessage) {
+	var ack protocol.AckData
+	if err := json.Unmarshal(data, &ack); err != nil {
+		slog.Warn("解析投递确认失败", "error", err)
+		return
+	}
+	a.outbox.ack(ack.Seq)
+}
+
+// eventSendLoop 事件类消息的可靠发送循环（每个连接一个）。
+// 连接建立后先从队头重放全部未确认消息（服务端按 seq 去重），之后增量
+// 发送新入队的消息；写失败直接返回并断开连接，未确认消息留给下一个
+// 连接重放。
+func (a *Agent) eventSendLoop(done chan struct{}, conn *safeConn) {
+	var lastSent uint64
+	for {
+		for _, msg := range a.outbox.after(lastSent) {
+			if err := conn.WriteJSON(msg); err != nil {
+				slog.Warn("事件消息发送失败，等待重连后重放", "error", err, "seq", msg.Seq)
+				a.killConn(conn)
+				return
+			}
+			lastSent = msg.Seq
+		}
+		if !a.outbox.wait(done) {
+			return
+		}
+	}
+}
+
+// killConn 写失败时立即关闭连接，让读循环马上感知并触发重连，而不是
+// 等 30 秒 pong 超时。Close 可安全并发调用，会以错误唤醒阻塞中的读写。
+func (a *Agent) killConn(conn *safeConn) {
+	if conn == nil {
+		return
+	}
+	_ = conn.Close()
 }
 
 func (a *Agent) handleMonitorConfig(data json.RawMessage) {
@@ -434,21 +511,16 @@ func (a *Agent) handleMonitorConfig(data json.RawMessage) {
 
 	slog.Info("收到服务监控配置，立即执行检测", "count", len(payload.Items))
 
-	// 立即执行一次监控检测，结果以单元素 batch 上报
+	// 立即执行一次监控检测，结果以单元素 batch 上报。
+	// 走 sendOutboundMessage（可靠模式下进 outbox），断线期间不丢
 	sample := manager.CollectMonitor(payload.Items)
-	conn := a.getActiveConn()
-	if conn == nil {
-		slog.Warn("当前连接不可用，无法上报监控结果")
-		return
-	}
-
-	if err := conn.WriteJSON(protocol.OutboundMessage{
+	if err := a.sendOutboundMessage(protocol.OutboundMessage{
 		Type: protocol.MessageTypeMetrics,
 		Data: protocol.MetricsBatch{Samples: []protocol.MetricSample{sample}},
 	}); err != nil {
 		slog.Warn("发送监控结果失败", "error", err)
 	} else {
-		slog.Info("服务监控检测完成，已上报监控项结果", "count", len(payload.Items))
+		slog.Info("服务监控检测完成，已提交监控项结果", "count", len(payload.Items))
 	}
 }
 
@@ -464,13 +536,22 @@ func (a *Agent) getActiveConn() *safeConn {
 	return a.activeConn
 }
 
+// sendOutboundMessage 发送事件类出站消息。服务端支持可靠投递时入队
+// outbox，由专用协程发送并等待服务端确认，断线期间不丢；否则降级为
+// 直接发送（与旧服务端兼容）。
 func (a *Agent) sendOutboundMessage(msg protocol.OutboundMessage) error {
+	if a.reliable.Load() {
+		a.outbox.enqueue(msg)
+		return nil
+	}
+
 	conn := a.getActiveConn()
 	if conn == nil {
 		return fmt.Errorf("connection is unavailable")
 	}
 
 	if err := conn.WriteJSON(msg); err != nil {
+		a.killConn(conn)
 		return err
 	}
 
@@ -571,7 +652,9 @@ func (a *Agent) sendMetricsOnce() {
 		Data: protocol.MetricsBatch{Samples: samples},
 	}); err != nil {
 		slog.Warn("发送指标 batch 失败", "error", err)
-		// 连接写失败仅用于日志，重连由 readLoop/pingLoop 驱动
+		// 写失败说明连接已不可用，主动断开让读循环立刻触发重连；
+		// 未确认样本由 metricsStore 在重连后重发（latest-wins）
+		a.killConn(conn)
 		return
 	}
 

pkg/agent/service/outbox.go

@@ -0,0 +1,134 @@
+package service
+
+import (
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/pika-monitor/pika/internal/protocol"
+)
+
+const (
+	// outboxMaxEntries 待确认队列上限。溢出时丢弃最旧消息：新事件（尤其是
+	// 安全事件）比旧事件更有价值，且队列只增不减通常意味着探针已离线过久。
+	outboxMaxEntries = 1024
+
+	// outboxDropLogInterval 溢出告警的日志节流，避免事件风暴时刷屏
+	outboxDropLogInterval = 30 * time.Second
+)
+
+type outboxEntry struct {
+	seq uint64
+	msg protocol.OutboundMessage
+}
+
+// outbox 事件类出站消息的有界待确认队列，实现 at-least-once 投递：
+//
+//   - 每条消息入队时分配单调递增 seq；
+//   - 网络写成功不代表删除，只有收到服务端累计 ack（或重连时注册响应
+//     携带的 AckSeq）才修剪队列；
+//   - 每次新连接从队头整体重放未确认消息，由服务端按 seq 去重。
+//
+// 纯内存实现：进程重启后清空，seq 从 1 重新分配。服务端通过注册请求
+// 携带的 BootID 识别探针重启并重置确认位点，避免新序号被旧位点误判
+// 为重复消息。
+type outbox struct {
+	mu       sync.Mutex
+	entries  []outboxEntry
+	nextSeq  uint64
+	ackedSeq uint64
+	dropped  uint64
+	lastDrop time.Time
+	notify   chan struct{}
+}
+
+func newOutbox() *outbox {
+	return &outbox{
+		notify: make(chan struct{}, 1),
+	}
+}
+
+// enqueue 分配序号并入队，返回分配的 seq。队列满时丢弃最旧消息。
+func (o *outbox) enqueue(msg protocol.OutboundMessage) uint64 {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	o.nextSeq++
+	msg.Seq = o.nextSeq
+	o.entries = append(o.entries, outboxEntry{seq: msg.Seq, msg: msg})
+
+	if len(o.entries) > outboxMaxEntries {
+		drop := len(o.entries) - outboxMaxEntries
+		o.entries = o.entries[drop:]
+		o.dropped += uint64(drop)
+		if time.Since(o.lastDrop) >= outboxDropLogInterval {
+			slog.Warn("事件待确认队列溢出，已丢弃最旧消息",
+				"droppedTotal", o.dropped, "queueSize", len(o.entries))
+			o.lastDrop = time.Now()
+		}
+	}
+
+	o.signal()
+	return msg.Seq
+}
+
+// after 返回 seq 大于 lastSent 的消息快照（按序），供发送协程增量发送或
+// 连接建立后整体重放。ack 修剪只从队头移除，不影响该判定。
+func (o *outbox) after(lastSent uint64) []protocol.OutboundMessage {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	msgs := make([]protocol.OutboundMessage, 0, len(o.entries))
+	for _, e := range o.entries {
+		if e.seq > lastSent {
+			msgs = append(msgs, e.msg)
+		}
+	}
+	return msgs
+}
+
+// ack 记录服务端累计确认并修剪队列，返回本次修剪的消息数。
+func (o *outbox) ack(seq uint64) int {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	if seq <= o.ackedSeq {
+		return 0
+	}
+	o.ackedSeq = seq
+
+	kept := o.entries[:0]
+	for _, e := range o.entries {
+		if e.seq > seq {
+			kept = append(kept, e)
+		}
+	}
+	trimmed := len(o.entries) - len(kept)
+	o.entries = kept
+	return trimmed
+}
+
+// stats 返回队列统计（用于日志与观测）。
+func (o *outbox) stats() (pending int, acked uint64, dropped uint64) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return len(o.entries), o.ackedSeq, o.dropped
+}
+
+// signal 唤醒发送协程（非阻塞，协程自带快照逻辑不会漏消息）。
+func (o *outbox) signal() {
+	select {
+	case o.notify <- struct{}{}:
+	default:
+	}
+}
+
+// wait 阻塞等待新消息或退出信号。
+func (o *outbox) wait(done <-chan struct{}) bool {
+	select {
+	case <-o.notify:
+		return true
+	case <-done:
+		return false
+	}
+}

pkg/agent/service/outbox_test.go

@@ -0,0 +1,106 @@
+package service
+
+import (
+	"testing"
+
+	"github.com/pika-monitor/pika/internal/protocol"
+)
+
+func TestOutboxSeqAssignment(t *testing.T) {
+	o := newOutbox()
+	for i := 1; i <= 3; i++ {
+		seq := o.enqueue(protocol.OutboundMessage{Type: protocol.MessageTypeSSHLoginEvent})
+		if seq != uint64(i) {
+			t.Fatalf("enqueue #%d got seq %d", i, seq)
+		}
+	}
+
+	msgs := o.after(0)
+	if len(msgs) != 3 {
+		t.Fatalf("after(0) returned %d messages, want 3", len(msgs))
+	}
+	for i, msg := range msgs {
+		if msg.Seq != uint64(i+1) {
+			t.Fatalf("msgs[%d].Seq = %d, want %d", i, msg.Seq, i+1)
+		}
+	}
+}
+
+func TestOutboxAckTrimsPrefix(t *testing.T) {
+	o := newOutbox()
+	for i := 0; i < 5; i++ {
+		o.enqueue(protocol.OutboundMessage{Type: protocol.MessageTypeTamperEvent})
+	}
+
+	if trimmed := o.ack(3); trimmed != 3 {
+		t.Fatalf("ack(3) trimmed %d, want 3", trimmed)
+	}
+	if pending, acked, _ := o.stats(); pending != 2 || acked != 3 {
+		t.Fatalf("after ack(3): pending=%d acked=%d, want pending=2 acked=3", pending, acked)
+	}
+
+	// 重复/回退的 ack 无效
+	if trimmed := o.ack(2); trimmed != 0 {
+		t.Fatalf("stale ack(2) trimmed %d, want 0", trimmed)
+	}
+
+	// 剩余消息 seq 保持原值，重放判定不受修剪影响
+	msgs := o.after(0)
+	if len(msgs) != 2 || msgs[0].Seq != 4 || msgs[1].Seq != 5 {
+		t.Fatalf("remaining messages = %+v, want seq 4,5", msgs)
+	}
+}
+
+func TestOutboxAfterSkipsSent(t *testing.T) {
+	o := newOutbox()
+	for i := 0; i < 4; i++ {
+		o.enqueue(protocol.OutboundMessage{Type: protocol.MessageTypeCommandResp})
+	}
+
+	msgs := o.after(2)
+	if len(msgs) != 2 || msgs[0].Seq != 3 || msgs[1].Seq != 4 {
+		t.Fatalf("after(2) = %+v, want seq 3,4", msgs)
+	}
+
+	// 全部已发送时返回空
+	if msgs := o.after(4); len(msgs) != 0 {
+		t.Fatalf("after(4) = %+v, want empty", msgs)
+	}
+}
+
+func TestOutboxOverflowDropsOldest(t *testing.T) {
+	o := newOutbox()
+	for i := 0; i < outboxMaxEntries+50; i++ {
+		o.enqueue(protocol.OutboundMessage{Type: protocol.MessageTypeTamperEvent})
+	}
+
+	pending, _, dropped := o.stats()
+	if pending != outboxMaxEntries {
+		t.Fatalf("pending = %d, want %d", pending, outboxMaxEntries)
+	}
+	if dropped != 50 {
+		t.Fatalf("dropped = %d, want 50", dropped)
+	}
+
+	// 队列保留的是最新的消息
+	msgs := o.after(0)
+	first := uint64(outboxMaxEntries + 50 - outboxMaxEntries + 1)
+	if len(msgs) != outboxMaxEntries || msgs[0].Seq != first {
+		t.Fatalf("oldest retained seq = %d, want %d", msgs[0].Seq, first)
+	}
+}
+
+func TestOutboxEnqueueSetsMsgSeq(t *testing.T) {
+	o := newOutbox()
+	msg := protocol.OutboundMessage{Type: protocol.MessageTypeCommandResp}
+	o.enqueue(msg)
+
+	// enqueue 内部复制赋值，入队消息带 seq
+	if msg.Seq != 0 {
+		t.Fatalf("caller message mutated, seq = %d", msg.Seq)
+	}
+	stored := o.after(0)
+	if len(stored) != 1 || stored[0].Seq != 1 {
+		t.Fatalf("stored = %+v, want one message with seq 1", stored)
+	}
+}