Honor remote delivery retry-after

Author: SqlRush

docs/cc-100-roadmap.md

@@ -131,6 +131,8 @@ M10 补充：remote poll/WebSocket 共用解码器现在兼容 `data`、`event`
 
 M10 补充：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 和服务端协议协商仍未完成。
 
+M10 补充：remote ack 和 lease renew 的 transient retry 现在会优先遵守服务端 `Retry-After` header（秒数或 HTTP-date），再回退到本地指数退避，并继续受最大退避上限约束，减少云端 429/503 限流时的协议偏差。
+
 M7 补充：interaction script paste payload 现在接受 ClipboardItem 风格的 `items[].getAsString`/`get_as_string` 以及 `stringData`/`textData` 文本字段，DOM clipboard 录制脚本可直接恢复 pasted text。
 
 M7 补充：scripted task runtime payload 和 task expectation 现在接受 `taskID`、`jobId`、`runId`、`label`、`displayName`、`phase`、`taskState`、`message`、`currentStep`、`percent`/`percentage`/`pct` 等相邻字段，并支持数字 task ID 与数字字符串 progress。

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

@@ -756,6 +756,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，并对 transport error、408/429/5xx 做一次短退避重试；pump state、structured result 和 `/status show remote` 会记录 ack event/sent/error 与 lease event 计数；非同源 ack URL 会被拒绝且脱敏；已过期 lease 会被跳过投递并 ack `expired`，同时记录 `lease_expired_count`。更深的服务端协议协商仍未完成。
+- 本轮补充：remote ack 和 lease renew 的 transient retry 现在会优先遵守服务端 `Retry-After` header（秒数或 HTTP-date），再回退到本地指数退避，并继续受最大退避上限约束，减少云端 429/503 限流时的协议偏差。
 - 本轮补充：daemon remote delivery 现在会对带 `lease_id` 且未过期的事件，在投递前向注册响应提供的同源 `lease_renew_url`/`lease_refresh_url` 做 best-effort POST，携带 Bearer auth、event_id 和 lease_id；renew 对 transport error、408/429/5xx 做一次短退避重试，成功/失败计数会写入 `remote-pump.json`、structured result 和 `/status show remote`。完整租约续期策略和云端协议演进策略仍未完成。
 
 ### M11: Bridge 和高级集成

internal/remote/pump.go

@@ -11,6 +11,7 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"time"
 
