From 2dc2d73b07dde8e76fa5cc478109f5b2a30091c1 Mon Sep 17 00:00:00 2001 From: zhang-guiping Date: Tue, 23 Jun 2026 04:02:58 +0800 Subject: [PATCH] fix(webchat): sessions persist after reconnects (#89017) * fix(gateway): preserve asserted webchat sessions * test(gateway): cover stale asserted webchat sessions * fix(gateway): scope webchat session resume * chore(protocol): refresh chat send models * fix: document reconnect session resume protocol * fix(gateway): keep reconnect resume internal * gateway: keep reconnect resume options internal * test(ui): avoid private resume marker lint access --- .../session-management-compaction.md | 1 + packages/gateway-protocol/src/index.test.ts | 10 ++ src/auto-reply/dispatch.ts | 29 +++--- .../reply/dispatch-from-config.types.ts | 8 +- src/auto-reply/reply/get-reply.ts | 12 ++- src/auto-reply/reply/get-reply.types.ts | 13 +++ src/auto-reply/reply/session.test.ts | 95 +++++++++++++++++++ src/auto-reply/reply/session.ts | 14 ++- src/gateway/server-methods/chat.ts | 29 +++++- .../server.chat.gateway-server-chat-b.test.ts | 71 ++++++++++++++ src/gateway/test-helpers.runtime-state.ts | 4 +- .../contracts/plugin-sdk-subpaths.test.ts | 8 ++ ui/src/ui/app-gateway.node.test.ts | 12 +++ ui/src/ui/app-gateway.ts | 7 ++ ui/src/ui/app-render.helpers.ts | 7 +- ui/src/ui/app.ts | 1 + ui/src/ui/controllers/chat.test.ts | 31 +++++- ui/src/ui/controllers/chat.ts | 14 +++ 18 files changed, 339 insertions(+), 27 deletions(-) diff --git a/docs/reference/session-management-compaction.md b/docs/reference/session-management-compaction.md index d279de436993..4c004f229a58 100644 --- a/docs/reference/session-management-compaction.md +++ b/docs/reference/session-management-compaction.md @@ -162,6 +162,7 @@ Rules of thumb: - **Reset** (`/new`, `/reset`) creates a new `sessionId` for that `sessionKey`. - **Daily reset** (default 4:00 AM local time on the gateway host) creates a new `sessionId` on the next message after the reset boundary. - **Idle expiry** (`session.reset.idleMinutes` or legacy `session.idleMinutes`) creates a new `sessionId` when a message arrives after the idle window. When daily + idle are both configured, whichever expires first wins. +- **Control UI reconnect resume** can preserve the currently visible session for one reconnect send when the Gateway receives the matching `sessionId` from an operator UI client. Ordinary stale sends still create a new `sessionId`. - **System events** (heartbeat, cron wakeups, exec notifications, gateway bookkeeping) may mutate the session row but do not extend daily/idle reset freshness. Reset rollover discards queued system-event notices for the previous session before the fresh prompt is built. - **Parent fork policy** uses OpenClaw's active branch when creating a thread or subagent fork. If that branch is too large, OpenClaw starts the child with isolated context instead of failing or inheriting unusable history. The sizing policy is automatic; legacy `session.parentForkMaxTokens` config is removed by `openclaw doctor --fix`. diff --git a/packages/gateway-protocol/src/index.test.ts b/packages/gateway-protocol/src/index.test.ts index 288f1b7aa678..3024b4bbdfbd 100644 --- a/packages/gateway-protocol/src/index.test.ts +++ b/packages/gateway-protocol/src/index.test.ts @@ -97,10 +97,20 @@ describe("lazy protocol validators", () => { validateChatSendParams({ sessionKey: "global", agentId: "work", + sessionId: "session-work", message: "hello", idempotencyKey: "run-global-work", }), ).toBe(true); + expect( + validateChatSendParams({ + sessionKey: "global", + sessionId: "session-work", + resumeSession: true, + message: "hello", + idempotencyKey: "run-global-work", + }), + ).toBe(false); expect( validateChatAbortParams({ sessionKey: "global", diff --git a/src/auto-reply/dispatch.ts b/src/auto-reply/dispatch.ts index 3a526ce265be..907ca9a969b1 100644 --- a/src/auto-reply/dispatch.ts +++ b/src/auto-reply/dispatch.ts @@ -24,7 +24,10 @@ import { copyReplyPayloadMetadata } from "./reply-payload.js"; import type { CommandSessionMetadataChange } from "./reply/command-session-metadata.js"; import { dispatchReplyFromConfig } from "./reply/dispatch-from-config.js"; import type { DispatchFromConfigResult } from "./reply/dispatch-from-config.types.js"; -import type { GetReplyFromConfig } from "./reply/get-reply.types.js"; +import type { + InternalGetReplyFromConfig, + InternalGetReplyOptions, +} from "./reply/get-reply.types.js"; import { finalizeInboundContext } from "./reply/inbound-context.js"; import { createReplyDispatcher, @@ -37,7 +40,9 @@ import type { ReplyDispatcher } from "./reply/reply-dispatcher.types.js"; import { runReplyPayloadSendingHook } from "./reply/reply-payload-sending-hook.js"; import { consumeReplyUsageState } from "./reply/reply-usage-state.js"; import type { FinalizedMsgContext, MsgContext } from "./templating.js"; -import type { GetReplyOptions, ReplyPayload } from "./types.js"; +import type { ReplyPayload } from "./types.js"; + +type InternalDispatchReplyOptions = Omit; type ForegroundReplyFenceState = { generation: number; @@ -60,9 +65,9 @@ const foregroundReplyFenceByKey = new Map(); const replyPayloadSendingDispatchers = new WeakSet(); function applyRuntimeToolsAllow( - replyOptions: Omit | undefined, + replyOptions: InternalDispatchReplyOptions | undefined, toolsAllow: string[] | undefined, -): Omit | undefined { +): InternalDispatchReplyOptions | undefined { if (toolsAllow === undefined) { return replyOptions; } @@ -377,9 +382,9 @@ function buildReplyPayloadSendingBeforeDeliver( } function bindReplyPayloadRunState( - replyOptions: Omit | undefined, + replyOptions: InternalDispatchReplyOptions | undefined, runState: ReplyPayloadRunState, -): Omit { +): InternalDispatchReplyOptions { const onAgentRunStart = replyOptions?.onAgentRunStart; return { ...replyOptions, @@ -499,8 +504,8 @@ export async function dispatchInboundMessage(params: { cfg: OpenClawConfig; dispatcher: ReplyDispatcher; toolsAllow?: string[]; - replyOptions?: Omit; - replyResolver?: GetReplyFromConfig; + replyOptions?: InternalDispatchReplyOptions; + replyResolver?: InternalGetReplyFromConfig; onSessionMetadataChanges?: (changes: CommandSessionMetadataChange[]) => void; replyPayloadRunState?: ReplyPayloadRunState; }): Promise { @@ -558,8 +563,8 @@ export async function dispatchInboundMessageWithBufferedDispatcher(params: { cfg: OpenClawConfig; dispatcherOptions: ReplyDispatcherWithTypingOptions; toolsAllow?: string[]; - replyOptions?: Omit; - replyResolver?: GetReplyFromConfig; + replyOptions?: InternalDispatchReplyOptions; + replyResolver?: InternalGetReplyFromConfig; onSessionMetadataChanges?: (changes: CommandSessionMetadataChange[]) => void; }): Promise { const finalized = finalizeInboundContext(params.ctx); @@ -660,8 +665,8 @@ export async function dispatchInboundMessageWithDispatcher(params: { cfg: OpenClawConfig; dispatcherOptions: ReplyDispatcherOptions; toolsAllow?: string[]; - replyOptions?: Omit; - replyResolver?: GetReplyFromConfig; + replyOptions?: InternalDispatchReplyOptions; + replyResolver?: InternalGetReplyFromConfig; }): Promise { const silentReplyContext = resolveDispatcherSilentReplyContext(params.ctx, params.cfg); const replyPayloadRunState = { diff --git a/src/auto-reply/reply/dispatch-from-config.types.ts b/src/auto-reply/reply/dispatch-from-config.types.ts index 6ca52c6c4033..c2b07adf8a4d 100644 --- a/src/auto-reply/reply/dispatch-from-config.types.ts +++ b/src/auto-reply/reply/dispatch-from-config.types.ts @@ -1,10 +1,10 @@ // Shared type contracts for dispatch-from-config runtime execution. import type { OpenClawConfig } from "../../config/types.openclaw.js"; -import type { GetReplyOptions, SourceReplyDeliveryMode } from "../get-reply-options.types.js"; +import type { SourceReplyDeliveryMode } from "../get-reply-options.types.js"; import type { FinalizedMsgContext } from "../templating.js"; import type { FormatAbortReplyText, TryFastAbortFromMessage } from "./abort.runtime-types.js"; import type { CommandSessionMetadataChange } from "./command-session-metadata.js"; -import type { GetReplyFromConfig } from "./get-reply.types.js"; +import type { InternalGetReplyFromConfig, InternalGetReplyOptions } from "./get-reply.types.js"; import type { ReplyDispatchKind, ReplyDispatcher } from "./reply-dispatcher.types.js"; export type DispatchFromConfigResult = { @@ -23,8 +23,8 @@ export type DispatchFromConfigParams = { ctx: FinalizedMsgContext; cfg: OpenClawConfig; dispatcher: ReplyDispatcher; - replyOptions?: Omit; - replyResolver?: GetReplyFromConfig; + replyOptions?: Omit; + replyResolver?: InternalGetReplyFromConfig; onSessionMetadataChanges?: (changes: CommandSessionMetadataChange[]) => void; fastAbortResolver?: TryFastAbortFromMessage; formatAbortReplyTextResolver?: FormatAbortReplyText; diff --git a/src/auto-reply/reply/get-reply.ts b/src/auto-reply/reply/get-reply.ts index 8b2e89fc6a5c..91c7f176968b 100644 --- a/src/auto-reply/reply/get-reply.ts +++ b/src/auto-reply/reply/get-reply.ts @@ -45,7 +45,10 @@ import { import { handleInlineActions } from "./get-reply-inline-actions.js"; import { maybeResolveNativeSlashCommandFastReply } from "./get-reply-native-slash-fast-path.js"; import { runPreparedReply } from "./get-reply-run.js"; -import type { ReplySessionBinding } from "./get-reply.types.js"; +import type { + InternalGetReplyOptions as BaseInternalGetReplyOptions, + ReplySessionBinding, +} from "./get-reply.types.js"; import { finalizeInboundContext } from "./inbound-context.js"; import { hasInboundMedia, hasInboundMediaForUnderstanding } from "./inbound-media.js"; import { emitPreAgentMessageHooks } from "./message-preprocess-hooks.js"; @@ -61,7 +64,7 @@ import { createTypingController } from "./typing.js"; type ResetCommandAction = "new" | "reset"; -type InternalGetReplyOptions = GetReplyOptions & { +type RuntimeInternalGetReplyOptions = BaseInternalGetReplyOptions & { onSessionPrepared?: (binding: ReplySessionBinding) => void; }; @@ -325,6 +328,7 @@ export async function getReplyFromConfig( ); const resolvedOpts = mergedSkillFilter !== undefined ? { ...opts, skillFilter: mergedSkillFilter } : opts; + const internalResolvedOpts = resolvedOpts as RuntimeInternalGetReplyOptions | undefined; const agentCfg = cfg.agents?.defaults; const sessionCfg = cfg.session; const { defaultProvider, defaultModel, aliasIndex } = resolverTiming.measureSync( @@ -480,6 +484,8 @@ export async function getReplyFromConfig( ctx: finalized, cfg, commandAuthorized, + requestedSessionId: internalResolvedOpts?.requestedSessionId, + resumeRequestedSession: internalResolvedOpts?.resumeRequestedSession, }), ); const { @@ -501,7 +507,7 @@ export async function getReplyFromConfig( } = sessionState; let { abortedLastRun } = sessionState; resolverTimingSessionKey = sessionKey ?? resolverTimingSessionKey; - (resolvedOpts as InternalGetReplyOptions | undefined)?.onSessionPrepared?.({ + internalResolvedOpts?.onSessionPrepared?.({ sessionKey, sessionId, storePath, diff --git a/src/auto-reply/reply/get-reply.types.ts b/src/auto-reply/reply/get-reply.types.ts index f67f388ba7df..95a5c6562cc3 100644 --- a/src/auto-reply/reply/get-reply.types.ts +++ b/src/auto-reply/reply/get-reply.types.ts @@ -10,9 +10,22 @@ export type ReplySessionBinding = { storePath?: string; }; +export type InternalReplySessionOptions = { + requestedSessionId?: string; + resumeRequestedSession?: boolean; +}; + +export type InternalGetReplyOptions = GetReplyOptions & InternalReplySessionOptions; + /** Reply resolver signature used by dispatchers and tests for dependency injection. */ export type GetReplyFromConfig = ( ctx: MsgContext, opts?: GetReplyOptions, configOverride?: OpenClawConfig, ) => Promise; + +export type InternalGetReplyFromConfig = ( + ctx: MsgContext, + opts?: InternalGetReplyOptions, + configOverride?: OpenClawConfig, +) => Promise; diff --git a/src/auto-reply/reply/session.test.ts b/src/auto-reply/reply/session.test.ts index 7d231f5a02b3..e9c582726217 100644 --- a/src/auto-reply/reply/session.test.ts +++ b/src/auto-reply/reply/session.test.ts @@ -1817,6 +1817,101 @@ describe("initSessionState reset policy", () => { expect(result.sessionId).not.toBe(existingSessionId); }); + it("preserves idle rollover when an ordinary send asserts the current session id", async () => { + vi.setSystemTime(new Date(2026, 0, 18, 5, 30, 0)); + const root = await makeCaseDir("openclaw-reset-idle-requested-session-ordinary-"); + const storePath = path.join(root, "sessions.json"); + const sessionKey = "agent:main:main"; + const existingSessionId = "webchat-ordinary-session-id"; + + await writeSessionStoreFast(storePath, { + [sessionKey]: { + sessionId: existingSessionId, + updatedAt: new Date(2026, 0, 18, 4, 45, 0).getTime(), + }, + }); + + const cfg = { + session: { + store: storePath, + reset: { mode: "daily", atHour: 4, idleMinutes: 30 }, + }, + } as OpenClawConfig; + const result = await initSessionState({ + ctx: { Body: "hello", SessionKey: sessionKey, Provider: "internal", Surface: "internal" }, + cfg, + commandAuthorized: true, + requestedSessionId: existingSessionId, + }); + + expect(result.isNewSession).toBe(true); + expect(result.sessionId).not.toBe(existingSessionId); + }); + + it("reuses an idle-expired session when a reconnecting client requests current session resume", async () => { + vi.setSystemTime(new Date(2026, 0, 18, 5, 30, 0)); + const root = await makeCaseDir("openclaw-reset-idle-requested-session-"); + const storePath = path.join(root, "sessions.json"); + const sessionKey = "agent:main:main"; + const existingSessionId = "webchat-reconnect-session-id"; + + await writeSessionStoreFast(storePath, { + [sessionKey]: { + sessionId: existingSessionId, + updatedAt: new Date(2026, 0, 18, 4, 45, 0).getTime(), + }, + }); + + const cfg = { + session: { + store: storePath, + reset: { mode: "daily", atHour: 4, idleMinutes: 30 }, + }, + } as OpenClawConfig; + const result = await initSessionState({ + ctx: { Body: "hello", SessionKey: sessionKey, Provider: "internal", Surface: "internal" }, + cfg, + commandAuthorized: true, + requestedSessionId: existingSessionId, + resumeRequestedSession: true, + }); + + expect(result.isNewSession).toBe(false); + expect(result.sessionId).toBe(existingSessionId); + }); + + it("does not reuse an idle-expired session for a stale asserted session id", async () => { + vi.setSystemTime(new Date(2026, 0, 18, 5, 30, 0)); + const root = await makeCaseDir("openclaw-reset-idle-stale-requested-session-"); + const storePath = path.join(root, "sessions.json"); + const sessionKey = "agent:main:main"; + const existingSessionId = "webchat-current-session-id"; + + await writeSessionStoreFast(storePath, { + [sessionKey]: { + sessionId: existingSessionId, + updatedAt: new Date(2026, 0, 18, 4, 45, 0).getTime(), + }, + }); + + const cfg = { + session: { + store: storePath, + reset: { mode: "daily", atHour: 4, idleMinutes: 30 }, + }, + } as OpenClawConfig; + const result = await initSessionState({ + ctx: { Body: "hello", SessionKey: sessionKey, Provider: "internal", Surface: "internal" }, + cfg, + commandAuthorized: true, + requestedSessionId: "webchat-stale-session-id", + resumeRequestedSession: true, + }); + + expect(result.isNewSession).toBe(true); + expect(result.sessionId).not.toBe(existingSessionId); + }); + it("drains stale system events when idle rollover creates a new session", async () => { vi.setSystemTime(new Date(2026, 0, 18, 5, 30, 0)); const root = await makeCaseDir("openclaw-reset-idle-system-events-"); diff --git a/src/auto-reply/reply/session.ts b/src/auto-reply/reply/session.ts index d3d6125ae238..2121551e1951 100644 --- a/src/auto-reply/reply/session.ts +++ b/src/auto-reply/reply/session.ts @@ -231,6 +231,8 @@ export async function initSessionState(params: { ctx: MsgContext; cfg: OpenClawConfig; commandAuthorized: boolean; + requestedSessionId?: string; + resumeRequestedSession?: boolean; }): Promise { const { ctx, cfg, commandAuthorized } = params; // Heartbeat, cron-event, and exec-event runs should NEVER trigger session @@ -433,6 +435,14 @@ export async function initSessionState(params: { Boolean(entry?.sessionId) && typeof entry?.updatedAt === "number" && Number.isFinite(entry.updatedAt); + const requestedSessionId = params.requestedSessionId?.trim() || undefined; + const requestedCurrentSession = Boolean( + requestedSessionId && entry?.sessionId && entry.sessionId === requestedSessionId, + ); + // Control UI sends sessionId on ordinary sends too, so only the one-shot reconnect + // resume signal is allowed to suppress configured idle/daily rollover. + const reconnectResumeRequested = + params.resumeRequestedSession === true && requestedCurrentSession; const skipImplicitExpiry = hasProviderOwnedSession(entry) && resetPolicy.configured !== true; const lifecycleTimestamps = resolveSessionLifecycleTimestamps({ entry, @@ -477,7 +487,9 @@ export async function initSessionState(params: { })); const freshEntry = (isSystemEvent && canReuseExistingEntry) || - (((entryFreshness?.fresh ?? false) || (softResetAllowed && canReuseExistingEntry)) && + (((reconnectResumeRequested && canReuseExistingEntry) || + (entryFreshness?.fresh ?? false) || + (softResetAllowed && canReuseExistingEntry)) && !terminalMainTranscriptNewerThanRegistry); // Capture the current session entry before any reset so its transcript can be // archived afterward. We need to do this for both explicit resets (/new, /reset) diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts index ff925e439896..7b3072b7ee22 100644 --- a/src/gateway/server-methods/chat.ts +++ b/src/gateway/server-methods/chat.ts @@ -283,6 +283,26 @@ function shouldIncludeChatSendAckServerTiming(client?: { return isOperatorUiClient(client); } +const CONTROL_UI_RECONNECT_RESUME_PARAM = "__controlUiReconnectResume"; + +function resolveControlUiReconnectResumeParams( + params: unknown, + clientInfo?: { id?: string | null; mode?: string | null }, +): { params: unknown; resumeRequested: boolean } { + if (!params || typeof params !== "object" || Array.isArray(params)) { + return { params, resumeRequested: false }; + } + const record = params as Record; + const resumeRequested = + record[CONTROL_UI_RECONNECT_RESUME_PARAM] === true && isOperatorUiClient(clientInfo); + if (!resumeRequested) { + return { params, resumeRequested: false }; + } + const validatedParams = { ...record }; + delete validatedParams[CONTROL_UI_RECONNECT_RESUME_PARAM]; + return { params: validatedParams, resumeRequested: true }; +} + function emitOperatorChatSendServerTiming(params: { context: Pick; client?: GatewayClient | null; @@ -3110,7 +3130,9 @@ export const chatHandlers: GatewayRequestHandlers = { }, "chat.send": async ({ params, respond, context, client }) => { const chatSendReceivedAtMs = performance.now(); - if (!validateChatSendParams(params)) { + const clientInfo = client?.connect?.client; + const controlUiReconnectResume = resolveControlUiReconnectResumeParams(params, clientInfo); + if (!validateChatSendParams(controlUiReconnectResume.params)) { respond( false, undefined, @@ -3121,7 +3143,7 @@ export const chatHandlers: GatewayRequestHandlers = { ); return; } - const p = params as { + const p = controlUiReconnectResume.params as { sessionKey: string; agentId?: string; sessionId?: string; @@ -3364,7 +3386,6 @@ export const chatHandlers: GatewayRequestHandlers = { }); return; } - const clientInfo = client?.connect?.client; const chatSendTraceAttributes = { runId: clientRunId, sessionKey, @@ -3957,6 +3978,8 @@ export const chatHandlers: GatewayRequestHandlers = { }), } : {}), + requestedSessionId, + resumeRequestedSession: controlUiReconnectResume.resumeRequested, abortSignal: activeRunAbort.controller.signal, images: replyOptionImages, imageOrder: imageOrder.length > 0 ? imageOrder : undefined, diff --git a/src/gateway/server.chat.gateway-server-chat-b.test.ts b/src/gateway/server.chat.gateway-server-chat-b.test.ts index ecfedaa7ffde..2f61f6d1edc4 100644 --- a/src/gateway/server.chat.gateway-server-chat-b.test.ts +++ b/src/gateway/server.chat.gateway-server-chat-b.test.ts @@ -5,6 +5,7 @@ import os from "node:os"; import path from "node:path"; import { afterAll, beforeAll, describe, expect, test, vi } from "vitest"; import type { GetReplyOptions } from "../auto-reply/get-reply-options.types.js"; +import type { InternalGetReplyOptions } from "../auto-reply/reply/get-reply.types.js"; import { clearConfigCache } from "../config/config.js"; import type { AgentModelConfig } from "../config/types.agents-shared.js"; import { createDeferred } from "../test-utils/deferred.js"; @@ -2416,6 +2417,76 @@ describe("gateway server chat", () => { ); }); + test("chat.send rejects Control UI reconnect resume marker from public WebChat clients", async () => { + await withGatewayChatHarness( + async ({ ws }) => { + await connectOk(ws, { + client: { + id: GATEWAY_CLIENT_NAMES.WEBCHAT_UI, + version: "1.0.0", + platform: "web", + mode: GATEWAY_CLIENT_MODES.WEBCHAT, + }, + }); + + const sendRes = await rpcReq(ws, "chat.send", { + sessionKey: "main", + sessionId: "sess-main", + __controlUiReconnectResume: true, + message: "hello after reconnect", + idempotencyKey: "idem-public-webchat-resume", + }); + expect(sendRes.ok).toBe(false); + }, + { + headers: { origin: `http://127.0.0.1:${harness.port}` }, + }, + ); + }); + + test("chat.send forwards Control UI reconnect resume internally", async () => { + await withGatewayChatHarness( + async ({ ws, createSessionDir }) => { + const spy = getReplyFromConfig; + await connectOk(ws, { + client: { + id: GATEWAY_CLIENT_NAMES.CONTROL_UI, + version: "1.0.0", + platform: "web", + mode: GATEWAY_CLIENT_MODES.WEBCHAT, + }, + }); + + await createSessionDir(); + await writeMainSessionStore(); + let capturedOpts: InternalGetReplyOptions | undefined; + mockGetReplyFromConfigOnce(async (_ctx, opts) => { + capturedOpts = opts; + return undefined; + }); + + const sendRes = await rpcReq(ws, "chat.send", { + sessionKey: "main", + sessionId: "sess-main", + __controlUiReconnectResume: true, + message: "hello after reconnect", + idempotencyKey: "idem-requested-session-id", + }); + expect(sendRes.ok).toBe(true); + + await vi.waitFor(() => { + expect(spy.mock.calls.length).toBeGreaterThan(0); + }, FAST_WAIT_OPTS); + + expect(capturedOpts?.requestedSessionId).toBe("sess-main"); + expect(capturedOpts?.resumeRequestedSession).toBe(true); + }, + { + headers: { origin: `http://127.0.0.1:${harness.port}` }, + }, + ); + }); + test("chat.history hard-caps single oversized nested payloads", async () => { await withGatewayChatHarness(async ({ ws, createSessionDir }) => { const historyMaxBytes = 64 * 1024; diff --git a/src/gateway/test-helpers.runtime-state.ts b/src/gateway/test-helpers.runtime-state.ts index 122e898d9c1a..530c1095175a 100644 --- a/src/gateway/test-helpers.runtime-state.ts +++ b/src/gateway/test-helpers.runtime-state.ts @@ -5,8 +5,8 @@ import os from "node:os"; import path from "node:path"; import { vi } from "vitest"; import type { Mock } from "vitest"; -import type { GetReplyOptions } from "../auto-reply/get-reply-options.types.js"; import type { ReplyPayload } from "../auto-reply/reply-payload.js"; +import type { InternalGetReplyOptions } from "../auto-reply/reply/get-reply.types.js"; import type { MsgContext } from "../auto-reply/templating.js"; import type { AgentBinding } from "../config/types.agents.js"; import type { HooksConfig } from "../config/types.hooks.js"; @@ -20,7 +20,7 @@ import { resolveGlobalSingleton } from "../shared/global-singleton.js"; */ export type GetReplyFromConfigFn = ( ctx: MsgContext, - opts?: GetReplyOptions, + opts?: InternalGetReplyOptions, configOverride?: OpenClawConfig, ) => Promise; type CronIsolatedRunFn = (...args: unknown[]) => Promise; diff --git a/src/plugins/contracts/plugin-sdk-subpaths.test.ts b/src/plugins/contracts/plugin-sdk-subpaths.test.ts index b2d58e3ef31b..9ab27160a7ce 100644 --- a/src/plugins/contracts/plugin-sdk-subpaths.test.ts +++ b/src/plugins/contracts/plugin-sdk-subpaths.test.ts @@ -23,6 +23,7 @@ import type { PluginRuntime as CorePluginRuntime, } from "openclaw/plugin-sdk/core"; import * as providerEntrySdk from "openclaw/plugin-sdk/provider-entry"; +import type { GetReplyOptions as ReplyRuntimeGetReplyOptions } from "openclaw/plugin-sdk/reply-runtime"; import * as zalouserSdk from "openclaw/plugin-sdk/zalouser"; import ts from "typescript"; import { beforeAll, describe, expect, expectTypeOf, it } from "vitest"; @@ -625,6 +626,8 @@ describe("plugin-sdk subpath exports", () => { "clearHistoryEntriesIfEnabled", "recordPendingHistoryEntryIfEnabled", "DEFAULT_GROUP_HISTORY_LIMIT", + "requestedSessionId", + "resumeRequestedSession", ], }); expectSourceMentions("account-helpers", ["createAccountListHelpers"]); @@ -1337,6 +1340,11 @@ describe("plugin-sdk subpath exports", () => { expectTypeOf().toMatchTypeOf(); expectTypeOf().toMatchTypeOf(); expectTypeOf().toMatchTypeOf(); + type PrivateResumeOptionKeys = Extract< + keyof ReplyRuntimeGetReplyOptions, + "requestedSessionId" | "resumeRequestedSession" + >; + expectTypeOf().toEqualTypeOf(); }); it("keeps runtime entry subpaths importable", async () => { diff --git a/ui/src/ui/app-gateway.node.test.ts b/ui/src/ui/app-gateway.node.test.ts index fc15883c9982..b40257da7aac 100644 --- a/ui/src/ui/app-gateway.node.test.ts +++ b/ui/src/ui/app-gateway.node.test.ts @@ -820,6 +820,18 @@ describe("connectGateway", () => { expect(host.lastErrorCode).toBeNull(); }); + it("marks the visible session for one reconnect resume after close", () => { + const host = createHost(); + host.currentSessionId = " session-before-reconnect "; + + connectGateway(host); + const client = requireGatewayClient(); + + client.emitClose({ code: 1006 }); + + expect(host.reconnectResumeSessionId).toBe("session-before-reconnect"); + }); + it("routes exec approval requested events with command spans", () => { const { host, client } = connectHostGateway(); diff --git a/ui/src/ui/app-gateway.ts b/ui/src/ui/app-gateway.ts index f73aca9bcb09..23aefce771ac 100644 --- a/ui/src/ui/app-gateway.ts +++ b/ui/src/ui/app-gateway.ts @@ -151,6 +151,8 @@ type GatewayHost = { execApprovalBusy: boolean; execApprovalError: string | null; updateAvailable: UpdateAvailable | null; + currentSessionId?: string | null; + reconnectResumeSessionId?: string | null; reconcileWebPushState?: () => Promise | void; realtimeTalkOptionsOpen?: boolean; fetchRealtimeTalkCatalog?: () => Promise; @@ -911,6 +913,11 @@ export function connectGateway(host: GatewayHost, options?: ConnectGatewayOption return; } host.connected = false; + const currentSessionId = + typeof host.currentSessionId === "string" ? host.currentSessionId.trim() : ""; + if (currentSessionId) { + host.reconnectResumeSessionId = currentSessionId; + } markQueuedChatSendsWaitingForReconnect( host as unknown as Parameters[0], ); diff --git a/ui/src/ui/app-render.helpers.ts b/ui/src/ui/app-render.helpers.ts index 292808f3d193..b9bd693a6a98 100644 --- a/ui/src/ui/app-render.helpers.ts +++ b/ui/src/ui/app-render.helpers.ts @@ -161,7 +161,12 @@ function resetChatStateForSessionSwitch(state: AppViewState, sessionKey: string) if (previousSessionKey !== sessionKey) { resetChatSessionPickerState(state); } - (state as unknown as { currentSessionId?: string | null }).currentSessionId = null; + const chatSessionState = state as unknown as { + currentSessionId?: string | null; + reconnectResumeSessionId?: string | null; + }; + chatSessionState.currentSessionId = null; + chatSessionState.reconnectResumeSessionId = null; state.chatMessage = ""; state.chatAttachments = []; state.chatMessages = restoreChatMessagesForSession(state, sessionKey); diff --git a/ui/src/ui/app.ts b/ui/src/ui/app.ts index b4d0e7dcdd10..ede73ca0ed5d 100644 --- a/ui/src/ui/app.ts +++ b/ui/src/ui/app.ts @@ -262,6 +262,7 @@ export class OpenClawApp extends LitElement { chatSessionMessageSubscriptionKey: string | null = null; chatSessionMessageSubscriptionRequestedKey: string | null = null; currentSessionId: string | null = null; + reconnectResumeSessionId: string | null = null; @state() chatLoading = false; @state() chatSending = false; @state() chatMessage = ""; diff --git a/ui/src/ui/controllers/chat.test.ts b/ui/src/ui/controllers/chat.test.ts index 04a4a48629d1..2915ff305788 100644 --- a/ui/src/ui/controllers/chat.test.ts +++ b/ui/src/ui/controllers/chat.test.ts @@ -1879,7 +1879,7 @@ describe("sendChatMessage", () => { expect(state.chatMessages).toHaveLength(1); }); - it("passes the backing session id from history when sending after reconnect", async () => { + it("passes the backing session id from history without resume for ordinary sends", async () => { const request = vi .fn() .mockResolvedValueOnce({ @@ -1904,9 +1904,38 @@ describe("sendChatMessage", () => { const sendParams = requireRecord(sendRequest?.[1]); expect(sendParams.sessionKey).toBe("main"); expect(sendParams.sessionId).toBe("session-before-reconnect"); + expect(sendParams.resumeSession).toBeUndefined(); + expect(sendParams).not.toHaveProperty("__controlUiReconnectResume"); expect(sendParams.message).toBe("continue"); }); + it("sends reconnect resume once when the current session matches the reconnect marker", async () => { + const request = vi.fn().mockResolvedValue({ runId: "run-1", status: "started" }); + const state = createState({ + connected: true, + client: { request } as unknown as ChatState["client"], + currentSessionId: "session-before-reconnect", + reconnectResumeSessionId: "session-before-reconnect", + }); + + await requestChatSend(state, { + message: "continue", + runId: "run-1", + }); + + expect(request).toHaveBeenCalledWith( + "chat.send", + expect.objectContaining({ + sessionKey: "main", + sessionId: "session-before-reconnect", + __controlUiReconnectResume: true, + message: "continue", + idempotencyKey: "run-1", + }), + ); + expect(state.reconnectResumeSessionId).toBeNull(); + }); + it("does not reuse another global agent's visible session id for queued sends", async () => { const request = vi.fn().mockResolvedValue({ runId: "run-work", status: "started" }); const state = createState({ diff --git a/ui/src/ui/controllers/chat.ts b/ui/src/ui/controllers/chat.ts index 0ab284946797..2866dada790b 100644 --- a/ui/src/ui/controllers/chat.ts +++ b/ui/src/ui/controllers/chat.ts @@ -385,6 +385,7 @@ export type ChatState = { connected: boolean; sessionKey: string; currentSessionId?: string | null; + reconnectResumeSessionId?: string | null; chatLoading: boolean; chatMessages: unknown[]; chatMessagesBySession?: ChatMessageCache; @@ -754,6 +755,12 @@ async function loadChatHistoryUncached( : typeof res.sessionId === "string" && res.sessionId.trim() ? res.sessionId : null; + if ( + state.reconnectResumeSessionId && + state.reconnectResumeSessionId !== state.currentSessionId + ) { + state.reconnectResumeSessionId = null; + } state.chatThinkingLevel = res.sessionInfo?.thinkingLevel ?? res.thinkingLevel ?? null; const resetStream = !state.chatRunId || state.chatRunId === previousRunId; if (resetStream) { @@ -956,17 +963,24 @@ export async function requestChatSend( }, ): Promise { const routing = resolveChatSendRouting(state, params); + const controlUiReconnectResume = Boolean( + routing.sessionId && state.reconnectResumeSessionId === routing.sessionId, + ); const payload = await state.client!.request("chat.send", { sessionKey: routing.sessionKey, ...(isGlobalSessionKey(routing.sessionKey) && routing.selectedAgentId ? { agentId: routing.selectedAgentId } : {}), ...(routing.sessionId ? { sessionId: routing.sessionId } : {}), + ...(controlUiReconnectResume ? { __controlUiReconnectResume: true } : {}), message: params.message, deliver: false, idempotencyKey: params.runId, attachments: buildApiAttachments(params.attachments), }); + if (controlUiReconnectResume) { + state.reconnectResumeSessionId = null; + } return normalizeChatSendAck(payload, params.runId); }