Merge remote-tracking branch 'origin/main' into fix-local-provider-lifecycle-20260804

* origin/main:
  fix(gateway): steer matching embedded turns (#119287)

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";

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.