From ed6ee79c56f6dfe98fe755ad814196a4d88e4e79 Mon Sep 17 00:00:00 2001 From: "Jason (Json)" <263060202+fuller-stack-dev@users.noreply.github.com> Date: Tue, 4 Aug 2026 12:21:40 -0600 Subject: [PATCH] fix(gateway): steer matching embedded turns (#119287) --- .../dispatch-from-config.base.test-utils.ts | 134 ++++++++++++++++++ .../reply/dispatch-from-config.lifecycle.ts | 18 ++- 2 files changed, 151 insertions(+), 1 deletion(-) diff --git a/src/auto-reply/reply/dispatch-from-config.base.test-utils.ts b/src/auto-reply/reply/dispatch-from-config.base.test-utils.ts index dbf83a844356..c0d600c777e6 100644 --- a/src/auto-reply/reply/dispatch-from-config.base.test-utils.ts +++ b/src/auto-reply/reply/dispatch-from-config.base.test-utils.ts @@ -2,6 +2,10 @@ import { AsyncResource } from "node:async_hooks"; import { expectDefined } from "@openclaw/normalization-core"; import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { + clearActiveEmbeddedRun, + setActiveEmbeddedRun, +} from "../../agents/embedded-agent-runner/runs.js"; import type { OpenClawConfig } from "../../config/config.js"; import { setActivePluginRegistry } from "../../plugins/runtime.js"; import { @@ -1222,6 +1226,136 @@ describe("dispatchReplyFromConfig", () => { } }); + it("lets Gateway-owned turns reach queue resolution while only an embedded run is active", async () => { + setNoAbort(); + const sessionKey = "agent:main:main"; + const sessionId = "active-embedded-session"; + const activeHandle = { + queueMessage: vi.fn(async () => {}), + isStreaming: () => true, + isCompacting: () => false, + abort: vi.fn(), + }; + setActiveEmbeddedRun(sessionId, activeHandle, sessionKey); + sessionStoreMocks.currentEntry = { sessionId, updatedAt: Date.now() }; + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => { + expect(replyRunRegistry.get(sessionKey)).toBeUndefined(); + return undefined; + }); + + try { + const result = await dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "webchat", + Surface: "webchat", + SessionKey: sessionKey, + BodyForAgent: "steer this active turn", + }), + cfg: emptyConfig, + dispatcher, + replyOptions: { + turnAdoptionLifecycle: { + onAdopted: async () => {}, + onDeferred: vi.fn(), + onSettled: vi.fn(), + }, + }, + replyResolver, + }); + + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + noVisibleReplyFallbackDelivered: true, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(replyRunRegistry.get(sessionKey)).toBeUndefined(); + } finally { + clearActiveEmbeddedRun(sessionId, activeHandle, sessionKey); + } + }); + + it("keeps non-Gateway turns on normal admission while only an embedded run is active", async () => { + setNoAbort(); + const sessionKey = "agent:main:telegram:direct:embedded-only"; + const sessionId = "active-non-gateway-embedded-session"; + const activeHandle = { + queueMessage: vi.fn(async () => {}), + isStreaming: () => true, + isCompacting: () => false, + abort: vi.fn(), + }; + setActiveEmbeddedRun(sessionId, activeHandle, sessionKey); + sessionStoreMocks.currentEntry = { sessionId, updatedAt: Date.now() }; + const replyResolver = vi.fn(async () => { + expect(replyRunRegistry.get(sessionKey)?.sessionId).toBe(sessionId); + return undefined; + }); + + try { + await dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + SessionKey: sessionKey, + BodyForAgent: "queue through normal admission", + }), + cfg: emptyConfig, + dispatcher: createDispatcher(), + replyResolver, + }); + + expect(replyResolver).toHaveBeenCalledTimes(1); + } finally { + clearActiveEmbeddedRun(sessionId, activeHandle, sessionKey); + } + }); + + it("keeps Gateway turns on normal admission when the embedded run belongs to an old session", async () => { + setNoAbort(); + const sessionKey = "agent:main:main"; + const staleSessionId = "stale-embedded-session"; + const currentSessionId = "current-session"; + const activeHandle = { + queueMessage: vi.fn(async () => {}), + isStreaming: () => true, + isCompacting: () => false, + abort: vi.fn(), + }; + setActiveEmbeddedRun(staleSessionId, activeHandle, sessionKey); + sessionStoreMocks.currentEntry = { sessionId: currentSessionId, updatedAt: Date.now() }; + const replyResolver = vi.fn(async () => { + expect(replyRunRegistry.get(sessionKey)?.sessionId).toBe(currentSessionId); + return undefined; + }); + + try { + await dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "webchat", + Surface: "webchat", + SessionKey: sessionKey, + BodyForAgent: "start on the current session", + }), + cfg: emptyConfig, + dispatcher: createDispatcher(), + replyOptions: { + turnAdoptionLifecycle: { + onAdopted: async () => {}, + onDeferred: vi.fn(), + onSettled: vi.fn(), + }, + }, + replyResolver, + }); + + expect(replyResolver).toHaveBeenCalledTimes(1); + } finally { + clearActiveEmbeddedRun(staleSessionId, activeHandle, sessionKey); + } + }); + it("clears stale active reply operations for terminal sessions and retries admission", async () => { setNoAbort(); const sessionKey = "agent:main:telegram:group:-1003774691294"; diff --git a/src/auto-reply/reply/dispatch-from-config.lifecycle.ts b/src/auto-reply/reply/dispatch-from-config.lifecycle.ts index 2b202d9d722b..6df6a48f2e37 100644 --- a/src/auto-reply/reply/dispatch-from-config.lifecycle.ts +++ b/src/auto-reply/reply/dispatch-from-config.lifecycle.ts @@ -1,4 +1,5 @@ import crypto from "node:crypto"; +import { resolveActiveEmbeddedRunSessionId } from "../../agents/embedded-agent-runner/run-state.js"; import { isRecoverableTerminalSessionStatus } from "../../config/sessions/terminal-status.js"; import type { SessionEntry } from "../../config/sessions/types.js"; import { logVerbose } from "../../globals.js"; @@ -157,12 +158,27 @@ export function createDispatchReplyOperationCoordinator(params: { params.operationSessionStoreEntry.entry?.sessionId ?? crypto.randomUUID(); const replyTurnKind = resolveReplyTurnKind(params.replyOptions); + const activeReplyOperation = replyRunRegistry.get(params.dispatchOperationSessionKey); + const activeEmbeddedSessionId = resolveActiveEmbeddedRunSessionId( + params.dispatchOperationSessionKey, + ); + const allowGatewayEmbeddedQueueResolution = + replyTurnKind === "visible" && + params.replyOptions?.turnAdoptionLifecycle !== undefined && + activeReplyOperation === undefined && + activeEmbeddedSessionId === operationSessionId; + if (allowGatewayEmbeddedQueueResolution) { + // An embedded owner can outlive its reply-operation registration. Do not + // create a competing operation for the same session before queue policy + // gets a chance to steer the active backend. + return { status: "ready" }; + } const allowActivePreDispatch = phase === "pre_dispatch" && replyTurnKind === "visible"; const allowGatewayQueueResolution = phase === "dispatch" && replyTurnKind === "visible" && params.replyOptions?.turnAdoptionLifecycle !== undefined && - replyRunRegistry.get(params.dispatchOperationSessionKey) !== undefined; + activeReplyOperation !== undefined; if (allowGatewayQueueResolution) { // Gateway turns need to reach getReplyFromConfig while the owner is active; // that layer applies the session's steer/followup/collect/drop policy.