From 2333c3e5fd05cd9fc27dcca7a882334e117ef255 Mon Sep 17 00:00:00 2001 From: Haitao Pan Date: Fri, 26 Jun 2026 18:52:41 +0800 Subject: [PATCH] =?UTF-8?q?feat(acp):=20durable=20per-session=20run=20regi?= =?UTF-8?q?stry=20=E2=80=94=20survive=20gateway=20WS=20loss=20(T7/T8/T9)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OpenClaw gateway turns are async: chat.send returns a runId fast, the app then polls tasks.get. Previously every tasks.get re-asked the gateway, so a WS blip / reconnect that lost the gateway's in-memory run state turned into not_found / socket_closed — the already-finished result was lost and the client either hard-failed or polled forever. Make tasks.get resilient by leaning on the per-session store (s.sessions), whose lifetime is independent of the bridge<->gateway WebSocket: - T8: cache a gateway-confirmed terminal result (final client-facing shape, after download-URL decoration + inline-content stripping) into sess.lastResult and serve it on subsequent polls, so a later gateway not_found cannot lose it. - T7: when the gateway can't confirm (unavailable / socket closed / not_found) but the run is still within budget, synthesize a running handle so the client keeps polling across a transient blip — run tracking decoupled from WS lifetime. - T9: when the run is past its DeadlineAt and the gateway still can't confirm, return a deterministic `interrupted` terminal (OPENCLAW_RUN_DEADLINE_EXCEEDED). Correctness guards: - startOpenClawGatewayTask resets State/ProgressTerminal when a session is reused for a new turn, so a prior turn's terminal can't be mis-served for a new runId. - cache lookups verify the cached runId matches the requested runId (defense in depth). Design note: T7 is handled at the tasks.get layer (re-correlate by runId via the durable session store) rather than rewiring gatewayruntime's pending map — lower risk, equivalent effect. A killed in-flight request surfaces as a gateway error that the new fallback absorbs. T9 only force-terminates when the gateway is unconfirmed, never when it explicitly reports running (avoids killing legit long runs; the client-side deadline T3 covers that case). Tests: internal/acp/openclaw_run_registry_test.go (terminal detection, within-budget keep-polling, past-deadline interrupt, cache hit/replay, cross-runId isolation, no-session not_found). go vet + full acp package green. Co-Authored-By: Claude Opus 4.8 --- internal/acp/openclaw_run_registry.go | 185 +++++++++++++++++++++ internal/acp/openclaw_run_registry_test.go | 154 +++++++++++++++++ internal/acp/orchestrator.go | 4 + internal/acp/rpc_handler.go | 23 ++- 4 files changed, 354 insertions(+), 12 deletions(-) create mode 100644 internal/acp/openclaw_run_registry.go create mode 100644 internal/acp/openclaw_run_registry_test.go diff --git a/internal/acp/openclaw_run_registry.go b/internal/acp/openclaw_run_registry.go new file mode 100644 index 0000000..7fa9f42 --- /dev/null +++ b/internal/acp/openclaw_run_registry.go @@ -0,0 +1,185 @@ +package acp + +import ( + "strings" + "time" + + "xworkmate-bridge/internal/shared" +) + +// 持久 run 仓 / run 关联与 WS 解耦（T7/T8/T9）。 +// +// 背景：OpenClaw gateway turn 采用异步模型——chat.send 快速返回 runId，bridge 把 +// run 记录(sess.openClaw)、预算(sess.task.DeadlineAt)、运行句柄(sess.lastResult) +// 落在「按 sessionID 维度」的 per-session store 里（s.sessions），其生命周期独立于 +// bridge↔gateway 的 WebSocket 连接。客户端随后轮询 tasks.get。 +// +// 此前 tasks.get 每次都强依赖 gateway 应答：一旦 WS 抖动 / 重连后 run 内存态丢失， +// tasks.get 回 not_found 或 socket_closed，已完成的结果就此丢失，客户端要么硬失败、 +// 要么（修复前）无限轮询。下列辅助把 tasks.get 改造为「优先用持久 run 仓兜底」： +// +// T8 已观察到的终态结果缓存进 sess.lastResult，gateway 之后查不到也不丢； +// T7 gateway 暂时无法确认（unavailable / socket closed / not_found）但 run 仍在预算内时， +// 合成一个 running 句柄让客户端继续轮询，跨越瞬时抖动（与 WS 生命周期解耦）； +// T9 run 超过 DeadlineAt 且 gateway 仍无法确认时，回确定性的 interrupted 终态。 + +// openClawTaskGetResultIsTerminal 判断一个 tasks.get 结果是否表示 run 已结束。 +// 注意：仍在 artifact 同步中的结果会被 normalizeOpenClawTaskGetResult 重写为 status=running， +// 因此这里只认显式终态，不会把「同步中」误判为终态。 +func openClawTaskGetResultIsTerminal(payload map[string]any) bool { + switch strings.ToLower(strings.TrimSpace(shared.StringArg(payload, "status", ""))) { + case string(TaskStateCompleted), string(TaskStateFailed), string(TaskStateCancelled), + "interrupted", "partially_delivered": + return true + } + return false +} + +// cacheOpenClawTaskGetResultIfTerminal 把一次 gateway 确认的终态结果落进 per-session 持久 run 仓（T8）。 +func (s *Server) cacheOpenClawTaskGetResultIfTerminal(params map[string]any, payload map[string]any) { + if len(payload) == 0 || !openClawTaskGetResultIsTerminal(payload) { + return + } + sess := s.findTaskSession(params) + if sess == nil { + return + } + sess.mu.Lock() + defer sess.mu.Unlock() + switch strings.ToLower(strings.TrimSpace(shared.StringArg(payload, "status", ""))) { + case string(TaskStateFailed): + sess.task.State = TaskStateFailed + case string(TaskStateCancelled): + sess.task.State = TaskStateCancelled + default: + sess.task.State = TaskStateCompleted + } + sess.task.ProgressTerminal = true + sess.task.ProgressStage = strings.ToLower(strings.TrimSpace(shared.StringArg(payload, "status", ""))) + sess.task.UpdatedAt = time.Now() + sess.lastResult = cloneMap(payload) +} + +// cachedTerminalOpenClawResult 返回某 run 此前已观察到的终态结果（若有）（T7/T8）。 +func (s *Server) cachedTerminalOpenClawResult(params map[string]any) (map[string]any, bool) { + sess := s.findTaskSession(params) + if sess == nil { + return nil, false + } + sess.mu.Lock() + defer sess.mu.Unlock() + return cachedTerminalForRunLocked(sess, params) +} + +// cachedTerminalForRunLocked 仅当缓存终态确实属于「本次请求的 runId」时才命中， +// 防止同一 session 复用后把旧 run 的终态错配给新 run。调用方须持有 sess.mu。 +func cachedTerminalForRunLocked(sess *session, params map[string]any) (map[string]any, bool) { + if !sess.task.ProgressTerminal || len(sess.lastResult) == 0 { + return nil, false + } + if !openClawTaskGetResultIsTerminal(sess.lastResult) { + return nil, false + } + requestedRun := strings.TrimSpace(shared.StringArg(params, "runId", "")) + if requestedRun == "" { + requestedRun = strings.TrimSpace(shared.StringArg(params, "taskId", "")) + } + if requestedRun != "" { + cachedRun := firstNonEmptyString(sess.lastResult, "runId", "taskId") + if cachedRun != "" && !strings.EqualFold(cachedRun, requestedRun) { + return nil, false + } + } + return cloneMap(sess.lastResult), true +} + +// openClawTaskGetGatewayUnconfirmedFallback 在 gateway 无法确认 run 时，用持久 run 仓兜底（T7/T9）： +// - 已有缓存终态 -> 直接返回； +// - run 仍在预算内 -> 合成 running 句柄，客户端继续轮询，跨越瞬时抖动； +// - run 超过 deadline -> 回确定性 interrupted 终态。 +// +// 没有任何 per-session 记录时退回旧行为（not_found），不改变无状态查询的语义。 +func (s *Server) openClawTaskGetGatewayUnconfirmedFallback(params map[string]any, code string, message string) map[string]any { + notFound := func() map[string]any { + return map[string]any{ + "ok": false, + "status": "not_found", + "code": fallbackString(code, "TASK_LOOKUP_FAILED"), + "message": fallbackString(message, "openclaw native task lookup failed"), + } + } + sess := s.findTaskSession(params) + if sess == nil { + return notFound() + } + sess.mu.Lock() + defer sess.mu.Unlock() + if cached, ok := cachedTerminalForRunLocked(sess, params); ok { + return cached + } + if sess.openClaw == nil { + return notFound() + } + now := time.Now() + if !sess.task.DeadlineAt.IsZero() && now.After(sess.task.DeadlineAt) { + return s.markOpenClawRunDeadlineInterruptedLocked(sess, code, message) + } + // 仍在预算内：合成 running 句柄让客户端继续轮询，不因一次瞬时抖动硬失败。 + running := openClawRunningTaskResult(sess.openClaw) + running["transportDegraded"] = true + if strings.TrimSpace(code) != "" { + running["transportDegradedCode"] = strings.TrimSpace(code) + } + sess.lastResult = cloneMap(running) + return running +} + +// markOpenClawRunDeadlineInterruptedLocked 为「超过预算且 gateway 无法确认」的 run 生成确定性 +// interrupted 终态（T9）。调用方须持有 sess.mu。 +func (s *Server) markOpenClawRunDeadlineInterruptedLocked(sess *session, code string, message string) map[string]any { + now := time.Now() + sess.task.State = TaskStateFailed + sess.task.ProgressTerminal = true + sess.task.ProgressStage = "interrupted" + sess.task.ProgressMessage = "OpenClaw run exceeded its budget and could not be confirmed" + sess.task.UpdatedAt = now + + result := map[string]any{ + "ok": true, + "success": false, + "status": "interrupted", + "event": "interrupted", + "pending": false, + "code": "OPENCLAW_RUN_DEADLINE_EXCEEDED", + "artifactSyncStatus": "interrupted", + "message": "OpenClaw 任务超过预算上限且网关无法确认结果，已结束本轮等待。任务可能已在后台完成，请重新发送请求以拿回结果。", + "artifacts": []any{}, + } + if strings.TrimSpace(code) != "" { + result["gatewayUnconfirmedCode"] = strings.TrimSpace(code) + } + if strings.TrimSpace(message) != "" { + result["gatewayUnconfirmedMessage"] = strings.TrimSpace(message) + } + if record := sess.openClaw; record != nil { + result["runId"] = record.RunID + result["taskId"] = record.RunID + result["turnId"] = record.TurnID + result["sessionId"] = record.SessionID + result["threadId"] = record.ThreadID + result["appThreadKey"] = record.ThreadID + result["openclawSessionKey"] = record.SessionKey + result["resolvedGatewayProviderId"] = record.GatewayProviderID + result["startedAt"] = record.StartedAt.UTC().Format(time.RFC3339Nano) + result["deadlineAt"] = record.DeadlineAt.UTC().Format(time.RFC3339Nano) + } + sess.lastResult = cloneMap(result) + return result +} + +func fallbackString(value string, fallback string) string { + if strings.TrimSpace(value) == "" { + return fallback + } + return strings.TrimSpace(value) +} diff --git a/internal/acp/openclaw_run_registry_test.go b/internal/acp/openclaw_run_registry_test.go new file mode 100644 index 0000000..e5a94cf --- /dev/null +++ b/internal/acp/openclaw_run_registry_test.go @@ -0,0 +1,154 @@ +package acp + +import ( + "testing" + "time" + + "xworkmate-bridge/internal/shared" +) + +func newRunRegistryTestServer(deadline time.Time) (*Server, map[string]any) { + sess := &session{sessionID: "s1", threadID: "t1"} + sess.task.RunID = "run-1" + sess.task.SessionKey = "sk" + sess.task.GatewayProviderID = "openclaw" + sess.task.DeadlineAt = deadline + sess.openClaw = &OpenClawTaskRecord{ + SessionID: "s1", + ThreadID: "t1", + TurnID: "turn-1", + RunID: "run-1", + SessionKey: "sk", + GatewayProviderID: "openclaw", + StartedAt: time.Now().Add(-time.Minute), + DeadlineAt: deadline, + } + srv := &Server{sessions: map[string]*session{"s1": sess}} + params := map[string]any{"sessionId": "s1", "runId": "run-1"} + return srv, params +} + +func TestOpenClawTaskGetResultIsTerminal(t *testing.T) { + cases := []struct { + status string + want bool + }{ + {"completed", true}, + {"failed", true}, + {"cancelled", true}, + {"interrupted", true}, + {"partially_delivered", true}, + {"running", false}, + {"syncing-artifacts", false}, + {"queued", false}, + {"", false}, + } + for _, tc := range cases { + if got := openClawTaskGetResultIsTerminal(map[string]any{"status": tc.status}); got != tc.want { + t.Errorf("status=%q: got %v, want %v", tc.status, got, tc.want) + } + } +} + +// T7: gateway 无法确认但 run 仍在预算内 -> 合成 running 句柄续轮询。 +func TestGatewayUnconfirmedFallbackWithinBudgetKeepsPolling(t *testing.T) { + srv, params := newRunRegistryTestServer(time.Now().Add(30 * time.Minute)) + got := srv.openClawTaskGetGatewayUnconfirmedFallback(params, "SOCKET_CLOSED", "socket closed") + if status := shared.StringArg(got, "status", ""); status != string(TaskStateRunning) { + t.Fatalf("status = %q, want running", status) + } + if !parseBool(got["transportDegraded"]) { + t.Fatalf("transportDegraded not set: %v", got) + } + if shared.StringArg(got, "runId", "") != "run-1" { + t.Fatalf("runId mismatch: %v", got["runId"]) + } +} + +// T9: run 超过 deadline 且 gateway 无法确认 -> 确定性 interrupted 终态。 +func TestGatewayUnconfirmedFallbackPastDeadlineInterrupts(t *testing.T) { + srv, params := newRunRegistryTestServer(time.Now().Add(-time.Minute)) + got := srv.openClawTaskGetGatewayUnconfirmedFallback(params, "SOCKET_CLOSED", "socket closed") + if status := shared.StringArg(got, "status", ""); status != "interrupted" { + t.Fatalf("status = %q, want interrupted", status) + } + if code := shared.StringArg(got, "code", ""); code != "OPENCLAW_RUN_DEADLINE_EXCEEDED" { + t.Fatalf("code = %q, want OPENCLAW_RUN_DEADLINE_EXCEEDED", code) + } + if parseBool(got["success"]) { + t.Fatalf("interrupted result must not be success") + } + sess := srv.findTaskSession(params) + if sess == nil || !sess.task.ProgressTerminal || sess.task.State != TaskStateFailed { + t.Fatalf("session terminal state not recorded: %+v", sess) + } +} + +// T8: 已观察到的终态被缓存，且即使之后 gateway 不可达也优先返回缓存终态。 +func TestTerminalResultCachedAndServedAfterGatewayLoss(t *testing.T) { + srv, params := newRunRegistryTestServer(time.Now().Add(30 * time.Minute)) + terminal := map[string]any{ + "ok": true, + "success": true, + "status": "completed", + "runId": "run-1", + "message": "done", + } + srv.cacheOpenClawTaskGetResultIfTerminal(params, terminal) + + cached, ok := srv.cachedTerminalOpenClawResult(params) + if !ok { + t.Fatalf("expected cached terminal result") + } + if shared.StringArg(cached, "status", "") != "completed" { + t.Fatalf("cached status = %q, want completed", cached["status"]) + } + + // 即使 run 已过 deadline + gateway 丢失，也应优先返回缓存终态而非 interrupted。 + sess := srv.findTaskSession(params) + sess.mu.Lock() + sess.task.DeadlineAt = time.Now().Add(-time.Hour) + sess.mu.Unlock() + got := srv.openClawTaskGetGatewayUnconfirmedFallback(params, "SOCKET_CLOSED", "socket closed") + if shared.StringArg(got, "status", "") != "completed" { + t.Fatalf("expected cached completed to win over deadline interrupt, got %v", got["status"]) + } +} + +// 同一 session 复用后，旧 run 的终态不得错配给新 runId 的查询。 +func TestCachedTerminalNotServedForDifferentRunId(t *testing.T) { + srv, params := newRunRegistryTestServer(time.Now().Add(30 * time.Minute)) + srv.cacheOpenClawTaskGetResultIfTerminal(params, map[string]any{ + "status": "completed", "success": true, "runId": "run-1", + }) + // 新一轮查询带不同 runId -> 不应命中旧缓存。 + newParams := map[string]any{"sessionId": "s1", "runId": "run-2"} + if _, ok := srv.cachedTerminalOpenClawResult(newParams); ok { + t.Fatalf("stale terminal for run-1 must not be served for run-2") + } + // 原 runId 仍应命中。 + if _, ok := srv.cachedTerminalOpenClawResult(params); !ok { + t.Fatalf("terminal for run-1 should still be served for run-1") + } +} + +// running 结果不应被当作终态缓存。 +func TestRunningResultNotCachedAsTerminal(t *testing.T) { + srv, params := newRunRegistryTestServer(time.Now().Add(30 * time.Minute)) + srv.cacheOpenClawTaskGetResultIfTerminal(params, map[string]any{"status": "running", "runId": "run-1"}) + if _, ok := srv.cachedTerminalOpenClawResult(params); ok { + t.Fatalf("running result must not be cached as terminal") + } +} + +// 无 per-session 记录时退回旧的 not_found 行为。 +func TestGatewayUnconfirmedFallbackWithoutSessionReturnsNotFound(t *testing.T) { + srv := &Server{sessions: map[string]*session{}} + got := srv.openClawTaskGetGatewayUnconfirmedFallback(map[string]any{"sessionId": "missing"}, "X", "y") + if parseBool(got["ok"]) { + t.Fatalf("expected ok=false not_found, got %v", got) + } + if shared.StringArg(got, "status", "") != "not_found" { + t.Fatalf("status = %q, want not_found", got["status"]) + } +} diff --git a/internal/acp/orchestrator.go b/internal/acp/orchestrator.go index 4de7330..dbfbbfa 100644 --- a/internal/acp/orchestrator.go +++ b/internal/acp/orchestrator.go @@ -417,6 +417,10 @@ func (o *SessionOrchestrator) startOpenClawGatewayTask( sess.task.DeadlineAt = record.DeadlineAt sess.task.ProgressStage = "running" sess.task.ProgressMessage = "OpenClaw task accepted" + // 新一轮 turn 复用同一 session 时，必须重置上一轮可能留下的终态标记， + // 否则持久 run 仓(T8)会把旧 runId 的终态错配给新 run。 + sess.task.State = TaskStateRunning + sess.task.ProgressTerminal = false sess.openClaw = record running := openClawRunningTaskResult(record) sess.lastResult = cloneMap(running) diff --git a/internal/acp/rpc_handler.go b/internal/acp/rpc_handler.go index 4f1a1ba..bebec26 100644 --- a/internal/acp/rpc_handler.go +++ b/internal/acp/rpc_handler.go @@ -132,13 +132,13 @@ func (s *Server) handleTaskGet(ctx context.Context, params map[string]any, notif if gatewayProvider == "" { gatewayProvider = "openclaw" } + // T7/T8: 一旦观察到终态就从持久 run 仓返回，避免之后 gateway 查不到导致结果丢失。 + if cached, ok := s.cachedTerminalOpenClawResult(params); ok { + return cached + } if rpcErr := ensureProductionGatewayConnected(s, gatewayProvider, notify); rpcErr != nil { - return map[string]any{ - "ok": false, - "status": "not_found", - "code": "GATEWAY_UNAVAILABLE", - "message": rpcErr.Message, - } + // T7/T9: gateway 不可达时按持久 run 仓兜底（续轮询 / deadline 终态），而非裸 not_found。 + return s.openClawTaskGetGatewayUnconfirmedFallback(params, "GATEWAY_UNAVAILABLE", rpcErr.Message) } result := s.gateway.RequestByMode( gatewayProvider, @@ -162,16 +162,15 @@ func (s *Server) handleTaskGet(ctx context.Context, params map[string]any, notif } s.decorateOpenClawArtifactDownloadURLs(payload, sessionKey, runID) stripOpenClawArtifactInlineContent(payload) + // T8: 缓存「最终客户端可见形态」（已 decorate 下载 URL + strip 内联内容）， + // 这样从缓存回放时与正常路径完全一致。 + s.cacheOpenClawTaskGetResultIfTerminal(params, payload) return payload } + // T7/T9: gateway 返回错误（socket closed / not_found / lookup failed）时同样走持久 run 仓兜底。 message := strings.TrimSpace(shared.StringArg(result.Error, "message", "openclaw native task lookup failed")) code := strings.TrimSpace(shared.StringArg(result.Error, "code", "TASK_LOOKUP_FAILED")) - return map[string]any{ - "ok": false, - "status": "not_found", - "code": code, - "message": message, - } + return s.openClawTaskGetGatewayUnconfirmedFallback(params, code, message) } func (s *Server) taskGetParamsWithSessionScope(params map[string]any) map[string]any {