ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-23 10:55:31 -06:00
Code Issues Packages Projects Releases Wiki Activity

refactor(plugins): root-owned conversation-binding pending requests (#118075)

Browse Source
This commit is contained in:
Peter Steinberger
2026-08-02 11:10:45 -07:00
committed by GitHub
parent 079d26c92c
commit 023ea4615d
5 changed files with 41 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/gateway/server-close.test.ts
+7
View File
@@ -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();
});
src/gateway/server-close.ts
+2
View File
@@ -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.
src/plugins/conversation-binding.test-fixtures.ts
+3 -4
View File
@@ -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<void>;
};
export function resetPluginConversationBindingStateForTest(): void {
resolveGlobalMap(Symbol.for("openclaw.pluginBindingPendingRequests")).clear();
export function resetPluginConversationBindingCachesForTest(): void {
const state = resolveGlobalSingleton<PluginBindingGlobalState>(
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();
}
src/plugins/conversation-binding.test.ts
+25 -33
View File
@@ -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({
src/plugins/conversation-binding.ts
+4 -7
View File
@@ -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<string, PendingPluginBindingRequest>();
const pendingRequests = resolveGlobalMap<string, PendingPluginBindingRequest>(
PLUGIN_BINDING_PENDING_REQUESTS_KEY,
);
export const clearPluginBindingPendingRequests = () => pendingRequests.clear();
type PluginBindingGlobalState = {
fallbackNoticeBindingIds: Set<string>;
@@ -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),