diff --git a/src/gateway/server-close.test.ts b/src/gateway/server-close.test.ts index 766f6f02325e..f5f90e8ad77e 100644 --- a/src/gateway/server-close.test.ts +++ b/src/gateway/server-close.test.ts @@ -26,6 +26,7 @@ const mocks = vi.hoisted(() => ({ disposeAllBundleLspRuntimes: vi.fn(async () => undefined), drainRetainedEmbeddingProviders: vi.fn(async () => undefined), clearSessionSuspensionTimers: vi.fn(() => 0), + clearPluginBindingPendingRequests: vi.fn(), })); const WEBSOCKET_CLOSE_GRACE_MS = 1_000; const WEBSOCKET_CLOSE_FORCE_CONTINUE_MS = 250; @@ -85,6 +86,10 @@ vi.mock("../agents/session-suspension.js", () => ({ clearSessionSuspensionTimers: mocks.clearSessionSuspensionTimers, })); +vi.mock("../plugins/conversation-binding.js", () => ({ + clearPluginBindingPendingRequests: mocks.clearPluginBindingPendingRequests, +})); + vi.mock("../logging/subsystem.js", () => ({ createSubsystemLogger: vi.fn(() => ({ debug: vi.fn(), @@ -186,6 +191,7 @@ describe("createGatewayCloseHandler", () => { mocks.drainRetainedEmbeddingProviders.mockResolvedValue(undefined); mocks.clearSessionSuspensionTimers.mockReset(); mocks.clearSessionSuspensionTimers.mockReturnValue(0); + mocks.clearPluginBindingPendingRequests.mockClear(); }); afterEach(() => { @@ -239,6 +245,7 @@ describe("createGatewayCloseHandler", () => { await close({ reason: "test" }); + expect(mocks.clearPluginBindingPendingRequests).toHaveBeenCalledOnce(); expect(getActivePluginRegistry()).toBeNull(); }); diff --git a/src/gateway/server-close.ts b/src/gateway/server-close.ts index c5ceb4a11d06..cdf1c9d7ef36 100644 --- a/src/gateway/server-close.ts +++ b/src/gateway/server-close.ts @@ -12,6 +12,7 @@ import { createInternalHookEvent, triggerInternalHook } from "../hooks/internal- import type { HeartbeatRunner } from "../infra/heartbeat-runner.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; import { closePluginStateDatabase } from "../plugin-state/plugin-state-store.js"; +import { clearPluginBindingPendingRequests } from "../plugins/conversation-binding.js"; import { clearActivePluginRegistry } from "../plugins/runtime.js"; import type { PluginServicesHandle } from "../plugins/services.js"; import { @@ -1053,6 +1054,7 @@ export function createGatewayCloseHandler( warnings, }); } finally { + await shutdownStep("plugin-binding-requests", clearPluginBindingPendingRequests, warnings); await shutdownStep("plugin-host-registry", clearActivePluginRegistry, warnings); // Channel and plugin teardown still resolve account credentials. Keep the // active snapshot until every teardown owner is done, then always scrub it. diff --git a/src/plugins/conversation-binding.test-fixtures.ts b/src/plugins/conversation-binding.test-fixtures.ts index 996712e4e298..d8930719c537 100644 --- a/src/plugins/conversation-binding.test-fixtures.ts +++ b/src/plugins/conversation-binding.test-fixtures.ts @@ -1,6 +1,6 @@ /** Test-only reset for process-global plugin conversation binding state. */ import { executeSqliteQuerySync, getNodeSqliteKysely } from "../infra/kysely-sync.js"; -import { resolveGlobalMap, resolveGlobalSingleton } from "../shared/global-singleton.js"; +import { resolveGlobalSingleton } from "../shared/global-singleton.js"; import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js"; import { runOpenClawStateWriteTransaction } from "../state/openclaw-state-db.js"; @@ -13,8 +13,7 @@ type PluginBindingGlobalState = { approvalsSaveChain: Promise; }; -export function resetPluginConversationBindingStateForTest(): void { - resolveGlobalMap(Symbol.for("openclaw.pluginBindingPendingRequests")).clear(); +export function resetPluginConversationBindingCachesForTest(): void { const state = resolveGlobalSingleton( Symbol.for("openclaw.plugins.binding.global-state"), () => ({ @@ -62,5 +61,5 @@ export function seedPluginConversationBindingApprovalForTest(params: { ); }); // Seeded rows must become visible even if another test loaded the process cache first. - resetPluginConversationBindingStateForTest(); + resetPluginConversationBindingCachesForTest(); } diff --git a/src/plugins/conversation-binding.test.ts b/src/plugins/conversation-binding.test.ts index d92ea55b0a5c..d361289d5844 100644 --- a/src/plugins/conversation-binding.test.ts +++ b/src/plugins/conversation-binding.test.ts @@ -14,7 +14,7 @@ import { runOpenClawStateWriteTransaction, } from "../state/openclaw-state-db.js"; import { - resetPluginConversationBindingStateForTest, + resetPluginConversationBindingCachesForTest, seedPluginConversationBindingApprovalForTest, } from "./conversation-binding.test-fixtures.js"; import { createEmptyPluginRegistry } from "./registry-empty.js"; @@ -118,6 +118,7 @@ vi.mock("./runtime.js", async () => { }); let buildPluginBindingApprovalCustomId: typeof import("./conversation-binding.js").buildPluginBindingApprovalCustomId; +let clearPluginBindingPendingRequests: typeof import("./conversation-binding.js").clearPluginBindingPendingRequests; let bindPluginSessionConversation: typeof import("./session-conversation-binding.js").bindPluginSessionConversation; let detachPluginConversationBinding: typeof import("./conversation-binding.js").detachPluginConversationBinding; let getCurrentPluginConversationBinding: typeof import("./conversation-binding.js").getCurrentPluginConversationBinding; @@ -175,6 +176,7 @@ afterAll(() => { beforeAll(async () => { ({ buildPluginBindingApprovalCustomId, + clearPluginBindingPendingRequests, detachPluginConversationBinding, getCurrentPluginConversationBinding, parsePluginBindingApprovalCustomId, @@ -287,7 +289,7 @@ async function approveBindingRequest( async function importDuplicateConversationBindingModules() { const first = await importConversationBindingModule(`first-${Date.now()}`); const second = await importConversationBindingModule(`second-${Date.now()}`); - resetPluginConversationBindingStateForTest(); + resetPluginConversationBindingCachesForTest(); return { first, second }; } @@ -458,10 +460,11 @@ function insertPluginBindingApprovalRow(params: { describe("plugin conversation binding approvals", () => { beforeEach(() => { + clearPluginBindingPendingRequests(); process.env.OPENCLAW_STATE_DIR = tempRoot; clearPluginBindingApprovalRows(); sessionBindingState.reset(); - resetPluginConversationBindingStateForTest(); + resetPluginConversationBindingCachesForTest(); setActivePluginRegistry(createEmptyPluginRegistry()); unregisterSessionBindingAdapter({ channel: "discord", accountId: "default" }); unregisterSessionBindingAdapter({ channel: "discord", accountId: "work" }); @@ -727,33 +730,6 @@ describe("plugin conversation binding approvals", () => { ]); }); - it("shares pending bind approvals across duplicate module instances", async () => { - const { first, second } = await importDuplicateConversationBindingModules(); - const request = await requestPendingBinding( - createTelegramCodexBindRequest( - "-10099:topic:77", - "77", - "Bind this conversation to Codex thread abc.", - ), - first.requestPluginConversationBinding, - ); - - const approved = await approveBindingRequest( - request.approvalId, - "allow-once", - second.resolvePluginConversationBindingApproval, - ); - expect(approved.status).toBe("approved"); - if (approved.status !== "approved") { - throw new Error("expected approved bind result"); - } - expect(approved.binding.pluginId).toBe("codex"); - expect(approved.binding.pluginRoot).toBe("/plugins/codex-a"); - expect(approved.binding.conversationId).toBe("-10099:topic:77"); - - resetPluginConversationBindingStateForTest(); - }); - it("shares persistent approvals across duplicate module instances", async () => { const { first, second } = await importDuplicateConversationBindingModules(); const request = await requestPendingBinding( @@ -768,7 +744,7 @@ describe("plugin conversation binding approvals", () => { const approved = await approveBindingRequest( request.approvalId, "allow-always", - second.resolvePluginConversationBindingApproval, + first.resolvePluginConversationBindingApproval, ); expect(approved.status).toBe("approved"); if (approved.status !== "approved") { @@ -776,7 +752,7 @@ describe("plugin conversation binding approvals", () => { } expect(approved.decision).toBe("allow-always"); - const rebound = await first.requestPluginConversationBinding( + const rebound = await second.requestPluginConversationBinding( createTelegramCodexBindRequest( "-10099:topic:78", "78", @@ -786,10 +762,26 @@ describe("plugin conversation binding approvals", () => { expect(rebound.status).toBe("bound"); - resetPluginConversationBindingStateForTest(); + resetPluginConversationBindingCachesForTest(); clearPluginBindingApprovalRows(); }); + it("expires pending approvals when their Gateway lifecycle closes", async () => { + const request = await requestPendingBinding( + createTelegramCodexBindRequest( + "-10099:topic:79", + "79", + "Bind this conversation to Codex thread ghi.", + ), + ); + + clearPluginBindingPendingRequests(); + + await expect(approveBindingRequest(request.approvalId, "allow-once")).resolves.toEqual({ + status: "expired", + }); + }); + it("does not share persistent approvals across plugin roots even with the same plugin id", async () => { const request = await requestPluginConversationBinding( createCodexBindRequest({ diff --git a/src/plugins/conversation-binding.ts b/src/plugins/conversation-binding.ts index 923070b1df22..6875f6d4dda9 100644 --- a/src/plugins/conversation-binding.ts +++ b/src/plugins/conversation-binding.ts @@ -12,7 +12,7 @@ import { formatErrorMessage } from "../infra/errors.js"; import { executeSqliteQuerySync, getNodeSqliteKysely } from "../infra/kysely-sync.js"; import type { ConversationRef } from "../infra/outbound/session-binding-service.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; -import { resolveGlobalMap, resolveGlobalSingleton } from "../shared/global-singleton.js"; +import { resolveGlobalSingleton } from "../shared/global-singleton.js"; import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js"; import { openOpenClawStateDatabase, @@ -71,7 +71,6 @@ type PendingPluginBindingRequest = { pluginName?: string; pluginRoot: string; conversation: PluginBindingConversation; - requestedAt: number; requestedBySenderId?: string; summary?: string; detachHint?: string; @@ -115,11 +114,10 @@ type PluginBindingResolveResult = status: "expired"; }; -const PLUGIN_BINDING_PENDING_REQUESTS_KEY = Symbol.for("openclaw.pluginBindingPendingRequests"); +// Rent: approvals cross request roots through deep SDK paths; unified emission makes this ambient slot safe, and Gateway close bounds its lifetime. +const pendingRequests = new Map(); -const pendingRequests = resolveGlobalMap( - PLUGIN_BINDING_PENDING_REQUESTS_KEY, -); +export const clearPluginBindingPendingRequests = () => pendingRequests.clear(); type PluginBindingGlobalState = { fallbackNoticeBindingIds: Set; @@ -831,7 +829,6 @@ export async function requestPluginConversationBinding(params: { pluginName: params.pluginName, pluginRoot: params.pluginRoot, conversation, - requestedAt: Date.now(), requestedBySenderId: normalizeOptionalString(params.requestedBySenderId), summary: normalizeOptionalString(params.binding?.summary), detachHint: normalizeOptionalString(params.binding?.detachHint),