diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts index 244278b6bf89..cdd88e40cbdc 100644 --- a/src/gateway/server-methods/chat.ts +++ b/src/gateway/server-methods/chat.ts @@ -194,7 +194,7 @@ import { loadOptionalServerMethodModelCatalog, startOptionalServerMethodModelCatalogLoad, } from "./optional-model-catalog.js"; -import { hasTrackedActiveSessionRun } from "./session-active-runs.js"; +import { hasTrackedActiveSessionRun, hasVisibleActiveSessionRun } from "./session-active-runs.js"; import { emitSessionsChanged } from "./session-change-event.js"; import type { GatewayClient, @@ -3042,10 +3042,11 @@ async function handleChatHistoryRequest({ }); const activeRunAgentId = canonicalKey === "global" ? (selectedAgent.agentId ?? defaultAgentId) : selectedAgent.agentId; - sessionInfo.hasActiveRun = hasTrackedActiveSessionRun({ + sessionInfo.hasActiveRun = hasVisibleActiveSessionRun({ context, requestedKey: sessionKey, canonicalKey, + sessionId: entry?.sessionId, ...(activeRunAgentId ? { agentId: activeRunAgentId } : {}), defaultAgentId, }); diff --git a/src/gateway/server-methods/session-active-runs.ts b/src/gateway/server-methods/session-active-runs.ts index 861d867ec013..9c6646569655 100644 --- a/src/gateway/server-methods/session-active-runs.ts +++ b/src/gateway/server-methods/session-active-runs.ts @@ -1,5 +1,6 @@ // Session active-run helpers decide whether session operations should treat a // session as busy based on Control UI-visible active chat/agent runs. +import { isEmbeddedAgentRunActive } from "../../agents/embedded-agent-runner/runs.js"; import { normalizeAgentId } from "../../routing/session-key.js"; import type { GatewayRequestContext } from "./types.js"; @@ -84,3 +85,18 @@ export function hasTrackedActiveSessionRun(params: { ), ); } + +export function hasVisibleActiveSessionRun(params: { + context: Partial>; + requestedKey: string; + canonicalKey: string; + sessionId?: string; + agentId?: string; + defaultAgentId?: string; +}): boolean { + if (hasTrackedActiveSessionRun(params)) { + return true; + } + const sessionId = params.sessionId?.trim(); + return sessionId ? isEmbeddedAgentRunActive(sessionId) : false; +} diff --git a/src/gateway/server-methods/session-change-event.ts b/src/gateway/server-methods/session-change-event.ts index e86b94e26259..38ee2b916101 100644 --- a/src/gateway/server-methods/session-change-event.ts +++ b/src/gateway/server-methods/session-change-event.ts @@ -2,7 +2,7 @@ import { resolveDefaultAgentId } from "../../agents/agent-scope.js"; import { buildGatewaySessionEventFields } from "../session-event-payload.js"; import { loadGatewaySessionRow } from "../session-utils.js"; -import { hasTrackedActiveSessionRun } from "./session-active-runs.js"; +import { hasVisibleActiveSessionRun } from "./session-active-runs.js"; import type { GatewayRequestContext } from "./types.js"; export type SessionChangedPayload = { @@ -45,10 +45,11 @@ export function emitSessionsChanged( ...buildGatewaySessionEventFields({ sessionRow, agentId: payload.agentId, - hasActiveRun: hasTrackedActiveSessionRun({ + hasActiveRun: hasVisibleActiveSessionRun({ context, requestedKey: payload.sessionKey ?? sessionRow.key, canonicalKey: sessionRow.key, + sessionId: sessionRow.sessionId, agentId: sessionRow.key === "global" ? payload.agentId : undefined, defaultAgentId, }), diff --git a/src/gateway/server-methods/sessions.abort-agent-scope.test.ts b/src/gateway/server-methods/sessions.abort-agent-scope.test.ts index e2d48f98f427..d5818c78874e 100644 --- a/src/gateway/server-methods/sessions.abort-agent-scope.test.ts +++ b/src/gateway/server-methods/sessions.abort-agent-scope.test.ts @@ -8,6 +8,7 @@ const chatAbortMock = vi.fn(); const resolveSessionKeyForRunMock = vi.fn(); const listSessionsFromStoreAsyncMock = vi.fn(); const loadCombinedSessionStoreForGatewayMock = vi.fn(); +const isEmbeddedAgentRunActiveMock = vi.fn(); const loadSessionEntryMock = vi.fn((sessionKey: string, _opts?: { agentId?: string }) => ({ canonicalKey: sessionKey, })); @@ -34,6 +35,16 @@ vi.mock("../session-utils.js", async () => { }; }); +vi.mock("../../agents/embedded-agent-runner/runs.js", async () => { + const actual = await vi.importActual( + "../../agents/embedded-agent-runner/runs.js", + ); + return { + ...actual, + isEmbeddedAgentRunActive: (...args: unknown[]) => isEmbeddedAgentRunActiveMock(...args), + }; +}); + import { sessionsHandlers } from "./sessions.js"; function createActiveRun(sessionKey: string, params: { agentId?: string } = {}) { @@ -170,6 +181,8 @@ describe("sessions.abort agent scope", () => { store: {}, }); loadSessionEntryMock.mockClear(); + isEmbeddedAgentRunActiveMock.mockReset(); + isEmbeddedAgentRunActiveMock.mockReturnValue(false); }); it("does not abort an active run whose session key belongs to another requested agent", async () => { @@ -192,6 +205,39 @@ describe("sessions.abort agent scope", () => { }); }); + it("marks listed sessions active when the embedded or channel reply run registry owns the session id", async () => { + const context = createContext({ + extra: { loadGatewayModelCatalog: vi.fn().mockResolvedValue([]) }, + }); + listSessionsFromStoreAsyncMock.mockResolvedValue({ + sessions: [{ key: "agent:main:openclaw-weixin:direct:user", sessionId: "sess-weixin" }], + }); + isEmbeddedAgentRunActiveMock.mockImplementation( + (sessionId: string) => sessionId === "sess-weixin", + ); + + const respond = await callSessions( + "sessions.list", + { agentId: "main" }, + { context, reqId: "req-channel-active" }, + ); + + expect(isEmbeddedAgentRunActiveMock).toHaveBeenCalledWith("sess-weixin"); + expect(respond).toHaveBeenCalledWith( + true, + expect.objectContaining({ + sessions: [ + expect.objectContaining({ + key: "agent:main:openclaw-weixin:direct:user", + sessionId: "sess-weixin", + hasActiveRun: true, + }), + ], + }), + undefined, + ); + }); + it("preserves runId-only aborts for active non-default agent runs", async () => { const activeRun = createActiveRun("agent:beta:dashboard:target"); const context = createBetaRunContext(activeRun); diff --git a/src/gateway/server-methods/sessions.ts b/src/gateway/server-methods/sessions.ts index e80132f5469d..8d6c99a80d54 100644 --- a/src/gateway/server-methods/sessions.ts +++ b/src/gateway/server-methods/sessions.ts @@ -124,7 +124,7 @@ import { resolveSessionKeyFromResolveParams } from "../sessions-resolve.js"; import { setGatewayDedupeEntry } from "./agent-wait-dedupe.js"; import { chatHandlers } from "./chat.js"; import { loadOptionalServerMethodModelCatalog } from "./optional-model-catalog.js"; -import { hasTrackedActiveSessionRun } from "./session-active-runs.js"; +import { hasTrackedActiveSessionRun, hasVisibleActiveSessionRun } from "./session-active-runs.js"; import { emitSessionsChanged } from "./session-change-event.js"; import type { GatewayClient, @@ -899,10 +899,11 @@ export const sessionsHandlers: GatewayRequestHandlers = { () => { return result.sessions.map((session) => Object.assign({}, session, { - hasActiveRun: hasTrackedActiveSessionRun({ + hasActiveRun: hasVisibleActiveSessionRun({ context, requestedKey: session.key, canonicalKey: session.key, + sessionId: session.sessionId, ...(session.key === "global" && p.agentId ? { agentId: p.agentId } : {}), defaultAgentId: resolveDefaultAgentId(cfg), }), diff --git a/src/gateway/server-session-events.test.ts b/src/gateway/server-session-events.test.ts index ffa6010548a8..6c3efc4ccbce 100644 --- a/src/gateway/server-session-events.test.ts +++ b/src/gateway/server-session-events.test.ts @@ -4,9 +4,11 @@ import type { ChatAbortControllerEntry } from "./chat-abort.js"; const sessionRow = vi.hoisted(() => ({ key: "agent:main:main", kind: "direct", + sessionId: "sess-main", status: "done", updatedAt: 1, })); +const isEmbeddedAgentRunActiveMock = vi.hoisted(() => vi.fn()); vi.mock("../config/io.js", () => ({ getRuntimeConfig: () => ({}) })); vi.mock("./chat-display-projection.js", () => ({ @@ -18,6 +20,15 @@ vi.mock("./session-utils.js", () => ({ loadSessionEntry: () => ({ entry: undefined, storePath: "" }), readSessionMessageCountAsync: vi.fn(), })); +vi.mock("../agents/embedded-agent-runner/runs.js", async () => { + const actual = await vi.importActual( + "../agents/embedded-agent-runner/runs.js", + ); + return { + ...actual, + isEmbeddedAgentRunActive: (...args: unknown[]) => isEmbeddedAgentRunActiveMock(...args), + }; +}); const { createTranscriptUpdateBroadcastHandler } = await import("./server-session-events.js"); @@ -62,6 +73,7 @@ async function emitAssistantTranscriptUpdate( describe("createTranscriptUpdateBroadcastHandler", () => { beforeEach(() => { vi.clearAllMocks(); + isEmbeddedAgentRunActiveMock.mockReturnValue(false); }); it("keeps transcript snapshots active while plugin finalization delays the terminal event", async () => { @@ -83,6 +95,17 @@ describe("createTranscriptUpdateBroadcastHandler", () => { }); }); + it("keeps transcript snapshots active for embedded or channel reply runs", async () => { + isEmbeddedAgentRunActiveMock.mockImplementation((sessionId) => sessionId === "sess-main"); + + await expect(emitAssistantTranscriptUpdate(false)).resolves.toMatchObject({ + sessionKey: "agent:main:main", + hasActiveRun: true, + session: { key: "agent:main:main", sessionId: "sess-main", hasActiveRun: true }, + }); + expect(isEmbeddedAgentRunActiveMock).toHaveBeenCalledWith("sess-main"); + }); + it("broadcasts user idempotency keys in session.message metadata", async () => { await expect( emitAssistantTranscriptUpdate(false, { diff --git a/src/gateway/server-session-events.ts b/src/gateway/server-session-events.ts index 5db03c58b4c5..aecb5fadc5a3 100644 --- a/src/gateway/server-session-events.ts +++ b/src/gateway/server-session-events.ts @@ -14,7 +14,7 @@ import type { SessionEventSubscriberRegistry, SessionMessageSubscriberRegistry, } from "./server-chat.js"; -import { hasTrackedActiveSessionRun } from "./server-methods/session-active-runs.js"; +import { hasVisibleActiveSessionRun } from "./server-methods/session-active-runs.js"; import { buildGatewaySessionEventFields } from "./session-event-payload.js"; import { resolveSessionKeyForTranscriptFile } from "./session-transcript-key.js"; import { @@ -179,10 +179,11 @@ async function handleTranscriptUpdateBroadcast( transcriptUsageMaxBytes: 64 * 1024, }); const hasActiveRun = sessionRow - ? hasTrackedActiveSessionRun({ + ? hasVisibleActiveSessionRun({ context: params, requestedKey: sessionKey, canonicalKey: sessionRow.key, + sessionId: sessionRow.sessionId, ...(sessionRow.key === "global" && visibleAgentId ? { agentId: visibleAgentId } : {}), defaultAgentId: normalizeAgentId(resolveDefaultAgentId(getRuntimeConfig())), })