[修复] 修复 WS 连接下更新无法下发

[优化] 更新 openresty_observability_port 描述，增强健康检查逻辑，使用 stub_status 代替 openresty -t
2026-09-29 05:56:38 +08:00 · 2026-05-29 11:08:05 +08:00 · 2026-05-29 10:52:25 +08:00 · 2026-05-29 10:50:13 +08:00 · 2026-05-29 10:44:14 +08:00 · 2026-05-29 10:34:25 +08:00
40 changed files with 1681 additions and 108 deletions
@@ -6,7 +6,7 @@

 </div>

-<p align="center">
+<p align="center
  <a href="https://raw.githubusercontent.com/Rain-kl/OpenFlare/main/LICENSE">
    <img src="https://img.shields.io/github/license/Rain-kl/OpenFlare?color=brightgreen" alt="license">
  </a>
@@ -118,7 +118,6 @@ Docker 部署可直接运行 Agent 镜像：
 ```bash
 docker run -d --name openflare-agent --restart unless-stopped \
  -p 80:80 -p 443:443 -p 127.0.0.1:18081:18081 \
-  -v openflare-agent-data:/data \
  -e OPENFLARE_SERVER_URL=http://your-server:3000 \
  -e OPENFLARE_AGENT_TOKEN=YOUR_AGENT_TOKEN \
  ghcr.io/rain-kl/openflare-agent:latest
@@ -139,7 +138,7 @@ curl -fsSL https://raw.githubusercontent.com/Rain-kl/OpenFlare/main/scripts/unin
 1. 登录管理端并新增反代规则
 2. 在发布前查看预览或变更摘要
 3. 激活新版本
-4. 等待 Agent 在后续 heartbeat 中拉取并应用配置
+4. Agent 通过 WebSocket 通知或后续 heartbeat 拉取并应用配置

 版本号格式固定为 `YYYYMMDD-NNN`，历史版本不可变，回滚通过重新激活旧版本完成。

@@ -17,6 +17,7 @@ services:
    environment:
      OPENFLARE_SERVER_URL: "http://host.docker.internal:3000"
      OPENFLARE_AGENT_TOKEN: "373956188ddead1df6dd7c86cd330b73"
+      LOG_LEVEL: "debug"

    extra_hosts:
      - "host.docker.internal:host-gateway"
@@ -89,6 +89,8 @@ Agent 执行 OpenResty 校验与 reload
 Agent 上报应用结果
 ```

+默认启用 WS 连接升级时，Agent 会先通过 HTTP heartbeat 获取设置，随后尝试连接 Agent WebSocket。WS 成功后，周期性状态上报改由 WS 承载；Server 发布或激活版本后会向已连接 Agent 广播激活版本摘要，使 Agent 立即进入既有同步流程。WS 断开或建立失败时，Agent 自动退回 HTTP heartbeat。
+
 ### 反向代理流

 ```text
@@ -234,11 +234,14 @@ Agent 必须满足：
 * 启动后读取或生成本地 `node_id`。
 * 周期性心跳与同步。
 * 常规同步优先依据 heartbeat 返回的版本摘要判断。
+* WS 连接升级开启且连接成功时，Agent 可通过 WS 接收激活版本摘要并立即同步；WS 失败或断开必须退回 HTTP heartbeat。
 * 发现新版本时先备份旧文件。
 * 写入主配置、路由配置与必要证书文件。
 * 写入新配置后执行 `openresty -t -c <main_config_path>`，再 reload；reload 发现运行时未启动时允许直接启动 OpenResty。
+* 周期性运行时健康检查不得调用 `openresty -t`，避免健康探针触发 upstream 域名同步解析；应优先请求本地 `openresty_observability_port` 上的 `/openflare/stub_status`，以 HTTP `200 OK` 作为 OpenResty 主进程和 worker 正在提供服务的判断依据。
 * 新配置激活失败时必须先尝试用目标配置恢复运行，再回滚到旧配置并重新拉起 OpenResty。
-* 回滚后 OpenResty 恢复正常时上报警告；回滚后仍无法恢复运行时上报失败。
+* 回滚后 OpenResty 恢复正常时上报警告；如果本地没有历史主配置可恢复，必须允许写入内置安全兜底配置并拉起对外只监听 `80` 端口、统一返回 `503` 的 OpenResty 运行态；兜底配置仍需保留本地 `stub_status` 健康检查入口。
+* 兜底运行态不得清除失败目标的阻断状态；应用记录必须能体现目标版本失败但 fallback runtime 已启动。存在历史主配置但回滚后仍无法恢复运行时上报失败。
 * 某个目标 `version + checksum` 一旦应用失败并回退，Agent 必须在本地状态中阻断该目标的重复应用。

 ## 前端请求、状态与类型
@@ -35,7 +35,7 @@ Server 发布时必须：

 全局同时只能有一个激活版本。当前不做按节点分组的差异化版本。

-Agent 通过 heartbeat 获取激活版本摘要；当远端版本或 checksum 与本地状态不一致时，Agent 才进入同步流程。
+Agent 通过 heartbeat 获取激活版本摘要；当远端版本或 checksum 与本地状态不一致时，Agent 才进入同步流程。当 Agent WS 连接升级开启且连接可用时，Server 在发布或激活版本成功后会广播最新激活版本摘要，Agent 收到后复用普通同步流程立即拉取并应用配置。WS 不可用时仍按 HTTP heartbeat 间隔发现变更。

 ## 不可变历史

@@ -58,7 +58,7 @@ Agent 发现新版本后会：
 5. reload；如果运行时未启动，则尝试用当前配置启动 OpenResty。
 6. 上报成功、警告或失败。

-如果新配置激活失败，Agent 必须尝试恢复运行；回滚成功时上报警告，回滚后仍无法恢复运行时上报失败。
+如果新配置激活失败，Agent 必须尝试恢复运行；回滚成功时上报警告。若本地没有历史主配置可回滚，Agent 会写入内置安全兜底配置并尝试拉起 OpenResty：该配置对外只监听 `80` 端口，不包含任何用户路由，统一返回 `503 Service Unavailable` 与 `OpenFlare: No Valid Configuration`，同时保留本地 `stub_status` 健康检查入口。兜底启动成功时仍阻断失败目标版本并上报警告；存在历史主配置但回滚后仍无法恢复运行时上报失败。

 某个目标 `version + checksum` 一旦应用失败并回退，Agent 会在本地状态中阻断该目标重复应用。只有远端激活版本或 checksum 发生变化，才允许再次尝试。

@@ -167,6 +167,8 @@ OpenResty runtime:
 ps aux | grep openresty
 ```

+Agent periodic health checks use local `http://127.0.0.1:<openresty_observability_port>/openflare/stub_status` instead of repeatedly running `openresty -t`. If a node is unhealthy, first confirm that the local observability port is listening. If `host not found in upstream` only appears during apply, the failure comes from config validation or reload, not the periodic health probe.
+
 Use the actual `openresty_path` and `main_config_path` from `agent.json`.

 ## HTTPS Does Not Work
@@ -63,7 +63,7 @@ Agent supports the `-config` CLI flag, an `agent.json` file, and the `LOG_LEVEL`
 | `openresty_path` | OpenResty binary path | no | `openresty` |
 | `openresty_container_name` | Deprecated Docker-control field, read for compatibility only | no | empty |
 | `openresty_docker_image` | Deprecated Docker-control field, read for compatibility only | no | empty |
-| `openresty_observability_port` | Local observability port | no | `18081` |
+| `openresty_observability_port` | Local observability and OpenResty health-check port | no | `18081` |
 | `docker_binary` | Deprecated Docker-control field, read for compatibility only | no | empty |
 | `data_dir` | Agent data directory | no | `data` under config directory |
 | `access_log_path` | OpenResty access log path | no | `data_dir/var/log/openflare/access.log` |
@@ -224,6 +224,8 @@ export LOG_LEVEL='info'

 未配置 `openresty_path` 时，Agent 默认调用 `openresty`。

+默认情况下，Agent 在 HTTP 心跳成功后会尝试升级为 WebSocket。升级成功时，Server 发布或激活配置会立即通知 Agent；如果 WebSocket 无法建立或意外断开，Agent 会自动退回 HTTP 心跳同步。
+
 ## 最小联调步骤

 1. 启动 Server 并完成首次登录。
@@ -143,6 +143,8 @@ journalctl -u openflare-agent -f

 注意：某个目标 `version + checksum` 一旦应用失败并回退，Agent 会在本地状态中阻断该目标重复应用。修正配置后需要重新发布生成新的 checksum，或激活旧版本回滚。

+如果这是 Agent 首次应用配置，且本地没有历史 `nginx.conf` 可回滚，失败目标仍会被阻断，但 Agent 会尝试进入安全兜底运行态。此时应用记录和 Agent 日志会包含 `fallback runtime started`，OpenResty 对外只监听 `80` 端口并统一返回 `503` 与 `OpenFlare: No Valid Configuration`，同时保留本地 `stub_status` 健康检查入口。修正配置并重新发布新版本后，Agent 会覆盖兜底配置并恢复正常代理。
+
 ## OpenResty 应用失败

 常见原因：
@@ -167,6 +169,8 @@ OpenResty 运行状态：
 ps aux | grep openresty
 ```

+Agent 周期性健康检查通过本地 `http://127.0.0.1:<openresty_observability_port>/openflare/stub_status` 判断 OpenResty 是否存活，不会反复执行 `openresty -t`。如果节点被标记为 unhealthy，优先确认该本地观测端口是否正在监听；如果只在应用配置时出现 `host not found in upstream`，说明失败来自配置校验或 reload，而不是周期性健康探针。
+
 实际二进制路径和主配置路径以 `agent.json` 中的 `openresty_path` 与 `main_config_path` 为准。

 ## HTTPS 不生效
@@ -72,6 +72,7 @@ go run . --port 3000 --log-dir ./logs
 | 配置项 | 作用 | 默认值 |
 | --- | --- | --- |
 | `AgentHeartbeatInterval` | Agent 心跳间隔（毫秒） | `10000` |
+| `AgentWebsocketUpgradeEnabled` | 是否允许 Agent 在 HTTP 心跳成功后升级为 WebSocket | `true` |
 | `NodeOfflineThreshold` | 节点离线阈值（毫秒） | `120000` |
 | `AgentUpdateRepo` | Agent 自更新仓库 | `Rain-kl/OpenFlare` |
 | `GeoIPProvider` | 节点/IP 归属解析方式 | `ipinfo` |
@@ -165,7 +166,7 @@ OpenResty 性能参数与缓存参数继续统一保存在 `Option` 表。当前
 | `openresty_path` | OpenResty 二进制路径 | 否 | `openresty` |
 | `openresty_container_name` | 旧 Docker 控制字段，仅兼容读取 | 否 | 空 |
 | `openresty_docker_image` | 旧 Docker 控制字段，仅兼容读取 | 否 | 空 |
-| `openresty_observability_port` | 本地观测端口 | 否 | `18081` |
+| `openresty_observability_port` | 本地观测与 OpenResty 健康检查端口 | 否 | `18081` |
 | `docker_binary` | 旧 Docker 控制字段，仅兼容读取 | 否 | 空 |
 | `data_dir` | Agent 数据目录 | 否 | 配置文件所在目录下的 `data` |
 | `main_config_path` | OpenResty 主配置写入路径 | 否 | `data_dir/etc/nginx/nginx.conf` |
@@ -186,7 +187,9 @@ OpenResty 性能参数与缓存参数继续统一保存在 `Option` 表。当前

 * `agent_token` 与 `discovery_token` 不能同时为空。
 * `heartbeat_interval` 与 `request_timeout` 支持毫秒整数或 Go duration 字符串。
+* Server 运行时配置 `AgentWebsocketUpgradeEnabled` 开启时，Agent 会在 HTTP 心跳成功后尝试升级为 WebSocket；连接失败或断开后自动退回 HTTP 心跳。
 * 未配置 `openresty_path` 时默认调用 `openresty`。
+* Agent 周期性健康检查会请求 `http://127.0.0.1:<openresty_observability_port>/openflare/stub_status`，不再通过高频 `openresty -t` 判断运行时健康；配置应用、启动恢复和 reload 前校验仍会执行 `openresty -t -c <main_config_path>`。
 * 如果 `agent.json` 不存在，但 `OPENFLARE_SERVER_URL` 与 Token 等环境变量足够，Agent 可以直接启动；两者同时存在时环境变量优先。
 * Agent 自动探测到私网 `node_ip` 时，Server 会在注册/心跳阶段优先保留 Agent 直连来源的公网地址，避免 NAT/多网卡场景误登记内网网卡地址。

