Skip to content

Commit 82de698

Browse files
author
SqlRush
committed
Audit remote stream lifecycle
1 parent 9ca0c92 commit 82de698

8 files changed

Lines changed: 93 additions & 51 deletions

File tree

‎cmd/claude/main.go‎

Lines changed: 26 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -857,13 +857,15 @@ func runDaemonRemoteStream(ctx context.Context, runner conversation.Runner, now
857857
if options.ReconnectMaxDelay <= 0 {
858858
options.ReconnectMaxDelay = 30 * time.Second
859859
}
860+
streamStartedAt := now.UTC().Format(time.RFC3339Nano)
860861
pumpState := remotepkg.PumpState{
861-
SessionID: runner.SessionID,
862-
RuntimeState: remotepkg.PumpRunning,
863-
Transport: "websocket_stream",
864-
PollURL: remotepkg.DisplayEndpoint(registration.PollURL),
865-
WebSocketURL: remotepkg.DisplayEndpoint(registration.WebSocketURL),
866-
LastPollAt: now.UTC().Format(time.RFC3339Nano),
862+
SessionID: runner.SessionID,
863+
RuntimeState: remotepkg.PumpRunning,
864+
Transport: "websocket_stream",
865+
PollURL: remotepkg.DisplayEndpoint(registration.PollURL),
866+
WebSocketURL: remotepkg.DisplayEndpoint(registration.WebSocketURL),
867+
LastPollAt: streamStartedAt,
868+
StreamStartedAt: streamStartedAt,
867869
}
868870
writeStreamState := func() {
869871
_ = remotepkg.WritePumpState(pumpPath, pumpState)
@@ -887,6 +889,8 @@ func runDaemonRemoteStream(ctx context.Context, runner conversation.Runner, now
887889
pumpState.ConnectCount = result.ConnectCount
888890
pumpState.ReconnectCount = result.ReconnectCount
889891
pumpState.LastPollAt = time.Now().UTC().Format(time.RFC3339Nano)
892+
pumpState.StreamEndedAt = pumpState.LastPollAt
893+
pumpState.StreamStopReason = daemonRemoteStreamStopReason(ctx, streamOptions, result)
890894
if result.Error != "" {
891895
pumpState.RuntimeState = remotepkg.PumpFailed
892896
pumpState.ErrorCount++
@@ -902,6 +906,9 @@ func runDaemonRemoteStream(ctx context.Context, runner conversation.Runner, now
902906
structured["frame_count"] = pumpState.FrameCount
903907
structured["connect_count"] = pumpState.ConnectCount
904908
structured["reconnect_count"] = pumpState.ReconnectCount
909+
structured["stream_started_at"] = pumpState.StreamStartedAt
910+
structured["stream_ended_at"] = pumpState.StreamEndedAt
911+
structured["stream_stop_reason"] = pumpState.StreamStopReason
905912
structured["event_count"] = pumpState.EventCount
906913
structured["delivered_count"] = pumpState.DeliveredCount
907914
structured["duplicate_count"] = pumpState.DuplicateCount
@@ -915,6 +922,19 @@ func runDaemonRemoteStream(ctx context.Context, runner conversation.Runner, now
915922
}
916923
}
917924

925+
func daemonRemoteStreamStopReason(ctx context.Context, options remotepkg.WebSocketOptions, result remotepkg.WebSocketResult) string {
926+
if strings.TrimSpace(result.Error) != "" {
927+
return "error"
928+
}
929+
if ctx.Err() != nil {
930+
return "context_cancelled"
931+
}
932+
if options.MaxFrames > 0 && result.FrameCount >= options.MaxFrames {
933+
return "max_frames"
934+
}
935+
return "closed"
936+
}
937+
918938
func runDaemonDueSchedules(ctx context.Context, runner conversation.Runner, now time.Time) (contracts.ToolResult, error) {
919939
return tasktools.RunDueSchedules(tool.Context{
920940
Context: ctx,

‎cmd/claude/main_test.go‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -494,14 +494,17 @@ func TestRunDaemonRemoteStreamInjectsRemoteTriggers(t *testing.T) {
494494
if result.StructuredContent["runtime_state"] != remotepkg.PumpRunning || result.StructuredContent["transport"] != "websocket_stream" || result.StructuredContent["frame_count"] != 1 || result.StructuredContent["connect_count"] != 1 || result.StructuredContent["delivered_count"] != 1 || result.StructuredContent["error_count"] != 0 {
495495
t.Fatalf("stream result = %#v", result.StructuredContent)
496496
}
497+
if result.StructuredContent["stream_started_at"] != "1970-01-01T00:03:20Z" || result.StructuredContent["stream_ended_at"] == "" || result.StructuredContent["stream_stop_reason"] != "max_frames" {
498+
t.Fatalf("stream lifecycle result = %#v", result.StructuredContent)
499+
}
497500
if len(auths) != 1 || auths[0] != "Bearer stream-token" {
498501
t.Fatalf("auths = %#v", auths)
499502
}
500503
pump, err := remotepkg.LoadPumpState(remotepkg.SessionPumpPath(transcriptPath, sessionID))
501504
if err != nil {
502505
t.Fatal(err)
503506
}
504-
if pump.Transport != "websocket_stream" || pump.FrameCount != 1 || pump.ConnectCount != 1 || pump.DeliveredCount != 1 || strings.Contains(pump.WebSocketURL, "token=secret") {
507+
if pump.Transport != "websocket_stream" || pump.FrameCount != 1 || pump.ConnectCount != 1 || pump.DeliveredCount != 1 || pump.StreamStartedAt != "1970-01-01T00:03:20Z" || pump.StreamEndedAt == "" || pump.StreamStopReason != "max_frames" || strings.Contains(pump.WebSocketURL, "token=secret") {
505508
t.Fatalf("pump = %#v", pump)
506509
}
507510
resume, err := manager.ResumeContext("agent/remote-lead", 3)

‎docs/cc-100-roadmap.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -119,7 +119,7 @@ M10 补充：remote WebSocket pump 现在支持单次 tick 内读取多帧事件
119119

120120
M10 补充：`internal/remote` 新增 callback 型 `StreamWebSocketEvents` primitive，可保持 WebSocket 连接逐帧解码并把事件批次交给调用方，支持 context 取消、可选帧上限、handler 错误传播、异常 close/读错后的 backoff 重连以及 `ReconnectAttempts < 0` 无限重连语义；该能力为 daemon 常驻 stream 托管接线打底。完整云端协议 hardening 仍未完成。
121121

122-
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 审计仍未完成。
122+
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，并在 pump state 与 `/status show remote` 中记录 stream start/end/stop reason。完整云端协议 hardening 仍未完成。
123123

124124
M7 补充：interaction script paste payload 现在接受 ClipboardItem 风格的 `items[].getAsString`/`get_as_string` 以及 `stringData`/`textData` 文本字段，DOM clipboard 录制脚本可直接恢复 pasted text。
125125

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -748,7 +748,7 @@ test/parity/ # golden tests against TS/official behavior
748748
- 本轮补充：新增 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 仍未完成。
749749
- 本轮补充：remote WebSocket pump 现在支持单次 tick 内读取多帧事件，并在握手/读帧失败或非正常 close 时按可配置 backoff 重连；daemon 默认读取最多 8 帧、最多重连 2 次，并把 frame/connect/reconnect 计数写入 `remote-pump.json` 和 `/status show remote`。完整 CCR 云端 WebSocket 常驻持久 stream 与更深协议 hardening 仍未完成。
750750
- 本轮补充：`internal/remote` 新增 callback 型 `StreamWebSocketEvents` primitive，可保持 WebSocket 连接逐帧解码并把事件批次交给调用方，支持 context 取消、可选帧上限、handler 错误传播、异常 close/读错后的 backoff 重连以及 `ReconnectAttempts < 0` 无限重连语义；该能力为 daemon 常驻 stream 托管接线打底。完整云端协议 hardening 仍未完成。
751-
- 本轮补充：`--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 审计仍未完成。
751+
- 本轮补充：`--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，并在 pump state 与 `/status show remote` 中记录 stream start/end/stop reason。完整云端协议 hardening 仍未完成。
752752

753753
### M11: Bridge 和高级集成
754754

‎internal/conversation/run.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1885,6 +1885,15 @@ func formatRemotePump(state remotepkg.PumpState) []string {
18851885
parts = append(parts, fmt.Sprintf("delivered %d", state.DeliveredCount))
18861886
parts = append(parts, fmt.Sprintf("duplicates %d", state.DuplicateCount))
18871887
parts = append(parts, fmt.Sprintf("errors %d", state.ErrorCount))
1888+
if state.StreamStartedAt != "" {
1889+
parts = append(parts, "stream started "+state.StreamStartedAt)
1890+
}
1891+
if state.StreamEndedAt != "" {
1892+
parts = append(parts, "stream ended "+state.StreamEndedAt)
1893+
}
1894+
if state.StreamStopReason != "" {
1895+
parts = append(parts, "stream stop "+state.StreamStopReason)
1896+
}
18881897
lines := []string{"Remote pump: " + strings.Join(parts, ": ")}
18891898
if state.LastError != "" {
18901899
lines = append(lines, "Remote pump error: "+state.LastError)

‎internal/conversation/run_test.go‎

Lines changed: 18 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -2819,20 +2819,23 @@ func TestRunnerExecutesStatusShowSectionsWithoutQuery(t *testing.T) {
28192819
t.Fatal(err)
28202820
}
28212821
if err := remotepkg.WritePumpState(remotepkg.SessionPumpPath(transcriptPath, "sess_status_show"), remotepkg.PumpState{
2822-
SessionID: "sess_status_show",
2823-
RuntimeState: remotepkg.PumpRunning,
2824-
Transport: "websocket",
2825-
WebSocketURL: "wss://remote.example/ws?token=secret",
2826-
PollURL: "https://remote.example/poll?token=secret",
2827-
LastCursor: "cursor-2",
2828-
StatusCode: http.StatusOK,
2829-
FrameCount: 2,
2830-
ConnectCount: 1,
2831-
ReconnectCount: 1,
2832-
EventCount: 3,
2833-
DeliveredCount: 2,
2834-
DuplicateCount: 1,
2835-
ErrorCount: 0,
2822+
SessionID: "sess_status_show",
2823+
RuntimeState: remotepkg.PumpRunning,
2824+
Transport: "websocket",
2825+
WebSocketURL: "wss://remote.example/ws?token=secret",
2826+
PollURL: "https://remote.example/poll?token=secret",
2827+
LastCursor: "cursor-2",
2828+
StreamStartedAt: "2026-06-17T10:03:00Z",
2829+
StreamEndedAt: "2026-06-17T10:08:00Z",
2830+
StreamStopReason: "context_cancelled",
2831+
StatusCode: http.StatusOK,
2832+
FrameCount: 2,
2833+
ConnectCount: 1,
2834+
ReconnectCount: 1,
2835+
EventCount: 3,
2836+
DeliveredCount: 2,
2837+
DuplicateCount: 1,
2838+
ErrorCount: 0,
28362839
}); err != nil {
28372840
t.Fatal(err)
28382841
}
@@ -2975,6 +2978,7 @@ func TestRunnerExecutesStatusShowSectionsWithoutQuery(t *testing.T) {
29752978
"Remote environment: env-status",
29762979
"Remote registration: registered: url https://remote.example/register: status 202: remote session remote-status: websocket wss://remote.example/ws: poll https://remote.example/poll",
29772980
"Remote pump: running: transport websocket: websocket wss://remote.example/ws: poll https://remote.example/poll: cursor cursor-2: status 200: frames 2: connects 1: reconnects 1: events 3: delivered 2: duplicates 1: errors 0",
2981+
"stream started 2026-06-17T10:03:00Z: stream ended 2026-06-17T10:08:00Z: stream stop context_cancelled",
29782982
"Remote services: 2",
29792983
"- bridge: running: endpoint http://127.0.0.1:8888: websocket ws://127.0.0.1:8888/ws: token required: commands 2: capabilities websocket_protocol, remote_trigger, remote_service",
29802984
"- daemon: running: endpoint http://127.0.0.1:7777: pid 4242: capabilities health, status, tick, stop",

‎internal/remote/pump.go‎

Lines changed: 19 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -26,22 +26,25 @@ const (
2626
)
2727

2828
type PumpState struct {
29-
SessionID contracts.ID `json:"session_id,omitempty"`
30-
RuntimeState string `json:"runtime_state"`
31-
Transport string `json:"transport,omitempty"`
32-
PollURL string `json:"poll_url,omitempty"`
33-
WebSocketURL string `json:"websocket_url,omitempty"`
34-
LastCursor string `json:"last_cursor,omitempty"`
35-
LastPollAt string `json:"last_poll_at,omitempty"`
36-
StatusCode int `json:"status_code,omitempty"`
37-
FrameCount int `json:"frame_count,omitempty"`
38-
ConnectCount int `json:"connect_count,omitempty"`
39-
ReconnectCount int `json:"reconnect_count,omitempty"`
40-
EventCount int `json:"event_count,omitempty"`
41-
DeliveredCount int `json:"delivered_count,omitempty"`
42-
DuplicateCount int `json:"duplicate_count,omitempty"`
43-
ErrorCount int `json:"error_count,omitempty"`
44-
LastError string `json:"last_error,omitempty"`
29+
SessionID contracts.ID `json:"session_id,omitempty"`
30+
RuntimeState string `json:"runtime_state"`
31+
Transport string `json:"transport,omitempty"`
32+
PollURL string `json:"poll_url,omitempty"`
33+
WebSocketURL string `json:"websocket_url,omitempty"`
34+
LastCursor string `json:"last_cursor,omitempty"`
35+
LastPollAt string `json:"last_poll_at,omitempty"`
36+
StreamStartedAt string `json:"stream_started_at,omitempty"`
37+
StreamEndedAt string `json:"stream_ended_at,omitempty"`
38+
StreamStopReason string `json:"stream_stop_reason,omitempty"`
39+
StatusCode int `json:"status_code,omitempty"`
40+
FrameCount int `json:"frame_count,omitempty"`
41+
ConnectCount int `json:"connect_count,omitempty"`
42+
ReconnectCount int `json:"reconnect_count,omitempty"`
43+
EventCount int `json:"event_count,omitempty"`
44+
DeliveredCount int `json:"delivered_count,omitempty"`
45+
DuplicateCount int `json:"duplicate_count,omitempty"`
46+
ErrorCount int `json:"error_count,omitempty"`
47+
LastError string `json:"last_error,omitempty"`
4548
}
4649

4750
type PollEvent struct {

‎internal/remote/pump_test.go‎

Lines changed: 15 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -91,17 +91,20 @@ func TestFetchPollEventsReportsFailedHTTPAndRedactsInvalidURL(t *testing.T) {
9191
func TestWriteAndLoadPumpState(t *testing.T) {
9292
path := filepath.Join(t.TempDir(), "sess_remote", pumpFileName)
9393
state := PumpState{
94-
SessionID: "sess_remote",
95-
RuntimeState: PumpRunning,
96-
Transport: "websocket",
97-
PollURL: "https://remote/poll",
98-
WebSocketURL: "wss://remote/ws",
99-
LastCursor: "cursor-1",
100-
FrameCount: 2,
101-
ConnectCount: 1,
102-
ReconnectCount: 1,
103-
EventCount: 2,
104-
DeliveredCount: 1,
94+
SessionID: "sess_remote",
95+
RuntimeState: PumpRunning,
96+
Transport: "websocket",
97+
PollURL: "https://remote/poll",
98+
WebSocketURL: "wss://remote/ws",
99+
LastCursor: "cursor-1",
100+
StreamStartedAt: "2026-06-17T10:00:00Z",
101+
StreamEndedAt: "2026-06-17T10:05:00Z",
102+
StreamStopReason: "max_frames",
103+
FrameCount: 2,
104+
ConnectCount: 1,
105+
ReconnectCount: 1,
106+
EventCount: 2,
107+
DeliveredCount: 1,
105108
}
106109
if err := WritePumpState(path, state); err != nil {
107110
t.Fatal(err)
@@ -110,7 +113,7 @@ func TestWriteAndLoadPumpState(t *testing.T) {
110113
if err != nil {
111114
t.Fatal(err)
112115
}
113-
if loaded.SessionID != "sess_remote" || loaded.RuntimeState != PumpRunning || loaded.Transport != "websocket" || loaded.WebSocketURL != "wss://remote/ws" || loaded.LastCursor != "cursor-1" || loaded.FrameCount != 2 || loaded.ConnectCount != 1 || loaded.ReconnectCount != 1 || loaded.LastPollAt == "" {
116+
if loaded.SessionID != "sess_remote" || loaded.RuntimeState != PumpRunning || loaded.Transport != "websocket" || loaded.WebSocketURL != "wss://remote/ws" || loaded.LastCursor != "cursor-1" || loaded.StreamStartedAt != "2026-06-17T10:00:00Z" || loaded.StreamEndedAt != "2026-06-17T10:05:00Z" || loaded.StreamStopReason != "max_frames" || loaded.FrameCount != 2 || loaded.ConnectCount != 1 || loaded.ReconnectCount != 1 || loaded.LastPollAt == "" {
114117
t.Fatalf("loaded = %#v", loaded)
115118
}
116119
data, err := json.Marshal(loaded)

0 commit comments

Comments
 (0)