Avoid duplicate websocket reads in daemon ticks

Author: SqlRush

cmd/claude/main.go

@@ -66,6 +66,10 @@ type daemonOptions struct {
 	HeartbeatInterval time.Duration
 }
 
+type daemonTickOptions struct {
+	SkipRemoteWhenWebSocket bool
+}
+
 type daemonControlOptions struct {
 	StatePath         string
 	Status            bool
@@ -690,6 +694,7 @@ func runDaemon(ctx context.Context, state *bootstrap.State, options daemonOption
 	var tickMu sync.Mutex
 	stopRequested := make(chan struct{})
 	var stopOnce sync.Once
+	remoteStreamMode := false
 	writeHeartbeat := func(now time.Time) error {
 		return daemonpkg.WriteState(statePath, daemonpkg.BuildState(runner.SessionID, runner.WorkingDirectory, daemonpkg.RuntimeRunning, os.Getpid(), endpoint, now, nil))
 	}
@@ -699,7 +704,7 @@ func runDaemon(ctx context.Context, state *bootstrap.State, options daemonOption
 		if err := writeHeartbeat(now); err != nil {
 			return contracts.ToolResult{}, err
 		}
-		return runDaemonTick(ctx, runner, now)
+		return runDaemonTickWithOptions(ctx, runner, now, daemonTickOptions{SkipRemoteWhenWebSocket: remoteStreamMode})
 	}
 	if !options.Once {
 		daemonServer, err = daemonpkg.StartServer(daemonpkg.ServerOptions{
@@ -737,6 +742,7 @@ func runDaemon(ctx context.Context, state *bootstrap.State, options daemonOption
 		fmt.Fprintf(stderr, "ccgo daemon: %v\n", err)
 		return 1
 	}
+	remoteStreamMode = true
 	fmt.Fprintf(stdout, "ccgo daemon running\nsession_id=%s\nstate_path=%s\n", runner.SessionID, statePath)
 	if options.Once {
 		return 0
@@ -921,11 +927,20 @@ func runDaemonDueSchedules(ctx context.Context, runner conversation.Runner, now
 }
 
 func runDaemonTick(ctx context.Context, runner conversation.Runner, now time.Time) (contracts.ToolResult, error) {
+	return runDaemonTickWithOptions(ctx, runner, now, daemonTickOptions{})
+}
+
+func runDaemonTickWithOptions(ctx context.Context, runner conversation.Runner, now time.Time, options daemonTickOptions) (contracts.ToolResult, error) {
 	scheduleResult, err := runDaemonDueSchedules(ctx, runner, now)
 	if err != nil {
 		return contracts.ToolResult{}, err
 	}
-	remoteResult := runDaemonRemotePoll(ctx, runner, now)
+	var remoteResult contracts.ToolResult
+	if options.SkipRemoteWhenWebSocket {
+		remoteResult = runDaemonRemotePollUnlessStream(ctx, runner, now)
+	} else {
+		remoteResult = runDaemonRemotePoll(ctx, runner, now)
+	}
 	schedule := scheduleResult.StructuredContent
 	remotePoll := remoteResult.StructuredContent
 	scheduleTriggered := intMapValue(schedule, "triggered_count")
@@ -953,6 +968,32 @@ func runDaemonTick(ctx context.Context, runner conversation.Runner, now time.Tim
 	}, nil
 }
 
+func runDaemonRemotePollUnlessStream(ctx context.Context, runner conversation.Runner, now time.Time) contracts.ToolResult {
+	registrationPath := remotepkg.SessionRegistrationPath(runner.SessionPath, runner.SessionID)
+	if registrationPath == "" {
+		return runDaemonRemotePoll(ctx, runner, now)
+	}
+	registration, err := remotepkg.LoadRegistrationState(registrationPath)
+	if err != nil || registration.RuntimeState != remotepkg.RegistrationRegistered || strings.TrimSpace(registration.WebSocketURL) == "" {
+		return runDaemonRemotePoll(ctx, runner, now)
+	}
+	structured := map[string]any{
+		"type":            "remote_poll",
+		"checked_at":      now.UTC().Format(time.RFC3339Nano),
+		"runtime_state":   remotepkg.PumpRunning,
+		"transport":       "websocket_stream",
+		"skipped":         true,
+		"event_count":     0,
+		"delivered_count": 0,
+		"duplicate_count": 0,
+		"error_count":     0,
+	}
+	return contracts.ToolResult{
+		Content:           "Remote stream is handling websocket events.",
+		StructuredContent: structured,
+	}
+}
+
 func runDaemonRemotePoll(ctx context.Context, runner conversation.Runner, now time.Time) contracts.ToolResult {
 	structured := map[string]any{
 		"type":            "remote_poll",

cmd/claude/main_test.go

@@ -513,6 +513,43 @@ func TestRunDaemonRemoteStreamInjectsRemoteTriggers(t *testing.T) {
 	}
 }
 
+func TestRunDaemonTickSkipsRemotePollWhenWebSocketStreamRegistered(t *testing.T) {
+	dir := t.TempDir()
+	transcriptPath := filepath.Join(dir, "session.jsonl")
+	sessionID := contracts.ID("sess_daemon_stream_skip")
+	if err := remotepkg.WriteRegistrationState(remotepkg.SessionRegistrationPath(transcriptPath, sessionID), remotepkg.RegistrationState{
+		SessionID:    sessionID,
+		RuntimeState: remotepkg.RegistrationRegistered,
+		WebSocketURL: "ws://127.0.0.1:1/stream?token=secret",
+		PollURL:      "https://poll.example.invalid/events?token=secret",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	runner := conversation.Runner{
+		SessionID:        sessionID,
+		SessionPath:      transcriptPath,
+		WorkingDirectory: dir,
+	}
+	result, err := runDaemonTickWithOptions(context.Background(), runner, time.Unix(200, 0).UTC(), daemonTickOptions{SkipRemoteWhenWebSocket: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	remotePoll, ok := result.StructuredContent["remote_poll"].(map[string]any)
+	if !ok {
+		t.Fatalf("remote poll = %#v", result.StructuredContent["remote_poll"])
+	}
+	if remotePoll["transport"] != "websocket_stream" || remotePoll["skipped"] != true || remotePoll["error_count"] != 0 {
+		t.Fatalf("remote poll = %#v", remotePoll)
+	}
+	pump, err := remotepkg.LoadPumpState(remotepkg.SessionPumpPath(transcriptPath, sessionID))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if pump.RuntimeState != "" {
+		t.Fatalf("pump should not be overwritten by skipped tick: %#v", pump)
+	}
+}
+
 func TestRunDaemonServesHealthEndpoint(t *testing.T) {
 	t.Setenv("CLAUDE_CONFIG_DIR", t.TempDir())
 	cwd := t.TempDir()

docs/cc-100-roadmap.md

@@ -119,7 +119,7 @@ M10 补充：remote WebSocket pump 现在支持单次 tick 内读取多帧事件
 
 M10 补充：`internal/remote` 新增 callback 型 `StreamWebSocketEvents` primitive，可保持 WebSocket 连接逐帧解码并把事件批次交给调用方，支持 context 取消、可选帧上限、handler 错误传播、异常 close/读错后的 backoff 重连以及 `ReconnectAttempts < 0` 无限重连语义；该能力为 daemon 常驻 stream 托管接线打底。完整云端协议 hardening 仍未完成。
 
-M10 补充：`--daemon` 常驻模式现在会在初始 tick 后启动 remote WebSocket stream goroutine，按 heartbeat 间隔重试注册状态，复用 `StreamWebSocketEvents` 和 `RemoteTrigger` delivery/dedupe，把推送事件实时注入 running team，并在 `remote-pump.json` 中持续更新 `websocket_stream` transport、frame/connect/reconnect、delivered/duplicate/error 计数；daemon stop/context cancel 会取消 stream。完整云端协议 hardening 和更细的 stream lifecycle 审计仍未完成。
+M10 补充：`--daemon` 常驻模式现在会在初始 tick 后启动 remote WebSocket stream goroutine，按 heartbeat 间隔重试注册状态，复用 `StreamWebSocketEvents` 和 `RemoteTrigger` delivery/dedupe，把推送事件实时注入 running team，并在 `remote-pump.json` 中持续更新 `websocket_stream` transport、frame/connect/reconnect、delivered/duplicate/error 计数；daemon heartbeat/tick 在已注册 `websocket_url` 时会跳过短 WebSocket 读取，避免 stream 和 tick 双连接/重复写 pump state，poll-only 注册仍走原 tick 路径；daemon stop/context cancel 会取消 stream。完整云端协议 hardening 和更细的 stream lifecycle 审计仍未完成。
 
 M7 补充：interaction script paste payload 现在接受 ClipboardItem 风格的 `items[].getAsString`/`get_as_string` 以及 `stringData`/`textData` 文本字段，DOM clipboard 录制脚本可直接恢复 pasted text。
 

docs/claude-code-go-rewrite-plan.md

@@ -748,7 +748,7 @@ test/parity/                 # golden tests against TS/official behavior
 - 本轮补充：新增 session-scoped `remote-pump.json` 和 daemon remote 消息泵；daemon tick 会在 ScheduleCron due tick 后优先读取 registered `websocket_url`，通过 Bearer auth 建立 WebSocket、读取单帧事件并复用 poll 解码/`RemoteTrigger` 注入路径；无 WebSocket 或 WebSocket 失败且存在 poll URL 时回退到带 cursor 的 poll 拉取。pump 状态会记录 transport、websocket/poll URL 脱敏值、cursor、HTTP status、event/delivered/duplicate/error 计数，`/status show remote` 可审计当前传输。完整 CCR WebSocket 常驻持久 stream 和云端协议 hardening 仍未完成。
 - 本轮补充：remote WebSocket pump 现在支持单次 tick 内读取多帧事件，并在握手/读帧失败或非正常 close 时按可配置 backoff 重连；daemon 默认读取最多 8 帧、最多重连 2 次，并把 frame/connect/reconnect 计数写入 `remote-pump.json` 和 `/status show remote`。完整 CCR 云端 WebSocket 常驻持久 stream 与更深协议 hardening 仍未完成。
 - 本轮补充：`internal/remote` 新增 callback 型 `StreamWebSocketEvents` primitive，可保持 WebSocket 连接逐帧解码并把事件批次交给调用方，支持 context 取消、可选帧上限、handler 错误传播、异常 close/读错后的 backoff 重连以及 `ReconnectAttempts < 0` 无限重连语义；该能力为 daemon 常驻 stream 托管接线打底。完整云端协议 hardening 仍未完成。
-- 本轮补充：`--daemon` 常驻模式现在会在初始 tick 后启动 remote WebSocket stream goroutine，按 heartbeat 间隔重试注册状态，复用 `StreamWebSocketEvents` 和 `RemoteTrigger` delivery/dedupe，把推送事件实时注入 running team，并在 `remote-pump.json` 中持续更新 `websocket_stream` transport、frame/connect/reconnect、delivered/duplicate/error 计数；daemon stop/context cancel 会取消 stream。完整云端协议 hardening 和更细的 stream lifecycle 审计仍未完成。
+- 本轮补充：`--daemon` 常驻模式现在会在初始 tick 后启动 remote WebSocket stream goroutine，按 heartbeat 间隔重试注册状态，复用 `StreamWebSocketEvents` 和 `RemoteTrigger` delivery/dedupe，把推送事件实时注入 running team，并在 `remote-pump.json` 中持续更新 `websocket_stream` transport、frame/connect/reconnect、delivered/duplicate/error 计数；daemon heartbeat/tick 在已注册 `websocket_url` 时会跳过短 WebSocket 读取，避免 stream 和 tick 双连接/重复写 pump state，poll-only 注册仍走原 tick 路径；daemon stop/context cancel 会取消 stream。完整云端协议 hardening 和更细的 stream lifecycle 审计仍未完成。
 
 ### M11: Bridge 和高级集成