Renew remote event leases before delivery

Author: SqlRush

cmd/claude/main.go

@@ -873,20 +873,22 @@ func runDaemonRemoteStream(ctx context.Context, runner conversation.Runner, now
 	writeStreamState()
 	deliveryOptions := newDaemonRemoteDeliveryOptions(registration, authToken, time.Time{})
 	result := remotepkg.StreamWebSocketEvents(ctx, options, func(events []remotepkg.PollEvent) error {
-		delivered, duplicates, ackEvents, ackSent, ackErrors, leaseEvents, leaseExpired, errorsOut := deliverDaemonRemoteEvents(ctx, runner, events, deliveryOptions)
+		delivery := deliverDaemonRemoteEvents(ctx, runner, events, deliveryOptions)
 		pumpState.RuntimeState = remotepkg.PumpRunning
 		pumpState.EventCount += len(events)
-		pumpState.AckEventCount += ackEvents
-		pumpState.AckSentCount += ackSent
-		pumpState.AckErrorCount += ackErrors
-		pumpState.LeaseEventCount += leaseEvents
-		pumpState.LeaseExpiredCount += leaseExpired
-		pumpState.DeliveredCount += delivered
-		pumpState.DuplicateCount += duplicates
-		pumpState.ErrorCount += len(errorsOut)
+		pumpState.AckEventCount += delivery.AckEvents
+		pumpState.AckSentCount += delivery.AckSent
+		pumpState.AckErrorCount += delivery.AckErrors
+		pumpState.LeaseEventCount += delivery.LeaseEvents
+		pumpState.LeaseExpiredCount += delivery.LeaseExpired
+		pumpState.LeaseRenewSent += delivery.LeaseRenewSent
+		pumpState.LeaseRenewErrors += delivery.LeaseRenewErrors
+		pumpState.DeliveredCount += delivery.Delivered
+		pumpState.DuplicateCount += delivery.Duplicates
+		pumpState.ErrorCount += len(delivery.ErrorsOut)
 		pumpState.LastPollAt = time.Now().UTC().Format(time.RFC3339Nano)
-		if len(errorsOut) > 0 {
-			pumpState.LastError = fmt.Sprint(errorsOut[0]["error"])
+		if len(delivery.ErrorsOut) > 0 {
+			pumpState.LastError = fmt.Sprint(delivery.ErrorsOut[0]["error"])
 		}
 		writeStreamState()
 		return nil
@@ -919,6 +921,8 @@ func runDaemonRemoteStream(ctx context.Context, runner conversation.Runner, now
 	structured["ack_error_count"] = pumpState.AckErrorCount
 	structured["lease_event_count"] = pumpState.LeaseEventCount
 	structured["lease_expired_count"] = pumpState.LeaseExpiredCount
+	structured["lease_renew_sent_count"] = pumpState.LeaseRenewSent
+	structured["lease_renew_error_count"] = pumpState.LeaseRenewErrors
 	structured["stream_started_at"] = pumpState.StreamStartedAt
 	structured["stream_ended_at"] = pumpState.StreamEndedAt
 	structured["stream_stop_reason"] = pumpState.StreamStopReason
@@ -1123,17 +1127,19 @@ func runDaemonRemotePoll(ctx context.Context, runner conversation.Runner, now ti
 		_ = remotepkg.WritePumpState(pumpPath, pumpState)
 		return contracts.ToolResult{Content: "Remote poll failed.", StructuredContent: structured}
 	}
-	delivered, duplicates, ackEvents, ackSent, ackErrors, leaseEvents, leaseExpired, errorsOut := deliverDaemonRemoteEvents(ctx, runner, remoteFetch.Events, newDaemonRemoteDeliveryOptions(registration, authToken, now))
-	pumpState.AckEventCount = ackEvents
-	pumpState.AckSentCount = ackSent
-	pumpState.AckErrorCount = ackErrors
-	pumpState.LeaseEventCount = leaseEvents
-	pumpState.LeaseExpiredCount = leaseExpired
-	pumpState.DeliveredCount = delivered
-	pumpState.DuplicateCount = duplicates
-	pumpState.ErrorCount = len(errorsOut)
-	if len(errorsOut) > 0 {
-		pumpState.LastError = fmt.Sprint(errorsOut[0]["error"])
+	delivery := deliverDaemonRemoteEvents(ctx, runner, remoteFetch.Events, newDaemonRemoteDeliveryOptions(registration, authToken, now))
+	pumpState.AckEventCount = delivery.AckEvents
+	pumpState.AckSentCount = delivery.AckSent
+	pumpState.AckErrorCount = delivery.AckErrors
+	pumpState.LeaseEventCount = delivery.LeaseEvents
+	pumpState.LeaseExpiredCount = delivery.LeaseExpired
+	pumpState.LeaseRenewSent = delivery.LeaseRenewSent
+	pumpState.LeaseRenewErrors = delivery.LeaseRenewErrors
+	pumpState.DeliveredCount = delivery.Delivered
+	pumpState.DuplicateCount = delivery.Duplicates
+	pumpState.ErrorCount = len(delivery.ErrorsOut)
+	if len(delivery.ErrorsOut) > 0 {
+		pumpState.LastError = fmt.Sprint(delivery.ErrorsOut[0]["error"])
 	}
 	_ = remotepkg.WritePumpState(pumpPath, pumpState)
 	structured["runtime_state"] = pumpState.RuntimeState
@@ -1151,16 +1157,18 @@ func runDaemonRemotePoll(ctx context.Context, runner conversation.Runner, now ti
 	structured["ack_error_count"] = pumpState.AckErrorCount
 	structured["lease_event_count"] = pumpState.LeaseEventCount
 	structured["lease_expired_count"] = pumpState.LeaseExpiredCount
+	structured["lease_renew_sent_count"] = pumpState.LeaseRenewSent
+	structured["lease_renew_error_count"] = pumpState.LeaseRenewErrors
 	structured["event_count"] = pumpState.EventCount
 	structured["delivered_count"] = pumpState.DeliveredCount
 	structured["duplicate_count"] = pumpState.DuplicateCount
 	structured["error_count"] = pumpState.ErrorCount
-	structured["errors"] = errorsOut
+	structured["errors"] = delivery.ErrorsOut
 	if remoteFetch.FallbackError != "" {
 		structured["fallback_error"] = remoteFetch.FallbackError
 	}
 	return contracts.ToolResult{
-		Content:           fmt.Sprintf("Remote %s delivered %d event(s); %d duplicate(s); %d error(s).", remoteFetch.Transport, delivered, duplicates, len(errorsOut)),
+		Content:           fmt.Sprintf("Remote %s delivered %d event(s); %d duplicate(s); %d error(s).", remoteFetch.Transport, delivery.Delivered, delivery.Duplicates, len(delivery.ErrorsOut)),
 		StructuredContent: structured,
 	}
 }
@@ -1181,17 +1189,20 @@ type daemonRemoteFetch struct {
 type daemonRemoteDeliveryOptions struct {
 	AuthToken      string
 	AllowedOrigins []string
+	LeaseRenewURL  string
 	Now            time.Time
 }
 
 func newDaemonRemoteDeliveryOptions(registration remotepkg.RegistrationState, authToken string, now time.Time) daemonRemoteDeliveryOptions {
 	return daemonRemoteDeliveryOptions{
 		AuthToken: authToken,
 		AllowedOrigins: []string{
+			registration.RegistrationURL,
 			registration.PollURL,
 			registration.WebSocketURL,
 		},
-		Now: now,
+		LeaseRenewURL: registration.LeaseRenewURL,
+		Now:           now,
 	}
 }
 
@@ -1249,40 +1260,52 @@ func fetchDaemonRemoteEvents(ctx context.Context, registration remotepkg.Registr
 	}
 }
 
-func deliverDaemonRemoteEvents(ctx context.Context, runner conversation.Runner, events []remotepkg.PollEvent, options daemonRemoteDeliveryOptions) (int, int, int, int, int, int, int, []map[string]any) {
-	errorsOut := make([]map[string]any, 0)
-	delivered := 0
-	duplicates := 0
-	ackEvents := 0
-	ackSent := 0
-	ackErrors := 0
-	leaseEvents := 0
-	leaseExpired := 0
+type daemonRemoteDeliveryResult struct {
+	Delivered        int
+	Duplicates       int
+	AckEvents        int
+	AckSent          int
+	AckErrors        int
+	LeaseEvents      int
+	LeaseExpired     int
+	LeaseRenewSent   int
+	LeaseRenewErrors int
+	ErrorsOut        []map[string]any
+}
+
+func deliverDaemonRemoteEvents(ctx context.Context, runner conversation.Runner, events []remotepkg.PollEvent, options daemonRemoteDeliveryOptions) daemonRemoteDeliveryResult {
+	result := daemonRemoteDeliveryResult{ErrorsOut: make([]map[string]any, 0)}
 	for _, event := range events {
 		if event.AckURL != "" {
-			ackEvents++
+			result.AckEvents++
 		}
 		if event.LeaseID != "" || event.LeaseExpiresAt != "" {
-			leaseEvents++
+			result.LeaseEvents++
 		}
 		if daemonRemoteLeaseExpired(event, options.Now) {
-			leaseExpired++
+			result.LeaseExpired++
 			errorText := "remote event lease expired"
-			errorsOut = append(errorsOut, map[string]any{
+			result.ErrorsOut = append(result.ErrorsOut, map[string]any{
 				"type":             "remote_lease",
 				"event_id":         event.EventID,
 				"lease_id":         event.LeaseID,
 				"lease_expires_at": event.LeaseExpiresAt,
 				"error":            errorText,
 			})
 			sent, ackErr := acknowledgeDaemonRemoteEvent(ctx, event, options, "expired", 0, false, errorText)
-			ackSent += sent
+			result.AckSent += sent
 			if ackErr != nil {
-				ackErrors++
-				errorsOut = append(errorsOut, ackErr)
+				result.AckErrors++
+				result.ErrorsOut = append(result.ErrorsOut, ackErr)
 			}
 			continue
 		}
+		sent, renewErr := renewDaemonRemoteLease(ctx, event, options)
+		result.LeaseRenewSent += sent
+		if renewErr != nil {
+			result.LeaseRenewErrors++
+			result.ErrorsOut = append(result.ErrorsOut, renewErr)
+		}
 		input, err := json.Marshal(map[string]any{
 			"team_id":  event.TeamID,
 			"target":   event.Target,
@@ -1292,16 +1315,16 @@ func deliverDaemonRemoteEvents(ctx context.Context, runner conversation.Runner,
 			"message":  event.Message,
 		})
 		if err != nil {
-			errorsOut = append(errorsOut, map[string]any{"event_id": event.EventID, "error": err.Error()})
+			result.ErrorsOut = append(result.ErrorsOut, map[string]any{"event_id": event.EventID, "error": err.Error()})
 			sent, ackErr := acknowledgeDaemonRemoteEvent(ctx, event, options, "failed", 0, false, err.Error())
-			ackSent += sent
+			result.AckSent += sent
 			if ackErr != nil {
-				ackErrors++
-				errorsOut = append(errorsOut, ackErr)
+				result.AckErrors++
+				result.ErrorsOut = append(result.ErrorsOut, ackErr)
 			}
 			continue
 		}
-		result, err := tasktools.RunRemoteTrigger(tool.Context{
+		toolResult, err := tasktools.RunRemoteTrigger(tool.Context{
 			Context:          ctx,
 			WorkingDirectory: runner.WorkingDirectory,
 			SessionID:        runner.SessionID,
@@ -1310,35 +1333,35 @@ func deliverDaemonRemoteEvents(ctx context.Context, runner conversation.Runner,
 			},
 		}, input, tool.NopProgressSink())
 		if err != nil {
-			errorsOut = append(errorsOut, map[string]any{"event_id": event.EventID, "team_id": event.TeamID, "error": err.Error()})
+			result.ErrorsOut = append(result.ErrorsOut, map[string]any{"event_id": event.EventID, "team_id": event.TeamID, "error": err.Error()})
 			sent, ackErr := acknowledgeDaemonRemoteEvent(ctx, event, options, "failed", 0, false, err.Error())
-			ackSent += sent
+			result.AckSent += sent
 			if ackErr != nil {
-				ackErrors++
-				errorsOut = append(errorsOut, ackErr)
+				result.AckErrors++
+				result.ErrorsOut = append(result.ErrorsOut, ackErr)
 			}
 			continue
 		}
-		if duplicate, _ := result.StructuredContent["duplicate"].(bool); duplicate {
-			duplicates++
+		if duplicate, _ := toolResult.StructuredContent["duplicate"].(bool); duplicate {
+			result.Duplicates++
 			sent, ackErr := acknowledgeDaemonRemoteEvent(ctx, event, options, "duplicate", 0, true, "")
-			ackSent += sent
+			result.AckSent += sent
 			if ackErr != nil {
-				ackErrors++
-				errorsOut = append(errorsOut, ackErr)
+				result.AckErrors++
+				result.ErrorsOut = append(result.ErrorsOut, ackErr)
 			}
 			continue
 		}
-		sentCount := intMapValue(result.StructuredContent, "sent_count")
-		delivered += sentCount
+		sentCount := intMapValue(toolResult.StructuredContent, "sent_count")
+		result.Delivered += sentCount
 		sent, ackErr := acknowledgeDaemonRemoteEvent(ctx, event, options, "delivered", sentCount, false, "")
-		ackSent += sent
+		result.AckSent += sent
 		if ackErr != nil {
-			ackErrors++
-			errorsOut = append(errorsOut, ackErr)
+			result.AckErrors++
+			result.ErrorsOut = append(result.ErrorsOut, ackErr)
 		}
 	}
-	return delivered, duplicates, ackEvents, ackSent, ackErrors, leaseEvents, leaseExpired, errorsOut
+	return result
 }
 
 func daemonRemoteLeaseExpired(event remotepkg.PollEvent, now time.Time) bool {
@@ -1355,6 +1378,28 @@ func daemonRemoteLeaseExpired(event remotepkg.PollEvent, now time.Time) bool {
 	return !expiresAt.After(now.UTC())
 }
 
+func renewDaemonRemoteLease(ctx context.Context, event remotepkg.PollEvent, options daemonRemoteDeliveryOptions) (int, map[string]any) {
+	if strings.TrimSpace(event.LeaseID) == "" || strings.TrimSpace(options.LeaseRenewURL) == "" {
+		return 0, nil
+	}
+	result := remotepkg.SendLeaseRenewal(ctx, remotepkg.LeaseRenewOptions{
+		LeaseRenewURL:  options.LeaseRenewURL,
+		AuthToken:      options.AuthToken,
+		EventID:        event.EventID,
+		LeaseID:        event.LeaseID,
+		AllowedOrigins: options.AllowedOrigins,
+	})
+	if result.Error != "" {
+		return 0, map[string]any{
+			"type":     "remote_lease_renew",
+			"event_id": event.EventID,
+			"lease_id": event.LeaseID,
+			"error":    result.Error,
+		}
+	}
+	return 1, nil
+}
+
 func acknowledgeDaemonRemoteEvent(ctx context.Context, event remotepkg.PollEvent, options daemonRemoteDeliveryOptions, status string, sentCount int, duplicate bool, errorText string) (int, map[string]any) {
 	if event.AckURL == "" {
 		return 0, nil

cmd/claude/main_test.go

@@ -466,6 +466,91 @@ func TestRunDaemonRemotePollSkipsExpiredLease(t *testing.T) {
 	}
 }
 
+func TestRunDaemonRemotePollRenewsLeaseBeforeDelivery(t *testing.T) {
+	dir := t.TempDir()
+	transcriptPath := filepath.Join(dir, "session.jsonl")
+	sessionID := contracts.ID("sess_daemon_remote_lease_renew")
+	manager := session.NewSidechainManager(transcriptPath, sessionID)
+	if _, err := manager.Start(session.SidechainOptions{ID: "agent/remote-lead", StartedAt: time.Unix(100, 0).UTC()}); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := manager.CreateTeam(session.TeamOptions{
+		ID:                "remote/team",
+		CoordinatorTaskID: "agent/remote-lead",
+		Timestamp:         time.Unix(102, 0).UTC(),
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var renewAuth string
+	var renewPayload map[string]any
+	var ackStatuses []string
+	var server *httptest.Server
+	server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case "/leases/renew":
+			renewAuth = r.Header.Get("Authorization")
+			if err := json.NewDecoder(r.Body).Decode(&renewPayload); err != nil {
+				t.Fatal(err)
+			}
+			w.Header().Set("content-type", "application/json")
+			w.WriteHeader(http.StatusAccepted)
+			_, _ = w.Write([]byte(`{"lease_expires_at":"2026-06-17T12:05:00Z"}`))
+		case "/ack":
+			var payload map[string]any
+			if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
+				t.Fatal(err)
+			}
+			ackStatuses = append(ackStatuses, fmt.Sprint(payload["status"]))
+			w.WriteHeader(http.StatusAccepted)
+		default:
+			w.Header().Set("content-type", "application/json")
+			_, _ = w.Write([]byte(fmt.Sprintf(`{"events":[{"id":"delivery-renew","team":"remote/team","message":"Renewed deploy.","ack_url":%q,"lease_id":"lease-renew","lease_expires_at":"2026-06-17T12:00:00Z"}]}`, server.URL+"/ack")))
+		}
+	}))
+	defer server.Close()
+	if err := remotepkg.WriteRegistrationState(remotepkg.SessionRegistrationPath(transcriptPath, sessionID), remotepkg.RegistrationState{
+		SessionID:       sessionID,
+		RuntimeState:    remotepkg.RegistrationRegistered,
+		PollURL:         server.URL + "/poll?token=secret",
+		LeaseRenewURL:   server.URL + "/leases/renew?token=secret",
+		RemoteSessionID: "remote-session",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	runner := conversation.Runner{
+		SessionID:        sessionID,
+		SessionPath:      transcriptPath,
+		WorkingDirectory: dir,
+		MCP: &conversation.MCPConfig{UserSettings: contracts.Settings{
+			Remote: &contracts.RemoteSetting{AuthToken: "renew-token"},
+		}},
+	}
+	result := runDaemonRemotePoll(context.Background(), runner, time.Date(2026, 6, 17, 11, 55, 0, 0, time.UTC))
+	if result.StructuredContent["delivered_count"] != 1 || result.StructuredContent["lease_event_count"] != 1 || result.StructuredContent["lease_expired_count"] != 0 || result.StructuredContent["lease_renew_sent_count"] != 1 || result.StructuredContent["lease_renew_error_count"] != 0 || result.StructuredContent["ack_sent_count"] != 1 || result.StructuredContent["error_count"] != 0 {
+		t.Fatalf("renew poll = %#v", result.StructuredContent)
+	}
+	if renewAuth != "Bearer renew-token" || renewPayload["event_id"] != "delivery-renew" || renewPayload["lease_id"] != "lease-renew" {
+		t.Fatalf("renew auth=%q payload=%#v", renewAuth, renewPayload)
+	}
+	if len(ackStatuses) != 1 || ackStatuses[0] != "delivered" {
+		t.Fatalf("ack statuses = %#v", ackStatuses)
+	}
+	pump, err := remotepkg.LoadPumpState(remotepkg.SessionPumpPath(transcriptPath, sessionID))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if pump.DeliveredCount != 1 || pump.LeaseRenewSent != 1 || pump.LeaseRenewErrors != 0 || pump.ErrorCount != 0 {
+		t.Fatalf("pump = %#v", pump)
+	}
+	resume, err := manager.ResumeContext("agent/remote-lead", 3)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(resume.Messages) != 2 || !strings.Contains(messages.TextContent(resume.Messages[1]), "Renewed deploy.") {
+		t.Fatalf("resume messages = %#v", resume.Messages)
+	}
+}
+
 func TestRunDaemonRemotePollPrefersWebSocket(t *testing.T) {
 	dir := t.TempDir()
 	transcriptPath := filepath.Join(dir, "session.jsonl")

docs/cc-100-roadmap.md

@@ -1488,6 +1488,8 @@ M7 补充：terminal input parser 和 configurable keybinding name parser 现在
 
 本轮补充：remote registration 响应会持久化协议版本、能力列表和 lease renew/refresh endpoint，并在 `/status show remote` 中脱敏展示，为后续租约刷新调用和协议兼容判断提供可审计状态；实际 lease renew/refresh 调用和协议版本强制兼容仍未完成。
 
+本轮补充：daemon remote delivery 会在投递未过期 leased event 前，对注册级同源 lease renew/refresh endpoint 做 best-effort POST，并把 renew sent/error 计数写入 pump state、structured result 和 `/status show remote`；完整续期策略、重试退避和协议版本强制兼容仍未完成。
+
 ### M11: Bridge, LSP, Telemetry, Advanced Integrations
 
 目标：补齐高级集成能力。

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

@@ -753,6 +753,7 @@ test/parity/                 # golden tests against TS/official behavior
 - 本轮补充：`--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 仍未完成。
 - 本轮补充：remote poll/WebSocket 共用解码器现在兼容 `data`、`event`、`remote_event`、`delivery`、`payload` 包裹的单条事件，以及这些 wrapper 下的 `events/items/messages/deliveries` 列表；云端可以用 envelope 协议携带 cursor 和事件内容，而无需强制把事件字段铺在顶层。更深的鉴权刷新、ack/lease 和服务端协议协商仍未完成。
 - 本轮补充：remote poll/WebSocket 事件现在会解析 `ack_url`/`ackUrl`/`acknowledge_url`/`receipt_url` 和 `lease_id`/`lease_expires_at` 及 `ack`/`lease` nested object 元数据；daemon 会在 delivered/duplicate/failed 后对注册 poll/websocket 同源的 ack URL 做 best-effort POST，带 Bearer auth 和 event/status/sent_count/duplicate/error payload，并在 pump state、structured result 和 `/status show remote` 记录 ack event/sent/error 与 lease event 计数；非同源 ack URL 会被拒绝且脱敏；已过期 lease 会被跳过投递并 ack `expired`，同时记录 `lease_expired_count`。更深的 lease renew/refresh 和服务端协议协商仍未完成。
+- 本轮补充：daemon remote delivery 现在会对带 `lease_id` 且未过期的事件，在投递前向注册响应提供的同源 `lease_renew_url`/`lease_refresh_url` 做 best-effort POST，携带 Bearer auth、event_id 和 lease_id；renew 成功/失败计数会写入 `remote-pump.json`、structured result 和 `/status show remote`。完整租约续期策略、失败重试和协议版本强制兼容仍未完成。
 
 ### M11: Bridge 和高级集成