diff --git a/extensions/acpx/src/service.test.ts b/extensions/acpx/src/service.test.ts index 02276398f7e9..2ffe0d9b5a74 100644 --- a/extensions/acpx/src/service.test.ts +++ b/extensions/acpx/src/service.test.ts @@ -80,6 +80,20 @@ const { acpxRuntimeConstructorMock, createAgentRegistryMock, createFileSessionSt runTurn: vi.fn(async function* () {}), setConfigOption: vi.fn(async () => {}), setMode: vi.fn(async () => {}), + startTurn: vi.fn((input: { requestId: string }) => ({ + requestId: input.requestId, + promptStarted: Promise.resolve(), + events: (async function* () { + yield { + type: "text_delta" as const, + stream: "output" as const, + text: "progress", + }; + })(), + result: Promise.resolve({ status: "completed" as const, stopReason: "end_turn" }), + cancel: vi.fn(async () => {}), + closeStream: vi.fn(async () => {}), + })), __options: options, }; }), @@ -680,7 +694,7 @@ describe("createAcpxRuntimeService", () => { await service.stop?.(ctx); }); - it("registers the backend lazily without importing ACPX runtime when startup probe is disabled", async () => { + it("registers the backend lazily and forwards sessions and turns when startup probe is disabled", async () => { process.env.OPENCLAW_ACPX_RUNTIME_STARTUP_PROBE = "0"; delete process.env.OPENCLAW_SKIP_ACPX_RUNTIME_PROBE; const workspaceDir = testWorkspace.dir; @@ -695,6 +709,16 @@ describe("createAcpxRuntimeService", () => { } const backendRuntime = backend.runtime as { ensureSession(input: { agent: string; mode: string; sessionKey: string }): Promise; + startTurn(input: { + handle: { sessionKey: string; backend: string; runtimeSessionName: string }; + text: string; + mode: string; + requestId: string; + }): { + promptStarted: Promise; + events: AsyncIterable; + result: Promise; + }; }; expect(typeof backendRuntime.ensureSession).toBe("function"); expect(backend.healthy).toBeUndefined(); @@ -711,61 +735,6 @@ describe("createAcpxRuntimeService", () => { expect.objectContaining({ elicitationModes: ["form", "url"] }), ); expect(backend.healthy).toBeUndefined(); - - await service.stop?.(ctx); - }); - - it("forwards startTurn through the lazily resolved default runtime", async () => { - process.env.OPENCLAW_ACPX_RUNTIME_STARTUP_PROBE = "0"; - const workspaceDir = testWorkspace.dir; - const ctx = createServiceContext(workspaceDir); - const startTurn = vi.fn((input: { requestId: string }) => ({ - requestId: input.requestId, - promptStarted: Promise.resolve(), - events: (async function* () { - yield { - type: "text_delta" as const, - stream: "output" as const, - text: "legacy progress", - }; - })(), - result: Promise.resolve({ status: "completed" as const, stopReason: "end_turn" }), - cancel: vi.fn(async () => {}), - closeStream: vi.fn(async () => {}), - })); - acpxRuntimeConstructorMock.mockImplementationOnce(function AcpxRuntime(options: unknown) { - return { - ...createMockRuntime({ - startTurn, - }), - getCapabilities: vi.fn(async () => ({ controls: [] })), - getStatus: vi.fn(async () => ({ summary: "ready" })), - prepareFreshSession: vi.fn(async () => {}), - setConfigOption: vi.fn(async () => {}), - setMode: vi.fn(async () => {}), - __options: options, - }; - }); - const service = createAcpxRuntimeService(ctx); - - await service.start(ctx); - - const backend = getAcpRuntimeBackend("acpx"); - if (!backend) { - throw new Error("expected ACPX runtime backend"); - } - const backendRuntime = backend.runtime as { - startTurn(input: { - handle: { sessionKey: string; backend: string; runtimeSessionName: string }; - text: string; - mode: string; - requestId: string; - }): { - promptStarted: Promise; - events: AsyncIterable; - result: Promise; - }; - }; const turn = backendRuntime.startTurn({ handle: { sessionKey: "agent:codex:acp:test", @@ -790,10 +759,10 @@ describe("createAcpxRuntimeService", () => { { type: "text_delta", stream: "output", - text: "legacy progress", + text: "progress", }, ]); - expect(startTurn).toHaveBeenCalledOnce(); + expect(acpxRuntimeConstructorMock.mock.results[0]?.value.startTurn).toHaveBeenCalledOnce(); await service.stop?.(ctx); }); diff --git a/extensions/imessage/src/conversation-route.ts b/extensions/imessage/src/conversation-route.ts index d39cc256b6f2..5325aa6cef67 100644 --- a/extensions/imessage/src/conversation-route.ts +++ b/extensions/imessage/src/conversation-route.ts @@ -36,23 +36,20 @@ export function resolveIMessageConversationRoute(params: { return { route, bindingResolution: null }; } + const conversation = { + channel: "imessage", + accountId: params.accountId, + conversationId, + }; const configuredRoute = resolveConfiguredBindingRoute({ cfg: params.cfg, route, - conversation: { - channel: "imessage", - accountId: params.accountId, - conversationId, - }, + conversation, }); const runtimeRoute = resolveRuntimeConversationBindingRoute({ route: configuredRoute.route, - conversation: { - channel: "imessage", - accountId: params.accountId, - conversationId, - }, + conversation, }); if (runtimeRoute.bindingRecord && !runtimeRoute.boundSessionKey) { logVerbose(`imessage: plugin-bound conversation ${conversationId}`); diff --git a/extensions/imessage/src/monitor/inbound-processing.systemPrompt.test-support.ts b/extensions/imessage/src/monitor/inbound-processing.systemPrompt.test-support.ts index c22d9286ec2a..c2a1d1e53957 100644 --- a/extensions/imessage/src/monitor/inbound-processing.systemPrompt.test-support.ts +++ b/extensions/imessage/src/monitor/inbound-processing.systemPrompt.test-support.ts @@ -212,6 +212,7 @@ describe("buildIMessageInboundContext forwards GroupSystemPrompt", () => { lastRoutePolicy: "main", matchedBy: "default", }, + bindingResolution: null, bodyText: "hi", createdAt: undefined, replyContext: null, diff --git a/extensions/imessage/src/monitor/inbound-processing.test.ts b/extensions/imessage/src/monitor/inbound-processing.test.ts index da740e9684fa..d6f2c15a8e0b 100644 --- a/extensions/imessage/src/monitor/inbound-processing.test.ts +++ b/extensions/imessage/src/monitor/inbound-processing.test.ts @@ -954,6 +954,7 @@ describe("buildIMessageInboundContext MessageSid handling (rowid-leak regression const decision = { kind: "dispatch" as const, route: { accountId: "default", agentId: "lobster", sessionKey: "k", mainSessionKey: "mk" }, + bindingResolution: null, isGroup: false, sender: "+15555550123", senderId: "+15555550123", diff --git a/extensions/imessage/src/monitor/inbound-processing.ts b/extensions/imessage/src/monitor/inbound-processing.ts index b5b13f8fdbf1..984c847171c8 100644 --- a/extensions/imessage/src/monitor/inbound-processing.ts +++ b/extensions/imessage/src/monitor/inbound-processing.ts @@ -362,7 +362,7 @@ type IMessageInboundDispatchDecision = { sender: string; senderNormalized: string; route: ReturnType; - bindingResolution?: ConfiguredBindingRouteResult["bindingResolution"]; + bindingResolution: ConfiguredBindingRouteResult["bindingResolution"]; bodyText: string; agentBodyText?: string; createdAt?: number; diff --git a/extensions/imessage/src/test-plugin.test.ts b/extensions/imessage/src/test-plugin.test.ts index 70c9b3abf879..b713a0d975cf 100644 --- a/extensions/imessage/src/test-plugin.test.ts +++ b/extensions/imessage/src/test-plugin.test.ts @@ -15,7 +15,6 @@ import { } from "openclaw/plugin-sdk/channel-test-helpers"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { drainPendingDeliveries } from "openclaw/plugin-sdk/delivery-queue-runtime"; -import { getSessionBindingService } from "openclaw/plugin-sdk/session-binding-runtime"; import { withStateDirEnv } from "openclaw/plugin-sdk/test-env"; import { beforeEach, describe, expect, it, vi } from "vitest"; import { @@ -124,55 +123,6 @@ describe("imessagePlugin contracts", () => { ).resolves.toMatchObject({ kind: "user", to: "auto:AliceSmith" }); }); - it("keeps account-prefixed conversation bindings and opaque metadata across manager recreation", async () => { - await withStateDirEnv("openclaw-imessage-conversation-binding-", async () => { - const createManager = imessagePlugin.conversationBindings?.createManager; - expect(createManager).toBeTypeOf("function"); - if (!createManager) { - throw new Error("iMessage conversation binding manager is unavailable"); - } - - const cfg = { - channels: { imessage: { accounts: { default: {} } } }, - } satisfies OpenClawConfig; - const conversation = { - channel: "imessage", - accountId: "default", - conversationId: "+15555550999", - }; - const opaqueMetadata = { - label: "persisted iMessage binding", - opaque: { owner: "channel", flags: ["durable", "account-scoped"] }, - }; - let manager = await createManager({ cfg, accountId: "default" }); - - try { - const binding = await getSessionBindingService().bind({ - conversation, - targetKind: "session", - targetSessionKey: "agent:main:acp:imessage-durable", - placement: "current", - metadata: opaqueMetadata, - }); - - expect(binding.bindingId).toBe("default:+15555550999"); - expect(binding.metadata).toMatchObject(opaqueMetadata); - - await manager.stop(); - manager = await createManager({ cfg, accountId: "default" }); - - expect(getSessionBindingService().resolveByConversation(conversation)).toMatchObject({ - bindingId: binding.bindingId, - targetKind: "session", - targetSessionKey: binding.targetSessionKey, - metadata: opaqueMetadata, - }); - } finally { - await manager.stop(); - } - }); - }); - it("declares durable final delivery capabilities", () => { expect(imessagePlugin.outbound?.deliveryCapabilities?.durableFinal).toStrictEqual({ text: true, diff --git a/packages/acp-core/src/session.ts b/packages/acp-core/src/session.ts index d50e7e95e877..7bc4b9648daa 100644 --- a/packages/acp-core/src/session.ts +++ b/packages/acp-core/src/session.ts @@ -165,11 +165,7 @@ export function createInMemorySessionStore( touchSession(session, now()); }; - const clearActiveRun: AcpSessionStore["clearActiveRun"] = (sessionId, expectedRunId) => { - const session = sessions.get(sessionId); - if (!session || (expectedRunId !== undefined && session.activeRunId !== expectedRunId)) { - return; - } + const releaseActiveRun = (session: AcpSession) => { if (session.activeRunId) { runIdToSessionId.delete(session.activeRunId); } @@ -178,6 +174,13 @@ export function createInMemorySessionStore( touchSession(session, now()); }; + const clearActiveRun: AcpSessionStore["clearActiveRun"] = (sessionId, expectedRunId) => { + const session = sessions.get(sessionId); + if (session && (expectedRunId === undefined || session.activeRunId === expectedRunId)) { + releaseActiveRun(session); + } + }; + const cancelActiveRun: AcpSessionStore["cancelActiveRun"] = (sessionId, expectedRunId) => { const session = sessions.get(sessionId); if ( @@ -187,12 +190,7 @@ export function createInMemorySessionStore( return false; } session.abortController.abort(); - if (session.activeRunId) { - runIdToSessionId.delete(session.activeRunId); - } - session.abortController = null; - session.activeRunId = null; - touchSession(session, now()); + releaseActiveRun(session); return true; }; diff --git a/src/acp/control-plane/manager.turn-results.test.ts b/src/acp/control-plane/manager.turn-results.test.ts index c4391a68a044..8295f7f57497 100644 --- a/src/acp/control-plane/manager.turn-results.test.ts +++ b/src/acp/control-plane/manager.turn-results.test.ts @@ -5,6 +5,7 @@ import { requireTaskByRunId, withAcpManagerTaskStateDir, } from "../../../test/helpers/acp-manager-task-state.js"; +import { createDeferred } from "../../../test/helpers/promise.js"; import { AcpRuntimeError, AcpSessionManager, @@ -121,17 +122,14 @@ describe("AcpSessionManager turn results", () => { it("finishes submitted work when its lifecycle observer fails", async () => { const { runtimeState, sessionKey } = setupPromptStartedRuntime(); const transitions: string[] = []; - let finishTurn!: (result: { status: "completed" }) => void; - const result = new Promise<{ status: "completed" }>((resolve) => { - finishTurn = resolve; - }); + const result = createDeferred<{ status: "completed" }>(); const startTurn = vi.fn>((input) => ({ requestId: input.requestId, promptStarted: Promise.resolve().then(() => { transitions.push("prompt-started"); }), events: (async function* () {})(), - result, + result: result.promise, cancel: vi.fn(async () => {}), closeStream: vi.fn(async () => {}), })); @@ -149,7 +147,7 @@ describe("AcpSessionManager turn results", () => { transitions.push("observer-failed"); queueMicrotask(() => { transitions.push("turn-cleaned-up"); - finishTurn({ status: "completed" }); + result.resolve({ status: "completed" }); }); throw new Error("lifecycle observer unavailable"); }, @@ -165,10 +163,7 @@ describe("AcpSessionManager turn results", () => { "settles a %s terminal result when prompt readiness never resolves", async (terminalStatus) => { const { runtimeState, sessionKey } = setupPromptStartedRuntime(); - let resolveAbandonedReadiness!: () => void; - const promptStarted = new Promise((resolve) => { - resolveAbandonedReadiness = resolve; - }); + const promptStarted = createDeferred(); const result = terminalStatus === "completed" ? { status: "completed" as const } @@ -178,7 +173,7 @@ describe("AcpSessionManager turn results", () => { }; runtimeState.runtime.startTurn = vi.fn((input) => ({ requestId: input.requestId, - promptStarted, + promptStarted: promptStarted.promise, events: (async function* () {})(), result: Promise.resolve(result), cancel: vi.fn(async () => {}), @@ -216,12 +211,12 @@ describe("AcpSessionManager turn results", () => { } expect(onLifecycle).not.toHaveBeenCalled(); - resolveAbandonedReadiness(); + promptStarted.resolve(); await Promise.resolve(); expect(onLifecycle).not.toHaveBeenCalled(); expect(runtimeState.ensureSession).toHaveBeenCalledOnce(); } finally { - resolveAbandonedReadiness(); + promptStarted.resolve(); await outcome; } }, @@ -229,17 +224,14 @@ describe("AcpSessionManager turn results", () => { it("retries cleaned-up terminal failures without publishing abandoned prompt readiness", async () => { const { runtimeState, sessionKey } = setupPromptStartedRuntime(); - let resolveAbandonedReadiness!: () => void; - const abandonedReadiness = new Promise((resolve) => { - resolveAbandonedReadiness = resolve; - }); + const abandonedReadiness = createDeferred(); let attempt = 0; const startTurn = vi.fn>((input) => { attempt += 1; const firstAttempt = attempt === 1; return { requestId: input.requestId, - promptStarted: firstAttempt ? abandonedReadiness : Promise.resolve(), + promptStarted: firstAttempt ? abandonedReadiness.promise : Promise.resolve(), events: (async function* () {})(), result: Promise.resolve( firstAttempt @@ -278,11 +270,11 @@ describe("AcpSessionManager turn results", () => { expect(runtimeState.ensureSession).toHaveBeenCalledTimes(2); expect(onLifecycle).toHaveBeenCalledOnce(); - resolveAbandonedReadiness(); + abandonedReadiness.resolve(); await Promise.resolve(); expect(onLifecycle).toHaveBeenCalledOnce(); } finally { - resolveAbandonedReadiness(); + abandonedReadiness.resolve(); await turn.catch(() => {}); } }); diff --git a/src/acp/translator.cancel-scoping.test.ts b/src/acp/translator.cancel-scoping.test.ts index f7379e58f140..bf645d154787 100644 --- a/src/acp/translator.cancel-scoping.test.ts +++ b/src/acp/translator.cancel-scoping.test.ts @@ -9,6 +9,7 @@ import { createInMemorySessionStore } from "@openclaw/acp-core/session"; import { expectDefined } from "@openclaw/normalization-core"; import { describe, expect, it, type Mock, vi } from "vitest"; import type { EventFrame } from "../../packages/gateway-protocol/src/index.js"; +import { createDeferred } from "../../test/helpers/promise.js"; import type { GatewayClient } from "../gateway/client.js"; import { AcpGatewayAgent } from "./translator.js"; import { createAcpConnection, createAcpGateway } from "./translator.test-helpers.js"; @@ -101,7 +102,8 @@ function createHarness( function blockAcceptedPromptAbort(harness: Harness) { const firstSettlement = vi.fn(); - let releaseAbort: (() => void) | undefined; + const abortStarted = createDeferred(); + const abortReleased = createDeferred(); harness.requestSpy.mockImplementation(async (method, params) => { if (method === "chat.send") { const runId = expectDefined( @@ -113,9 +115,8 @@ function blockAcceptedPromptAbort(harness: Harness) { } if (method === "chat.abort") { expect(firstSettlement).not.toHaveBeenCalled(); - await new Promise((resolve) => { - releaseAbort = resolve; - }); + abortStarted.resolve(); + await abortReleased.promise; } return {}; }); @@ -123,14 +124,8 @@ function blockAcceptedPromptAbort(harness: Harness) { observeFirstSettlement(promise: Promise) { void promise.then(firstSettlement); }, - async waitForAbort() { - await vi.waitFor(() => { - expect(releaseAbort).toBeDefined(); - }); - }, - releaseAbort() { - expectDefined(releaseAbort, "blocked exact Gateway abort")(); - }, + waitForAbort: () => abortStarted.promise, + releaseAbort: abortReleased.resolve, }; } @@ -241,27 +236,25 @@ describe("acp translator cancel and run scoping", () => { }, ])( "does not submit a replacement closed by $closure while its prior abort is pending", - async ({ close, closure }) => { + async ({ close }) => { const sessionKey = "agent:main:shared"; const harness = createHarness([{ sessionId: "session-1", sessionKey }]); const first = await startPendingPrompt(harness, "session-1"); - let releaseAbort: (() => void) | undefined; + const abortStarted = createDeferred(); + const abortReleased = createDeferred(); harness.requestSpy.mockImplementationOnce(async (method: string) => { expect(method).toBe("chat.abort"); - await new Promise((resolve) => { - releaseAbort = resolve; - }); + abortStarted.resolve(); + await abortReleased.promise; return {}; }); const replacement = harness.agent.prompt(createPromptRequest("session-1")); - await vi.waitFor(() => { - expect(releaseAbort).toBeDefined(); - }); + await abortStarted.promise; const closed = close(harness); expect(harness.sentRunIds).toEqual([first.runId]); - expectDefined(releaseAbort, `${closure} owns the blocked prior abort`)(); + abortReleased.resolve(); await closed; await Promise.resolve(); @@ -306,7 +299,8 @@ describe("acp translator cancel and run scoping", () => { it("closes an admitted prompt when shutdown interrupts its blocked final snapshot", async () => { const sessionKey = "agent:main:shared"; const harness = createHarness([{ sessionId: "session-1", sessionKey }]); - let releaseSnapshot: (() => void) | undefined; + const snapshotStarted = createDeferred(); + const snapshot = createDeferred>(); harness.requestSpy.mockImplementation(async (method, params) => { if (method === "chat.send") { const runId = expectDefined( @@ -317,9 +311,8 @@ describe("acp translator cancel and run scoping", () => { return { runId, status: "started" }; } if (method === "sessions.list") { - return await new Promise>((resolve) => { - releaseSnapshot = () => resolve({ sessions: [] }); - }); + snapshotStarted.resolve(); + return await snapshot.promise; } return {}; }); @@ -327,9 +320,7 @@ describe("acp translator cancel and run scoping", () => { const terminalEvent = harness.agent.handleGatewayEvent( createChatEvent({ runId: pending.runId, sessionKey, seq: 1, state: "final" }), ); - await vi.waitFor(() => { - expect(releaseSnapshot).toBeDefined(); - }); + await snapshotStarted.promise; const shutdownSettled = vi.fn(); const shutdown = harness.agent.shutdown().then(shutdownSettled); @@ -339,7 +330,7 @@ describe("acp translator cancel and run scoping", () => { }); await expect(pending.promptPromise).resolves.toEqual({ stopReason: "cancelled" }); } finally { - expectDefined(releaseSnapshot, "blocked terminal session snapshot")(); + snapshot.resolve({ sessions: [] }); await terminalEvent; await shutdown; } @@ -349,23 +340,21 @@ describe("acp translator cancel and run scoping", () => { const sessionKey = "agent:main:shared"; const harness = createHarness([{ sessionId: "session-1", sessionKey }]); const first = await startPendingPrompt(harness, "session-1"); - let releaseAbort: (() => void) | undefined; + const abortStarted = createDeferred(); + const abortReleased = createDeferred(); harness.requestSpy.mockImplementationOnce(async (method: string) => { expect(method).toBe("chat.abort"); - await new Promise((resolve) => { - releaseAbort = resolve; - }); + abortStarted.resolve(); + await abortReleased.promise; return {}; }); const second = harness.agent.prompt(createPromptRequest("session-1")); - await vi.waitFor(() => { - expect(releaseAbort).toBeDefined(); - }); + await abortStarted.promise; const third = harness.agent.prompt(createPromptRequest("session-1")); const cancellation = harness.agent.cancel({ sessionId: "session-1" }); - expectDefined(releaseAbort, "blocked prior abort")(); + abortReleased.resolve(); await cancellation; await Promise.resolve(); @@ -426,7 +415,7 @@ describe("acp translator cancel and run scoping", () => { const harness = createHarness([{ sessionId: "session-1", sessionKey }], { provenanceMode: "meta", }); - let rejectFirstSend: ((error: Error) => void) | undefined; + const firstSend = createDeferred>(); harness.requestSpy.mockImplementation(async (method, params) => { if (method !== "chat.send") { return {}; @@ -436,9 +425,7 @@ describe("acp translator cancel and run scoping", () => { harness.sentRunIds.push(runId); } if (harness.sentRunIds.length === 1) { - return await new Promise>((_, reject) => { - rejectFirstSend = reject; - }); + return await firstSend.promise; } return await new Promise>(() => {}); }); @@ -446,10 +433,7 @@ describe("acp translator cancel and run scoping", () => { const replacement = await startPendingPrompt(harness, "session-1"); await expect(first.promptPromise).resolves.toEqual({ stopReason: "cancelled" }); - expectDefined( - rejectFirstSend, - "blocked first chat.send request", - )( + firstSend.reject( Object.assign(new Error("system provenance fields require admin scope"), { name: "GatewayClientRequestError", gatewayCode: "INVALID_REQUEST", @@ -468,13 +452,12 @@ describe("acp translator cancel and run scoping", () => { const sessionKey = "agent:main:shared"; const harness = createHarness([{ sessionId: "session-1", sessionKey }]); const first = await startPendingPrompt(harness, "session-1"); - let releaseDelivery: (() => void) | undefined; - harness.sessionUpdateSpy.mockImplementationOnce( - () => - new Promise((resolve) => { - releaseDelivery = resolve; - }), - ); + const deliveryStarted = createDeferred(); + const deliveryReleased = createDeferred(); + harness.sessionUpdateSpy.mockImplementationOnce(async () => { + deliveryStarted.resolve(); + await deliveryReleased.promise; + }); const staleFinal = harness.agent.handleGatewayEvent( createChatEvent({ @@ -485,12 +468,10 @@ describe("acp translator cancel and run scoping", () => { message: { content: [{ type: "text", text: "old response" }] }, }), ); - await vi.waitFor(() => { - expect(releaseDelivery).toBeDefined(); - }); + await deliveryStarted.promise; const replacement = await startPendingPrompt(harness, "session-1"); - expectDefined(releaseDelivery, "blocked session-update delivery")(); + deliveryReleased.resolve(); await staleFinal; expect(harness.sessionStore.getSession("session-1")?.activeRunId).toBe(replacement.runId); @@ -502,22 +483,20 @@ describe("acp translator cancel and run scoping", () => { const sessionKey = "agent:main:shared"; const harness = createHarness([{ sessionId: "session-1", sessionKey }]); const first = await startPendingPrompt(harness, "session-1"); - let releaseAbort: (() => void) | undefined; + const abortStarted = createDeferred(); + const abortReleased = createDeferred(); harness.requestSpy.mockImplementationOnce(async (method: string) => { expect(method).toBe("chat.abort"); - await new Promise((resolve) => { - releaseAbort = resolve; - }); + abortStarted.resolve(); + await abortReleased.promise; return {}; }); const cancellation = harness.agent.cancel({ sessionId: "session-1" } as CancelNotification); - await vi.waitFor(() => { - expect(releaseAbort).toBeDefined(); - }); + await abortStarted.promise; const replacement = await startPendingPrompt(harness, "session-1"); - expectDefined(releaseAbort, "blocked chat.abort request")(); + abortReleased.resolve(); await cancellation; await expect(first.promptPromise).resolves.toEqual({ stopReason: "cancelled" }); expect(harness.sessionStore.getSession("session-1")?.activeRunId).toBe(replacement.runId); @@ -649,13 +628,12 @@ describe("acp translator cancel and run scoping", () => { const sessionKey = "agent:main:shared"; const harness = createHarness([{ sessionId: "session-1", sessionKey }]); const first = await startPendingPrompt(harness, "session-1"); - let releaseThought: (() => void) | undefined; - harness.sessionUpdateSpy.mockImplementationOnce( - () => - new Promise((resolve) => { - releaseThought = resolve; - }), - ); + const thoughtStarted = createDeferred(); + const thoughtReleased = createDeferred(); + harness.sessionUpdateSpy.mockImplementationOnce(async () => { + thoughtStarted.resolve(); + await thoughtReleased.promise; + }); const staleSnapshot = harness.agent.handleGatewayEvent( createChatEvent({ @@ -671,12 +649,10 @@ describe("acp translator cancel and run scoping", () => { }, }), ); - await vi.waitFor(() => { - expect(releaseThought).toBeDefined(); - }); + await thoughtStarted.promise; const replacement = await startPendingPrompt(harness, "session-1"); - expectDefined(releaseThought, "blocked stale thought delivery")(); + thoughtReleased.resolve(); await staleSnapshot; const visibleChunks = harness.sessionUpdateSpy.mock.calls.filter( diff --git a/src/acp/translator.prompt-stream.ts b/src/acp/translator.prompt-stream.ts index 3c9c5f368edb..c33d70135820 100644 --- a/src/acp/translator.prompt-stream.ts +++ b/src/acp/translator.prompt-stream.ts @@ -197,9 +197,9 @@ export class AcpTranslatorPromptStream { settled: createDeferredCore(), }; this.pendingPromptAdmissions.set(params.sessionId, admission); - // Supersession keeps the predecessor's abort barrier; explicit closure alone releases it. - for (let previous = admission.previous; previous; previous = previous.previous) { - previous.closed = true; + // Each predecessor already closed its own predecessor; keep its abort barrier intact. + if (admission.previous) { + admission.previous.closed = true; } try { @@ -222,6 +222,8 @@ export class AcpTranslatorPromptStream { if (!this.ownsPromptAdmission(admission)) { return { stopReason: "cancelled" }; } + // Closure traversal no longer needs a settled predecessor or its retained ancestors. + admission.previous = undefined; } return await Promise.race([ this.submitPrompt(params, session), diff --git a/src/agents/subagents/spawn/acp-spawn-target.ts b/src/agents/subagents/spawn/acp-spawn-target.ts index eb9335021171..411ce24b3305 100644 --- a/src/agents/subagents/spawn/acp-spawn-target.ts +++ b/src/agents/subagents/spawn/acp-spawn-target.ts @@ -2,7 +2,7 @@ import { normalizeOptionalString } from "@openclaw/normalization-core/string-coe import { getAcpRuntimeBackend } from "../../../acp/runtime/registry.js"; import type { OpenClawConfig } from "../../../config/types.openclaw.js"; import { normalizeAgentIdStrict, normalizeOptionalAgentId } from "../../../routing/session-key.js"; -import { listAgentEntries } from "../../agent-scope-config.js"; +import { listAgentEntries, resolveAgentEntry } from "../../agent-scope-config.js"; import { listAgentIds } from "../../agent-scope.js"; type ResolvedAcpAgentTarget = { @@ -41,9 +41,7 @@ export function resolveTargetAcpAgentId(params: { } const requested = normalizedRequest?.value; if (requested) { - const configuredAgent = listAgentEntries(params.cfg).find( - (agent) => normalizeOptionalAgentId(agent.id) === requested, - ); + const configuredAgent = resolveAgentEntry(params.cfg, requested); if (configuredAgent?.runtime?.type === "acp") { return resolveAcpAgentTarget({ cfg: params.cfg, @@ -70,9 +68,7 @@ export function resolveTargetAcpAgentId(params: { const configuredDefault = normalizeOptionalAgentId(params.cfg.acp?.defaultAgent); if (configuredDefault) { - const configuredAgent = listAgentEntries(params.cfg).find( - (agent) => normalizeOptionalAgentId(agent.id) === configuredDefault, - ); + const configuredAgent = resolveAgentEntry(params.cfg, configuredDefault); return resolveAcpAgentTarget({ cfg: params.cfg, agentId: configuredDefault, diff --git a/src/channels/plugins/contracts/test-helpers/registry-session-binding.ts b/src/channels/plugins/contracts/test-helpers/registry-session-binding.ts index 7033614e98f1..005e658267cb 100644 --- a/src/channels/plugins/contracts/test-helpers/registry-session-binding.ts +++ b/src/channels/plugins/contracts/test-helpers/registry-session-binding.ts @@ -7,7 +7,6 @@ import fs from "node:fs"; import path from "node:path"; import { expect } from "vitest"; import type { OpenClawConfig } from "../../../../config/config.js"; -import { testing as currentConversationBindingTesting } from "../../../../infra/outbound/current-conversation-bindings.js"; import { getSessionBindingService, type SessionBindingRecord, @@ -20,6 +19,7 @@ import { resetPluginStateStoreForTests, } from "../../../../plugin-sdk/plugin-state-test-runtime.js"; import { setActivePluginRegistry } from "../../../../plugins/runtime.js"; +import { closeOpenClawStateDatabaseForTest } from "../../../../state/openclaw-state-db.js"; import { loadBundledPluginFacade } from "../../../../test-utils/bundled-plugin-public-surface.js"; import { createTestRegistry } from "../../../../test-utils/channel-plugins.js"; import { getChannelPlugin } from "../../registry.js"; @@ -83,6 +83,7 @@ function expectResolvedSessionBinding(params: { conversationId: string; parentConversationId?: string; targetSessionKey: string; + metadata?: Record; }) { expect( getSessionBindingService().resolveByConversation({ @@ -93,6 +94,7 @@ function expectResolvedSessionBinding(params: { }), )?.toMatchObject({ targetSessionKey: params.targetSessionKey, + ...(params.metadata ? { metadata: params.metadata } : {}), }); } @@ -309,12 +311,13 @@ type SessionBindingContractFixture = { expectedBindingId?: string; targetKind: SessionBindingRecord["targetKind"]; label: string; + metadata?: Record; placements: SessionBindingCapabilities["placements"]; preload: () => Promise; beforeEach: () => Promise; ensureManager: () => Promise; stopManager?: () => Promise; - resetProcessMemory?: () => Promise; + restartBindingManager?: () => Promise; }; function createSessionBindingContractEntry( @@ -352,20 +355,24 @@ function createSessionBindingContractEntry( targetKind: fixture.targetKind, conversation, placement: "current", - metadata: { agentId: fixture.id, label: fixture.label }, + metadata: { agentId: fixture.id, label: fixture.label, ...fixture.metadata }, }); if (fixture.expectedBindingId) { expect(binding.bindingId).toBe(fixture.expectedBindingId); } + if (fixture.metadata) { + expect(binding.metadata).toMatchObject(fixture.metadata); + } expectResolvedSessionBinding({ ...conversation, targetSessionKey: fixture.targetSessionKey, }); - if (fixture.resetProcessMemory) { - await fixture.resetProcessMemory(); + if (fixture.restartBindingManager) { + await fixture.restartBindingManager(); expectResolvedSessionBinding({ ...conversation, targetSessionKey: fixture.targetSessionKey, + metadata: fixture.metadata, }); } return binding; @@ -430,14 +437,15 @@ const sessionBindingContractEntries = { expectedBindingId: "default:+15555550124", targetKind: "session", label: "imessage-main", + metadata: { opaque: { ownerEpoch: 7, capabilities: ["approve", "resume"] } }, placements: ["current"], preload: getIMessageContractApi, beforeEach: prepareIMessageSessionBindingContract, ensureManager: ensureIMessageSessionBindingManager, stopManager: stopIMessageSessionBindingManager, - resetProcessMemory: async () => { + restartBindingManager: async () => { await stopIMessageSessionBindingManager(); - currentConversationBindingTesting.resetCurrentConversationBindingsForTests(); + closeOpenClawStateDatabaseForTest(); await ensureIMessageSessionBindingManager(); }, }), diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts index 3397cd44ba96..de268543c07f 100644 --- a/src/gateway/server-methods/chat.ts +++ b/src/gateway/server-methods/chat.ts @@ -72,26 +72,17 @@ export const chatHandlers: GatewayRequestHandlers = { respond(false, undefined, requestedAgent.error); return; } - const selectedAgent = validateChatSelectedAgent({ - cfg, - requestedSessionKey: params.sessionKey, - explicitAgentId: agentIdOverride, - }); - if (!selectedAgent.ok) { - respond(false, undefined, errorShape(ErrorCodes.INVALID_REQUEST, selectedAgent.error)); - return; - } const sessionAgentId = resolveSessionAgentId({ sessionKey: params.sessionKey, config: cfg, - agentId: selectedAgent.agentId, + agentId: requestedAgent.agentId, }); // Session entry carries per-session model overrides; utility routing must // derive its small-model default from the provider this session actually // uses, not the agent's configured default. const { cfg: sessionCfg, entry } = loadGatewaySessionEntryReadOnly( params.sessionKey, - selectedAgent.agentId ? { agentId: selectedAgent.agentId } : undefined, + requestedAgent.agentId ? { agentId: requestedAgent.agentId } : undefined, ); const sessionModel = resolveSessionModelRef(sessionCfg, entry, sessionAgentId); // Title generation pulls in the simple-completion runtime; load it lazily diff --git a/src/infra/outbound/account-scoped-conversation-bindings.test.ts b/src/infra/outbound/account-scoped-conversation-bindings.test.ts index f8971dcb6085..21e151c03713 100644 --- a/src/infra/outbound/account-scoped-conversation-bindings.test.ts +++ b/src/infra/outbound/account-scoped-conversation-bindings.test.ts @@ -63,16 +63,12 @@ describe("account-scoped conversation binding expiry", () => { testStateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-account-bindings-")); process.env.OPENCLAW_STATE_DIR = testStateDir; resetAccountScopedConversationBindingsForTests({ stateKey }); - currentConversationBindingTesting.resetCurrentConversationBindingsForTests({ - deletePersistedFile: true, - }); + currentConversationBindingTesting.clearPersistedCurrentConversationBindingsForTests(); }); afterEach(async () => { resetAccountScopedConversationBindingsForTests({ stateKey }); - currentConversationBindingTesting.resetCurrentConversationBindingsForTests({ - deletePersistedFile: true, - }); + currentConversationBindingTesting.clearPersistedCurrentConversationBindingsForTests(); closeOpenClawStateDatabaseForTest(); if (previousStateDir === undefined) { delete process.env.OPENCLAW_STATE_DIR; @@ -97,7 +93,6 @@ describe("account-scoped conversation binding expiry", () => { ); manager.stop(); - currentConversationBindingTesting.resetCurrentConversationBindingsForTests(); closeOpenClawStateDatabaseForTest(); const restarted = createManager(); @@ -136,7 +131,6 @@ describe("account-scoped conversation binding expiry", () => { expect(manager.getByConversationId(conversation.conversationId)?.targetKind).toBe("acp"); manager.stop(); - currentConversationBindingTesting.resetCurrentConversationBindingsForTests(); closeOpenClawStateDatabaseForTest(); createManager(); diff --git a/src/infra/outbound/current-conversation-bindings.test.ts b/src/infra/outbound/current-conversation-bindings.test.ts index 2e8ff914da9e..2159564ba060 100644 --- a/src/infra/outbound/current-conversation-bindings.test.ts +++ b/src/infra/outbound/current-conversation-bindings.test.ts @@ -215,16 +215,12 @@ describe("generic current-conversation bindings", () => { testStateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-current-bindings-")); process.env.OPENCLAW_STATE_DIR = testStateDir; setMinimalCurrentConversationRegistry(); - testing.resetCurrentConversationBindingsForTests({ - deletePersistedFile: true, - }); + testing.clearPersistedCurrentConversationBindingsForTests(); }); afterEach(async () => { vi.useRealTimers(); - testing.resetCurrentConversationBindingsForTests({ - deletePersistedFile: true, - }); + testing.clearPersistedCurrentConversationBindingsForTests(); closeOpenClawStateDatabaseForTest(); if (previousStateDir == null) { delete process.env.OPENCLAW_STATE_DIR; @@ -303,7 +299,7 @@ describe("generic current-conversation bindings", () => { ); }); - it("reloads persisted bindings after the in-memory cache is cleared", async () => { + it("preserves persisted bindings after the state database reopens", async () => { const bound = await bindGenericCurrentConversation({ targetSessionKey: "agent:codex:acp:workspace-dm", targetKind: "session", @@ -322,7 +318,7 @@ describe("generic current-conversation bindings", () => { targetSessionKey: "agent:codex:acp:workspace-dm", }); - testing.resetCurrentConversationBindingsForTests(); + closeOpenClawStateDatabaseForTest(); const resolved = resolveGenericCurrentConversationBinding({ channel: "workspace", @@ -395,7 +391,6 @@ describe("generic current-conversation bindings", () => { opaque: { nested: true }, }); - testing.resetCurrentConversationBindingsForTests(); closeOpenClawStateDatabaseForTest(); expect(resolveWorkspaceConversation("user:inserted")?.targetSessionKey).toBe( "agent:codex:acp:inserted-target", @@ -650,7 +645,6 @@ describe("generic current-conversation bindings", () => { bindingId: "generic:forum\u241fdefault\u241f\u241f6098642967", }); - testing.resetCurrentConversationBindingsForTests(); expect( resolveGenericCurrentConversationBinding({ channel: "forum", @@ -754,8 +748,6 @@ describe("generic current-conversation bindings", () => { reason: "test cleanup", }); - testing.resetCurrentConversationBindingsForTests(); - expect( resolveGenericCurrentConversationBinding({ channel: "googlechat", @@ -813,7 +805,7 @@ describe("generic current-conversation bindings", () => { ).toBeNull(); }); - it("persists touched activity across reloads", async () => { + it("persists touched activity after the state database reopens", async () => { const bound = await bindGenericCurrentConversation({ targetSessionKey: "agent:codex:acp:workspace-dm", targetKind: "session", @@ -834,7 +826,7 @@ describe("generic current-conversation bindings", () => { 1_234_567_890, ); - testing.resetCurrentConversationBindingsForTests(); + closeOpenClawStateDatabaseForTest(); expectBindingMetadata( resolveGenericCurrentConversationBinding({ @@ -850,7 +842,7 @@ describe("generic current-conversation bindings", () => { }); describe("SQLite write failures", () => { - it("keeps a replacement bind out of memory and disk", async () => { + it("keeps the committed binding when its replacement write fails", async () => { await bindWorkspaceConversation("user:U1", { targetSessionKey: "agent:codex:acp:session-a", }); @@ -866,14 +858,13 @@ describe("generic current-conversation bindings", () => { expect(resolveWorkspaceConversation("user:U1")?.targetSessionKey).toBe( "agent:codex:acp:session-a", ); - testing.resetCurrentConversationBindingsForTests(); closeOpenClawStateDatabaseForTest(); expect(resolveWorkspaceConversation("user:U1")?.targetSessionKey).toBe( "agent:codex:acp:session-a", ); }); - it("keeps a failed touch out of memory and disk", async () => { + it("keeps committed activity unchanged when a touch write fails", async () => { const bound = expectSessionBinding( await bindWorkspaceConversation("user:U1", { metadata: { label: "workspace-dm" } }), ); @@ -888,10 +879,6 @@ describe("generic current-conversation bindings", () => { expect(resolveWorkspaceConversation("user:U1")?.metadata?.lastActivityAt).toBe( originalActivity, ); - testing.resetCurrentConversationBindingsForTests(); - expect(resolveWorkspaceConversation("user:U1")?.metadata?.lastActivityAt).toBe( - originalActivity, - ); }); it("keeps a binding when unbind by id fails", async () => { @@ -907,8 +894,6 @@ describe("generic current-conversation bindings", () => { ).rejects.toThrow(); expect(resolveWorkspaceConversation("user:U1")).not.toBeNull(); - testing.resetCurrentConversationBindingsForTests(); - expect(resolveWorkspaceConversation("user:U1")).not.toBeNull(); }); it("keeps every matching binding when unbind by session fails", async () => { @@ -926,8 +911,6 @@ describe("generic current-conversation bindings", () => { ).rejects.toThrow(); expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2); - testing.resetCurrentConversationBindingsForTests(); - expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2); }); it("keeps an expired binding when prune-on-resolve fails", async () => { @@ -942,8 +925,6 @@ describe("generic current-conversation bindings", () => { vi.setSystemTime(new Date(1_000_500)); expect(resolveWorkspaceConversation("user:U1")).not.toBeNull(); - testing.resetCurrentConversationBindingsForTests(); - expect(resolveWorkspaceConversation("user:U1")).not.toBeNull(); }); it("keeps expired list entries when their cleanup write fails", async () => { @@ -962,8 +943,6 @@ describe("generic current-conversation bindings", () => { vi.setSystemTime(new Date(1_000_500)); expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2); - testing.resetCurrentConversationBindingsForTests(); - expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2); }); it("does not partially prune an unbind-by-session batch", async () => { @@ -985,13 +964,10 @@ describe("generic current-conversation bindings", () => { vi.setSystemTime(new Date(1_000_500)); expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2); - testing.resetCurrentConversationBindingsForTests(); - expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2); }); it("reads an unexpired binding without requiring a SQLite cleanup write", async () => { await bindWorkspaceConversation("user:U1"); - testing.resetCurrentConversationBindingsForTests(); await expect( withReadOnlyStateDatabase(() => resolveWorkspaceConversation("user:U1")), diff --git a/src/infra/outbound/current-conversation-bindings.ts b/src/infra/outbound/current-conversation-bindings.ts index 70b3dbc6796e..c1130160dd15 100644 --- a/src/infra/outbound/current-conversation-bindings.ts +++ b/src/infra/outbound/current-conversation-bindings.ts @@ -48,13 +48,9 @@ type CurrentConversationBindingRow = { }; function buildConversationKey(ref: ConversationRef): string { - const normalized = normalizeConversationRef(ref); - return [ - normalized.channel, - normalized.accountId, - normalized.parentConversationId ?? "", - normalized.conversationId, - ].join("\u241f"); + return [ref.channel, ref.accountId, ref.parentConversationId ?? "", ref.conversationId].join( + "\u241f", + ); } function buildBindingId(ref: ConversationRef): string { @@ -112,11 +108,10 @@ function targetAgentIdForSessionKey(targetSessionKey: string): string { function readCurrentConversationBindingRow( db: DatabaseSync, - ref: ConversationRef, + conversation: ConversationRef, + bindingKey: string, ): CurrentConversationBindingRow | undefined { const bindingDb = getNodeSqliteKysely(db); - const conversation = normalizeConversationRef(ref); - const bindingKey = buildConversationKey(conversation); const exact = executeSqliteQueryTakeFirstSync( db, bindingDb @@ -145,10 +140,13 @@ function readCurrentConversationBindingRow( }); } -function currentConversationBindingRow(record: SessionBindingRecord) { - const conversation = normalizeConversationRef(record.conversation); +function currentConversationBindingRow( + record: SessionBindingRecord, + conversation: ConversationRef, + bindingKey: string, +) { return { - binding_key: buildConversationKey(conversation), + binding_key: bindingKey, binding_id: record.bindingId, target_agent_id: targetAgentIdForSessionKey(record.targetSessionKey), target_session_id: null, @@ -182,8 +180,9 @@ export function updateCurrentConversationBindingRecord( update: (current: SessionBindingRecord | null) => SessionBindingRecord | null, ): { previous: SessionBindingRecord | null; current: SessionBindingRecord | null } { const conversation = normalizeConversationRef(ref); + const bindingKey = buildConversationKey(conversation); return runOpenClawStateWriteTransaction(({ db }) => { - const existingRow = readCurrentConversationBindingRow(db, conversation); + const existingRow = readCurrentConversationBindingRow(db, conversation, bindingKey); const existing = existingRow ? (bindingRowsToRecords([existingRow])[0] ?? null) : null; const previous = existing && !isBindingExpired(existing) ? existing : null; const current = update(previous); @@ -194,14 +193,13 @@ export function updateCurrentConversationBindingRecord( return { previous, current: null }; } - const bindingKey = buildConversationKey(conversation); - if (buildConversationKey(current.conversation) !== bindingKey) { + if (buildConversationKey(normalizeConversationRef(current.conversation)) !== bindingKey) { throw new Error("Current conversation binding update changed its conversation owner"); } if (existingRow && existingRow.binding_key !== bindingKey) { deleteCurrentConversationBindingRow(db, existingRow.binding_key); } - const row = currentConversationBindingRow(current); + const row = currentConversationBindingRow(current, conversation, bindingKey); const bindingDb = getNodeSqliteKysely(db); executeSqliteQuerySync( db, @@ -219,7 +217,9 @@ export function resolveCurrentConversationBindingRecord( ref: ConversationRef, ): SessionBindingRecord | null { const { db } = openOpenClawStateDatabase(); - const row = readCurrentConversationBindingRow(db, ref); + const conversation = normalizeConversationRef(ref); + const bindingKey = buildConversationKey(conversation); + const row = readCurrentConversationBindingRow(db, conversation, bindingKey); if (!row) { return null; } @@ -228,14 +228,14 @@ export function resolveCurrentConversationBindingRecord( return null; } if (isBindingExpired(record)) { - return updateCurrentConversationBindingRecord(ref, (current) => current).current; + return updateCurrentConversationBindingRecord(conversation, (current) => current).current; } if ( row.binding_key !== buildConversationKey(record.conversation) || row.binding_id !== record.bindingId || row.target_session_key !== record.targetSessionKey ) { - return updateCurrentConversationBindingRecord(ref, (current) => current).current; + return updateCurrentConversationBindingRecord(conversation, (current) => current).current; } return record; } @@ -262,6 +262,9 @@ function listCurrentConversationBindingRowsBySession( query = query .where("channel", "=", normalized.channel) .where("account_id", "=", normalized.accountId); + } else { + // Generic lookups must not load or decode rows belonging to account-owned adapters. + query = query.where("binding_id", "like", `${CURRENT_BINDINGS_ID_PREFIX}%`); } return executeSqliteQuerySync(db, query.orderBy("binding_id", "asc")).rows; } @@ -296,16 +299,19 @@ export function listCurrentConversationBindingRecordsBySession( }); } -/** Deletes one account's exact current session rows without disturbing sibling owners. */ +/** Deletes exact account-owned or generic session rows without disturbing sibling owners. */ export function deleteCurrentConversationBindingRecordsBySession( targetSessionKey: string, - scope: CurrentConversationBindingScope, + scope?: CurrentConversationBindingScope, ): SessionBindingRecord[] { return runOpenClawStateWriteTransaction(({ db }) => { const rows = listCurrentConversationBindingRowsBySession(db, targetSessionKey, scope); const removed: SessionBindingRecord[] = []; for (const row of rows) { const record = bindingRowsToRecords([row])[0]; + if (!scope && !record?.bindingId.startsWith(CURRENT_BINDINGS_ID_PREFIX)) { + continue; + } deleteCurrentConversationBindingRow(db, row.binding_key); if (record && !isBindingExpired(record)) { removed.push(record); @@ -471,11 +477,11 @@ export function listGenericCurrentConversationBindingsBySession( /** Persists last-activity metadata for an existing generic current-conversation binding. */ export function touchGenericCurrentConversationBinding(bindingId: string, at = Date.now()): void { - const record = resolveGenericCurrentConversationBindingById(bindingId); - if (!record) { + const conversation = bindingRefFromId(bindingId); + if (!conversation || !supportsGenericCurrentConversationBinding(conversation)) { return; } - updateCurrentConversationBindingRecord(record.conversation, (current) => + updateCurrentConversationBindingRecord(conversation, (current) => current?.bindingId === bindingId ? { ...current, @@ -488,53 +494,17 @@ export function touchGenericCurrentConversationBinding(bindingId: string, at = D ); } -function resolveGenericCurrentConversationBindingById( - bindingId: string, -): SessionBindingRecord | undefined { - const bindingRef = bindingRefFromId(bindingId); - if (!bindingRef || !supportsGenericCurrentConversationBinding(bindingRef)) { - return undefined; - } - const { db } = openOpenClawStateDatabase(); - const row = readCurrentConversationBindingRow(db, bindingRef); - const record = row ? bindingRowsToRecords([row])[0] : undefined; - return record?.bindingId === bindingId ? record : undefined; -} - function unbindCurrentConversationBindingById(bindingId: string): SessionBindingRecord[] { - const record = resolveGenericCurrentConversationBindingById(bindingId); - if (!record) { + const conversation = bindingRefFromId(bindingId); + if (!conversation || !supportsGenericCurrentConversationBinding(conversation)) { return []; } - const { previous, current } = updateCurrentConversationBindingRecord( - record.conversation, - (latest) => (latest?.bindingId === bindingId ? null : latest), + const { previous, current } = updateCurrentConversationBindingRecord(conversation, (latest) => + latest?.bindingId === bindingId ? null : latest, ); return previous && !current ? [previous] : []; } -function unbindGenericCurrentConversationBindingsBySession( - targetSessionKey: string, -): SessionBindingRecord[] { - return runOpenClawStateWriteTransaction(({ db }) => { - const rows = listCurrentConversationBindingRowsBySession(db, targetSessionKey); - const removed: SessionBindingRecord[] = []; - for (const row of rows) { - const record = bindingRowsToRecords([row])[0]; - if (!record?.bindingId.startsWith(CURRENT_BINDINGS_ID_PREFIX)) { - continue; - } - if (isBindingExpired(record)) { - deleteCurrentConversationBindingRow(db, row.binding_key); - continue; - } - deleteCurrentConversationBindingRow(db, row.binding_key); - removed.push(record); - } - return removed; - }); -} - /** Removes generic current-conversation bindings by binding id or target session key. */ export async function unbindGenericCurrentConversationBindings( input: SessionBindingUnbindInput, @@ -545,23 +515,15 @@ export async function unbindGenericCurrentConversationBindings( } const normalizedTargetSessionKey = input.targetSessionKey?.trim(); return normalizedTargetSessionKey - ? unbindGenericCurrentConversationBindingsBySession(normalizedTargetSessionKey) + ? deleteCurrentConversationBindingRecordsBySession(normalizedTargetSessionKey) : []; } export const testing = { - resetCurrentConversationBindingsForTests(params?: { - deletePersistedFile?: boolean; - env?: NodeJS.ProcessEnv; - }) { - if (params?.deletePersistedFile) { - runOpenClawStateWriteTransaction( - ({ db }) => { - const bindingDb = getNodeSqliteKysely(db); - executeSqliteQuerySync(db, bindingDb.deleteFrom("current_conversation_bindings")); - }, - params.env ? { env: params.env } : undefined, - ); - } + clearPersistedCurrentConversationBindingsForTests() { + runOpenClawStateWriteTransaction(({ db }) => { + const bindingDb = getNodeSqliteKysely(db); + executeSqliteQuerySync(db, bindingDb.deleteFrom("current_conversation_bindings")); + }); }, }; diff --git a/src/infra/outbound/session-binding-service.ts b/src/infra/outbound/session-binding-service.ts index 5bcede0b3dd2..3fe7586d7d17 100644 --- a/src/infra/outbound/session-binding-service.ts +++ b/src/infra/outbound/session-binding-service.ts @@ -344,9 +344,7 @@ export function getSessionBindingService(): SessionBindingService { export const testing = { resetSessionBindingAdaptersForTests() { ADAPTERS_BY_CHANNEL_ACCOUNT.clear(); - genericCurrentConversationBindingTesting.resetCurrentConversationBindingsForTests({ - deletePersistedFile: true, - }); + genericCurrentConversationBindingTesting.clearPersistedCurrentConversationBindingsForTests(); }, getRegisteredAdapterKeys() { return [...ADAPTERS_BY_CHANNEL_ACCOUNT.keys()];