@@ -17,6 +17,7 @@ import (
 	"openflare-agent/internal/state"
 	syncservice "openflare-agent/internal/sync"
 	"openflare-agent/internal/updater"
+	"openflare-agent/internal/wsclient"
 )

 func main() {
@@ -56,6 +57,7 @@ func main() {
 	)

 	client := httpclient.New(cfg.ServerURL, cfg.InitialAuthToken(), cfg.RequestTimeout.Duration())
+	wsClient := wsclient.New(cfg.ServerURL, cfg.InitialAuthToken(), cfg.RequestTimeout.Duration())
 	stateStore := state.NewStore(cfg.StatePath)
 	observabilityBuffer := state.NewObservabilityBufferStore(cfg.ObservabilityBufferPath)
 	runtimeManager := &nginx.Manager{
@@ -93,6 +95,7 @@ func main() {
 		SyncService:         syncservice.New(client, runtimeManager, stateStore),
 		Updater:             updater.New(),
 		RuntimeManager:      runtimeManager,
+		WebSocketService:    wsClient,
 	}

 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
@@ -2,6 +2,9 @@ module openflare-agent

 go 1.25.0

-require openflare v0.0.0
+require (
+	golang.org/x/net v0.53.0
+	openflare v0.0.0
+)

 replace openflare => ../openflare_server
@@ -0,0 +1,2 @@
+golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
+golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
@@ -2,6 +2,7 @@ package agent

 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"log/slog"
 	"strings"
@@ -22,6 +23,7 @@ type HeartbeatService interface {
 type SyncService interface {
 	SyncOnStartup(ctx context.Context, target *protocol.ActiveConfigMeta) error
 	SyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error
+	ForceSyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error
 }

 type Updater interface {
@@ -33,6 +35,12 @@ type RuntimeManager interface {
 	Restart(ctx context.Context) error
 }

+type WebSocketService interface {
+	Connect(ctx context.Context) (protocol.WebSocketConnection, error)
+	SetToken(token string)
+	URL() string
+}
+
 type UpdateOptions struct {
 	Channel string
 	TagName string
@@ -47,13 +55,15 @@ type Runner struct {
 	SyncService         SyncService
 	Updater             Updater
 	RuntimeManager      RuntimeManager
+	WebSocketService    WebSocketService

-	autoUpdate          bool
-	updateNow           bool
-	updateRepo          string
-	updateChan          string
-	updateTag           string
-	restartOpenrestyNow bool
+	autoUpdate              bool
+	updateNow               bool
+	updateRepo              string
+	updateChan              string
+	updateTag               string
+	restartOpenrestyNow     bool
+	websocketUpgradeEnabled bool
 }

 func (r *Runner) Run(ctx context.Context) error {
@@ -63,26 +73,8 @@ func (r *Runner) Run(ctx context.Context) error {
 	}
 	slog.Info("agent runner started", "node_id", nodeID, "node", r.Config.NodeName, "ip", r.Config.NodeIP)
 	if r.hasAgentToken() {
-		r.refreshOpenrestyHealth(ctx)
-		payload, ackWindows := r.prepareHeartbeatPayload(nodeID)
-		heartbeatResult, hbErr := r.HeartbeatService.Heartbeat(ctx, payload)
-		if hbErr != nil {
+		if _, hbErr := r.performHeartbeatCycle(ctx, nodeID, true); hbErr != nil {
 			slog.Error("agent startup heartbeat failed", "error", hbErr)
-		} else {
-			r.ackObservabilityWindows(ackWindows)
-			if heartbeatResult == nil {
-				heartbeatResult = &protocol.HeartbeatResult{}
-			}
-			slog.Debug("agent startup heartbeat succeeded", "node_id", nodeID)
-			r.applySettings(heartbeatResult.AgentSettings)
-			if err = r.SyncService.SyncOnStartup(ctx, heartbeatResult.ActiveConfig); err != nil {
-				r.recordSyncError(err)
-				slog.Error("agent startup sync failed", "error", err)
-			} else {
-				slog.Debug("agent startup sync completed")
-			}
-			r.tryRestartOpenresty(ctx)
-			r.tryAutoUpdate(ctx)
 		}
 	} else if err = r.tryRegister(ctx, &nodeID); err != nil {
 		slog.Error("agent initial discovery register failed", "error", err)
@@ -90,43 +82,278 @@ func (r *Runner) Run(ctx context.Context) error {

 	heartbeatTicker := time.NewTicker(r.Config.HeartbeatInterval.Duration())
 	defer heartbeatTicker.Stop()
+	var wsDone <-chan error
+	wsBackoff := newWebSocketBackoff()
+	nextWSAttempt := time.Now()
+	tryStartWebSocket := func() {
+		if wsDone != nil || !r.shouldUseWebSocket() || time.Now().Before(nextWSAttempt) {
+			return
+		}
+		done, startErr := r.startWebSocket(ctx, nodeID)
+		if startErr != nil {
+			delay := wsBackoff.Next()
+			nextWSAttempt = time.Now().Add(delay)
+			slog.Debug("agent ws upgrade failed; falling back to http heartbeat",
+				"enabled", r.websocketUpgradeEnabled,
+				"url", r.websocketURL(),
+				"retry_after", delay,
+				"error", startErr,
+			)
+			return
+		}
+		wsBackoff.Reset()
+		wsDone = done
+		slog.Debug("agent switched to websocket mode", "url", r.websocketURL())
+	}
+	tryStartWebSocket()

 	for {
 		select {
 		case <-ctx.Done():
 			slog.Info("agent runner shutting down", "error", ctx.Err())
 			return ctx.Err()
+		case wsErr := <-wsDone:
+			wsDone = nil
+			delay := wsBackoff.Next()
+			nextWSAttempt = time.Now().Add(delay)
+			slog.Debug("agent ws disconnected; resuming http heartbeat", "retry_after", delay, "error", wsErr)
+			if r.hasAgentToken() {
+				if _, hbErr := r.performHeartbeatCycle(ctx, nodeID, false); hbErr != nil {
+					slog.Error("agent heartbeat after ws disconnect failed", "error", hbErr)
+				}
+			}
 		case <-heartbeatTicker.C:
+			if wsDone != nil {
+				continue
+			}
 			if !r.hasAgentToken() {
 				if err = r.tryRegister(ctx, &nodeID); err != nil {
 					slog.Error("agent discovery register failed", "error", err)
 				}
 				continue
 			}
-			r.refreshOpenrestyHealth(ctx)
-			payload, ackWindows := r.prepareHeartbeatPayload(nodeID)
-			heartbeatResult, hbErr := r.HeartbeatService.Heartbeat(ctx, payload)
-			if hbErr != nil {
+			if changed, hbErr := r.performHeartbeatCycle(ctx, nodeID, false); hbErr != nil {
 				slog.Error("agent heartbeat failed", "error", hbErr)
 			} else {
-				r.ackObservabilityWindows(ackWindows)
-				if heartbeatResult == nil {
-					heartbeatResult = &protocol.HeartbeatResult{}
-				}
-				if changed := r.applySettings(heartbeatResult.AgentSettings); changed {
+				if changed {
 					heartbeatTicker.Reset(r.Config.HeartbeatInterval.Duration())
 				}
-				if err = r.SyncService.SyncOnce(ctx, heartbeatResult.ActiveConfig); err != nil {
-					r.recordSyncError(err)
-					slog.Error("agent sync failed", "error", err)
-				}
-				r.tryRestartOpenresty(ctx)
-				r.tryAutoUpdate(ctx)
+				tryStartWebSocket()
 			}
 		}
 	}
 }

+func (r *Runner) performHeartbeatCycle(ctx context.Context, nodeID string, startup bool) (bool, error) {
+	r.refreshOpenrestyHealth(ctx)
+	payload, ackWindows := r.prepareHeartbeatPayload(nodeID)
+	heartbeatResult, err := r.HeartbeatService.Heartbeat(ctx, payload)
+	if err != nil {
+		return false, err
+	}
+	r.ackObservabilityWindows(ackWindows)
+	if heartbeatResult == nil {
+		heartbeatResult = &protocol.HeartbeatResult{}
+	}
+	mode := "periodic"
+	if startup {
+		mode = "startup"
+	}
+	slog.Debug("agent heartbeat succeeded", "mode", mode, "node_id", nodeID)
+	changed := r.applySettings(heartbeatResult.AgentSettings)
+	if startup {
+		if err = r.SyncService.SyncOnStartup(ctx, heartbeatResult.ActiveConfig); err != nil {
+			r.recordSyncError(err)
+			slog.Error("agent startup sync failed", "error", err)
+		} else {
+			slog.Debug("agent startup sync completed")
+		}
+	} else if err = r.SyncService.SyncOnce(ctx, heartbeatResult.ActiveConfig); err != nil {
+		r.recordSyncError(err)
+		slog.Error("agent sync failed", "error", err)
+	}
+	r.tryRestartOpenresty(ctx)
+	r.tryAutoUpdate(ctx)
+	return changed, nil
+}
+
+func (r *Runner) shouldUseWebSocket() bool {
+	enabled := r.WebSocketService != nil && r.websocketUpgradeEnabled && r.hasAgentToken()
+	slog.Debug("agent ws upgrade eligibility checked", "enabled", enabled, "server_enabled", r.websocketUpgradeEnabled, "url", r.websocketURL())
+	return enabled
+}
+
+func (r *Runner) websocketURL() string {
+	if r.WebSocketService == nil {
+		return ""
+	}
+	return r.WebSocketService.URL()
+}
+
+func (r *Runner) startWebSocket(ctx context.Context, nodeID string) (<-chan error, error) {
+	if r.WebSocketService == nil {
+		return nil, errors.New("websocket service is not configured")
+	}
+	conn, err := r.WebSocketService.Connect(ctx)
+	if err != nil {
+		return nil, err
+	}
+	done := make(chan error, 1)
+	go func() {
+		defer func() {
+			_ = conn.Close()
+		}()
+		done <- r.runWebSocket(ctx, nodeID, conn)
+	}()
+	return done, nil
+}
+
+func (r *Runner) runWebSocket(ctx context.Context, nodeID string, conn protocol.WebSocketConnection) error {
+	slog.Debug("agent ws connected", "url", conn.URL(), "node_id", nodeID)
+	statusTicker := time.NewTicker(r.Config.HeartbeatInterval.Duration())
+	defer statusTicker.Stop()
+
+	messages := make(chan protocol.WSMessage, 8)
+	readDone := make(chan error, 1)
+	go func() {
+		for {
+			message, err := conn.Receive()
+			if err != nil {
+				readDone <- err
+				return
+			}
+			select {
+			case messages <- message:
+			case <-ctx.Done():
+				readDone <- ctx.Err()
+				return
+			}
+		}
+	}()
+
+	if err := r.sendWebSocketStatus(ctx, nodeID, conn); err != nil {
+		return err
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case err := <-readDone:
+			return err
+		case <-statusTicker.C:
+			if err := r.sendWebSocketStatus(ctx, nodeID, conn); err != nil {
+				return err
+			}
+		case message := <-messages:
+			changed, err := r.handleWebSocketMessage(ctx, message, conn)
+			if err != nil {
+				return err
+			}
+			if changed {
+				statusTicker.Reset(r.Config.HeartbeatInterval.Duration())
+			}
+		}
+	}
+}
+
+func (r *Runner) sendWebSocketStatus(ctx context.Context, nodeID string, conn protocol.WebSocketConnection) error {
+	r.refreshOpenrestyHealth(ctx)
+	payload, ackWindows := r.prepareHeartbeatPayload(nodeID)
+	if err := conn.SendStatus(payload); err != nil {
+		return err
+	}
+	r.ackObservabilityWindows(ackWindows)
+	return nil
+}
+
+func (r *Runner) handleWebSocketMessage(ctx context.Context, message protocol.WSMessage, conn protocol.WebSocketConnection) (bool, error) {
+	switch message.Type {
+	case protocol.WSMessageTypeSettings:
+		var settings protocol.AgentSettings
+		if err := json.Unmarshal(message.Payload, &settings); err != nil {
+			slog.Debug("agent ws settings decode failed", "error", err)
+			return false, nil
+		}
+		changed := r.applySettings(&settings)
+		r.tryRestartOpenresty(ctx)
+		r.tryAutoUpdate(ctx)
+		if !r.websocketUpgradeEnabled {
+			slog.Debug("agent ws disabled by server settings; falling back to http heartbeat")
+			return changed, errors.New("websocket upgrade disabled by server")
+		}
+		return changed, nil
+	case protocol.WSMessageTypeActiveConfig:
+		var target protocol.ActiveConfigMeta
+		if err := json.Unmarshal(message.Payload, &target); err != nil {
+			slog.Debug("agent ws active config decode failed", "error", err)
+			return false, nil
+		}
+		slog.Debug("agent ws active config received", "version", target.Version, "checksum", target.Checksum, "trigger_sync", true)
+		if err := r.SyncService.SyncOnce(ctx, &target); err != nil {
+			r.recordSyncError(err)
+			slog.Error("agent ws triggered sync failed", "version", target.Version, "error", err)
+		}
+		return false, nil
+	case protocol.WSMessageTypeForceSyncConfig:
+		var target protocol.ActiveConfigMeta
+		if err := json.Unmarshal(message.Payload, &target); err != nil {
+			slog.Debug("agent ws force sync config decode failed", "error", err)
+			return false, nil
+		}
+		slog.Debug("agent ws force sync config received", "version", target.Version, "checksum", target.Checksum, "trigger_sync", true)
+		if err := r.SyncService.ForceSyncOnce(ctx, &target); err != nil {
+			r.recordSyncError(err)
+			slog.Error("agent ws triggered force sync failed", "version", target.Version, "error", err)
+		}
+		return false, nil
+	case protocol.WSMessageTypePing:
+		slog.Debug("agent ws ping received")
+		return false, conn.SendPong()
+	case protocol.WSMessageTypePong:
+		slog.Debug("agent ws pong received")
+		return false, nil
+	default:
+		slog.Debug("agent ws unsupported message type", "type", message.Type)
+		return false, nil
+	}
+}
+
+type webSocketBackoff struct {
+	delays []time.Duration
+	index  int
+}
+
+func newWebSocketBackoff() *webSocketBackoff {
+	return &webSocketBackoff{
+		delays: []time.Duration{
+			time.Second,
+			2 * time.Second,
+			5 * time.Second,
+			10 * time.Second,
+			30 * time.Second,
+		},
+	}
+}
+
+func (backoff *webSocketBackoff) Next() time.Duration {
+	if backoff == nil || len(backoff.delays) == 0 {
+		return 30 * time.Second
+	}
+	if backoff.index >= len(backoff.delays) {
+		return backoff.delays[len(backoff.delays)-1]
+	}
+	delay := backoff.delays[backoff.index]
+	backoff.index++
+	return delay
+}
+
+func (backoff *webSocketBackoff) Reset() {
+	if backoff != nil {
+		backoff.index = 0
+	}
+}
+
 func (r *Runner) hasAgentToken() bool {
 	return strings.TrimSpace(r.Config.AgentToken) != ""
 }
@@ -144,6 +371,10 @@ func (r *Runner) applySettings(settings *protocol.AgentSettings) bool {
 			changed = true
 		}
 	}
+	if settings.WebsocketUpgradeEnabled != r.websocketUpgradeEnabled {
+		slog.Debug("agent websocket upgrade setting updated", "from", r.websocketUpgradeEnabled, "to", settings.WebsocketUpgradeEnabled)
+	}
+	r.websocketUpgradeEnabled = settings.WebsocketUpgradeEnabled
 	r.autoUpdate = settings.AutoUpdate
 	r.updateNow = settings.UpdateNow
 	r.updateRepo = strings.TrimSpace(settings.UpdateRepo)
@@ -222,6 +453,9 @@ func (r *Runner) tryRegister(ctx context.Context, nodeID *string) error {
 		return err
 	}
 	r.HeartbeatService.SetToken(response.AgentToken)
+	if r.WebSocketService != nil {
+		r.WebSocketService.SetToken(response.AgentToken)
+	}
 	*nodeID = response.NodeID
 	slog.Info("agent discovery registration succeeded", "node_id", response.NodeID)
 	r.refreshOpenrestyHealth(ctx)
@@ -2,6 +2,7 @@ package agent

 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"os"
 	"path/filepath"
@@ -67,6 +68,7 @@ type fakeSyncService struct {
 	syncOnceErr    error
 	startupCalls   int
 	syncOnceCalls  int
+	lastTarget     *protocol.ActiveConfigMeta
 	onSyncOnceCall func(int)
 }

@@ -104,6 +106,10 @@ func (f *fakeSyncService) SyncOnStartup(ctx context.Context, target *protocol.Ac
 func (f *fakeSyncService) SyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error {
 	f.mu.Lock()
 	f.syncOnceCalls++
+	if target != nil {
+		copied := *target
+		f.lastTarget = &copied
+	}
 	callIndex := f.syncOnceCalls
 	callback := f.onSyncOnceCall
 	f.mu.Unlock()
@@ -113,6 +119,47 @@ func (f *fakeSyncService) SyncOnce(ctx context.Context, target *protocol.ActiveC
 	return f.syncOnceErr
 }

+func (f *fakeSyncService) ForceSyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error {
+	f.mu.Lock()
+	f.syncOnceCalls++
+	if target != nil {
+		copied := *target
+		f.lastTarget = &copied
+	}
+	callIndex := f.syncOnceCalls
+	callback := f.onSyncOnceCall
+	f.mu.Unlock()
+	if callback != nil {
+		callback(callIndex)
+	}
+	return f.syncOnceErr
+}
+
+type fakeWebSocketConnection struct {
+	pongCalls int
+}
+
+func (f *fakeWebSocketConnection) URL() string {
+	return "ws://127.0.0.1/api/agent/ws"
+}
+
+func (f *fakeWebSocketConnection) SendStatus(payload protocol.NodePayload) error {
+	return nil
+}
+
+func (f *fakeWebSocketConnection) SendPong() error {
+	f.pongCalls++
+	return nil
+}
+
+func (f *fakeWebSocketConnection) Receive() (protocol.WSMessage, error) {
+	return protocol.WSMessage{}, errors.New("not implemented")
+}
+
+func (f *fakeWebSocketConnection) Close() error {
+	return nil
+}
+
 func TestRunnerKeepsHeartbeatWhenStartupSyncFails(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -495,3 +542,83 @@ func TestRunnerDiscoveryRegisterUpdatesTokenAndNodeID(t *testing.T) {
 		t.Fatal("expected config token rotation to complete")
 	}
 }
+
+func TestRunnerHandlesWebSocketActiveConfigMessage(t *testing.T) {
+	syncService := &fakeSyncService{}
+	runner := &Runner{SyncService: syncService}
+	payload, err := json.Marshal(protocol.ActiveConfigMeta{
+		Version:  "20260529-001",
+		Checksum: "checksum-ws",
+	})
+	if err != nil {
+		t.Fatalf("marshal active config: %v", err)
+	}
+
+	changed, err := runner.handleWebSocketMessage(context.Background(), protocol.WSMessage{
+		Type:    protocol.WSMessageTypeActiveConfig,
+		Payload: payload,
+	}, &fakeWebSocketConnection{})
+	if err != nil {
+		t.Fatalf("handle websocket active config: %v", err)
+	}
+	if changed {
+		t.Fatal("active config message should not change heartbeat interval")
+	}
+	if syncService.syncOnceCalls != 1 {
+		t.Fatalf("expected one sync call, got %d", syncService.syncOnceCalls)
+	}
+	if syncService.lastTarget == nil || syncService.lastTarget.Version != "20260529-001" || syncService.lastTarget.Checksum != "checksum-ws" {
+		t.Fatalf("unexpected sync target: %+v", syncService.lastTarget)
+	}
+}
+
+func TestRunnerHandlesWebSocketSettingsDisabled(t *testing.T) {
+	runner := &Runner{
+		Config: &config.Config{
+			HeartbeatInterval: config.MillisecondDuration(10 * time.Second),
+		},
+		websocketUpgradeEnabled: true,
+	}
+	payload, err := json.Marshal(protocol.AgentSettings{
+		HeartbeatInterval:       15000,
+		WebsocketUpgradeEnabled: false,
+	})
+	if err != nil {
+		t.Fatalf("marshal settings: %v", err)
+	}
+
+	changed, err := runner.handleWebSocketMessage(context.Background(), protocol.WSMessage{
+		Type:    protocol.WSMessageTypeSettings,
+		Payload: payload,
+	}, &fakeWebSocketConnection{})
+	if err == nil {
+		t.Fatal("expected disabled websocket setting to request fallback")
+	}
+	if !changed {
+		t.Fatal("expected heartbeat interval change to be reported")
+	}
+	if runner.websocketUpgradeEnabled {
+		t.Fatal("expected websocket upgrade to be disabled")
+	}
+}
+
+func TestWebSocketBackoffSequence(t *testing.T) {
+	backoff := newWebSocketBackoff()
+	expected := []time.Duration{
+		time.Second,
+		2 * time.Second,
+		5 * time.Second,
+		10 * time.Second,
+		30 * time.Second,
+		30 * time.Second,
+	}
+	for _, want := range expected {
+		if got := backoff.Next(); got != want {
+			t.Fatalf("unexpected backoff: got %s want %s", got, want)
+		}
+	}
+	backoff.Reset()
+	if got := backoff.Next(); got != time.Second {
+		t.Fatalf("expected reset backoff to return 1s, got %s", got)
+	}
+}
@@ -9,6 +9,7 @@ import (
 	"io/fs"
 	"log/slog"
 	"net"
+	"net/http"
 	"net/url"
 	"os"
 	"os/exec"
@@ -16,6 +17,7 @@ import (
 	"regexp"
 	"sort"
 	"strings"
+	"time"

 	"openflare-agent/internal/protocol"
 )
@@ -155,6 +157,38 @@ const (
 	ApplyStatusFatal   ApplyStatus = "fatal"
 )

+const safeDefaultFallbackMainConfig = `# This file is generated by OpenFlare safe default fallback.
+worker_processes auto;
+pid logs/nginx.pid;
+
+events {
+    worker_connections 1024;
+}
+
+http {
+    default_type text/plain;
+
+    server {
+        listen 80 default_server;
+        server_name _;
+        return 503 "OpenFlare: No Valid Configuration\n";
+    }
+%s
+}
+`
+
+const safeDefaultFallbackObservabilityServerBlock = `
+    server {
+        listen %s;
+        server_name openflare-observability;
+        access_log off;
+
+        location = /openflare/stub_status {
+            stub_status;
+        }
+    }
+`
+
 type ApplyOutcome struct {
 	Status  ApplyStatus
 	Message string
@@ -219,7 +253,18 @@ func (m *Manager) rollbackAfterFailedApply(ctx context.Context, backup *backupSt
 		return fatalApplyOutcome(fmt.Errorf("restore openresty backup failed after apply error %v: %w", applyErr, err))
 	}
 	if err := m.activateConfig(ctx); err != nil {
-		return fatalApplyOutcome(fmt.Errorf("apply failed: %v; rollback recovery failed: %w", applyErr, err))
+		if backup != nil && backup.MainExisted {
+			return fatalApplyOutcome(fmt.Errorf("apply failed: %v; rollback recovery failed: %w", applyErr, err))
+		}
+		if fallbackErr := m.EnsureSafeFallbackRuntime(ctx, fmt.Sprintf("apply failed: %v; rollback recovery failed: %v", applyErr, err)); fallbackErr != nil {
+			return fatalApplyOutcome(fmt.Errorf("apply failed: %v; rollback recovery failed: %w; fallback recovery failed: %v", applyErr, err, fallbackErr))
+		}
+		message := fmt.Sprintf("apply failed, but fallback runtime started: %v; rollback recovery failed: %v", applyErr, err)
+		slog.Warn("openresty apply recovered with safe default fallback", "message", message)
+		return ApplyOutcome{
+			Status:  ApplyStatusWarning,
+			Message: message,
+		}
 	}
 	message := fmt.Sprintf("apply failed, rolled back to previous config: %v", applyErr)
 	slog.Warn("openresty apply rolled back successfully", "message", message)
@@ -276,6 +321,25 @@ func (m *Manager) EnsureRuntime(ctx context.Context, recreate bool) error {
 	return m.Executor.EnsureRuntime(ctx, recreate)
 }

+func (m *Manager) EnsureSafeFallbackRuntime(ctx context.Context, reason string) error {
+	if m.Executor == nil {
+		return errors.New("executor 未配置")
+	}
+	trimmedReason := strings.TrimSpace(reason)
+	if trimmedReason == "" {
+		trimmedReason = "no valid local openresty config is available"
+	}
+	slog.Warn("starting openresty safe default fallback runtime", "reason", trimmedReason)
+	if err := m.writeSafeDefaultFallbackFiles(); err != nil {
+		return fmt.Errorf("write safe default fallback config failed: %w", err)
+	}
+	if err := m.activateConfig(ctx); err != nil {
+		return fmt.Errorf("activate safe default fallback runtime failed: %w", err)
+	}
+	slog.Warn("openresty safe default fallback runtime started", "main_config", m.MainConfigPath, "route_config", m.RouteConfigPath)
+	return nil
+}
+
 func (m *Manager) CheckHealth(ctx context.Context) error {
 	if m.Executor == nil {
 		return errors.New("executor 未配置")
@@ -285,7 +349,10 @@ func (m *Manager) CheckHealth(ctx context.Context) error {
 			return errors.New("openresty config not exists: waiting for initial sync")
 		}
 	}
-	return m.Executor.CheckHealth(ctx)
+	if m.OpenrestyObservabilityPort <= 0 {
+		return m.Executor.CheckHealth(ctx)
+	}
+	return m.checkStubStatus(ctx)
 }

 func (m *Manager) Restart(ctx context.Context) error {
@@ -675,6 +742,56 @@ func (m *Manager) restorePowConfig(state *backupState) error {
 	return os.WriteFile(configPath, []byte(state.PowConfig.Content), 0o644)
 }

+func (m *Manager) writeSafeDefaultFallbackFiles() error {
+	if strings.TrimSpace(m.MainConfigPath) == "" {
+		return errors.New("main config path 不能为空")
+	}
+	if strings.TrimSpace(m.RouteConfigPath) == "" {
+		return errors.New("route config path 不能为空")
+	}
+	if err := os.MkdirAll(filepath.Dir(m.MainConfigPath), 0o755); err != nil {
+		return err
+	}
+	if err := os.MkdirAll(filepath.Dir(m.RouteConfigPath), 0o755); err != nil {
+		return err
+	}
+	if err := os.WriteFile(m.RouteConfigPath, nil, 0o644); err != nil {
+		return err
+	}
+	if err := os.WriteFile(m.MainConfigPath, []byte(m.safeDefaultFallbackMainConfig()), 0o644); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (m *Manager) safeDefaultFallbackMainConfig() string {
+	observabilityBlock := ""
+	if listen := strings.TrimSpace(m.OpenrestyObservabilityListen); listen != "" {
+		observabilityBlock = fmt.Sprintf(safeDefaultFallbackObservabilityServerBlock, listen)
+	}
+	return fmt.Sprintf(safeDefaultFallbackMainConfig, observabilityBlock)
+}
+
+func (m *Manager) checkStubStatus(ctx context.Context) error {
+	ctx, cancel := context.WithTimeout(ctx, 1500*time.Millisecond)
+	defer cancel()
+	openrestyStubUrl := fmt.Sprintf("http://127.0.0.1:%d/openflare/stub_status", m.OpenrestyObservabilityPort)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, openrestyStubUrl, nil)
+	if err != nil {
+		return err
+	}
+	resp, err := (&http.Client{}).Do(req)
+	if err != nil {
+		return fmt.Errorf("openresty health endpoint unreachable: %w", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("openresty health endpoint returned %s", resp.Status)
+	}
+	slog.Debug("openresty health endpoint is healthy", "url", openrestyStubUrl)
+	return nil
+}
+
 func removeLegacyPowConfig(path string) error {
 	if strings.TrimSpace(path) == "" {
 		return nil
@@ -3,6 +3,8 @@ package nginx
 import (
 	"context"
 	"errors"
+	"net"
+	"net/http"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -29,6 +31,8 @@ type fakeExecutor struct {
 }

 type scriptedExecutor struct {
+	testErrors   []error
+	testCalls    int
 	reloadErrors []error
 	reloadCalls  int
 }
@@ -62,7 +66,12 @@ func (e *fakeExecutor) Restart(ctx context.Context) error {
 }

 func (e *scriptedExecutor) Test(ctx context.Context) error {
-	return nil
+	index := e.testCalls
+	e.testCalls++
+	if index >= len(e.testErrors) {
+		return nil
+	}
+	return e.testErrors[index]
 }

 func (e *scriptedExecutor) Reload(ctx context.Context) error {
@@ -321,6 +330,67 @@ func TestManagerApplyWritesSupportFilesAndReplacesPlaceholder(t *testing.T) {
 	}
 }

+func TestManagerCheckHealthUsesStubStatusInsteadOfConfigTest(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("Listen failed: %v", err)
+	}
+	port := listener.Addr().(*net.TCPAddr).Port
+	server := &http.Server{
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if r.URL.Path != "/openflare/stub_status" {
+				http.NotFound(w, r)
+				return
+			}
+			w.WriteHeader(http.StatusOK)
+			_, _ = w.Write([]byte("Active connections: 1\n"))
+		}),
+	}
+	go func() {
+		_ = server.Serve(listener)
+	}()
+	defer server.Shutdown(context.Background())
+
+	mainPath := filepath.Join(t.TempDir(), "nginx.conf")
+	if err := os.WriteFile(mainPath, []byte("main"), 0o644); err != nil {
+		t.Fatalf("WriteFile failed: %v", err)
+	}
+	manager := &Manager{
+		MainConfigPath:             mainPath,
+		OpenrestyObservabilityPort: port,
+		Executor: &fakeExecutor{
+			testErr: errors.New("openresty -t should not be called"),
+		},
+	}
+	if err := manager.CheckHealth(context.Background()); err != nil {
+		t.Fatalf("CheckHealth failed: %v", err)
+	}
+}
+
+func TestManagerCheckHealthFailsWhenStubStatusUnavailable(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("Listen failed: %v", err)
+	}
+	port := listener.Addr().(*net.TCPAddr).Port
+	if err := listener.Close(); err != nil {
+		t.Fatalf("listener close failed: %v", err)
+	}
+
+	mainPath := filepath.Join(t.TempDir(), "nginx.conf")
+	if err := os.WriteFile(mainPath, []byte("main"), 0o644); err != nil {
+		t.Fatalf("WriteFile failed: %v", err)
+	}
+	manager := &Manager{
+		MainConfigPath:             mainPath,
+		OpenrestyObservabilityPort: port,
+		Executor:                   &fakeExecutor{},
+	}
+	if err := manager.CheckHealth(context.Background()); err == nil {
+		t.Fatal("expected CheckHealth to fail when stub_status is unavailable")
+	}
+}
+
 func TestResolverDirectiveUsesExplicitResolvers(t *testing.T) {
 	got := ResolverDirective("", []string{"10.0.0.2", "1.1.1.1"})
 	if !strings.Contains(got, "resolver 10.0.0.2 1.1.1.1") {
@@ -733,6 +803,55 @@ func TestManagerApplyReturnsWarningWhenRollbackRecoversRuntime(t *testing.T) {
 	}
 }

+func TestManagerApplyStartsSafeFallbackWhenNoRollbackConfigExists(t *testing.T) {
+	tempDir := t.TempDir()
+	routePath := filepath.Join(tempDir, "routes.conf")
+	mainPath := filepath.Join(tempDir, "nginx.conf")
+	executor := &scriptedExecutor{
+		testErrors: []error{errors.New("target config failed"), errors.New("rollback config missing"), nil},
+	}
+	manager := &Manager{
+		MainConfigPath:               mainPath,
+		RouteConfigPath:              routePath,
+		OpenrestyObservabilityListen: "127.0.0.1:18081",
+		Executor:                     executor,
+	}
+
+	outcome := manager.Apply(context.Background(), "bad-main", "bad-route", nil)
+	if outcome.Status != ApplyStatusWarning {
+		t.Fatalf("expected warning apply outcome, got %#v", outcome)
+	}
+	if !strings.Contains(outcome.Message, "fallback runtime started") {
+		t.Fatalf("expected fallback message, got %q", outcome.Message)
+	}
+	if executor.testCalls != 3 {
+		t.Fatalf("expected target, rollback, and fallback tests, got %d", executor.testCalls)
+	}
+	mainData, err := os.ReadFile(mainPath)
+	if err != nil {
+		t.Fatalf("failed to read main config: %v", err)
+	}
+	if !strings.Contains(string(mainData), "OpenFlare: No Valid Configuration") {
+		t.Fatalf("expected safe fallback main config, got %s", string(mainData))
+	}
+	if !strings.Contains(string(mainData), "listen 80 default_server") {
+		t.Fatalf("expected fallback to listen on port 80, got %s", string(mainData))
+	}
+	if !strings.Contains(string(mainData), "listen 127.0.0.1:18081") {
+		t.Fatalf("expected fallback to expose local stub_status port, got %s", string(mainData))
+	}
+	if !strings.Contains(string(mainData), "stub_status;") {
+		t.Fatalf("expected fallback to expose stub_status, got %s", string(mainData))
+	}
+	routeData, err := os.ReadFile(routePath)
+	if err != nil {
+		t.Fatalf("failed to read route config: %v", err)
+	}
+	if len(routeData) != 0 {
+		t.Fatalf("expected fallback route config to be empty, got %q", string(routeData))
+	}
+}
+
 func TestManagerCertFileTargetPathRejectsEscapes(t *testing.T) {
 	manager := &Manager{CertDir: filepath.Join(t.TempDir(), "certs")}
 	if err := os.MkdirAll(manager.CertDir, 0o755); err != nil {
@@ -1,5 +1,7 @@
 package protocol

+import "encoding/json"
+
 type APIResponse[T any] struct {
 	Success bool   `json:"success"`
 	Message string `json:"message"`
@@ -20,13 +22,41 @@ type HeartbeatResult struct {
 }

 type AgentSettings struct {
-	HeartbeatInterval   int    `json:"heartbeat_interval"`
-	AutoUpdate          bool   `json:"auto_update"`
-	UpdateRepo          string `json:"update_repo"`
-	UpdateNow           bool   `json:"update_now"`
-	UpdateChannel       string `json:"update_channel"`
-	UpdateTag           string `json:"update_tag"`
-	RestartOpenrestyNow bool   `json:"restart_openresty_now"`
+	HeartbeatInterval       int    `json:"heartbeat_interval"`
+	WebsocketUpgradeEnabled bool   `json:"websocket_upgrade_enabled"`
+	AutoUpdate              bool   `json:"auto_update"`
+	UpdateRepo              string `json:"update_repo"`
+	UpdateNow               bool   `json:"update_now"`
+	UpdateChannel           string `json:"update_channel"`
+	UpdateTag               string `json:"update_tag"`
+	RestartOpenrestyNow     bool   `json:"restart_openresty_now"`
+}
+
+const (
+	WSMessageTypeStatus          = "status"
+	WSMessageTypeSettings        = "settings"
+	WSMessageTypeActiveConfig    = "active_config"
+	WSMessageTypeForceSyncConfig = "force_sync_config"
+	WSMessageTypePing            = "ping"
+	WSMessageTypePong            = "pong"
+)
+
+type WSMessage struct {
+	Type    string          `json:"type"`
+	Payload json.RawMessage `json:"payload,omitempty"`
+}
+
+type WSOutboundMessage struct {
+	Type    string `json:"type"`
+	Payload any    `json:"payload,omitempty"`
+}
+
+type WebSocketConnection interface {
+	URL() string
+	SendStatus(payload NodePayload) error
+	SendPong() error
+	Receive() (WSMessage, error)
+	Close() error
 }

 const (
@@ -27,6 +27,7 @@ type ConfigClient interface {
 type NginxManager interface {
 	Apply(ctx context.Context, mainConfig string, routeConfig string, supportFiles []protocol.SupportFile) nginx.ApplyOutcome
 	EnsureRuntime(ctx context.Context, recreate bool) error
+	EnsureSafeFallbackRuntime(ctx context.Context, reason string) error
 	CurrentChecksum() (string, error)
 }

@@ -136,6 +137,18 @@ func (s *Service) sync(ctx context.Context, startup bool, target *protocol.Activ
 	return s.applyIfNeeded(ctx, mode, startup, snapshot, currentChecksum, target, config)
 }

+func (s *Service) ForceSyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error {
+	snapshot, err := s.stateStore.Load()
+	if err != nil {
+		return err
+	}
+	if hasBlockedTarget(snapshot) {
+		clearBlockedTarget(snapshot)
+		_ = s.stateStore.Save(snapshot)
+	}
+	return s.SyncOnce(ctx, target)
+}
+
 func (s *Service) applyIfNeeded(ctx context.Context, mode string, startup bool, snapshot *state.Snapshot, currentChecksum string, target *protocol.ActiveConfigMeta, config *protocol.ActiveConfigResponse) error {
 	if currentChecksum == config.Checksum {
 		slog.Debug("local openresty config already up to date", "mode", mode, "version", config.Version)
@@ -264,10 +277,29 @@ func outcomeError(version string, message string) error {
 func (s *Service) ensureRuntimeForCurrentConfig(ctx context.Context, mode string, snapshot *state.Snapshot, currentChecksum string) error {
 	if strings.TrimSpace(currentChecksum) == "" {
 		slog.Warn("blocked config cannot be retried and no local checksum is available for runtime recovery", "mode", mode, "blocked_version", snapshot.BlockedVersion)
+		reason := fmt.Sprintf("blocked config %s has no valid local config available for runtime recovery", strings.TrimSpace(snapshot.BlockedVersion))
+		if err := s.nginxManager.EnsureSafeFallbackRuntime(ctx, reason); err != nil {
+			snapshot.OpenrestyStatus = protocol.OpenrestyStatusUnhealthy
+			snapshot.OpenrestyMessage = err.Error()
+			_ = s.stateStore.Save(snapshot)
+			return err
+		}
+		snapshot.OpenrestyStatus = protocol.OpenrestyStatusHealthy
+		snapshot.OpenrestyMessage = "safe default fallback runtime started"
 		return nil
 	}
 	slog.Info("ensuring runtime with current local config while active target remains blocked", "mode", mode, "current_version", snapshot.CurrentVersion, "current_checksum", currentChecksum, "blocked_version", snapshot.BlockedVersion)
 	if err := s.nginxManager.EnsureRuntime(ctx, true); err != nil {
+		if strings.TrimSpace(snapshot.CurrentChecksum) == "" {
+			reason := fmt.Sprintf("blocked config %s has no historical config and current local config cannot start: %v", strings.TrimSpace(snapshot.BlockedVersion), err)
+			if fallbackErr := s.nginxManager.EnsureSafeFallbackRuntime(ctx, reason); fallbackErr == nil {
+				snapshot.OpenrestyStatus = protocol.OpenrestyStatusHealthy
+				snapshot.OpenrestyMessage = "safe default fallback runtime started"
+				return nil
+			} else {
+				err = fmt.Errorf("%v; fallback recovery failed: %w", err, fallbackErr)
+			}
+		}
 		snapshot.OpenrestyStatus = protocol.OpenrestyStatusUnhealthy
 		snapshot.OpenrestyMessage = err.Error()
 		_ = s.stateStore.Save(snapshot)
@@ -28,7 +28,9 @@ type fakeManager struct {
 	currentChecksum    string
 	currentChecksumErr error
 	ensureErr          error
+	fallbackErr        error
 	ensureCalls        []bool
+	fallbackReasons    []string
 	applyMainContents  []string
 	applyRouteContents []string
 	applyFiles         [][]protocol.SupportFile
@@ -79,6 +81,11 @@ func (m *fakeManager) EnsureRuntime(ctx context.Context, recreate bool) error {
 	return m.ensureErr
 }

+func (m *fakeManager) EnsureSafeFallbackRuntime(ctx context.Context, reason string) error {
+	m.fallbackReasons = append(m.fallbackReasons, reason)
+	return m.fallbackErr
+}
+
 func (m *fakeManager) CurrentChecksum() (string, error) {
 	return m.currentChecksum, m.currentChecksumErr
 }
@@ -480,6 +487,120 @@ func TestSyncOnStartupKeepsBlockedVersionSuppressedUntilNewTargetArrives(t *test
 	}
 }

+func TestSyncOnStartupStartsFallbackWhenBlockedVersionHasNoLocalConfig(t *testing.T) {
+	client := &fakeClient{
+		config: protocol.ActiveConfigResponse{
+			Version:        "20260309-007",
+			Checksum:       "checksum-7",
+			MainConfig:     "worker_processes 7;",
+			RouteConfig:    "server { listen 87; }",
+			RenderedConfig: "server { listen 87; }",
+			CreatedAt:      time.Now().Format(time.RFC3339),
+		},
+	}
+	stateStore := state.NewStore(filepath.Join(t.TempDir(), "state.json"))
+	nodeID, err := stateStore.EnsureNodeID()
+	if err != nil {
+		t.Fatalf("EnsureNodeID failed: %v", err)
+	}
+	if err = stateStore.Save(&state.Snapshot{
+		NodeID:           nodeID,
+		BlockedVersion:   "20260309-007",
+		BlockedChecksum:  "checksum-7",
+		BlockedReason:    "apply failed, but fallback runtime started",
+		OpenrestyStatus:  protocol.OpenrestyStatusUnhealthy,
+		OpenrestyMessage: "apply failed, but fallback runtime started",
+		LastError:        "apply failed, but fallback runtime started",
+	}); err != nil {
+		t.Fatalf("failed to seed state: %v", err)
+	}
+
+	manager := &fakeManager{}
+	service := New(client, manager, stateStore)
+	if err = service.SyncOnStartup(context.Background(), &protocol.ActiveConfigMeta{
+		Version:  "20260309-007",
+		Checksum: "checksum-7",
+	}); err != nil {
+		t.Fatalf("expected blocked startup target to start fallback, got %v", err)
+	}
+	if len(manager.fallbackReasons) != 1 {
+		t.Fatalf("expected fallback runtime to be started once, got %d", len(manager.fallbackReasons))
+	}
+	if client.fetchCalls != 0 {
+		t.Fatalf("expected blocked startup target to skip fetch, got %d", client.fetchCalls)
+	}
+	if len(client.reports) != 0 {
+		t.Fatal("expected blocked startup target to skip duplicate apply report")
+	}
+	snapshot, err := stateStore.Load()
+	if err != nil {
+		t.Fatalf("failed to load state: %v", err)
+	}
+	if snapshot.BlockedVersion != "20260309-007" || snapshot.BlockedChecksum != "checksum-7" {
+		t.Fatalf("expected blocked target to remain recorded, got %+v", snapshot)
+	}
+	if snapshot.OpenrestyStatus != protocol.OpenrestyStatusHealthy {
+		t.Fatalf("expected fallback startup recovery to mark openresty healthy, got %q", snapshot.OpenrestyStatus)
+	}
+	if snapshot.OpenrestyMessage != "safe default fallback runtime started" {
+		t.Fatalf("expected fallback status message, got %q", snapshot.OpenrestyMessage)
+	}
+}
+
+func TestSyncOnStartupStartsFallbackWhenResidualConfigCannotRecover(t *testing.T) {
+	client := &fakeClient{
+		config: protocol.ActiveConfigResponse{
+			Version:        "20260309-007",
+			Checksum:       "checksum-7",
+			MainConfig:     "worker_processes 7;",
+			RouteConfig:    "server { listen 87; }",
+			RenderedConfig: "server { listen 87; }",
+			CreatedAt:      time.Now().Format(time.RFC3339),
+		},
+	}
+	stateStore := state.NewStore(filepath.Join(t.TempDir(), "state.json"))
+	nodeID, err := stateStore.EnsureNodeID()
+	if err != nil {
+		t.Fatalf("EnsureNodeID failed: %v", err)
+	}
+	if err = stateStore.Save(&state.Snapshot{
+		NodeID:          nodeID,
+		BlockedVersion:  "20260309-007",
+		BlockedChecksum: "checksum-7",
+		BlockedReason:   "apply failed, but fallback runtime started",
+	}); err != nil {
+		t.Fatalf("failed to seed state: %v", err)
+	}
+
+	manager := &fakeManager{
+		currentChecksum: "residual-checksum",
+		ensureErr:       context.DeadlineExceeded,
+	}
+	service := New(client, manager, stateStore)
+	if err = service.SyncOnStartup(context.Background(), &protocol.ActiveConfigMeta{
+		Version:  "20260309-007",
+		Checksum: "checksum-7",
+	}); err != nil {
+		t.Fatalf("expected residual config failure to start fallback, got %v", err)
+	}
+	if len(manager.ensureCalls) != 1 {
+		t.Fatalf("expected residual config to be tested once, got %d", len(manager.ensureCalls))
+	}
+	if len(manager.fallbackReasons) != 1 {
+		t.Fatalf("expected fallback runtime to be started once, got %d", len(manager.fallbackReasons))
+	}
+	snapshot, err := stateStore.Load()
+	if err != nil {
+		t.Fatalf("failed to load state: %v", err)
+	}
+	if snapshot.OpenrestyStatus != protocol.OpenrestyStatusHealthy {
+		t.Fatalf("expected fallback startup recovery to mark openresty healthy, got %q", snapshot.OpenrestyStatus)
+	}
+	if snapshot.BlockedVersion != "20260309-007" || snapshot.BlockedChecksum != "checksum-7" {
+		t.Fatalf("expected blocked target to remain recorded, got %+v", snapshot)
+	}
+}
+
 func TestSyncOnceClearsBlockedTargetWhenNewVersionArrives(t *testing.T) {
 	client := &fakeClient{
 		config: protocol.ActiveConfigResponse{
@@ -0,0 +1,165 @@
+package wsclient
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"golang.org/x/net/websocket"
+
+	"openflare-agent/internal/protocol"
+)
+
+type Client struct {
+	baseURL string
+	token   string
+	timeout time.Duration
+}
+
+type Connection struct {
+	conn        *websocket.Conn
+	url         string
+	readTimeout time.Duration
+}
+
+func New(baseURL string, token string, timeout time.Duration) *Client {
+	return &Client{
+		baseURL: strings.TrimRight(baseURL, "/"),
+		token:   strings.TrimSpace(token),
+		timeout: timeout,
+	}
+}
+
+func (c *Client) SetToken(token string) {
+	c.token = strings.TrimSpace(token)
+	slog.Debug("agent ws client token updated")
+}
+
+func (c *Client) URL() string {
+	wsURL, err := buildWebsocketURL(c.baseURL)
+	if err != nil {
+		return ""
+	}
+	return wsURL
+}
+
+func (c *Client) Connect(ctx context.Context) (protocol.WebSocketConnection, error) {
+	wsURL, err := buildWebsocketURL(c.baseURL)
+	if err != nil {
+		return nil, err
+	}
+	if strings.TrimSpace(c.token) == "" {
+		return nil, errors.New("agent ws token is empty")
+	}
+	origin := strings.TrimSpace(c.baseURL)
+	if origin == "" {
+		origin = "http://localhost"
+	}
+	config, err := websocket.NewConfig(wsURL, origin)
+	if err != nil {
+		return nil, err
+	}
+	config.Header = http.Header{}
+	config.Header.Set("X-Agent-Token", c.token)
+	if c.timeout > 0 {
+		config.Dialer = &net.Dialer{Timeout: c.timeout}
+	}
+	slog.Debug("agent ws dialing server", "url", wsURL)
+	conn, err := config.DialContext(ctx)
+	if err != nil {
+		return nil, err
+	}
+	slog.Debug("agent ws dial succeeded", "url", wsURL)
+	return &Connection{conn: conn, url: wsURL, readTimeout: websocketReadTimeout(c.timeout)}, nil
+}
+
+func buildWebsocketURL(baseURL string) (string, error) {
+	parsed, err := url.Parse(strings.TrimRight(baseURL, "/"))
+	if err != nil {
+		return "", err
+	}
+	switch parsed.Scheme {
+	case "http":
+		parsed.Scheme = "ws"
+	case "https":
+		parsed.Scheme = "wss"
+	case "ws", "wss":
+	default:
+		return "", errors.New("server_url scheme must be http, https, ws, or wss")
+	}
+	parsed.Path = strings.TrimRight(parsed.Path, "/") + "/api/agent/ws"
+	parsed.RawQuery = ""
+	parsed.Fragment = ""
+	return parsed.String(), nil
+}
+
+func (conn *Connection) URL() string {
+	if conn == nil {
+		return ""
+	}
+	return conn.url
+}
+
+func (conn *Connection) SendStatus(payload protocol.NodePayload) error {
+	if conn == nil || conn.conn == nil {
+		return errors.New("agent ws connection is nil")
+	}
+	slog.Debug("agent ws sending status",
+		"node_id", payload.NodeID,
+		"current_version", payload.CurrentVersion,
+		"openresty_status", payload.OpenrestyStatus,
+	)
+	return websocket.JSON.Send(conn.conn, protocol.WSOutboundMessage{
+		Type:    protocol.WSMessageTypeStatus,
+		Payload: payload,
+	})
+}
+
+func (conn *Connection) SendPong() error {
+	if conn == nil || conn.conn == nil {
+		return errors.New("agent ws connection is nil")
+	}
+	slog.Debug("agent ws sending pong")
+	return websocket.JSON.Send(conn.conn, protocol.WSOutboundMessage{
+		Type: protocol.WSMessageTypePong,
+	})
+}
+
+func (conn *Connection) Receive() (protocol.WSMessage, error) {
+	var message protocol.WSMessage
+	if conn == nil || conn.conn == nil {
+		return message, errors.New("agent ws connection is nil")
+	}
+	if conn.readTimeout > 0 {
+		_ = conn.conn.SetReadDeadline(time.Now().Add(conn.readTimeout))
+	}
+	err := websocket.JSON.Receive(conn.conn, &message)
+	if err != nil {
+		if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
+			slog.Debug("agent ws receive timeout waiting for server message", "timeout", conn.readTimeout)
+		}
+		return message, err
+	}
+	slog.Debug("agent ws received message", "type", message.Type)
+	return message, nil
+}
+
+func websocketReadTimeout(requestTimeout time.Duration) time.Duration {
+	timeout := requestTimeout * 6
+	if timeout < 75*time.Second {
+		return 75 * time.Second
+	}
+	return timeout
+}
+
+func (conn *Connection) Close() error {
+	if conn == nil || conn.conn == nil {
+		return nil
+	}
+	return conn.conn.Close()
+}
@@ -53,6 +53,7 @@ var NodeOfflineThreshold = 2 * time.Minute

 // V3 operational settings (hot-reloadable via Option table)
 var AgentHeartbeatInterval = 10000 // milliseconds
+var AgentWebsocketUpgradeEnabled = true
 var AgentUpdateRepo = "Rain-kl/OpenFlare"
 var GeoIPProvider = "ipinfo"
 var DatabaseAutoCleanupEnabled = false
@@ -1,11 +1,17 @@
 package controller

 import (
+	"encoding/json"
+	"log/slog"
+	"net"
+	"openflare/common"
 	"openflare/model"
 	"openflare/service"
 	"strconv"
+	"time"

 	"github.com/gin-gonic/gin"
+	"golang.org/x/net/websocket"
 )

 // AgentRegister godoc
@@ -122,6 +128,126 @@ func AgentReportApplyLog(c *gin.Context) {
 	respondSuccess(c, log)
 }

+// AgentWebSocket godoc
+// @Summary Upgrade agent connection to websocket
+// @Tags Agent
+// @Security AgentTokenAuth
+// @Router /api/agent/ws [get]
+func AgentWebSocket(c *gin.Context) {
+	authNode, ok := c.Get("agent_node")
+	if !ok {
+		respondUnauthorized(c, "无权进行此操作，Agent Token 无效")
+		return
+	}
+	node := authNode.(*model.Node)
+	slog.Debug("agent ws upgrade requested", "node_id", node.NodeID, "remote", c.Request.RemoteAddr)
+	websocket.Handler(func(conn *websocket.Conn) {
+		client := service.RegisterAgentWSClient(node.NodeID)
+		defer service.UnregisterAgentWSClient(client)
+		defer func() {
+			_ = conn.Close()
+			slog.Debug("agent ws connection closed", "node_id", node.NodeID)
+		}()
+
+		slog.Debug("agent ws upgrade succeeded", "node_id", node.NodeID, "remote", c.Request.RemoteAddr)
+		go streamAgentWSMessages(c, conn, client)
+
+		for {
+			var message service.AgentWSInboundMessage
+			_ = conn.SetReadDeadline(time.Now().Add(agentWSReadTimeout()))
+			if err := websocket.JSON.Receive(conn, &message); err != nil {
+				if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
+					slog.Debug("agent ws receive timeout waiting for status or pong", "node_id", node.NodeID, "timeout", agentWSReadTimeout())
+					return
+				}
+				slog.Debug("agent ws receive failed", "node_id", node.NodeID, "error", err)
+				return
+			}
+			slog.Debug("agent ws message received", "node_id", node.NodeID, "type", message.Type)
+			switch message.Type {
+			case service.AgentWSMessageTypeStatus:
+				handleAgentWSStatus(c, node, message)
+			case service.AgentWSMessageTypePing:
+				if !service.SendAgentWSPong(node.NodeID) {
+					slog.Debug("agent ws pong enqueue failed", "node_id", node.NodeID)
+				}
+			case service.AgentWSMessageTypePong:
+				slog.Debug("agent ws pong received", "node_id", node.NodeID)
+			default:
+				slog.Debug("agent ws unsupported message type", "node_id", node.NodeID, "type", message.Type)
+			}
+		}
+	}).ServeHTTP(c.Writer, c.Request)
+}
+
+func agentWSReadTimeout() time.Duration {
+	timeout := time.Duration(common.AgentHeartbeatInterval) * time.Millisecond * 3
+	if timeout < 30*time.Second {
+		return 30 * time.Second
+	}
+	return timeout
+}
+
+func streamAgentWSMessages(c *gin.Context, conn *websocket.Conn, client *service.AgentWSClient) {
+	pingTicker := time.NewTicker(30 * time.Second)
+	defer pingTicker.Stop()
+	for {
+		select {
+		case message := <-client.Messages():
+			slog.Debug("agent ws sending message", "node_id", client.NodeID(), "type", message.Type)
+			if err := websocket.JSON.Send(conn, message); err != nil {
+				slog.Debug("agent ws send failed", "node_id", client.NodeID(), "type", message.Type, "error", err)
+				client.Close()
+				return
+			}
+		case <-pingTicker.C:
+			message := service.AgentWSOutboundMessage{Type: service.AgentWSMessageTypePing}
+			slog.Debug("agent ws sending ping", "node_id", client.NodeID())
+			if err := websocket.JSON.Send(conn, message); err != nil {
+				slog.Debug("agent ws ping failed", "node_id", client.NodeID(), "error", err)
+				client.Close()
+				return
+			}
+		case <-client.Done():
+			return
+		case <-c.Request.Context().Done():
+			client.Close()
+			return
+		}
+	}
+}
+
+func handleAgentWSStatus(c *gin.Context, node *model.Node, message service.AgentWSInboundMessage) {
+	var payload service.AgentNodePayload
+	if err := json.Unmarshal(message.Payload, &payload); err != nil {
+		slog.Debug("agent ws status payload decode failed", "node_id", node.NodeID, "error", err)
+		return
+	}
+	freshNode, err := model.GetNodeByNodeID(node.NodeID)
+	if err != nil {
+		slog.Debug("agent ws status reload node failed", "node_id", node.NodeID, "error", err)
+		return
+	}
+	payload.IP = service.ResolveReportedNodeIP(payload.IP, c.Request.RemoteAddr)
+	response, err := service.HeartbeatNode(freshNode, payload)
+	if err != nil {
+		slog.Debug("agent ws status handling failed", "node_id", node.NodeID, "error", err)
+		return
+	}
+	settingsSent := service.SendAgentWSSettings(node.NodeID, response.AgentSettings)
+	activeConfigSent := false
+	if response.ActiveConfig != nil {
+		activeConfigSent = service.SendAgentWSActiveConfig(node.NodeID, response.ActiveConfig)
+	}
+	slog.Debug("agent ws status processed",
+		"node_id", node.NodeID,
+		"current_version", payload.CurrentVersion,
+		"openresty_status", payload.OpenrestyStatus,
+		"settings_sent", settingsSent,
+		"active_config_sent", activeConfigSent,
+	)
+}
+
 // GetNodes godoc
 // @Summary List nodes
 // @Tags Nodes
@@ -188,6 +188,30 @@ func RequestNodeOpenrestyRestart(c *gin.Context) {
 	respondSuccess(c, node)
 }

+// RequestNodeForceSync godoc
+// @Summary Request force sync config on node
+// @Tags Nodes
+// @Produce json
+// @Security BearerAuth
+// @Param id path int true "Node ID"
+// @Success 200 {object} map[string]interface{}
+// @Failure 400 {object} map[string]interface{}
+// @Router /api/nodes/{id}/force-sync [post]
+func RequestNodeForceSync(c *gin.Context) {
+	id, err := strconv.ParseUint(c.Param("id"), 10, 64)
+	if err != nil || id == 0 {
+		respondBadRequest(c, "")
+		return
+	}
+
+	node, err := service.RequestNodeForceSync(uint(id))
+	if err != nil {
+		respondFailure(c, err.Error())
+		return
+	}
+	respondSuccess(c, node)
+}
+
 // GetNodeAgentRelease godoc
 // @Summary Check latest agent release for node
 // @Tags Nodes
@@ -92,6 +92,15 @@ func validateDatabaseCleanupOption(key string, value string) error {
 	}
 }

+func validateAgentOption(key string, value string) error {
+	switch key {
+	case "AgentWebsocketUpgradeEnabled":
+		return validateBooleanOption(key, strings.TrimSpace(value))
+	default:
+		return nil
+	}
+}
+
 func validateOpenRestyOption(key string, value string) error {
 	trimmed := strings.TrimSpace(value)

@@ -245,6 +254,9 @@ func validateOptionWithState(option model.Option, state map[string]string) error
 	if err := validateDatabaseCleanupOption(option.Key, option.Value); err != nil {
 		return err
 	}
+	if err := validateAgentOption(option.Key, option.Value); err != nil {
+		return err
+	}
 	return nil
 }

@@ -366,6 +378,13 @@ func UpdateOption(c *gin.Context) {
 		})
 		return
 	}
+	if err = validateAgentOption(option.Key, option.Value); err != nil {
+		c.JSON(http.StatusOK, gin.H{
+			"success": false,
+			"message": err.Error(),
+		})
+		return
+	}
 	err = model.UpdateOption(option.Key, option.Value)
 	if err != nil {
 		c.JSON(http.StatusOK, gin.H{
@@ -46,3 +46,15 @@ func TestValidateOpenRestyOption(t *testing.T) {
 		}
 	}
 }
+
+func TestValidateAgentOption(t *testing.T) {
+	if err := validateAgentOption("AgentWebsocketUpgradeEnabled", "true"); err != nil {
+		t.Fatalf("expected websocket upgrade option to accept true: %v", err)
+	}
+	if err := validateAgentOption("AgentWebsocketUpgradeEnabled", "false"); err != nil {
+		t.Fatalf("expected websocket upgrade option to accept false: %v", err)
+	}
+	if err := validateAgentOption("AgentWebsocketUpgradeEnabled", "on"); err == nil {
+		t.Fatal("expected websocket upgrade option to reject non-boolean value")
+	}
+}
@@ -54,6 +54,7 @@ func InitOptionMap() {
 	common.OptionMap["TurnstileSecretKey"] = ""
 	common.OptionMap["AgentDiscoveryToken"] = ""
 	common.OptionMap["AgentHeartbeatInterval"] = strconv.Itoa(common.AgentHeartbeatInterval)
+	common.OptionMap["AgentWebsocketUpgradeEnabled"] = strconv.FormatBool(common.AgentWebsocketUpgradeEnabled)
 	common.OptionMap["NodeOfflineThreshold"] = strconv.Itoa(int(common.NodeOfflineThreshold.Milliseconds()))
 	common.OptionMap["AgentUpdateRepo"] = common.AgentUpdateRepo
 	common.OptionMap["GeoIPProvider"] = common.GeoIPProvider
@@ -221,6 +222,8 @@ func updateOptionMap(key string, value string) {
 		if v, err := strconv.Atoi(value); err == nil && v > 0 {
 			common.AgentHeartbeatInterval = v
 		}
+	case "AgentWebsocketUpgradeEnabled":
+		common.AgentWebsocketUpgradeEnabled = value == "true"
 	case "NodeOfflineThreshold":
 		if v, err := strconv.Atoi(value); err == nil && v > 0 {
 			common.NodeOfflineThreshold = time.Duration(v) * time.Millisecond
@@ -165,12 +165,13 @@ func SetApiRouter(router *gin.Engine) {
 			nodeRoute.GET("/", controller.GetNodes)
 			nodeRoute.POST("/", controller.CreateNode)
 			nodeRoute.GET("/:id/agent-release", controller.GetNodeAgentRelease)
-			nodeRoute.GET("/:id/observability", controller.GetNodeObservability)
-			nodeRoute.POST("/:id/observability/cleanup", controller.CleanupNodeHealthEvents)
-			nodeRoute.POST("/:id/agent-update", controller.RequestNodeAgentUpdate)
-			nodeRoute.POST("/:id/openresty-restart", controller.RequestNodeOpenrestyRestart)
 			nodeRoute.POST("/:id/update", controller.UpdateNode)
 			nodeRoute.POST("/:id/delete", controller.DeleteNode)
+			nodeRoute.POST("/:id/agent-update", controller.RequestNodeAgentUpdate)
+			nodeRoute.POST("/:id/openresty-restart", controller.RequestNodeOpenrestyRestart)
+			nodeRoute.POST("/:id/force-sync", controller.RequestNodeForceSync)
+			nodeRoute.GET("/:id/observability", controller.GetNodeObservability)
+			nodeRoute.POST("/:id/observability/cleanup", controller.CleanupNodeHealthEvents)
 		}
 		applyLogRoute := apiRouter.Group("/apply-logs")
 		applyLogRoute.Use(middleware.AdminAuth())
@@ -197,6 +198,7 @@ func SetApiRouter(router *gin.Engine) {
 			authorizedRoute := agentRoute.Group("/")
 			authorizedRoute.Use(middleware.AgentAuth())
 			{
+				authorizedRoute.GET("/ws", controller.AgentWebSocket)
 				authorizedRoute.POST("/nodes/heartbeat", controller.AgentHeartbeat)
 				authorizedRoute.GET("/config-versions/active", controller.AgentGetActiveConfig)
 				authorizedRoute.POST("/apply-logs", controller.AgentReportApplyLog)
@@ -89,13 +89,14 @@ type AgentConfigResponse struct {
 }

 type AgentSettings struct {
-	HeartbeatInterval   int    `json:"heartbeat_interval"`
-	AutoUpdate          bool   `json:"auto_update"`
-	UpdateRepo          string `json:"update_repo"`
-	UpdateNow           bool   `json:"update_now"`
-	UpdateChannel       string `json:"update_channel"`
-	UpdateTag           string `json:"update_tag"`
-	RestartOpenrestyNow bool   `json:"restart_openresty_now"`
+	HeartbeatInterval       int    `json:"heartbeat_interval"`
+	WebsocketUpgradeEnabled bool   `json:"websocket_upgrade_enabled"`
+	AutoUpdate              bool   `json:"auto_update"`
+	UpdateRepo              string `json:"update_repo"`
+	UpdateNow               bool   `json:"update_now"`
+	UpdateChannel           string `json:"update_channel"`
+	UpdateTag               string `json:"update_tag"`
+	RestartOpenrestyNow     bool   `json:"restart_openresty_now"`
 }

 type ActiveConfigMeta struct {
@@ -130,7 +131,7 @@ type NodeView struct {
 	OpenrestyMessage          string     `json:"openresty_message"`
 	Status                    string     `json:"status"`
 	CurrentVersion            string     `json:"current_version"`
-	LastSeenAt                time.Time  `json:"last_seen_at"`
+	LastSeenAt                any        `json:"last_seen_at"`
 	LastError                 string     `json:"last_error"`
 	LatestApplyResult         string     `json:"latest_apply_result"`
 	LatestApplyMessage        string     `json:"latest_apply_message"`
@@ -177,20 +178,32 @@ func HeartbeatNode(node *model.Node, payload AgentNodePayload) (*HeartbeatRespon
 		return nil, err
 	}
 	return &HeartbeatResponse{
-		Node: node,
-		AgentSettings: &AgentSettings{
-			HeartbeatInterval:   common.AgentHeartbeatInterval,
-			AutoUpdate:          node.AutoUpdateEnabled,
-			UpdateRepo:          common.AgentUpdateRepo,
-			UpdateNow:           updateNow,
-			UpdateChannel:       updateChannel.String(),
-			UpdateTag:           updateTag,
-			RestartOpenrestyNow: restartOpenrestyNow,
-		},
-		ActiveConfig: activeConfig,
+		Node:          node,
+		AgentSettings: buildAgentSettings(node, updateNow, updateChannel.String(), updateTag, restartOpenrestyNow),
+		ActiveConfig:  activeConfig,
 	}, nil
 }

+func buildAgentSettings(node *model.Node, updateNow bool, updateChannel string, updateTag string, restartOpenrestyNow bool) *AgentSettings {
+	autoUpdate := false
+	if node != nil {
+		autoUpdate = node.AutoUpdateEnabled
+	}
+	if strings.TrimSpace(updateChannel) == "" {
+		updateChannel = ReleaseChannelStable.String()
+	}
+	return &AgentSettings{
+		HeartbeatInterval:       common.AgentHeartbeatInterval,
+		WebsocketUpgradeEnabled: common.AgentWebsocketUpgradeEnabled,
+		AutoUpdate:              autoUpdate,
+		UpdateRepo:              common.AgentUpdateRepo,
+		UpdateNow:               updateNow,
+		UpdateChannel:           updateChannel,
+		UpdateTag:               strings.TrimSpace(updateTag),
+		RestartOpenrestyNow:     restartOpenrestyNow,
+	}
+}
+
 func GetActiveConfigMetaForAgent() (*ActiveConfigMeta, error) {
 	version, err := model.GetActiveConfigVersion()
 	if err != nil {
@@ -438,6 +451,9 @@ func computeNodeStatus(node *model.Node) string {
 	if node == nil {
 		return NodeStatusOffline
 	}
+	if IsAgentWSConnected(node.NodeID) {
+		return NodeStatusOnline
+	}
 	if node.LastSeenAt.IsZero() {
 		return NodeStatusPending
 	}
@@ -0,0 +1,237 @@
+package service
+
+import (
+	"encoding/json"
+	"log/slog"
+	"sync"
+)
+
+const (
+	AgentWSMessageTypeStatus          = "status"
+	AgentWSMessageTypeSettings        = "settings"
+	AgentWSMessageTypeActiveConfig    = "active_config"
+	AgentWSMessageTypeForceSyncConfig = "force_sync_config"
+	AgentWSMessageTypePing            = "ping"
+	AgentWSMessageTypePong            = "pong"
+
+	AgentWSConnectedLastSeenValue = "__OPENFLARE_WS_CONNECTED__"
+)
+
+type AgentWSInboundMessage struct {
+	Type    string          `json:"type"`
+	Payload json.RawMessage `json:"payload,omitempty"`
+}
+
+type AgentWSOutboundMessage struct {
+	Type    string `json:"type"`
+	Payload any    `json:"payload,omitempty"`
+}
+
+type AgentWSBroadcastResult struct {
+	Version      string   `json:"version"`
+	Checksum     string   `json:"checksum"`
+	ClientCount  int      `json:"client_count"`
+	SuccessCount int      `json:"success_count"`
+	FailedNodes  []string `json:"failed_nodes"`
+}
+
+type AgentWSClient struct {
+	nodeID string
+	send   chan AgentWSOutboundMessage
+	done   chan struct{}
+	once   sync.Once
+}
+
+func (client *AgentWSClient) NodeID() string {
+	if client == nil {
+		return ""
+	}
+	return client.nodeID
+}
+
+func (client *AgentWSClient) Messages() <-chan AgentWSOutboundMessage {
+	if client == nil {
+		return nil
+	}
+	return client.send
+}
+
+func (client *AgentWSClient) Done() <-chan struct{} {
+	if client == nil {
+		return nil
+	}
+	return client.done
+}
+
+func (client *AgentWSClient) Send(message AgentWSOutboundMessage) bool {
+	if client == nil {
+		return false
+	}
+	select {
+	case <-client.done:
+		return false
+	case client.send <- message:
+		return true
+	default:
+		return false
+	}
+}
+
+func (client *AgentWSClient) Close() {
+	if client == nil {
+		return
+	}
+	client.once.Do(func() {
+		close(client.done)
+	})
+}
+
+type agentWSHub struct {
+	mu      sync.RWMutex
+	clients map[string]*AgentWSClient
+}
+
+var defaultAgentWSHub = &agentWSHub{
+	clients: make(map[string]*AgentWSClient),
+}
+
+func RegisterAgentWSClient(nodeID string) *AgentWSClient {
+	client := &AgentWSClient{
+		nodeID: nodeID,
+		send:   make(chan AgentWSOutboundMessage, 16),
+		done:   make(chan struct{}),
+	}
+	defaultAgentWSHub.mu.Lock()
+	if existing := defaultAgentWSHub.clients[nodeID]; existing != nil {
+		slog.Debug("agent ws replacing existing connection", "node_id", nodeID)
+		existing.Close()
+	}
+	defaultAgentWSHub.clients[nodeID] = client
+	count := len(defaultAgentWSHub.clients)
+	defaultAgentWSHub.mu.Unlock()
+	slog.Debug("agent ws connection registered", "node_id", nodeID, "client_count", count)
+	return client
+}
+
+func UnregisterAgentWSClient(client *AgentWSClient) {
+	if client == nil {
+		return
+	}
+	defaultAgentWSHub.mu.Lock()
+	if current := defaultAgentWSHub.clients[client.nodeID]; current == client {
+		delete(defaultAgentWSHub.clients, client.nodeID)
+	}
+	count := len(defaultAgentWSHub.clients)
+	defaultAgentWSHub.mu.Unlock()
+	client.Close()
+	slog.Debug("agent ws connection unregistered", "node_id", client.nodeID, "client_count", count)
+}
+
+func IsAgentWSConnected(nodeID string) bool {
+	defaultAgentWSHub.mu.RLock()
+	client := defaultAgentWSHub.clients[nodeID]
+	defaultAgentWSHub.mu.RUnlock()
+	if client == nil {
+		return false
+	}
+	select {
+	case <-client.done:
+		return false
+	default:
+		return true
+	}
+}
+
+func AgentWSClientCount() int {
+	defaultAgentWSHub.mu.RLock()
+	defer defaultAgentWSHub.mu.RUnlock()
+	return len(defaultAgentWSHub.clients)
+}
+
+func SendAgentWSSettings(nodeID string, settings *AgentSettings) bool {
+	if settings == nil {
+		return false
+	}
+	return sendAgentWSMessage(nodeID, AgentWSOutboundMessage{
+		Type:    AgentWSMessageTypeSettings,
+		Payload: settings,
+	})
+}
+
+func SendAgentWSActiveConfig(nodeID string, activeConfig *ActiveConfigMeta) bool {
+	if activeConfig == nil {
+		return false
+	}
+	return sendAgentWSMessage(nodeID, AgentWSOutboundMessage{
+		Type:    AgentWSMessageTypeActiveConfig,
+		Payload: activeConfig,
+	})
+}
+
+func SendAgentWSForceSyncConfig(nodeID string, activeConfig *ActiveConfigMeta) bool {
+	if activeConfig == nil {
+		return false
+	}
+	return sendAgentWSMessage(nodeID, AgentWSOutboundMessage{
+		Type:    AgentWSMessageTypeForceSyncConfig,
+		Payload: activeConfig,
+	})
+}
+
+func SendAgentWSPong(nodeID string) bool {
+	return sendAgentWSMessage(nodeID, AgentWSOutboundMessage{
+		Type: AgentWSMessageTypePong,
+	})
+}
+
+func sendAgentWSMessage(nodeID string, message AgentWSOutboundMessage) bool {
+	defaultAgentWSHub.mu.RLock()
+	client := defaultAgentWSHub.clients[nodeID]
+	defaultAgentWSHub.mu.RUnlock()
+	if client == nil {
+		return false
+	}
+	ok := client.Send(message)
+	if !ok {
+		slog.Debug("agent ws send queued message failed", "node_id", nodeID, "type", message.Type)
+	}
+	return ok
+}
+
+func BroadcastAgentWSActiveConfig(activeConfig *ActiveConfigMeta) AgentWSBroadcastResult {
+	result := AgentWSBroadcastResult{}
+	if activeConfig == nil {
+		slog.Debug("agent ws broadcast skipped because active config is nil")
+		return result
+	}
+	result.Version = activeConfig.Version
+	result.Checksum = activeConfig.Checksum
+
+	defaultAgentWSHub.mu.RLock()
+	clients := make([]*AgentWSClient, 0, len(defaultAgentWSHub.clients))
+	for _, client := range defaultAgentWSHub.clients {
+		clients = append(clients, client)
+	}
+	defaultAgentWSHub.mu.RUnlock()
+
+	result.ClientCount = len(clients)
+	message := AgentWSOutboundMessage{
+		Type:    AgentWSMessageTypeActiveConfig,
+		Payload: activeConfig,
+	}
+	for _, client := range clients {
+		if client.Send(message) {
+			result.SuccessCount++
+			continue
+		}
+		result.FailedNodes = append(result.FailedNodes, client.NodeID())
+	}
+	slog.Debug("agent ws broadcast active config",
+		"version", result.Version,
+		"checksum", result.Checksum,
+		"client_count", result.ClientCount,
+		"success_count", result.SuccessCount,
+		"failed_nodes", result.FailedNodes,
+	)
+	return result
+}
@@ -385,6 +385,10 @@ func PublishConfigVersion(createdBy string, force bool) (*ReleaseResult, error)
 		}
 		return nil, err
 	}
+	BroadcastAgentWSActiveConfig(&ActiveConfigMeta{
+		Version:  record.Version,
+		Checksum: record.Checksum,
+	})
 	return &ReleaseResult{
 		Version: record,
 		Routes:  bundle.Routes,
@@ -409,6 +413,10 @@ func ActivateConfigVersion(id uint) (*model.ConfigVersion, error) {
 		return nil, err
 	}
 	version.IsActive = true
+	BroadcastAgentWSActiveConfig(&ActiveConfigMeta{
+		Version:  version.Version,
+		Checksum: version.Checksum,
+	})
 	return version, nil
 }

@@ -48,23 +48,23 @@ type DashboardTrends struct {
 }

 type DashboardNodeHealth struct {
-	ID                  uint      `json:"id"`
-	NodeID              string    `json:"node_id"`
-	Name                string    `json:"name"`
-	GeoName             string    `json:"geo_name"`
-	GeoLatitude         *float64  `json:"geo_latitude"`
-	GeoLongitude        *float64  `json:"geo_longitude"`
-	Status              string    `json:"status"`
-	OpenrestyStatus     string    `json:"openresty_status"`
-	CurrentVersion      string    `json:"current_version"`
-	LastSeenAt          time.Time `json:"last_seen_at"`
-	ActiveEventCount    int       `json:"active_event_count"`
-	CPUUsagePercent     float64   `json:"cpu_usage_percent"`
-	MemoryUsagePercent  float64   `json:"memory_usage_percent"`
-	StorageUsagePercent float64   `json:"storage_usage_percent"`
-	RequestCount        int64     `json:"request_count"`
-	ErrorCount          int64     `json:"error_count"`
-	UniqueVisitorCount  int64     `json:"unique_visitor_count"`
+	ID                  uint     `json:"id"`
+	NodeID              string   `json:"node_id"`
+	Name                string   `json:"name"`
+	GeoName             string   `json:"geo_name"`
+	GeoLatitude         *float64 `json:"geo_latitude"`
+	GeoLongitude        *float64 `json:"geo_longitude"`
+	Status              string   `json:"status"`
+	OpenrestyStatus     string   `json:"openresty_status"`
+	CurrentVersion      string   `json:"current_version"`
+	LastSeenAt          any      `json:"last_seen_at"`
+	ActiveEventCount    int      `json:"active_event_count"`
+	CPUUsagePercent     float64  `json:"cpu_usage_percent"`
+	MemoryUsagePercent  float64  `json:"memory_usage_percent"`
+	StorageUsagePercent float64  `json:"storage_usage_percent"`
+	RequestCount        int64    `json:"request_count"`
+	ErrorCount          int64    `json:"error_count"`
+	UniqueVisitorCount  int64    `json:"unique_visitor_count"`
 }

 func GetDashboardOverview() (*DashboardOverviewView, error) {
@@ -139,7 +139,7 @@ func GetDashboardOverview() (*DashboardOverviewView, error) {
 			Status:           computedStatus,
 			OpenrestyStatus:  node.OpenrestyStatus,
 			CurrentVersion:   node.CurrentVersion,
-			LastSeenAt:       node.LastSeenAt,
+			LastSeenAt:       nodeViewLastSeenAt(node),
 			ActiveEventCount: len(nodeActiveEvents),
 		}

@@ -171,6 +171,11 @@ func RequestNodeAgentUpdate(id uint, input NodeAgentUpdateInput) (*NodeView, err
 		return nil, err
 	}
 	refreshAgentTokenCache(node)
+	if SendAgentWSSettings(node.NodeID, buildAgentSettings(node, true, channel.String(), tagName, node.RestartOpenrestyRequested)) {
+		slog.Debug("agent manual update pushed via ws", "node_id", node.NodeID, "channel", channel.String(), "tag", tagName)
+	} else {
+		slog.Debug("agent manual update waiting for next heartbeat", "node_id", node.NodeID, "channel", channel.String(), "tag", tagName)
+	}
 	slog.Info("agent manual update requested", "node_id", node.NodeID, "name", node.Name, "channel", channel.String(), "tag", tagName)
 	return buildNodeView(node), nil
 }
@@ -189,6 +194,22 @@ func RequestNodeOpenrestyRestart(id uint) (*NodeView, error) {
 	return buildNodeView(node), nil
 }

+func RequestNodeForceSync(id uint) (*NodeView, error) {
+	node, err := model.GetNodeByID(id)
+	if err != nil {
+		return nil, err
+	}
+	activeConfig, err := GetActiveConfigMetaForAgent()
+	if err != nil {
+		return nil, errors.New("无法获取当前激活的配置版本：" + err.Error())
+	}
+	if !SendAgentWSForceSyncConfig(node.NodeID, activeConfig) {
+		return nil, errors.New("节点不在线或通过 WebSocket 发送同步指令失败")
+	}
+	slog.Info("force sync requested via ws", "node_id", node.NodeID, "name", node.Name)
+	return buildNodeView(node), nil
+}
+
 func AuthenticateAgentToken(token string) (*model.Node, error) {
 	token = strings.TrimSpace(token)
 	if token == "" {
@@ -275,7 +296,7 @@ func buildNodeView(node *model.Node) *NodeView {
 		OpenrestyMessage:          strings.TrimSpace(node.OpenrestyMessage),
 		Status:                    status,
 		CurrentVersion:            node.CurrentVersion,
-		LastSeenAt:                node.LastSeenAt,
+		LastSeenAt:                nodeViewLastSeenAt(node),
 		LastError:                 node.LastError,
 		CreatedAt:                 node.CreatedAt,
 		UpdatedAt:                 node.UpdatedAt,
@@ -288,6 +309,16 @@ func buildNodeView(node *model.Node) *NodeView {
 	return view
 }

+func nodeViewLastSeenAt(node *model.Node) any {
+	if node != nil && IsAgentWSConnected(node.NodeID) {
+		return AgentWSConnectedLastSeenValue
+	}
+	if node == nil {
+		return time.Time{}
+	}
+	return node.LastSeenAt
+}
+
 func normalizeNodeInput(input NodeInput) (string, string, string, *float64, *float64, bool, error) {
 	name := strings.TrimSpace(input.Name)
 	ip := strings.TrimSpace(input.IP)
@@ -120,6 +120,9 @@ func TestRequestNodeAgentPreviewUpdate(t *testing.T) {
 		SetUpdateHTTPClientForTest(originalClient)
 	})

+	wsClient := RegisterAgentWSClient(node.NodeID)
+	defer UnregisterAgentWSClient(wsClient)
+
 	updated, err := RequestNodeAgentUpdate(node.ID, NodeAgentUpdateInput{
 		Channel: "preview",
 		TagName: "v0.5.0-rc.1",
@@ -136,6 +139,21 @@ func TestRequestNodeAgentPreviewUpdate(t *testing.T) {
 	if updated.UpdateTag != "v0.5.0-rc.1" {
 		t.Fatalf("unexpected update tag: %s", updated.UpdateTag)
 	}
+	select {
+	case message := <-wsClient.Messages():
+		if message.Type != AgentWSMessageTypeSettings {
+			t.Fatalf("expected settings message, got %s", message.Type)
+		}
+		settings, ok := message.Payload.(*AgentSettings)
+		if !ok {
+			t.Fatalf("expected agent settings payload, got %T", message.Payload)
+		}
+		if !settings.UpdateNow || settings.UpdateChannel != "preview" || settings.UpdateTag != "v0.5.0-rc.1" {
+			t.Fatalf("unexpected pushed settings: %+v", settings)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("expected websocket settings push for manual update")
+	}
 }

 func TestHeartbeatNodeReturnsPreviewUpdateSettings(t *testing.T) {
@@ -615,6 +633,43 @@ func TestListNodeViewsDoesNotPersistComputedStatus(t *testing.T) {
 	}
 }

+func TestAgentWSConnectionMarksNodeViewOnline(t *testing.T) {
+	node := &model.Node{
+		ID:         99,
+		NodeID:     "node-ws-view",
+		Name:       "edge-ws",
+		IP:         "10.0.0.9",
+		Status:     NodeStatusOffline,
+		LastSeenAt: time.Now().Add(-common.NodeOfflineThreshold - time.Minute),
+	}
+	client := RegisterAgentWSClient(node.NodeID)
+	defer UnregisterAgentWSClient(client)
+
+	view := buildNodeView(node)
+	if view.Status != NodeStatusOnline {
+		t.Fatalf("expected websocket-connected node to be online, got %s", view.Status)
+	}
+	if view.LastSeenAt != AgentWSConnectedLastSeenValue {
+		t.Fatalf("expected websocket special last_seen_at, got %#v", view.LastSeenAt)
+	}
+
+	result := BroadcastAgentWSActiveConfig(&ActiveConfigMeta{
+		Version:  "20260529-001",
+		Checksum: "checksum-ws",
+	})
+	if result.ClientCount != 1 || result.SuccessCount != 1 || len(result.FailedNodes) != 0 {
+		t.Fatalf("unexpected broadcast result: %+v", result)
+	}
+	select {
+	case message := <-client.Messages():
+		if message.Type != AgentWSMessageTypeActiveConfig {
+			t.Fatalf("unexpected websocket message type: %s", message.Type)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("expected websocket broadcast message")
+	}
+}
+
 func TestHeartbeatNodePersistsObservabilityPayload(t *testing.T) {
 	setupServiceTestDB(t)
 	withFakeAccessLogGeoProvider(t, &geoip.GeoInfo{
@@ -18,6 +18,7 @@ import {
  getNodeStatusVariant,
  getOpenrestyStatusLabel,
  getOpenrestyStatusVariant,
+  isWSConnectedLastSeen,
 } from '@/features/nodes/utils';
 import { formatDateTime, formatRelativeTime } from '@/lib/utils/date';
 import {
@@ -133,7 +134,9 @@ function NodeHealthRow({ node }: { node: DashboardNodeHealth }) {
        <p>当前版本：{node.current_version || '未应用'}</p>
        <p>
          最近心跳：
-          {node.last_seen_at
+          {isWSConnectedLastSeen(node.last_seen_at)
+            ? ' WS 已连接'
+            : node.last_seen_at
            ? ` ${formatRelativeTime(node.last_seen_at)} · ${formatDateTime(node.last_seen_at)}`
            : ' 暂无'}
        </p>
@@ -63,6 +63,12 @@ export function requestNodeAgentUpdate(
  });
 }

+export function requestNodeForceSync(id: number) {
+	return apiRequest<NodeItem>(`/nodes/${id}/force-sync`, {
+		method: 'POST',
+	});
+}
+
 export function requestNodeOpenrestyRestart(id: number) {
  return apiRequest<NodeItem>(`/nodes/${id}/openresty-restart`, {
    method: 'POST',
@@ -25,8 +25,10 @@ import {
  getNodeAgentRelease,
  getNodeObservability,
  getNodes,
+  requestNodeForceSync,
  requestNodeOpenrestyRestart,
  requestNodeAgentUpdate,
+  rotateNodeBootstrapToken,
  updateNode,
 } from '@/features/nodes/api/nodes';
 import { NodeEditorModal } from '@/features/nodes/components/node-editor-modal';
@@ -365,6 +367,20 @@ export function NodeDetailPage({ nodeId }: { nodeId: string }) {
    },
  });

+  const forceSyncMutation = useMutation({
+    mutationFn: () => requestNodeForceSync(Number(nodeId)),
+    onSuccess: async (updatedNode) => {
+      setFeedback({
+        tone: 'success',
+        message: `已向节点 ${updatedNode.name} 下发强制同步指令，无视当前错误拦截。`,
+      });
+      await queryClient.invalidateQueries({ queryKey: nodesQueryKey });
+    },
+    onError: (error) => {
+      setFeedback({ tone: 'danger', message: getErrorMessage(error) });
+    },
+  });
+
  const deleteMutation = useMutation({
    mutationFn: () => deleteNode(Number(nodeId)),
    onSuccess: async () => {
@@ -650,6 +666,16 @@ export function NodeDetailPage({ nodeId }: { nodeId: string }) {
              >
                {isRefreshing ? '刷新中...' : '刷新'}
              </SecondaryButton>
+              <SecondaryButton
+                type="button"
+                onClick={() => {
+                  setFeedback(null);
+                  forceSyncMutation.mutate();
+                }}
+                disabled={forceSyncMutation.isPending}
+              >
+                {forceSyncMutation.isPending ? '同步中...' : '同步'}
+              </SecondaryButton>
              <PrimaryButton
                type="button"
                onClick={handleOpenAgentUpdateModal}
@@ -34,6 +34,7 @@ import {
  getOpenrestyStatusLabel,
  getOpenrestyStatusVariant,
  isMeaningfulTime,
+  isWSConnectedLastSeen,
 } from '@/features/nodes/utils';

 const nodesQueryKey = ['nodes'];
@@ -350,7 +351,9 @@ export function NodesPage() {
                          </div>
                        </td>
                        <td className="px-3 py-4 text-[var(--foreground-secondary)]">
-                          {isMeaningfulTime(node.last_seen_at)
+                          {isWSConnectedLastSeen(node.last_seen_at)
+                            ? 'WS 已连接'
+                            : isMeaningfulTime(node.last_seen_at)
                            ? `${formatRelativeTime(
                                node.last_seen_at,
                              )} · ${formatDateTime(node.last_seen_at)}`
@@ -1,8 +1,18 @@
-import type { NodeItem } from '@/features/nodes/types';
-
-export function isMeaningfulTime(value: string | null | undefined) {
-  return Boolean(value) && !String(value).startsWith('0001-01-01');
-}
+import type { NodeItem } from '@/features/nodes/types';
+
+export const WS_CONNECTED_LAST_SEEN = '__OPENFLARE_WS_CONNECTED__';
+
+export function isWSConnectedLastSeen(value: string | null | undefined) {
+  return value === WS_CONNECTED_LAST_SEEN;
+}
+
+export function isMeaningfulTime(value: string | null | undefined) {
+  return (
+    Boolean(value) &&
+    !isWSConnectedLastSeen(value) &&
+    !String(value).startsWith('0001-01-01')
+  );
+}

 export function getNodeStatusVariant(status: NodeItem['status']) {
  if (status === 'online') {
@@ -87,6 +87,7 @@ const defaultSystemFields = {

 const defaultOperationFields = {
  AgentHeartbeatInterval: '10000',
+  AgentWebsocketUpgradeEnabled: true,
  NodeOfflineThreshold: '120000',
  AgentUpdateRepo: 'Rain-kl/OpenFlare',
  GeoIPProvider: 'ipinfo',
@@ -379,6 +380,10 @@ export function SettingsPage() {

    setOperationFields({
      AgentHeartbeatInterval: optionMap.AgentHeartbeatInterval ?? '10000',
+      AgentWebsocketUpgradeEnabled: toBoolean(
+        optionMap.AgentWebsocketUpgradeEnabled,
+        true,
+      ),
      NodeOfflineThreshold: optionMap.NodeOfflineThreshold ?? '120000',
      AgentUpdateRepo: optionMap.AgentUpdateRepo ?? 'Rain-kl/OpenFlare',
      GeoIPProvider: optionMap.GeoIPProvider ?? 'ipinfo',
@@ -1059,6 +1064,12 @@ export function SettingsPage() {
                      await saveOptionEntries(
                        [
                          ['AgentHeartbeatInterval', String(heartbeat)],
+                          [
+                            'AgentWebsocketUpgradeEnabled',
+                            String(
+                              operationFields.AgentWebsocketUpgradeEnabled,
+                            ),
+                          ],
                          ['NodeOfflineThreshold', String(offline)],
                          [
                            'AgentUpdateRepo',
@@ -1115,6 +1126,17 @@ export function SettingsPage() {
                      />
                    </ResourceField>
                  </div>
+                  <ToggleField
+                    label="开启 WS 连接升级"
+                    description="开启后 Agent 会在 HTTP 心跳成功后尝试升级为 WebSocket，发布配置时可立即收到同步通知。"
+                    checked={operationFields.AgentWebsocketUpgradeEnabled}
+                    onChange={(checked) =>
+                      setOperationFields((previous) => ({
+                        ...previous,
+                        AgentWebsocketUpgradeEnabled: checked,
+                      }))
+                    }
+                  />
                </div>

                <div className="border-t border-[var(--border-default)] pt-6">
Author	SHA1	Message	Date
ryan	806863f303	[修复] 修复 WS 连接下更新无法下发	2026-05-29 11:08:05 +08:00
ryan	ab8e3d4705	[优化] 更新 openresty_observability_port 描述，增强健康检查逻辑，使用 stub_status 代替 openresty -t	2026-05-29 10:52:25 +08:00
ryan	fe7f7da537	[优化] 更新 openresty_observability_port 描述，增强健康检查逻辑，使用 stub_status 代替 openresty -t	2026-05-29 10:50:13 +08:00
ryan	944b98d4d0	[新增] 实现节点强制同步功能，允许通过 API 请求强制同步配置	2026-05-29 10:44:14 +08:00
ryan	32dc7ef68e	[新增] 实现节点强制同步功能，允许通过 API 请求强制同步配置	2026-05-29 10:34:25 +08:00
ryan	32762fdf3c	[新增] 实现安全兜底配置功能，允许在无历史配置时启动 OpenResty 并返回 503 状态	2026-05-29 10:07:49 +08:00
ryan	79ed8fd6ab	[新增] 实现 Agent WebSocket 连接升级功能，支持状态上报和配置广播	2026-05-29 09:52:34 +08:00
ryan	4257b6fd5a	[优化] 移除 Docker 运行命令中的数据卷挂载	2026-05-29 09:39:49 +08:00