@@ -193,7 +194,7 @@ func SendAck(ctx context.Context, options AckOptions) AckResult {
 		result.StatusCode = resp.StatusCode
 		if resp.StatusCode < 200 || resp.StatusCode >= 300 {
 			result.Error = remoteRegistrationError(resp.Status, body)
-			if remoteDeliveryStatusRetryable(resp.StatusCode) && attempt < retries && sleepRemoteDeliveryRetry(ctx, options.RetryInitialDelay, options.RetryMaxDelay, attempt) {
+			if remoteDeliveryStatusRetryable(resp.StatusCode) && attempt < retries && sleepRemoteDeliveryRetryAfter(ctx, resp.Header.Get("Retry-After"), options.RetryInitialDelay, options.RetryMaxDelay, attempt, time.Now()) {
 				continue
 			}
 			return result
@@ -261,7 +262,7 @@ func SendLeaseRenewal(ctx context.Context, options LeaseRenewOptions) LeaseRenew
 		result.StatusCode = resp.StatusCode
 		if resp.StatusCode < 200 || resp.StatusCode >= 300 {
 			result.Error = remoteRegistrationError(resp.Status, body)
-			if remoteDeliveryStatusRetryable(resp.StatusCode) && attempt < retries && sleepRemoteDeliveryRetry(ctx, options.RetryInitialDelay, options.RetryMaxDelay, attempt) {
+			if remoteDeliveryStatusRetryable(resp.StatusCode) && attempt < retries && sleepRemoteDeliveryRetryAfter(ctx, resp.Header.Get("Retry-After"), options.RetryInitialDelay, options.RetryMaxDelay, attempt, time.Now()) {
 				continue
 			}
 			return result
@@ -384,6 +385,27 @@ func remoteDeliveryStatusRetryable(status int) bool {
 }
 
 func sleepRemoteDeliveryRetry(ctx context.Context, initialDelay, maxDelay time.Duration, attempt int) bool {
+	return sleepRemoteDeliveryRetryDelay(ctx, remoteDeliveryRetryDelay(initialDelay, maxDelay, attempt))
+}
+
+func sleepRemoteDeliveryRetryAfter(ctx context.Context, retryAfter string, initialDelay, maxDelay time.Duration, attempt int, now time.Time) bool {
+	delay := remoteDeliveryRetryDelay(initialDelay, maxDelay, attempt)
+	if retryDelay, ok := remoteRetryAfterDelay(retryAfter, now); ok {
+		delay = retryDelay
+		if maxDelay <= 0 {
+			maxDelay = time.Second
+		}
+		if delay > maxDelay {
+			delay = maxDelay
+		}
+		if delay < 0 {
+			delay = 0
+		}
+	}
+	return sleepRemoteDeliveryRetryDelay(ctx, delay)
+}
+
+func remoteDeliveryRetryDelay(initialDelay, maxDelay time.Duration, attempt int) time.Duration {
 	delay := initialDelay
 	if delay <= 0 {
 		delay = 100 * time.Millisecond
@@ -401,6 +423,35 @@ func sleepRemoteDeliveryRetry(ctx context.Context, initialDelay, maxDelay time.D
 	if delay > maxDelay {
 		delay = maxDelay
 	}
+	return delay
+}
+
+func remoteRetryAfterDelay(value string, now time.Time) (time.Duration, bool) {
+	value = strings.TrimSpace(value)
+	if value == "" {
+		return 0, false
+	}
+	if seconds, err := strconv.ParseInt(value, 10, 64); err == nil {
+		if seconds < 0 {
+			return 0, false
+		}
+		return time.Duration(seconds) * time.Second, true
+	}
+	after, err := http.ParseTime(value)
+	if err != nil {
+		return 0, false
+	}
+	if now.IsZero() {
+		now = time.Now()
+	}
+	delay := after.Sub(now)
+	if delay < 0 {
+		delay = 0
+	}
+	return delay, true
+}
+
+func sleepRemoteDeliveryRetryDelay(ctx context.Context, delay time.Duration) bool {
 	timer := time.NewTimer(delay)
 	defer timer.Stop()
 	select {

internal/remote/pump_test.go

@@ -168,6 +168,37 @@ func TestSendAckRetriesTransientFailure(t *testing.T) {
 	}
 }
 
+func TestSendAckHonorsRetryAfterHeader(t *testing.T) {
+	calls := 0
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls++
+		if calls == 1 {
+			w.Header().Set("Retry-After", "0")
+			http.Error(w, "slow down", http.StatusTooManyRequests)
+			return
+		}
+		w.WriteHeader(http.StatusAccepted)
+	}))
+	defer server.Close()
+
+	start := time.Now()
+	result := SendAck(context.Background(), AckOptions{
+		AckURL:            server.URL + "/ack",
+		EventID:           "evt-ack",
+		Status:            "delivered",
+		AllowedOrigins:    []string{server.URL + "/poll"},
+		RetryAttempts:     1,
+		RetryInitialDelay: time.Second,
+		RetryMaxDelay:     time.Second,
+	})
+	if result.Error != "" || result.StatusCode != http.StatusAccepted || result.AttemptCount != 2 || calls != 2 {
+		t.Fatalf("ack result = %#v calls=%d", result, calls)
+	}
+	if elapsed := time.Since(start); elapsed >= 500*time.Millisecond {
+		t.Fatalf("ack retry ignored Retry-After header; elapsed=%s", elapsed)
+	}
+}
+
 func TestSendAckRejectsDisallowedOriginAndRedactsURL(t *testing.T) {
 	result := SendAck(context.Background(), AckOptions{
 		AckURL:         "https://user:pass@example.invalid/ack?token=secret",
@@ -240,6 +271,39 @@ func TestSendLeaseRenewalRetriesTransientFailure(t *testing.T) {
 	}
 }
 
+func TestSendLeaseRenewalHonorsRetryAfterHeader(t *testing.T) {
+	calls := 0
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls++
+		if calls == 1 {
+			w.Header().Set("Retry-After", "0")
+			http.Error(w, "slow down", http.StatusTooManyRequests)
+			return
+		}
+		w.Header().Set("content-type", "application/json")
+		w.WriteHeader(http.StatusAccepted)
+		_, _ = w.Write([]byte(`{"lease_expires_at":"2026-06-17T12:15:00Z"}`))
+	}))
+	defer server.Close()
+
+	start := time.Now()
+	result := SendLeaseRenewal(context.Background(), LeaseRenewOptions{
+		LeaseRenewURL:     server.URL + "/leases/renew",
+		EventID:           "evt-renew",
+		LeaseID:           "lease-1",
+		AllowedOrigins:    []string{server.URL + "/poll"},
+		RetryAttempts:     1,
+		RetryInitialDelay: time.Second,
+		RetryMaxDelay:     time.Second,
+	})
+	if result.Error != "" || result.StatusCode != http.StatusAccepted || result.LeaseExpiresAt != "2026-06-17T12:15:00Z" || result.AttemptCount != 2 || calls != 2 {
+		t.Fatalf("renew result = %#v calls=%d", result, calls)
+	}
+	if elapsed := time.Since(start); elapsed >= 500*time.Millisecond {
+		t.Fatalf("lease renew retry ignored Retry-After header; elapsed=%s", elapsed)
+	}
+}
+
 func TestSendLeaseRenewalRejectsDisallowedOriginAndRedactsURL(t *testing.T) {
 	result := SendLeaseRenewal(context.Background(), LeaseRenewOptions{
 		LeaseRenewURL:  "https://user:pass@example.invalid/leases/renew?token=secret",