diff --git a/extensions/slack/src/monitor/message-handler/prepare.test.ts b/extensions/slack/src/monitor/message-handler/prepare.test.ts index 157ec54b3288..a0c689f08468 100644 --- a/extensions/slack/src/monitor/message-handler/prepare.test.ts +++ b/extensions/slack/src/monitor/message-handler/prepare.test.ts @@ -3703,6 +3703,8 @@ Second paragraph should still reach the agent after Slack's preview cutoff.`; expect(root.ctxPayload.SessionKey).toBe(expectedSessionKey); expect(followUp.ctxPayload.SessionKey).toBe(expectedSessionKey); expect(new Set([root.ctxPayload.SessionKey, followUp.ctxPayload.SessionKey]).size).toBe(1); + expect(root.ctxPayload).not.toHaveProperty("SystemEventSessionKey"); + expect(followUp.ctxPayload).not.toHaveProperty("SystemEventSessionKey"); if (expectedAgentId) { expect(root.route.agentId).toBe(expectedAgentId); } diff --git a/extensions/slack/src/monitor/message-handler/prepare.ts b/extensions/slack/src/monitor/message-handler/prepare.ts index 77d482de6f0b..c673697abb51 100644 --- a/extensions/slack/src/monitor/message-handler/prepare.ts +++ b/extensions/slack/src/monitor/message-handler/prepare.ts @@ -1592,7 +1592,8 @@ export async function prepareSlackMessage(params: { agentId: route.agentId, dmScope: route.dmScope, accountId: route.accountId, - routeSessionKey: sessionKey, + routeSessionKey: route.sessionKey, + dispatchSessionKey: sessionKey, parentSessionKey: threadKeys.parentSessionKey, }, reply: { diff --git a/src/auto-reply/reply/get-reply-run-admission.ts b/src/auto-reply/reply/get-reply-run-admission.ts index 517908aae9d5..a8b174eb2de2 100644 --- a/src/auto-reply/reply/get-reply-run-admission.ts +++ b/src/auto-reply/reply/get-reply-run-admission.ts @@ -48,6 +48,7 @@ import { resolveRoutedDeliveryThreadId, } from "./routed-delivery-thread.js"; import { drainFormattedSystemEvents } from "./session-system-events.js"; +import { getReplySystemEventSessionKey } from "./system-event-session-key.js"; export async function prepareReplyRunAdmission(context: PreparedReplyRunContext) { const { @@ -130,19 +131,30 @@ export async function prepareReplyRunAdmission(context: PreparedReplyRunContext) ? `[Thread starter - for context]\n${threadStarterBody}` : undefined; const drainedSystemEventBlocks: string[] = []; - const rebuildPromptBodies = async () => { - if (!useFastReplyRuntime && heartbeatRunScope !== "commitment-only") { + const drainSystemEventBlocks = async () => { + if (useFastReplyRuntime || heartbeatRunScope === "commitment-only") { + return; + } + const routeSystemEventSessionKey = normalizeOptionalString(getReplySystemEventSessionKey(opts)); + const systemEventSessionKeys = + routeSystemEventSessionKey && routeSystemEventSessionKey !== sessionKey + ? [routeSystemEventSessionKey, sessionKey] + : [sessionKey]; + for (const systemEventSessionKey of systemEventSessionKeys) { + const isCurrentSession = systemEventSessionKey === sessionKey; const eventsBlock = await drainFormattedSystemEvents({ cfg, - sessionKey, - isMainSession, - isNewSession, + sessionKey: systemEventSessionKey, + isMainSession: isCurrentSession && isMainSession, + isNewSession: isCurrentSession && isNewSession, suppressHeartbeatOwnedEvents: context.isHeartbeat, }); if (eventsBlock) { drainedSystemEventBlocks.push(eventsBlock); } } + }; + const rebuildPromptBodies = () => { const { activeGoalContext, inboundUserContext } = context.getInboundContext(); return buildReplyPromptEnvelope({ ctx, @@ -558,6 +570,17 @@ export async function prepareReplyRunAdmission(context: PreparedReplyRunContext) return { kind: "reply", reply: queueState.reply } as const; } } + if (activeRunQueueAction !== "drop") { + await traceRunPhase("reply.drain_system_events", () => drainSystemEventBlocks()); + ({ + prefixedCommandBody, + queuedBody, + transcriptBody, + transcriptCommandBody, + media: promptMedia, + currentInboundContext, + } = await traceRunPhase("reply.build_prompt_bodies", () => rebuildPromptBodies())); + } return { kind: "ready", diff --git a/src/auto-reply/reply/get-reply-run.media-only.test.ts b/src/auto-reply/reply/get-reply-run.media-only.test.ts index a13e21ff3bd7..21c247a06157 100644 --- a/src/auto-reply/reply/get-reply-run.media-only.test.ts +++ b/src/auto-reply/reply/get-reply-run.media-only.test.ts @@ -10,6 +10,11 @@ import { } from "../../agents/embedded-agent-runner/runs.js"; import type { SessionEntry } from "../../config/sessions.js"; import { HEARTBEAT_RUN_SCOPE } from "../../infra/heartbeat-run-scope.js"; +import { + enqueueSystemEvent, + peekSystemEventEntries, + resetSystemEventsForTest, +} from "../../infra/system-events.js"; import { MESSAGE_TOOL_ONLY_DELIVERY_HINT } from "../../plugin-sdk/message-tool-delivery-hints.js"; import { normalizeSessionDeliveryState } from "../../utils/delivery-context.shared.js"; import { runReplyAgent } from "./agent-runner.runtime.js"; @@ -31,6 +36,7 @@ import { testing as replyRunTesting } from "./reply-run-registry.test-support.js import { routeReply } from "./route-reply.runtime.js"; import { drainFormattedSystemEvents } from "./session-system-events.js"; import { buildChannelSourceTurnId } from "./source-turn-id.js"; +import { withReplySystemEventSessionKey } from "./system-event-session-key.js"; import { resolveTypingMode } from "./typing-mode.js"; vi.mock("../../agents/auth-profiles/session-override.js", () => ({ @@ -371,6 +377,7 @@ describe("runPreparedReply media-only handling", () => { afterEach(() => { vi.useRealTimers(); + resetSystemEventsForTest(); const paths = cleanupPaths.splice(0); return Promise.all(paths.map((entry) => rm(entry, { recursive: true, force: true }))); }); @@ -2253,12 +2260,72 @@ describe("runPreparedReply media-only handling", () => { nextRun.complete(); }); - it("re-drains system events after waiting behind an active run", async () => { + it("keeps route and dispatch system events queued when busy admission returns", async () => { + vi.useFakeTimers(); + const actualSystemEvents = await vi.importActual( + "./session-system-events.js", + ); + vi.mocked(drainFormattedSystemEvents).mockImplementation( + actualSystemEvents.drainFormattedSystemEvents, + ); const queueSettings = await import("./queue/settings-runtime.js"); vi.mocked(queueSettings.resolveQueueSettings).mockReturnValueOnce({ mode: "interrupt" }); - vi.mocked(drainFormattedSystemEvents) - .mockResolvedValueOnce("System: [t] Initial event.") - .mockResolvedValueOnce("System: [t] Post-compaction context."); + const routeSessionKey = "agent:main:slack:channel:c123"; + const dispatchSessionKey = `${routeSessionKey}:thread:123.456`; + enqueueSystemEvent("Slack reaction added: :eyes:", { sessionKey: routeSessionKey }); + enqueueSystemEvent("Slack message in #claw-test from Alice", { + sessionKey: dispatchSessionKey, + }); + const previousRun = createReplyOperation({ + sessionId: "session-before-wait", + sessionKey: dispatchSessionKey, + resetTriggered: false, + }); + previousRun.setPhase("running"); + + const runPromise = runPreparedReply( + baseParams({ + isNewSession: false, + sessionId: "session-before-wait", + sessionKey: dispatchSessionKey, + opts: withReplySystemEventSessionKey({}, routeSessionKey), + }), + ); + + await Promise.resolve(); + previousRun.complete(); + const nextRun = createReplyOperation({ + sessionId: "session-after-wait", + sessionKey: dispatchSessionKey, + resetTriggered: false, + }); + nextRun.setPhase("running"); + + const assertion = expect(runPromise).resolves.toEqual({ + text: "⚠️ Previous run is still shutting down. Please try again in a moment.", + }); + await vi.advanceTimersByTimeAsync(15_000); + await assertion; + expect(vi.mocked(runReplyAgent)).not.toHaveBeenCalled(); + expect(peekSystemEventEntries(routeSessionKey).map((event) => event.text)).toEqual([ + "Slack reaction added: :eyes:", + ]); + expect(peekSystemEventEntries(dispatchSessionKey).map((event) => event.text)).toEqual([ + "Slack message in #claw-test from Alice", + ]); + + nextRun.complete(); + }); + it("drains system events only after waiting behind an active run", async () => { + const actualSystemEvents = await vi.importActual( + "./session-system-events.js", + ); + vi.mocked(drainFormattedSystemEvents).mockImplementation( + actualSystemEvents.drainFormattedSystemEvents, + ); + const queueSettings = await import("./queue/settings-runtime.js"); + vi.mocked(queueSettings.resolveQueueSettings).mockReturnValueOnce({ mode: "interrupt" }); + enqueueSystemEvent("System event after active run", { sessionKey: "session-key" }); const previousRun = createReplyOperation({ sessionId: "session-events-after-wait", @@ -2275,16 +2342,18 @@ describe("runPreparedReply media-only handling", () => { ); await Promise.resolve(); + expect(peekSystemEventEntries("session-key").map((event) => event.text)).toEqual([ + "System event after active run", + ]); previousRun.complete(); await expect(runPromise).resolves.toEqual({ text: "ok" }); const call = requireLastRunReplyAgentCall(); - expect(call?.commandBody).toContain("System: [t] Initial event."); - expect(call?.commandBody).not.toContain("System: [t] Post-compaction context."); - expect(call?.transcriptCommandBody).not.toContain("System: [t] Initial event."); - expect(call?.followupRun.prompt).toContain("System: [t] Initial event."); - expect(call?.followupRun.prompt).not.toContain("System: [t] Post-compaction context."); - expect(call?.followupRun.transcriptPrompt).not.toContain("System: [t] Initial event."); + expect(call?.commandBody).toContain("System event after active run"); + expect(call?.transcriptCommandBody).not.toContain("System event after active run"); + expect(call?.followupRun.prompt).toContain("System event after active run"); + expect(call?.followupRun.transcriptPrompt).not.toContain("System event after active run"); + expect(peekSystemEventEntries("session-key")).toStrictEqual([]); }); it("threads inbound context as current-turn context without changing transcript text", async () => { @@ -3488,6 +3557,38 @@ describe("runPreparedReply media-only handling", () => { expect(applySessionHints).not.toHaveBeenCalled(); }); + it("includes route system events in a thread-scoped turn", async () => { + const actualSystemEvents = await vi.importActual( + "./session-system-events.js", + ); + vi.mocked(drainFormattedSystemEvents).mockImplementation( + actualSystemEvents.drainFormattedSystemEvents, + ); + enqueueSystemEvent("Slack reaction added: :eyes:", { + sessionKey: "agent:main:slack:channel:c123", + }); + enqueueSystemEvent("Slack message in #claw-test from Alice", { + sessionKey: "agent:main:slack:channel:c123:thread:123.456", + }); + + await runPreparedReply( + baseParams({ + ctx: createInboundBody("report queued reactions"), + isNewSession: false, + opts: withReplySystemEventSessionKey({}, "agent:main:slack:channel:c123"), + sessionKey: "agent:main:slack:channel:c123:thread:123.456", + }), + ); + + const prompt = requireRunReplyAgentCall().followupRun.prompt; + expect(prompt).toContain("Slack reaction added: :eyes:"); + expect(prompt).toContain("Slack message in #claw-test from Alice"); + expect(peekSystemEventEntries("agent:main:slack:channel:c123")).toStrictEqual([]); + expect(peekSystemEventEntries("agent:main:slack:channel:c123:thread:123.456")).toStrictEqual( + [], + ); + }); + it("keeps sender ownership when queued system events are prepended", async () => { vi.mocked(drainFormattedSystemEvents).mockResolvedValueOnce( "System: [t] External webhook payload.", diff --git a/src/auto-reply/reply/system-event-session-key.ts b/src/auto-reply/reply/system-event-session-key.ts new file mode 100644 index 000000000000..db8f0af83212 --- /dev/null +++ b/src/auto-reply/reply/system-event-session-key.ts @@ -0,0 +1,21 @@ +const REPLY_SYSTEM_EVENT_SESSION_KEY = Symbol("openclaw.reply.systemEventSessionKey"); + +/** Attach route-owned system-event state without widening public reply option contracts. */ +export function withReplySystemEventSessionKey( + options: T, + sessionKey: string, +): T { + return { + ...options, + [REPLY_SYSTEM_EVENT_SESSION_KEY]: sessionKey, + }; +} + +/** Read route-owned system-event state after it crosses internal reply-option spreads. */ +export function getReplySystemEventSessionKey(options: object | undefined): string | undefined { + if (!options) { + return undefined; + } + const value = (options as Record)[REPLY_SYSTEM_EVENT_SESSION_KEY]; + return typeof value === "string" ? value : undefined; +} diff --git a/src/channels/turn/kernel.test.ts b/src/channels/turn/kernel.test.ts index f09da39ce2e8..ee8272b9656f 100644 --- a/src/channels/turn/kernel.test.ts +++ b/src/channels/turn/kernel.test.ts @@ -3,6 +3,7 @@ import { afterEach, beforeEach, describe, expect, expectTypeOf, it, vi } from "v import type { ReplyPayload } from "../../auto-reply/reply-payload.js"; import type { HistoryEntry } from "../../auto-reply/reply/history.types.js"; import type { DispatchReplyWithBufferedBlockDispatcher } from "../../auto-reply/reply/provider-dispatcher.types.js"; +import { getReplySystemEventSessionKey } from "../../auto-reply/reply/system-event-session-key.js"; import type { FinalizedMsgContext } from "../../auto-reply/templating.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { @@ -296,6 +297,49 @@ describe("channel turn kernel", () => { } }); + it.each([ + { + channel: "slack", + routeSessionKey: "agent:main:slack:channel:c1", + dispatchSessionKey: "agent:main:slack:channel:c1:thread:123.456", + }, + { + channel: "discord", + routeSessionKey: "agent:main:discord:channel:c1", + dispatchSessionKey: "agent:main:discord:channel:c1:thread:t1", + }, + ])("carries $channel route system-event ownership privately into dispatch", async (scenario) => { + const { channel, routeSessionKey, dispatchSessionKey } = scenario; + const dispatchReplyWithBufferedBlockDispatcher = vi.fn( + async (params: Parameters[0]) => { + expect(params.ctx).not.toHaveProperty("SystemEventSessionKey"); + expect(getReplySystemEventSessionKey({ ...params.replyOptions })).toBe(routeSessionKey); + await params.dispatcherOptions.deliver({ text: "reply" }, { kind: "final" }); + return { queuedFinal: true, counts: { tool: 0, block: 0, final: 1 } }; + }, + ) as DispatchReplyWithBufferedBlockDispatcher; + + await dispatchAssembledChannelTurn({ + cfg, + channel, + agentId: "main", + routeSessionKey, + storePath: "/tmp/sessions.json", + ctxPayload: createCtx({ + SessionKey: dispatchSessionKey, + Surface: channel, + Provider: channel, + }), + recordInboundSession: createRecordInboundSession(), + dispatchReplyWithBufferedBlockDispatcher, + delivery: { + deliver: async () => ({ visibleReplySent: true }), + }, + }); + + expect(dispatchReplyWithBufferedBlockDispatcher).toHaveBeenCalledOnce(); + }); + it("runs routed direct message hooks after payload preparation", async () => { const events: string[] = []; const runMessageSending = vi.fn(async (event: { content: string }) => { diff --git a/src/channels/turn/lifecycle.ts b/src/channels/turn/lifecycle.ts index 1a7342a3ebab..d1b5c5fc9966 100644 --- a/src/channels/turn/lifecycle.ts +++ b/src/channels/turn/lifecycle.ts @@ -3,6 +3,7 @@ import type { ReplyPayload } from "../../auto-reply/reply-payload.js"; import type { DispatchFromConfigResult } from "../../auto-reply/reply/dispatch-from-config.types.js"; import type { ReplyDispatchKind } from "../../auto-reply/reply/reply-dispatcher.types.js"; import { runWithSessionInitConflictRetry } from "../../auto-reply/reply/session-init-conflict-retry.js"; +import { withReplySystemEventSessionKey } from "../../auto-reply/reply/system-event-session-key.js"; import { resolveStorePath } from "../../config/sessions/paths.js"; import { deriveInboundMessageHookContext, @@ -96,14 +97,17 @@ export function assembleResolvedChannelTurn< function resolveAssembledReplyPipeline( params: DispatchableChannelTurn, ): Pick { - const turnAdoptionLifecycle = - params.turnAdoptionLifecycle ?? params.replyOptions?.turnAdoptionLifecycle; + const adoption = params.turnAdoptionLifecycle ?? params.replyOptions?.turnAdoptionLifecycle; + let replyOptions = adoption + ? { ...params.replyOptions, turnAdoptionLifecycle: adoption } + : params.replyOptions; + if (params.routeSessionKey !== params.ctxPayload.SessionKey) { + replyOptions = withReplySystemEventSessionKey(replyOptions ?? {}, params.routeSessionKey); + } if (!params.replyPipeline) { return { dispatcherOptions: params.dispatcherOptions, - replyOptions: turnAdoptionLifecycle - ? { ...params.replyOptions, turnAdoptionLifecycle } - : params.replyOptions, + replyOptions, }; } const { onModelSelected, ...replyPipeline } = createChannelReplyPipeline({ @@ -120,8 +124,7 @@ function resolveAssembledReplyPipeline( }, replyOptions: { onModelSelected, - ...params.replyOptions, - ...(turnAdoptionLifecycle ? { turnAdoptionLifecycle } : {}), + ...replyOptions, }, }; }