diff --git a/src/gateway/server-methods/shared-types.ts b/src/gateway/server-methods/shared-types.ts index 573e0746d6ac..588d4a7b2858 100644 --- a/src/gateway/server-methods/shared-types.ts +++ b/src/gateway/server-methods/shared-types.ts @@ -164,6 +164,7 @@ type GatewaySystemAgentSession = { historyLength: () => number; historySince: (index: number) => SystemAgentHistoryTurn[]; hasPendingQrCode: () => boolean; + hasRecoverableQrReply: () => boolean; getPendingOperatorProposal: () => { operation: SystemAgentOperation; hash: string } | null; resolveOperatorApproval: ( decision: "allow-once" | "allow-always" | "deny" | null, diff --git a/src/gateway/server-methods/system-agent-session-ownership.test.ts b/src/gateway/server-methods/system-agent-session-ownership.test.ts index d8c0b5c837a5..f4c52f1f524f 100644 --- a/src/gateway/server-methods/system-agent-session-ownership.test.ts +++ b/src/gateway/server-methods/system-agent-session-ownership.test.ts @@ -40,6 +40,7 @@ type FakeEngine = { cancelWizard: ReturnType; pollStep: ReturnType; hasPendingQrCode: ReturnType; + hasRecoverableQrReply: ReturnType; handle: ReturnType; seedHistory: ReturnType; historyLength: ReturnType; @@ -63,6 +64,7 @@ function makeEngine(): FakeEngine { throw new SystemAgentWizardAnswerError("The hosted wizard step is no longer active."); }), hasPendingQrCode: vi.fn(() => false), + hasRecoverableQrReply: vi.fn(() => false), handle: vi.fn(async () => ({ text: "did the thing", action: "none" })), seedHistory: vi.fn(), historyLength: vi.fn(() => 0), diff --git a/src/gateway/server-methods/system-agent.test.ts b/src/gateway/server-methods/system-agent.test.ts index 5d926d1a2742..3557c01a1dde 100644 --- a/src/gateway/server-methods/system-agent.test.ts +++ b/src/gateway/server-methods/system-agent.test.ts @@ -889,6 +889,29 @@ describe("openclaw.chat", () => { client: defaultClient, } as never); + const lostDeliverySession = expectDefined(sessions.get("s1"), "lost delivery session"); + lostDeliverySession.lastUsedAt = 0; + const otherRetainedEngines = await Promise.all( + Array.from({ length: 7 }, async () => await makeDeliveredQrEngine()), + ); + for (const [index, retainedEngine] of otherRetainedEngines.entries()) { + sessions.set( + `delivered-${index}`, + seededSession({ + engine: retainedEngine, + lastUsedAt: index + 1, + ownerKey: `device:owner-${index}`, + }), + ); + } + stubEngineOverview(); + + const admission = await callChat(context, { sessionId: "new-session" }); + expect(admission).toMatchObject({ ok: true }); + expect(sessions.has("s1")).toBe(true); + expect(sessions.has("new-session")).toBe(true); + expect(sessions.size).toBe(9); + const persistedTerminal = transcriptStoreMocks.appendTranscriptTurn.mock.calls.filter( ([turn]) => turn.role === "assistant" && turn.text.includes("is configured"), ); @@ -1238,31 +1261,25 @@ describe("openclaw.chat", () => { ); }); - it("admits a new session when eight retained QR terminals were already delivered", async () => { - const engines = await Promise.all( - Array.from({ length: 8 }, async () => await makeDeliveredQrEngine()), - ); - expect(engines.map((engine) => engine.hasPendingQrCode())).toEqual( - Array.from({ length: 8 }, () => false), - ); - const sessions = new Map( - engines.map((engine, index) => [ - `delivered-${index}`, - seededSession({ - engine, - lastUsedAt: index, - ownerKey: `device:owner-${index}`, - }), - ]), - ); + it("bounds retained QR recovery sessions without evicting their live leases", async () => { + const sessions = new Map(); + for (let index = 0; index < 16; index += 1) { + const recoverySession = seededSession({ + lastUsedAt: index, + ownerKey: `device:recovery-owner-${index}`, + }); + vi.spyOn(recoverySession.engine, "hasRecoverableQrReply").mockReturnValue(true); + sessions.set(`recovery-${index}`, recoverySession); + } stubEngineOverview(); - const result = await callChat(makeContext(sessions), { sessionId: "new-session" }); + const result = await callChat(makeContext(sessions), { sessionId: "overflow" }); - expect(result).toMatchObject({ ok: true }); - expect(sessions.size).toBe(8); - expect(sessions.has("delivered-0")).toBe(false); - expect(sessions.has("new-session")).toBe(true); + expect(result).toMatchObject({ ok: false, error: { code: "UNAVAILABLE" } }); + expect(sessions.size).toBe(16); + expect([...sessions.keys()]).toEqual( + Array.from({ length: 16 }, (_value, index) => `recovery-${index}`), + ); }); it("protects terminal QR delivery through audit, then expires abandoned retention", async () => { diff --git a/src/gateway/server-methods/system-agent.ts b/src/gateway/server-methods/system-agent.ts index d186d73a8d89..40c227d70b6a 100644 --- a/src/gateway/server-methods/system-agent.ts +++ b/src/gateway/server-methods/system-agent.ts @@ -88,7 +88,9 @@ import { assertValidParams } from "./validation.js"; export type SystemAgentChatSession = GatewayRequestContext["systemAgentSessions"] extends Map ? Session : never; -const MAX_SYSTEM_AGENT_SESSIONS = 8; +const MAX_SYSTEM_AGENT_ACTIVE_SESSIONS = 8; +// Recovery entries free active capacity without making retained engines unbounded. +const MAX_SYSTEM_AGENT_SESSION_ENTRIES = 16; const SYSTEM_AGENT_SEED_HISTORY_LIMIT = 30; const DEFAULT_SYSTEM_AGENT_HISTORY_LIMIT = 100; const PROVIDER_AUTH_SESSION_TIMEOUT_MS = 25 * 60 * 1000; @@ -160,11 +162,15 @@ async function evictOldestSession( sessions: Map, context: GatewayRequestContext, ): Promise { - if (sessions.size < MAX_SYSTEM_AGENT_SESSIONS) { - return true; - } + let activeSessionCount = 0; const protectedQrSessions = new Map(); + const recoveryKeys = new Set(); for (const [key, session] of sessions) { + if (session.engine.hasRecoverableQrReply()) { + recoveryKeys.add(key); + continue; + } + activeSessionCount += 1; if (!session.engine.hasPendingQrCode()) { continue; } @@ -173,11 +179,19 @@ async function evictOldestSession( protectedQrSessions.set(session.ownerKey, { key, lastUsedAt: session.lastUsedAt }); } } + if ( + activeSessionCount < MAX_SYSTEM_AGENT_ACTIVE_SESSIONS && + sessions.size < MAX_SYSTEM_AGENT_SESSION_ENTRIES + ) { + return true; + } const protectedKeys = new Set([...protectedQrSessions.values()].map(({ key }) => key)); let oldestKey: string | undefined; let oldestAt = Number.POSITIVE_INFINITY; for (const [key, session] of sessions) { - if (protectedKeys.has(key)) { + // Recovery leases are a separate bounded pool: admission may use the freed + // active slot, but cannot destroy a terminal reply whose delivery was lost. + if (protectedKeys.has(key) || recoveryKeys.has(key)) { continue; } if (session.lastUsedAt < oldestAt) { @@ -702,7 +716,7 @@ export const systemAgentHandlers: GatewayRequestHandlers = { undefined, errorShape( ErrorCodes.UNAVAILABLE, - "OpenClaw chat capacity is reserved for active QR operations; try again after one completes.", + "OpenClaw chat capacity is reserved for active or recoverable QR operations; try again after one completes.", { retryable: true }, ), ); diff --git a/src/system-agent/chat-engine.passive-poll.test.ts b/src/system-agent/chat-engine.passive-poll.test.ts index 6d6814b56d65..5ced07288fcb 100644 --- a/src/system-agent/chat-engine.passive-poll.test.ts +++ b/src/system-agent/chat-engine.passive-poll.test.ts @@ -63,6 +63,53 @@ describe("SystemAgentChatEngine passive QR polling", () => { } }); + it("lets cancellation interrupt an active passive QR observation", async () => { + const owner = createDeferred(); + const observationStarted = createDeferred(); + const releaseObservation = createDeferred(); + let abortObserved = false; + const engine = createQrEngine(async (_channel, prompter, _beforePersistentApply, signal) => { + try { + await prompter.qrCode?.({ + title: "Link a device", + message: "Scan this QR code and approve the device.", + text: QR_TEXT, + dismissed: owner.promise, + }); + } finally { + abortObserved = signal.aborted; + } + }); + const poll = vi.spyOn(ChatWizardHost.prototype, "pollStep").mockImplementation(async () => { + observationStarted.resolve(); + await releaseObservation.promise; + return { text: "Observed", configWritten: false }; + }); + + let cancellation: ReturnType | undefined; + try { + const prompt = await engine.handle("connect telegram"); + const stepId = expectDefined(prompt.step, "QR step").id; + const observation = engine.pollStep(stepId); + await observationStarted.promise; + await expect(observation).resolves.toMatchObject({ wizardSettling: true }); + + cancellation = engine.cancelWizard({ stepId }); + await vi.waitFor(() => expect(abortObserved).toBe(true)); + releaseObservation.resolve(); + await expect(cancellation).resolves.toMatchObject({ + text: expect.stringContaining("setup cancelled"), + }); + await expect(engine.pollStep(stepId)).rejects.toThrow("no longer active"); + } finally { + poll.mockRestore(); + releaseObservation.resolve(); + owner.resolve(); + await cancellation?.catch(() => undefined); + await engine.dispose(); + } + }); + it("replays a dropped follow-up until the wizard advances", async () => { const owner = createDeferred(); const releaseFollowUp = createDeferred(); diff --git a/src/system-agent/chat-engine.ts b/src/system-agent/chat-engine.ts index 85ccfc87ed0a..11b6fcc1a96e 100644 --- a/src/system-agent/chat-engine.ts +++ b/src/system-agent/chat-engine.ts @@ -151,6 +151,14 @@ export class SystemAgentChatEngine { ); } + /** A delivered terminal poll remains replayable until its bounded recovery lease expires. */ + hasRecoverableQrReply(): boolean { + this.pruneExpiredPollReplies(); + return [...this.retainedPollReplies.values()].some( + ({ reply, terminalHistoryRecorded }) => isTerminalPollReply(reply) && terminalHistoryRecorded, + ); + } + getPersistentApplySettlement(): Promise | null { return this.persistentApplySettlement; } @@ -326,10 +334,19 @@ export class SystemAgentChatEngine { async cancelWizard(cancel: SystemAgentWizardCancel): Promise { this.assertActive(); + // Only an in-flight passive observation may be interrupted out of queue: it can + // be waiting for the same runner that cancellation must release. Other turns + // retain their accepted ordering on the single execution queue. + const interruption = this.passivePollObservations.has(cancel.stepId) + ? this.wizard.requestCancellation(cancel) + : null; const turn = this.turnQueue.then(async () => { this.assertActive(); - const result = await this.router.answerWizard(this.wizard.cancel(cancel)); + const result = await this.router.answerWizard( + interruption ? interruption.finish() : this.wizard.cancel(cancel), + ); this.assertActive(); + this.retainedPollReplies.delete(cancel.stepId); return this.completeTurn({ text: result.text, action: "none" }, result.userHistoryText); }); this.turnQueue = turn.catch(() => undefined); diff --git a/src/system-agent/chat-wizard-host.ts b/src/system-agent/chat-wizard-host.ts index 829a1893dd62..d9780b0922d7 100644 --- a/src/system-agent/chat-wizard-host.ts +++ b/src/system-agent/chat-wizard-host.ts @@ -53,6 +53,10 @@ export type ChatWizardAnswerResult = ChatWizardResult & { userHistoryText: string; }; +export type ChatWizardCancellation = { + finish: () => Promise; +}; + export type ChatWizardHostDependencies = { runChannelSetupWizard?: ( channel: string, @@ -111,6 +115,13 @@ function loadHostedRuntime(): Promise { return (hostedRuntimePromise ??= import("./hosted-setup.runtime.js")); } +function renderWizardCancellation(label: string): ChatWizardResult { + return { + text: `${label[0]?.toUpperCase() ?? "S"}${label.slice(1)} setup cancelled. Nothing was changed beyond completed steps.`, + configWritten: false, + }; +} + export class SystemAgentWizardAnswerError extends Error {} export class ChatWizardHost { @@ -217,7 +228,7 @@ export class ChatWizardHost { }; } - async cancel(cancel: SystemAgentWizardCancel): Promise { + requestCancellation(cancel: SystemAgentWizardCancel): ChatWizardCancellation { const bridge = this.bridge; const step = bridge?.step; if (!bridge) { @@ -233,8 +244,20 @@ export class ChatWizardHost { if (!bridge.session.cancel()) { throw new SystemAgentWizardAnswerError("The hosted wizard cannot be cancelled right now."); } - await bridge.session.whenSettled(); - return { ...(await this.pump()), userHistoryText: "Cancel" }; + return { + finish: async () => { + await bridge.session.whenSettled(); + const result = await this.pump(); + return { + ...(result.text ? result : renderWizardCancellation(bridge.label)), + userHistoryText: "Cancel", + }; + }, + }; + } + + async cancel(cancel: SystemAgentWizardCancel): Promise { + return await this.requestCancellation(cancel).finish(); } /** Observe a QR-owned step without answering the dependency-owned prompt. */ @@ -689,10 +712,7 @@ export class ChatWizardHost { } } if (result.status === "cancelled") { - return { - text: `${label[0]?.toUpperCase() ?? "S"}${label.slice(1)} setup cancelled. Nothing was changed beyond completed steps.`, - configWritten: false, - }; + return renderWizardCancellation(label); } return { text: `${label[0]?.toUpperCase() ?? "S"}${label.slice(1)} setup stopped: ${result.error ?? "unknown error"}`,