diff --git a/extensions/slack/src/monitor/ingress.ts b/extensions/slack/src/monitor/ingress.ts index ea0ef9db6854..6766cb982aa1 100644 --- a/extensions/slack/src/monitor/ingress.ts +++ b/extensions/slack/src/monitor/ingress.ts @@ -11,6 +11,7 @@ import { extractErrorCode, formatErrorMessage, } from "openclaw/plugin-sdk/error-runtime"; +import { createDeferred } from "openclaw/plugin-sdk/extension-shared"; import type { PluginJsonValue } from "openclaw/plugin-sdk/plugin-entry"; import { asOptionalRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; import { getSlackRuntime } from "../runtime.js"; @@ -281,16 +282,15 @@ export function createSlackDurableIngress( let releaseSession: (() => void) | undefined; let releaseChannel: (() => void) | undefined; let routedSession: string | undefined; - let migrationAwaitedChannelTurns = false; - let downstreamDeferred = false; - let settled = false; + let adoptOnCompletion = false; const settleSession = () => { - settled = true; + adoptOnCompletion = false; releaseSession?.(); }; const settleTurn = () => { settleSession(); releaseChannel?.(); + lifecycle.abortSignal.removeEventListener("abort", settleTurn); }; const routedLifecycle: SlackIngressTurnLifecycle = { ...lifecycle, @@ -303,45 +303,31 @@ export function createSlackDurableIngress( } lifecycle.abortSignal.throwIfAborted(); routedSession = sessionKey; + adoptOnCompletion = true; const previousTurn = activeSessionTurns.get(sessionKey); - let resolveCurrentTurn: () => void = () => {}; - const releasedCurrentTurn = new Promise((resolve) => { - resolveCurrentTurn = resolve; - }); + const releasedCurrentTurn = createDeferred(); const currentTurn = previousTurn - ? previousTurn.then(() => releasedCurrentTurn) - : releasedCurrentTurn; + ? previousTurn.then(() => releasedCurrentTurn.promise) + : releasedCurrentTurn.promise; activeSessionTurns.set(sessionKey, currentTurn); - let resolveChannelTurn: () => void = () => {}; - const channelTurn = new Promise((resolve) => { - resolveChannelTurn = resolve; - }); + const channelTurn = createDeferred(); const channelTurns = activeChannelTurns.get(laneKey) ?? new Set>(); - channelTurns.add(channelTurn); + channelTurns.add(channelTurn.promise); activeChannelTurns.set(laneKey, channelTurns); - const releaseCurrentSession = () => { - lifecycle.abortSignal.removeEventListener("abort", releaseCurrentSession); - resolveCurrentTurn(); - }; - const releaseCurrentChannel = () => { - lifecycle.abortSignal.removeEventListener("abort", releaseCurrentChannel); - resolveChannelTurn(); - }; void currentTurn.then(() => { if (activeSessionTurns.get(sessionKey) === currentTurn) { activeSessionTurns.delete(sessionKey); } }); - void channelTurn.then(() => { - channelTurns.delete(channelTurn); + void channelTurn.promise.then(() => { + channelTurns.delete(channelTurn.promise); if (channelTurns.size === 0 && activeChannelTurns.get(laneKey) === channelTurns) { activeChannelTurns.delete(laneKey); } }); - releaseSession = releaseCurrentSession; - releaseChannel = releaseCurrentChannel; - lifecycle.abortSignal.addEventListener("abort", releaseSession, { once: true }); - lifecycle.abortSignal.addEventListener("abort", releaseChannel, { once: true }); + releaseSession = () => releasedCurrentTurn.resolve(); + releaseChannel = () => channelTurn.resolve(); + lifecycle.abortSignal.addEventListener("abort", settleTurn, { once: true }); // Preserve shipped channel lanes until the prepared route proves its // session; channel-ID migration therefore still fences all traffic. lifecycle.onDeferred(); @@ -362,7 +348,6 @@ export function createSlackDurableIngress( } }, onDeferred: () => { - downstreamDeferred = true; lifecycle.onDeferred(); // Reply handoff releases session order; migration remains fenced // until the durable turn is actually adopted or abandoned. @@ -384,7 +369,7 @@ export function createSlackDurableIngress( if (channelTurns && channelTurns.size > 0) { // A migration owns the channel lane while earlier routed sessions // settle; later channel traffic cannot overtake the config change. - migrationAwaitedChannelTurns = true; + adoptOnCompletion = true; lifecycle.onAdoptionFinalizing(); await Promise.all(channelTurns); lifecycle.abortSignal.throwIfAborted(); @@ -411,11 +396,7 @@ export function createSlackDurableIngress( }, }); } - if ( - (routedSession !== undefined || migrationAwaitedChannelTurns) && - !settled && - !downstreamDeferred - ) { + if (adoptOnCompletion) { await routedLifecycle.onAdopted(); } } catch (error) { diff --git a/extensions/slack/src/monitor/message-handler.ts b/extensions/slack/src/monitor/message-handler.ts index 03eda4bc9ca5..fd4a189777e8 100644 --- a/extensions/slack/src/monitor/message-handler.ts +++ b/extensions/slack/src/monitor/message-handler.ts @@ -4,6 +4,7 @@ import { shouldDebounceTextInbound, } from "openclaw/plugin-sdk/channel-inbound"; import { collectErrorGraphCandidates, formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import { createDeferred } from "openclaw/plugin-sdk/extension-shared"; import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime"; import { getRuntimeConfigSnapshot, @@ -51,11 +52,7 @@ export type SlackMessageHandler = ( }, ) => Promise; -type SlackDispatchCompletion = { - promise: Promise; - resolve: () => void; - reject: (error: unknown) => void; -}; +type SlackDispatchCompletion = ReturnType>; type IngressSlackMessageOptions = Parameters[1] & { retryAttempt?: number; @@ -65,16 +62,6 @@ type QueuedSlackMessageOptions = IngressSlackMessageOptions & { dispatchCompletion?: Omit; }; -function createSlackDispatchCompletion(): SlackDispatchCompletion { - let resolve!: () => void; - let reject!: (error: unknown) => void; - const promise = new Promise((nextResolve, nextReject) => { - resolve = nextResolve; - reject = nextReject; - }); - return { promise, resolve, reject }; -} - const RETRYABLE_FLUSH_MAX_ATTEMPTS = 3; const RETRYABLE_FLUSH_RETRY_DELAY_MS = 1_000; const REPLY_SESSION_INIT_CONFLICT_MESSAGE_RE = /reply session initialization conflicted for \S+/u; @@ -468,7 +455,7 @@ export function createSlackMessageHandler(params: { pendingKeys.add(debounceKey); pendingTopLevelDebounceKeys.set(conversationKey, pendingKeys); } - const dispatchCompletion = opts.awaitDispatch ? createSlackDispatchCompletion() : undefined; + const dispatchCompletion = opts.awaitDispatch ? createDeferred() : undefined; await debouncer.enqueue({ message: resolvedMessage, opts: {