diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index 18a4f5e17f07..fa704e9344ad 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -67,7 +67,7 @@ e8adcff47c1b677cd2c01a2130fe4d226ac30ab3c88a3ceb4df5a8660316c4a9 module/discord f65408d85477bb362ebe6ed9148c1bb6b9eb7839733e5e3119f4ff8f1cfd0567 module/error-runtime b013053a61e7d9be3d0c683c02baf57fa7e4393ec54e0df6a46ab0f2fe2348fd module/extension-shared ceacad83db01c66e7be6aa21a291597020f13f737b697690eae7d47098e6499a module/gateway-method-runtime -1b8eea4bad785031864822f0f07dc03cbe8670e76e0c3ee8fa51c7a2ee5c5e15 module/gateway-runtime +8fa198fde6cb5f651caaa7349bc37992642ea827da6a4c52c94017a0e5dafd6a module/gateway-runtime 6062869202e3d7fdaa1f6801d6c1df7bf7406b8118cc4c7167ed409f5257b05c module/group-access d117ebba8cc490501725778676a9d75855872b6e5fe2b2f64b1270d4808a2277 module/health 182dc685f2103ff66c1a4839a48f4f40d2eeb0070cd74e449b47aacc4e6f1c22 module/hook-runtime @@ -143,7 +143,7 @@ a94c9ff59cc361b04f8731a47d722c4bc88cfabf942221369a8cb67bdcda7249 module/tool-pl 89846974257b7551d46a514902998d0592fc67b1d8ce036c6514d7b9cd1bc159 module/tool-results 9d6ab352913a573b226e054e1dc8c6d088493aea9954950c65585923b5b6895a module/tool-send 541df9dea799f25e83ea483d481ecebc5b91c016effab593c54d3efe3ee6517b module/web-media -0d5ae2f16bb33c0fe4f85eea3da80283e89e2995a7406b98d24805feeee4f7ca module/webhook-ingress +7c43c2cd0f09d72cf5f7cd7df582767e436acb697a81722dde47dbe171c66d2e module/webhook-ingress af8c5e1c84ec9d365a7aa0d15eb63f0dd9e7b50526023e6212bba020c229f224 module/webhook-request-guards 1cc469eacda2818a116ab7b63847b9ddf86225e8d02e62fac48e618ba41a9852 module/widget-html 9161b36ec0ab062ea41b363c894fcd672a7727f21cb726739f99f9c184fce69d module/zod diff --git a/extensions/buzz/src/gateway.lifecycle.test.ts b/extensions/buzz/src/gateway.lifecycle.test.ts index 1e3b6c66fea3..a9082375e758 100644 --- a/extensions/buzz/src/gateway.lifecycle.test.ts +++ b/extensions/buzz/src/gateway.lifecycle.test.ts @@ -173,12 +173,15 @@ describe("Buzz gateway lifecycle", () => { expect(setStatus).toHaveBeenCalledWith({ accountId: account.accountId, running: true, + connected: true, lifecycle: "ready", + lastConnectedAt: expect.any(Number), configured: true, enabled: account.enabled, baseUrl: account.relayUrl, publicKey: BOT_PUBLIC_KEY, lastError: null, + terminalDisconnect: undefined, }); gatewayMocks.onFatalError?.(new Error("relay failed")); diff --git a/extensions/buzz/src/gateway.ts b/extensions/buzz/src/gateway.ts index f05ada9aeab9..23ed26a04a30 100644 --- a/extensions/buzz/src/gateway.ts +++ b/extensions/buzz/src/gateway.ts @@ -1,6 +1,7 @@ import { waitUntilAbort } from "openclaw/plugin-sdk/channel-outbound"; import { attachChannelToResult } from "openclaw/plugin-sdk/channel-send-result"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { computeBackoff, sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import type { ChannelGatewayContext } from "../runtime-api.js"; import { sendBuzzTextOneShot, startBuzzBus, type BuzzBus } from "./buzz-bus.js"; @@ -138,16 +139,15 @@ export async function startBuzzGatewayAccount(ctx: ChannelGatewayContext { socket.emit("open"); expect(ctx.setStatus).toHaveBeenCalledWith({ accountId: "default", + running: true, connected: true, lifecycle: "ready", lastConnectedAt: expect.any(Number), diff --git a/extensions/clickclack/src/gateway.ts b/extensions/clickclack/src/gateway.ts index 4126d132d473..63b3d576fc4e 100644 --- a/extensions/clickclack/src/gateway.ts +++ b/extensions/clickclack/src/gateway.ts @@ -4,6 +4,7 @@ */ import type { ChannelGatewayContext } from "openclaw/plugin-sdk/channel-contract"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import { channelReadyPatch, channelStoppedPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import type { RawData } from "ws"; import { resolveClickClackInboundAccess } from "./access.js"; @@ -289,14 +290,7 @@ export async function startClickClackGatewayAccount( ctx.abortSignal.addEventListener("abort", abort, { once: true }); removeAbortListener = () => ctx.abortSignal.removeEventListener("abort", abort); socket.on("open", () => { - ctx.setStatus({ - accountId: account.accountId, - connected: true, - lifecycle: "ready", - lastConnectedAt: Date.now(), - lastError: null, - terminalDisconnect: undefined, - }); + ctx.setStatus(channelReadyPatch({ accountId: account.accountId })); }); socket.on("message", (data) => { if (closing || settled) { @@ -364,11 +358,6 @@ export async function startClickClackGatewayAccount( } } } finally { - ctx.setStatus({ - accountId: account.accountId, - running: false, - connected: false, - lifecycle: "stopped", - }); + ctx.setStatus(channelStoppedPatch({ accountId: account.accountId })); } } diff --git a/extensions/discord/src/components-registry-state.ts b/extensions/discord/src/components-registry-state.ts index ce9cc17ab6ed..95ef0bbdab6d 100644 --- a/extensions/discord/src/components-registry-state.ts +++ b/extensions/discord/src/components-registry-state.ts @@ -1,4 +1,4 @@ -import { resolveGlobalMap } from "openclaw/plugin-sdk/global-singleton"; +import { resolveGlobalSingleton } from "openclaw/plugin-sdk/global-singleton"; import type { DiscordComponentEntry, DiscordModalEntry } from "./components.js"; type PersistedDiscordRegistryEntry = { @@ -17,21 +17,20 @@ export type DiscordRegistryStore = DiscordPersistentSt PersistedDiscordRegistryEntry >; -export const discordComponentRegistryState = { - componentEntries: resolveGlobalMap( - Symbol.for("openclaw.discord.componentEntries"), - ), - modalEntries: resolveGlobalMap( - Symbol.for("openclaw.discord.modalEntries"), - ), - persistentComponentStore: undefined as DiscordRegistryStore | undefined, - persistentModalStore: undefined as DiscordRegistryStore | undefined, - persistentRegistryDisabled: false, - reset(): void { - this.componentEntries.clear(); - this.modalEntries.clear(); - this.persistentComponentStore = undefined; - this.persistentModalStore = undefined; - this.persistentRegistryDisabled = false; +export const discordComponentRegistryState = resolveGlobalSingleton( + Symbol.for("openclaw.discord.componentRegistryState"), + () => ({ + componentEntries: new Map(), + modalEntries: new Map(), + persistentComponentStore: undefined as DiscordRegistryStore | undefined, + persistentModalStore: undefined as DiscordRegistryStore | undefined, + persistentRegistryDisabled: false, + }), + (state) => { + state.componentEntries.clear(); + state.modalEntries.clear(); + state.persistentComponentStore = undefined; + state.persistentModalStore = undefined; + state.persistentRegistryDisabled = false; }, -}; +); diff --git a/extensions/discord/src/components-registry.test-support.ts b/extensions/discord/src/components-registry.test-support.ts index 159149852d77..1aed3f1ab2df 100644 --- a/extensions/discord/src/components-registry.test-support.ts +++ b/extensions/discord/src/components-registry.test-support.ts @@ -1,5 +1,9 @@ import { discordComponentRegistryState } from "./components-registry-state.js"; export function clearDiscordComponentEntriesForTest(): void { - discordComponentRegistryState.reset(); + discordComponentRegistryState.componentEntries.clear(); + discordComponentRegistryState.modalEntries.clear(); + discordComponentRegistryState.persistentComponentStore = undefined; + discordComponentRegistryState.persistentModalStore = undefined; + discordComponentRegistryState.persistentRegistryDisabled = false; } diff --git a/extensions/discord/src/components.test.ts b/extensions/discord/src/components.test.ts index 8433d40e2df1..70edfaf7c765 100644 --- a/extensions/discord/src/components.test.ts +++ b/extensions/discord/src/components.test.ts @@ -330,6 +330,10 @@ describe("discord component registry", () => { }); const componentsRegistryModuleUrl = new URL("./components-registry.ts", import.meta.url).href; + const componentsRegistryStateModuleUrl = new URL( + "./components-registry-state.ts", + import.meta.url, + ).href; it("registers and consumes component entries", async () => { registerDiscordComponentEntries({ @@ -426,6 +430,21 @@ describe("discord component registry", () => { clearDiscordComponentEntriesForTest(); }); + it("shares persistent registry state across duplicate state modules", async () => { + const first = (await import( + `${componentsRegistryStateModuleUrl}?t=first-${Date.now()}` + )) as typeof import("./components-registry-state.js"); + const second = (await import( + `${componentsRegistryStateModuleUrl}?t=second-${Date.now()}` + )) as typeof import("./components-registry-state.js"); + + first.discordComponentRegistryState.persistentRegistryDisabled = true; + + expect(second.discordComponentRegistryState).toBe(first.discordComponentRegistryState); + expect(second.discordComponentRegistryState.persistentRegistryDisabled).toBe(true); + clearDiscordComponentEntriesForTest(); + }); + it("expires component entries registered while the process clock is invalid", async () => { const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(Number.NaN); try { diff --git a/extensions/discord/src/monitor/status.ts b/extensions/discord/src/monitor/status.ts index 301963b6d0a7..bea38ee0a547 100644 --- a/extensions/discord/src/monitor/status.ts +++ b/extensions/discord/src/monitor/status.ts @@ -1,5 +1,5 @@ // Discord plugin module implements status behavior. -import { createConnectedChannelStatusPatch } from "openclaw/plugin-sdk/gateway-runtime"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; type DiscordMonitorStatusPatch = { connected?: boolean; @@ -28,11 +28,9 @@ export type DiscordMonitorStatusSink = (patch: DiscordMonitorStatusPatch) => voi /** READY proves a prior terminal failure was repaired, so the account is restartable again. */ export function createDiscordReadyStatusPatch(at: number = Date.now()) { - return { - ...createConnectedChannelStatusPatch(at), - lifecycle: "ready" as const, - terminalDisconnect: undefined, + return channelReadyPatch({ + lastConnectedAt: at, + lastEventAt: at, lastDisconnect: null, - lastError: null, - }; + }); } diff --git a/extensions/feishu/src/monitor.transport.ts b/extensions/feishu/src/monitor.transport.ts index b971cf41f98f..15ea186b600a 100644 --- a/extensions/feishu/src/monitor.transport.ts +++ b/extensions/feishu/src/monitor.transport.ts @@ -2,6 +2,7 @@ import crypto from "node:crypto"; import * as http from "node:http"; import * as Lark from "@larksuiteoapi/node-sdk"; +import { channelBlockedPatch, channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; import { waitForAbortableDelay } from "./async.js"; import { createFeishuWSClient } from "./client.js"; @@ -253,14 +254,12 @@ export async function monitorWebSocket({ }; const publishWsConnected = () => { const connectedAt = Date.now(); - statusSink?.({ - connected: true, - lifecycle: "ready", - terminalDisconnect: undefined, - lastConnectedAt: connectedAt, - lastEventAt: connectedAt, - lastError: null, - }); + statusSink?.( + channelReadyPatch({ + lastConnectedAt: connectedAt, + lastEventAt: connectedAt, + }), + ); }; const publishWsReconnecting = () => { const reconnectingAt = Date.now(); @@ -303,13 +302,12 @@ export async function monitorWebSocket({ // WS cycle ended via terminal error (not abort) — publish disconnected // so the health monitor can flag the channel before the next reconnect. const disconnectedAt = Date.now(); - statusSink?.({ - connected: false, - lifecycle: "blocked", - terminalDisconnect: true, - lastEventAt: disconnectedAt, - lastError: formatFeishuWsErrorForLog(cycleEnd), - }); + statusSink?.( + channelBlockedPatch(formatFeishuWsErrorForLog(cycleEnd), { + connected: false, + lastEventAt: disconnectedAt, + }), + ); attempt += 1; const delayMs = getFeishuWsReconnectDelayMs(attempt); @@ -525,14 +523,12 @@ export async function monitorWebhook({ // this, the gateway health monitor has no transport signal for webhook // mode and will not detect a server crash. See PROPOSAL.md. const webhookConnectedAt = Date.now(); - statusSink?.({ - connected: true, - lifecycle: "ready", - terminalDisconnect: undefined, - lastConnectedAt: webhookConnectedAt, - lastEventAt: webhookConnectedAt, - lastError: null, - }); + statusSink?.( + channelReadyPatch({ + lastConnectedAt: webhookConnectedAt, + lastEventAt: webhookConnectedAt, + }), + ); }); server.on("error", (err) => { diff --git a/extensions/googlechat/src/gateway.ts b/extensions/googlechat/src/gateway.ts index a7e2c25dda4f..305cda6d8b35 100644 --- a/extensions/googlechat/src/gateway.ts +++ b/extensions/googlechat/src/gateway.ts @@ -7,6 +7,7 @@ import { } from "openclaw/plugin-sdk/channel-outbound"; import { registerChannelRuntimeContext } from "openclaw/plugin-sdk/channel-runtime-context"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { channelBlockedPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { createLazyRuntimeNamedExport } from "openclaw/plugin-sdk/lazy-runtime"; import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/status-helpers"; import type { ResolvedGoogleChatAccount } from "./accounts.js"; @@ -52,11 +53,9 @@ export async function startGoogleChatGatewayAccount(ctx: { lastStartAt: Date.now(), ...(webhookPath ? { webhookPath, lifecycle: "starting" as const } - : { + : channelBlockedPatch(UNRESOLVED_WEBHOOK_URL_ERROR, { webhookPath: undefined, - lifecycle: "blocked" as const, - lastError: UNRESOLVED_WEBHOOK_URL_ERROR, - }), + })), audienceType: account.config.audienceType, audience: account.config.audience, }); diff --git a/extensions/googlechat/src/monitor.lifecycle.test.ts b/extensions/googlechat/src/monitor.lifecycle.test.ts index 4b86f599564c..4e958dbd870c 100644 --- a/extensions/googlechat/src/monitor.lifecycle.test.ts +++ b/extensions/googlechat/src/monitor.lifecycle.test.ts @@ -47,6 +47,7 @@ describe("Google Chat monitor lifecycle", () => { expect(mocks.registerTarget).toHaveBeenCalledOnce(); expect(statusSink).toHaveBeenCalledWith({ + running: true, connected: true, lifecycle: "ready", lastConnectedAt: expect.any(Number), diff --git a/extensions/googlechat/src/monitor.ts b/extensions/googlechat/src/monitor.ts index 061e16945ad9..d4e6cda9ddb0 100644 --- a/extensions/googlechat/src/monitor.ts +++ b/extensions/googlechat/src/monitor.ts @@ -6,6 +6,7 @@ import { type ChannelBotLoopProtectionFacts, type ChannelInboundMediaInput, } from "openclaw/plugin-sdk/channel-inbound"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { mergePairLoopGuardConfig } from "openclaw/plugin-sdk/pair-loop-guard-runtime"; import { normalizeOptionalLowercaseString } from "openclaw/plugin-sdk/string-coerce-runtime"; import type { OpenClawConfig } from "../runtime-api.js"; @@ -529,13 +530,7 @@ async function monitorGoogleChatProvider( let unregisterTarget: (() => void) | undefined; try { unregisterTarget = registerGoogleChatWebhookTarget(target); - options.statusSink?.({ - connected: true, - lifecycle: "ready", - lastConnectedAt: Date.now(), - lastError: null, - terminalDisconnect: undefined, - }); + options.statusSink?.(channelReadyPatch()); } catch (error) { await ingress.stop(); throw error; diff --git a/extensions/imessage/src/monitor/monitor-provider.ts b/extensions/imessage/src/monitor/monitor-provider.ts index cd5cd5b2aa60..eeb79cbf3a3f 100644 --- a/extensions/imessage/src/monitor/monitor-provider.ts +++ b/extensions/imessage/src/monitor/monitor-provider.ts @@ -25,6 +25,7 @@ import { upsertChannelPairingRequest, } from "openclaw/plugin-sdk/conversation-runtime"; import { expectDefined } from "openclaw/plugin-sdk/expect-runtime"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { normalizeScpRemoteHost } from "openclaw/plugin-sdk/host-runtime"; import { isInboundPathAllowed, kindFromMime } from "openclaw/plugin-sdk/media-runtime"; import { DEFAULT_GROUP_HISTORY_LIMIT, type HistoryEntry } from "openclaw/plugin-sdk/reply-history"; @@ -1552,13 +1553,7 @@ export async function monitorIMessageProvider(opts: MonitorIMessageOpts = {}): P { timeoutMs: probeTimeoutMs }, ); attemptSubscriptionId = result?.subscription ?? null; - opts.statusSink?.({ - connected: true, - lifecycle: "ready", - lastConnectedAt: Date.now(), - lastError: null, - terminalDisconnect: undefined, - }); + opts.statusSink?.(channelReadyPatch()); client = attemptClient; detachAbortHandler = attemptDetachAbortHandler; keepAttemptClient = true; diff --git a/extensions/irc/src/monitor.test.ts b/extensions/irc/src/monitor.test.ts index 607fc161a51d..84b929b02aeb 100644 --- a/extensions/irc/src/monitor.test.ts +++ b/extensions/irc/src/monitor.test.ts @@ -369,6 +369,17 @@ describe("irc monitor reconnect", () => { patch.lifecycle ? [patch.lifecycle as string] : [], ), ).toEqual(["ready", "recovering", "recovering", "ready"]); + for (const [readyPatch] of statusSink.mock.calls.filter( + ([statusPatch]) => statusPatch.lifecycle === "ready", + )) { + expect(readyPatch).toMatchObject({ + running: true, + connected: true, + lastConnectedAt: expect.any(Number), + lastError: null, + terminalDisconnect: undefined, + }); + } } finally { if (monitor) { await monitor.stop(); diff --git a/extensions/irc/src/monitor.ts b/extensions/irc/src/monitor.ts index 12159e431cd0..0794b339e888 100644 --- a/extensions/irc/src/monitor.ts +++ b/extensions/irc/src/monitor.ts @@ -1,5 +1,6 @@ // Irc plugin module implements monitor behavior. import { resolveLoggerBackedRuntime } from "openclaw/plugin-sdk/extension-shared"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/status-helpers"; import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime"; import { resolveIrcAccount } from "./accounts.js"; @@ -20,11 +21,7 @@ type IrcMonitorOptions = { config?: CoreConfig; runtime?: RuntimeEnv; abortSignal?: AbortSignal; - statusSink?: (patch: { - lastInboundAt?: number; - lastOutboundAt?: number; - lifecycle?: ChannelAccountSnapshot["lifecycle"]; - }) => void; + statusSink?: (patch: Omit) => void; onMessage?: (message: IrcInboundMessage, client: IrcClient) => void | Promise; ingressQueue?: NonNullable[0]["queue"]>; }; @@ -221,7 +218,7 @@ export async function monitorIrcProvider( return; } ingress.start(); - opts.statusSink?.({ lifecycle: "ready" }); + opts.statusSink?.(channelReadyPatch()); logger.info( `[${account.accountId}] connected to ${account.host}:${account.port}${account.tls ? " (tls)" : ""} as ${nextClient.nick}`, diff --git a/extensions/line/src/monitor.ts b/extensions/line/src/monitor.ts index 9b884879fe02..b41afbd76c43 100644 --- a/extensions/line/src/monitor.ts +++ b/extensions/line/src/monitor.ts @@ -4,6 +4,7 @@ import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contrac import { hasFinalInboundReplyDispatch } from "openclaw/plugin-sdk/channel-inbound"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import { channelReadyPatch, channelStoppedPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { chunkMarkdownText } from "openclaw/plugin-sdk/reply-runtime"; import { danger, @@ -431,13 +432,7 @@ export async function monitorLineProvider( const { unregister: unregisterHttp } = await registerLineWebhookTarget(registrationParams, bot); logVerbose(`line: registered webhook handler at ${normalizedPath}`); - statusSink?.({ - connected: true, - lifecycle: "ready", - lastConnectedAt: Date.now(), - lastError: null, - terminalDisconnect: undefined, - }); + statusSink?.(channelReadyPatch()); let stopped = false; let stopPromise: Promise | undefined; @@ -452,7 +447,7 @@ export async function monitorLineProvider( logVerbose(`line: stopping provider for account ${resolvedAccountId}`); unregisterHttp(); stopPromise = bot.stop().finally(() => { - statusSink?.({ running: false, connected: false, lifecycle: "stopped" }); + statusSink?.(channelStoppedPatch()); }); return stopPromise; }; diff --git a/extensions/matrix/src/matrix/monitor/status.ts b/extensions/matrix/src/matrix/monitor/status.ts index af1d3d144aa4..b8e9641e4318 100644 --- a/extensions/matrix/src/matrix/monitor/status.ts +++ b/extensions/matrix/src/matrix/monitor/status.ts @@ -2,7 +2,9 @@ import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contract"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { - createConnectedChannelStatusPatch, + channelBlockedPatch, + channelReadyPatch, + channelStoppedPatch, createTransportActivityStatusPatch, } from "openclaw/plugin-sdk/gateway-runtime"; import { isMatrixAccessTokenInvalidatedError } from "../sdk/client-support.js"; @@ -59,19 +61,19 @@ export function createMatrixMonitorStatusController(params: { }; const noteConnected = (at = Date.now(), options?: { transportActivity?: boolean }) => { - if (status.connected === true) { - status.lastEventAt = at; - } else { - Object.assign(status, createConnectedChannelStatusPatch(at)); - } + const lastConnectedAt = status.connected === true ? (status.lastConnectedAt ?? at) : at; + Object.assign( + status, + channelReadyPatch({ + lastConnectedAt, + lastEventAt: at, + lastDisconnect: null, + healthState: "healthy", + }), + ); if (options?.transportActivity) { Object.assign(status, createTransportActivityStatusPatch(at)); } - status.lastError = null; - status.lastDisconnect = null; - status.healthState = "healthy"; - status.lifecycle = "ready"; - status.terminalDisconnect = undefined; emit(); }; @@ -97,10 +99,14 @@ export function createMatrixMonitorStatusController(params: { at, ...(error ? { error } : {}), }; - status.lastError = error; status.healthState = paramsLocal.state.toLowerCase(); - status.lifecycle = tokenInvalidated ? "blocked" : "recovering"; - status.terminalDisconnect = tokenInvalidated || undefined; + if (tokenInvalidated) { + Object.assign(status, channelBlockedPatch(error ?? "Matrix access token invalidated")); + } else { + status.lastError = error; + status.lifecycle = "recovering"; + status.terminalDisconnect = undefined; + } emit(); }; @@ -129,11 +135,11 @@ export function createMatrixMonitorStatusController(params: { noteDisconnected({ state: "ERROR", at, error }); }, markStopped(at = Date.now()) { - status.connected = false; - status.lastEventAt = at; if (status.lifecycle !== "blocked" && status.healthState !== "error") { - status.healthState = "stopped"; - status.lifecycle = "stopped"; + Object.assign(status, channelStoppedPatch({ lastEventAt: at, healthState: "stopped" })); + } else { + status.connected = false; + status.lastEventAt = at; } emit(); }, diff --git a/extensions/mattermost/src/mattermost/monitor-websocket.test.ts b/extensions/mattermost/src/mattermost/monitor-websocket.test.ts index 80473fee4e0b..2e9275ba270a 100644 --- a/extensions/mattermost/src/mattermost/monitor-websocket.test.ts +++ b/extensions/mattermost/src/mattermost/monitor-websocket.test.ts @@ -249,6 +249,7 @@ describe("mattermost websocket monitor", () => { socket.emitMessage(Buffer.from(JSON.stringify({ status: "OK", seq_reply: 7 }))); expect(patches).toContainEqual({ + running: true, connected: true, lifecycle: "ready", lastConnectedAt: expect.any(Number), diff --git a/extensions/mattermost/src/mattermost/monitor-websocket.ts b/extensions/mattermost/src/mattermost/monitor-websocket.ts index f8348b701962..ac0dd86fd9be 100644 --- a/extensions/mattermost/src/mattermost/monitor-websocket.ts +++ b/extensions/mattermost/src/mattermost/monitor-websocket.ts @@ -1,6 +1,7 @@ // Mattermost plugin module implements monitor websocket behavior. import { randomUUID } from "node:crypto"; import { safeParseJsonWithSchema, safeParseWithSchema } from "openclaw/plugin-sdk/extension-shared"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { captureWsEvent, createDebugProxyWebSocketAgent, @@ -350,13 +351,7 @@ export function createMattermostConnectOnce( } if (payload.status === "OK" && payload.seq_reply === authenticationSeq) { - opts.statusSink?.({ - connected: true, - lifecycle: "ready", - lastConnectedAt: Date.now(), - lastError: null, - terminalDisconnect: undefined, - }); + opts.statusSink?.(channelReadyPatch()); return; } diff --git a/extensions/msteams/src/monitor-status.ts b/extensions/msteams/src/monitor-status.ts index dcbdbedd8091..bc5315f05efe 100644 --- a/extensions/msteams/src/monitor-status.ts +++ b/extensions/msteams/src/monitor-status.ts @@ -1,4 +1,9 @@ import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contract"; +import { + channelBlockedPatch, + channelReadyPatch, + channelStoppedPatch, +} from "openclaw/plugin-sdk/gateway-runtime"; export type MSTeamsStatusSink = (patch: Omit) => void; @@ -6,23 +11,11 @@ export function publishMSTeamsBlocked( statusSink: MSTeamsStatusSink | undefined, lastError: string, ) { - statusSink?.({ - running: true, - lifecycle: "blocked", - terminalDisconnect: true, - lastError, - }); + statusSink?.(channelBlockedPatch(lastError, { running: true })); } export function publishMSTeamsReady(statusSink: MSTeamsStatusSink | undefined, now = Date.now()) { - statusSink?.({ - running: true, - connected: true, - lifecycle: "ready", - lastConnectedAt: now, - lastError: null, - terminalDisconnect: undefined, - }); + statusSink?.(channelReadyPatch({ lastConnectedAt: now })); } export function publishMSTeamsRecovering( @@ -33,5 +26,5 @@ export function publishMSTeamsRecovering( } export function publishMSTeamsStopped(statusSink: MSTeamsStatusSink | undefined) { - statusSink?.({ running: false, connected: false, lifecycle: "stopped" }); + statusSink?.(channelStoppedPatch()); } diff --git a/extensions/msteams/src/pending-uploads.lifecycle.test.ts b/extensions/msteams/src/pending-uploads.lifecycle.test.ts new file mode 100644 index 000000000000..93c6d9835855 --- /dev/null +++ b/extensions/msteams/src/pending-uploads.lifecycle.test.ts @@ -0,0 +1,40 @@ +// Msteams tests cover lifecycle cleanup for pending upload timers. +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const singleton = vi.hoisted(() => ({ reset: undefined as (() => void) | undefined })); + +vi.mock("openclaw/plugin-sdk/global-singleton", () => ({ + resolveGlobalSingleton: (_key: symbol, create: () => T, reset?: (value: T) => void): T => { + const value = create(); + singleton.reset = () => reset?.(value); + return value; + }, +})); + +import { getPendingUpload, storePendingUpload } from "./pending-uploads.js"; + +describe("pending upload lifecycle reset", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + singleton.reset?.(); + vi.useRealTimers(); + }); + + it("cancels every TTL timer before clearing pending uploads", () => { + const id = storePendingUpload({ + buffer: Buffer.from("data"), + filename: "file.txt", + conversationId: "conv-1", + }); + expect(getPendingUpload(id)).toBeDefined(); + expect(vi.getTimerCount()).toBe(1); + + singleton.reset?.(); + + expect(getPendingUpload(id)).toBeUndefined(); + expect(vi.getTimerCount()).toBe(0); + }); +}); diff --git a/extensions/msteams/src/pending-uploads.ts b/extensions/msteams/src/pending-uploads.ts index f6aeaae65f65..ed7cc1e99533 100644 --- a/extensions/msteams/src/pending-uploads.ts +++ b/extensions/msteams/src/pending-uploads.ts @@ -7,6 +7,7 @@ */ import crypto from "node:crypto"; +import { resolveGlobalSingleton } from "openclaw/plugin-sdk/global-singleton"; interface PendingUpload { id: string; @@ -19,9 +20,21 @@ interface PendingUpload { createdAt: number; } -const pendingUploads = new Map(); -/** Timer handles keyed by upload ID, cleared on explicit removal to prevent ghost cleanup */ -const pendingUploadTimers = new Map>(); +const { pendingUploads, pendingUploadTimers } = resolveGlobalSingleton( + Symbol.for("openclaw.msteams.pendingUploadState"), + () => ({ + pendingUploads: new Map(), + /** Timer handles keyed by upload ID, cleared on explicit removal to prevent ghost cleanup. */ + pendingUploadTimers: new Map>(), + }), + (state) => { + for (const timer of state.pendingUploadTimers.values()) { + clearTimeout(timer); + } + state.pendingUploadTimers.clear(); + state.pendingUploads.clear(); + }, +); /** TTL for pending uploads: 5 minutes */ const PENDING_UPLOAD_TTL_MS = 5 * 60 * 1000; diff --git a/extensions/nextcloud-talk/src/monitor-runtime.abort.test.ts b/extensions/nextcloud-talk/src/monitor-runtime.abort.test.ts index f8795ad00c36..2ba37d7a26ab 100644 --- a/extensions/nextcloud-talk/src/monitor-runtime.abort.test.ts +++ b/extensions/nextcloud-talk/src/monitor-runtime.abort.test.ts @@ -42,7 +42,14 @@ describe("Nextcloud Talk monitor abort", () => { expect(createSpool).toHaveBeenCalledWith( expect.objectContaining({ abortSignal: abortController.signal }), ); - expect(statusSink).toHaveBeenCalledExactlyOnceWith({ lifecycle: "ready" }); + expect(statusSink).toHaveBeenCalledExactlyOnceWith({ + running: true, + connected: true, + lifecycle: "ready", + lastConnectedAt: expect.any(Number), + lastError: null, + terminalDisconnect: undefined, + }); abortController.abort(); await vi.waitFor(() => expect(spoolStop).toHaveBeenCalledOnce()); await monitor.stop(); diff --git a/extensions/nextcloud-talk/src/monitor-runtime.ts b/extensions/nextcloud-talk/src/monitor-runtime.ts index 7f66dee7252f..d0cfa5f6efb6 100644 --- a/extensions/nextcloud-talk/src/monitor-runtime.ts +++ b/extensions/nextcloud-talk/src/monitor-runtime.ts @@ -1,5 +1,6 @@ // Nextcloud Talk plugin module implements monitor runtime behavior. import { resolveLoggerBackedRuntime } from "openclaw/plugin-sdk/extension-shared"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime"; import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/status-helpers"; import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime"; @@ -34,11 +35,7 @@ type NextcloudTalkMonitorOptions = { message: NextcloudTalkInboundMessage, lifecycle: NextcloudTalkIngressLifecycle, ) => void | Promise; - statusSink?: (patch: { - lastInboundAt?: number; - lastOutboundAt?: number; - lifecycle?: ChannelAccountSnapshot["lifecycle"]; - }) => void; + statusSink?: (patch: Omit) => void; createSpool?: typeof createNextcloudTalkWebhookSpool; createServer?: typeof createNextcloudTalkWebhookServer; }; @@ -143,7 +140,7 @@ export async function monitorNextcloudTalkProvider( await stop(); return { stop }; } - opts.statusSink?.({ lifecycle: "ready" }); + opts.statusSink?.(channelReadyPatch()); const publicUrl = account.config.webhookPublicUrl ?? diff --git a/extensions/nostr/src/channel.lifecycle.test.ts b/extensions/nostr/src/channel.lifecycle.test.ts index 5672b4dd754b..82f662290620 100644 --- a/extensions/nostr/src/channel.lifecycle.test.ts +++ b/extensions/nostr/src/channel.lifecycle.test.ts @@ -133,8 +133,11 @@ describe("nostr gateway lifecycle", () => { options?.onConnect?.("wss://relay-one.example/"); expect(statusEvents.at(-1)).toMatchObject({ + running: true, lifecycle: "ready", connected: true, + lastConnectedAt: expect.any(Number), + lastError: null, terminalDisconnect: undefined, }); options?.onConnect?.("wss://relay-two.example/"); diff --git a/extensions/nostr/src/gateway.ts b/extensions/nostr/src/gateway.ts index cb230ae5d809..5c8918434433 100644 --- a/extensions/nostr/src/gateway.ts +++ b/extensions/nostr/src/gateway.ts @@ -10,6 +10,7 @@ import { import { createChannelPairingController } from "openclaw/plugin-sdk/channel-pairing"; import { attachChannelToResult } from "openclaw/plugin-sdk/channel-send-result"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { chunkTextForOutbound, sanitizeAssistantVisibleText, @@ -242,14 +243,7 @@ export const startNostrGatewayAccount: NostrGatewayStart = async (ctx) => { connectedRelays.add(normalizeRelayLifecycleKey(relay)); // Treat >=1 connected relay as ready. This favors partial availability over quorum // fidelity; circuit-breaker health stays private to nostr-bus, so ready is not all-relays. - ctx.setStatus({ - accountId: account.accountId, - connected: true, - lifecycle: "ready", - lastConnectedAt: Date.now(), - lastError: null, - terminalDisconnect: undefined, - }); + ctx.setStatus(channelReadyPatch({ accountId: account.accountId })); ctx.log?.debug?.(`[${account.accountId}] Connected to relay: ${relay}`); }, onDisconnect: (relay) => { diff --git a/extensions/qa-channel/src/gateway.test.ts b/extensions/qa-channel/src/gateway.test.ts index 88f1b776801b..f3aeca4ea792 100644 --- a/extensions/qa-channel/src/gateway.test.ts +++ b/extensions/qa-channel/src/gateway.test.ts @@ -209,6 +209,7 @@ describe("qa-channel gateway", () => { await vi.waitFor(() => expect(setStatus).toHaveBeenCalledWith({ accountId: "default", + running: true, connected: true, lifecycle: "ready", lastConnectedAt: expect.any(Number), diff --git a/extensions/qa-channel/src/gateway.ts b/extensions/qa-channel/src/gateway.ts index 676ce6212a0e..cc0f9dac3856 100644 --- a/extensions/qa-channel/src/gateway.ts +++ b/extensions/qa-channel/src/gateway.ts @@ -1,4 +1,5 @@ // Qa Channel plugin module implements gateway behavior. +import { channelReadyPatch, channelStoppedPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { pollQaBus } from "./bus-client.js"; import { handleQaInbound } from "./inbound.js"; import type { ChannelGatewayContext } from "./runtime-api.js"; @@ -62,14 +63,7 @@ export async function startQaGatewayAccount( }); if (!ready) { ready = true; - ctx.setStatus({ - accountId: account.accountId, - connected: true, - lifecycle: "ready", - lastConnectedAt: Date.now(), - lastError: null, - terminalDisconnect: undefined, - }); + ctx.setStatus(channelReadyPatch({ accountId: account.accountId })); } cursor = result.cursor; for (const event of result.events) { @@ -98,12 +92,7 @@ export async function startQaGatewayAccount( } } finally { await Promise.all([queuedInbound, ...controlTasks]); - ctx.setStatus({ - accountId: account.accountId, - running: false, - connected: false, - lifecycle: "stopped", - }); + ctx.setStatus(channelStoppedPatch({ accountId: account.accountId })); } if (inboundError) { throw inboundError; diff --git a/extensions/qqbot/src/channel.gateway-status.test.ts b/extensions/qqbot/src/channel.gateway-status.test.ts index 047531bf299a..07dcbd3e95fd 100644 --- a/extensions/qqbot/src/channel.gateway-status.test.ts +++ b/extensions/qqbot/src/channel.gateway-status.test.ts @@ -119,6 +119,7 @@ describe("qqbot channel gateway status", () => { expect(getStatus().connected).toBe(true); expect(getStatus().lastError).toBeNull(); expect(getStatus().lifecycle).toBe("ready"); + expect(getStatus().terminalDisconnect).toBeUndefined(); options.onDisconnected?.({ reason: "offline/sandbox-only", fatal: true }); options.onReady?.({}); @@ -126,5 +127,6 @@ describe("qqbot channel gateway status", () => { expect(getStatus().connected).toBe(true); expect(getStatus().lastError).toBeNull(); expect(getStatus().lifecycle).toBe("ready"); + expect(getStatus().terminalDisconnect).toBeUndefined(); }); }); diff --git a/extensions/qqbot/src/channel.ts b/extensions/qqbot/src/channel.ts index 6a388fb52417..35a71771fe44 100644 --- a/extensions/qqbot/src/channel.ts +++ b/extensions/qqbot/src/channel.ts @@ -9,6 +9,7 @@ import { } from "openclaw/plugin-sdk/channel-outbound"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import type { ChannelPlugin } from "openclaw/plugin-sdk/core"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime"; // Register the PlatformAdapter before any core/ module is used. import "./bridge/bootstrap.js"; @@ -381,28 +382,14 @@ export const qqbotPlugin: ChannelPlugin = { channelRuntime: ctx.channelRuntime as GatewayContext["channelRuntime"], onReady: () => { log?.info(`[qqbot:${account.accountId}] Gateway ready`); - ctx.setStatus({ - ...ctx.getStatus(), - running: true, - connected: true, - lastConnectedAt: Date.now(), - lastError: null, - lifecycle: "ready", - }); + ctx.setStatus(channelReadyPatch({ accountId: account.accountId })); // Snapshot credentials so we can recover from the next hot // upgrade that might wipe openclaw.json mid-flight. persistAccountCredentialSnapshot(account); }, onResumed: () => { log?.info(`[qqbot:${account.accountId}] Gateway resumed`); - ctx.setStatus({ - ...ctx.getStatus(), - running: true, - connected: true, - lastConnectedAt: Date.now(), - lastError: null, - lifecycle: "ready", - }); + ctx.setStatus(channelReadyPatch({ accountId: account.accountId })); persistAccountCredentialSnapshot(account); }, onError: (error) => { diff --git a/extensions/raft/src/gateway.test.ts b/extensions/raft/src/gateway.test.ts index 1af26d57b64b..904df00726ff 100644 --- a/extensions/raft/src/gateway.test.ts +++ b/extensions/raft/src/gateway.test.ts @@ -198,7 +198,14 @@ describe("Raft wake gateway", () => { const wakeEndpoint = await waitFor(() => endpoint); const bridgeToken = await waitFor(() => token); - expect(ctx.getStatus()).toMatchObject({ lifecycle: "ready" }); + expect(ctx.getStatus()).toMatchObject({ + running: true, + connected: true, + lifecycle: "ready", + lastConnectedAt: expect.any(Number), + lastError: null, + terminalDisconnect: undefined, + }); await expect(fetch(wakeEndpoint.replace("/wake", "/health"))).resolves.toMatchObject({ status: 200, }); diff --git a/extensions/raft/src/gateway.ts b/extensions/raft/src/gateway.ts index 4ebada72f945..1243d7919715 100644 --- a/extensions/raft/src/gateway.ts +++ b/extensions/raft/src/gateway.ts @@ -6,6 +6,7 @@ import { createServer, type IncomingMessage, type Server, type ServerResponse } import type { Socket } from "node:net"; import type { ChannelGatewayContext } from "openclaw/plugin-sdk/channel-contract"; import { keepHttpServerTaskAlive, waitUntilAbort } from "openclaw/plugin-sdk/channel-outbound"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue"; import { createChannelReplayGuard } from "openclaw/plugin-sdk/persistent-dedupe"; import { safeEqualSecret } from "openclaw/plugin-sdk/security-runtime"; @@ -377,14 +378,12 @@ export async function startRaftGatewayAccount( } }); - ctx.setStatus({ - accountId: ctx.accountId, - running: true, - lifecycle: "ready", - connected: true, - lastStartAt: Date.now(), - lastError: null, - }); + ctx.setStatus( + channelReadyPatch({ + accountId: ctx.accountId, + lastStartAt: Date.now(), + }), + ); ctx.log?.info?.(`Raft bridge started for profile "${profile}".`); await keepHttpServerTaskAlive({ diff --git a/extensions/reef/src/channel.ts b/extensions/reef/src/channel.ts index fef91a2dfe45..89e4dfb10b87 100644 --- a/extensions/reef/src/channel.ts +++ b/extensions/reef/src/channel.ts @@ -10,6 +10,7 @@ import { type ChannelPlugin, } from "openclaw/plugin-sdk/core"; import { createChannelDirectoryAdapter } from "openclaw/plugin-sdk/directory-runtime"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { runReefChannelLifecycle } from "./channel-lifecycle.js"; import { ReefChannelConfigSchema, @@ -438,14 +439,7 @@ export const reefPlugin: ChannelPlugin = { } ctx.setStatus( state === "connected" - ? { - accountId: "default", - running: true, - connected: true, - lifecycle: "ready", - lastConnectedAt: Date.now(), - lastError: null, - } + ? channelReadyPatch({ accountId: "default" }) : { accountId: "default", running: true, diff --git a/extensions/signal/src/sse-reconnect.ts b/extensions/signal/src/sse-reconnect.ts index 89b7a4fcb300..37179d0ca239 100644 --- a/extensions/signal/src/sse-reconnect.ts +++ b/extensions/signal/src/sse-reconnect.ts @@ -1,5 +1,6 @@ // Signal plugin module implements sse reconnect behavior. import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contract"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { computeBackoff, logVerbose, @@ -82,13 +83,7 @@ export async function runSignalSseLoop({ timeoutMs, transportKind, onStreamOpen: () => { - statusSink?.({ - connected: true, - lifecycle: "ready", - lastConnectedAt: Date.now(), - lastError: null, - terminalDisconnect: undefined, - }); + statusSink?.(channelReadyPatch()); }, onEvent: async (event: SignalSseEvent) => { reconnectAttempts = 0; diff --git a/extensions/slack/src/monitor/provider-support.ts b/extensions/slack/src/monitor/provider-support.ts index c2ba75e4d0aa..3371ca04f7cc 100644 --- a/extensions/slack/src/monitor/provider-support.ts +++ b/extensions/slack/src/monitor/provider-support.ts @@ -1,5 +1,6 @@ // Slack provider module implements model/runtime integration. import { toErrorObject } from "openclaw/plugin-sdk/error-runtime"; +import { channelBlockedPatch, channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { asOptionalRecord as asRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; import type { SlackChannelResolution } from "../resolve-channels.js"; import type { SlackUserResolution } from "../resolve-users.js"; @@ -193,12 +194,12 @@ export function publishSlackConnectedStatus( if (!setStatus) { return; } - setStatus({ - connected: true, - lastConnectedAt: Date.now(), - terminalDisconnect: undefined, - ...identityHealth, - }); + const lastConnectedAt = Date.now(); + setStatus( + identityHealth.lifecycle === "blocked" + ? channelBlockedPatch(identityHealth.lastError, { connected: true, lastConnectedAt }) + : channelReadyPatch({ lastConnectedAt }), + ); } export function publishSlackBlockedStatus( @@ -208,12 +209,11 @@ export function publishSlackBlockedStatus( if (!setStatus) { return; } - setStatus({ - connected: false, - lifecycle: "blocked", - terminalDisconnect: true, - lastError: formatUnknownError(error), - }); + setStatus( + channelBlockedPatch(formatUnknownError(error), { + connected: false, + }), + ); } export function publishSlackDisconnectedStatus( diff --git a/extensions/slack/src/monitor/provider.auth-test-token.test.ts b/extensions/slack/src/monitor/provider.auth-test-token.test.ts index 4fda13233635..3bb66e4c8e79 100644 --- a/extensions/slack/src/monitor/provider.auth-test-token.test.ts +++ b/extensions/slack/src/monitor/provider.auth-test-token.test.ts @@ -702,7 +702,9 @@ describe("connected identity health", () => { expect(setStatus).toHaveBeenCalledWith({ connected: true, lastConnectedAt: expect.any(Number), - terminalDisconnect: undefined, + ...(expected.lifecycle === "ready" + ? { running: true, terminalDisconnect: undefined } + : { terminalDisconnect: true }), ...expected, }); }); @@ -715,6 +717,7 @@ describe("connected identity health", () => { await stopSlackMonitor(monitor); expect(setStatus).toHaveBeenCalledWith({ + running: true, connected: true, lastConnectedAt: expect.any(Number), terminalDisconnect: undefined, @@ -749,7 +752,7 @@ describe("connected identity health", () => { expect(setStatus).toHaveBeenCalledWith({ connected: true, lastConnectedAt: expect.any(Number), - terminalDisconnect: undefined, + terminalDisconnect: true, lifecycle: "blocked", lastError: "request_timeout", }); @@ -773,6 +776,7 @@ describe("connected identity health", () => { }); expect(setStatus).toHaveBeenCalledWith({ + running: true, connected: true, lastConnectedAt: expect.any(Number), terminalDisconnect: undefined, diff --git a/extensions/slack/src/monitor/provider.reconnect-loop.test.ts b/extensions/slack/src/monitor/provider.reconnect-loop.test.ts index c928e97753f5..f422c52a4415 100644 --- a/extensions/slack/src/monitor/provider.reconnect-loop.test.ts +++ b/extensions/slack/src/monitor/provider.reconnect-loop.test.ts @@ -179,6 +179,7 @@ describe("slack socket reconnect loop", () => { await Promise.resolve(); expect(setStatus).toHaveBeenCalledWith({ + running: true, connected: true, lastConnectedAt: expect.any(Number), terminalDisconnect: undefined, diff --git a/extensions/slack/src/monitor/provider.reconnect.test.ts b/extensions/slack/src/monitor/provider.reconnect.test.ts index 7f1c92c4b2ee..5849a0b7fcf0 100644 --- a/extensions/slack/src/monitor/provider.reconnect.test.ts +++ b/extensions/slack/src/monitor/provider.reconnect.test.ts @@ -64,9 +64,11 @@ describe("slack socket reconnect helpers", () => { expect(setStatus).toHaveBeenCalledTimes(1); const status = statusCallAt(setStatus, 0); expect(status?.connected).toBe(true); + expect(status?.running).toBe(true); expect(status?.lastConnectedAt).toBe(1_711_406_400_000); expect(status?.lifecycle).toBe("ready"); expect(status?.lastError).toBeNull(); + expect(status?.terminalDisconnect).toBeUndefined(); expect(status).not.toHaveProperty("lastEventAt"); }); @@ -83,7 +85,7 @@ describe("slack socket reconnect helpers", () => { expect(setStatus).toHaveBeenCalledWith({ connected: true, lastConnectedAt: 1_711_406_400_500, - terminalDisconnect: undefined, + terminalDisconnect: true, lifecycle: "blocked", lastError: "auth.test returned no user_id", }); diff --git a/extensions/sms/src/gateway.ts b/extensions/sms/src/gateway.ts index 303cb29ee24a..5034e613783f 100644 --- a/extensions/sms/src/gateway.ts +++ b/extensions/sms/src/gateway.ts @@ -1,6 +1,11 @@ // Sms plugin module implements gateway behavior. import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contract"; import { waitUntilAbort } from "openclaw/plugin-sdk/channel-outbound"; +import { + channelBlockedPatch, + channelReadyPatch, + channelStoppedPatch, +} from "openclaw/plugin-sdk/gateway-runtime"; import { registerPluginHttpRoute } from "openclaw/plugin-sdk/webhook-ingress"; import { createSmsIngressSpool, type SmsIngressLog } from "./ingress-spool.js"; import type { ResolvedSmsAccount } from "./types.js"; @@ -160,7 +165,7 @@ export async function startSmsGatewayAccount(params: { params.statusSink?.({ lifecycle: "starting" }); if (!params.account.enabled) { params.log?.info?.(`SMS account ${params.account.accountId} is disabled`); - params.statusSink?.({ running: false, connected: false, lifecycle: "stopped" }); + params.statusSink?.(channelStoppedPatch()); return waitUntilAbort(params.abortSignal); } const warnings = collectSmsStartupWarnings(params.account); @@ -168,13 +173,12 @@ export async function startSmsGatewayAccount(params: { for (const warning of warnings) { params.log?.warn?.(warning); } - params.statusSink?.({ - running: true, - connected: false, - lifecycle: "blocked", - terminalDisconnect: true, - lastError: warnings.join("; "), - }); + params.statusSink?.( + channelBlockedPatch(warnings.join("; "), { + running: true, + connected: false, + }), + ); return waitUntilAbort(params.abortSignal); } for (const warning of warnings) { @@ -185,16 +189,9 @@ export async function startSmsGatewayAccount(params: { params.log?.info?.( `Registered SMS webhook route ${params.account.webhookPath} for account ${params.account.accountId}`, ); - params.statusSink?.({ - running: true, - connected: true, - lifecycle: "ready", - lastConnectedAt: Date.now(), - lastError: null, - terminalDisconnect: undefined, - }); + params.statusSink?.(channelReadyPatch()); } return registration.lifecycle.finally(() => { - params.statusSink?.({ running: false, connected: false, lifecycle: "stopped" }); + params.statusSink?.(channelStoppedPatch()); }); } diff --git a/extensions/synology-chat/src/channel.ts b/extensions/synology-chat/src/channel.ts index be8eb9f1c5e7..23ff806cd3ff 100644 --- a/extensions/synology-chat/src/channel.ts +++ b/extensions/synology-chat/src/channel.ts @@ -29,6 +29,11 @@ import { projectAccountWarningCollector, } from "openclaw/plugin-sdk/channel-policy"; import { createEmptyChannelDirectoryAdapter } from "openclaw/plugin-sdk/directory-runtime"; +import { + channelBlockedPatch, + channelReadyPatch, + channelStoppedPatch, +} from "openclaw/plugin-sdk/gateway-runtime"; import { parseStrictNonNegativeInteger } from "openclaw/plugin-sdk/number-runtime"; import { createComputedAccountStatusAdapter, @@ -396,13 +401,12 @@ function createSynologyChatPlugin(): SynologyChatPlugin { const { cfg, accountId, log, abortSignal } = ctx; const account = resolveAccount(cfg, accountId); if (!validateSynologyGatewayAccountStartup({ cfg, account, accountId, log }).ok) { - ctx.setStatus?.({ - accountId, - running: true, - lifecycle: "blocked", - terminalDisconnect: true, - lastError: "Synology Chat account failed startup validation", - }); + ctx.setStatus?.( + channelBlockedPatch("Synology Chat account failed startup validation", { + accountId, + running: true, + }), + ); return waitUntilAbort(abortSignal); } @@ -418,15 +422,7 @@ function createSynologyChatPlugin(): SynologyChatPlugin { }); log?.info?.(`Registered HTTP route: ${account.webhookPath} for Synology Chat`); - ctx.setStatus?.({ - accountId, - running: true, - connected: true, - lifecycle: "ready", - lastConnectedAt: Date.now(), - lastError: null, - terminalDisconnect: undefined, - }); + ctx.setStatus?.(channelReadyPatch({ accountId })); // Keep alive until abort signal fires. // The gateway expects a Promise that stays pending while the channel is running. @@ -434,12 +430,7 @@ function createSynologyChatPlugin(): SynologyChatPlugin { return waitUntilAbort(abortSignal, async () => { log?.info?.(`Stopping Synology Chat channel (account: ${accountId})`); await cleanup(); - ctx.setStatus?.({ - accountId, - running: false, - connected: false, - lifecycle: "stopped", - }); + ctx.setStatus?.(channelStoppedPatch({ accountId })); }); }, diff --git a/extensions/telegram/src/channel.ts b/extensions/telegram/src/channel.ts index adce6a7519a8..f0cdfca63717 100644 --- a/extensions/telegram/src/channel.ts +++ b/extensions/telegram/src/channel.ts @@ -27,6 +27,7 @@ import { import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { createChannelDirectoryAdapter } from "openclaw/plugin-sdk/directory-runtime"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import { channelBlockedPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime"; import type { RoutePeer } from "openclaw/plugin-sdk/routing"; import { @@ -1133,11 +1134,7 @@ export const telegramPlugin = createChatChannelPlugin({ } if (unauthorizedTokenReason) { ctx.log?.error?.(`[${account.accountId}] ${unauthorizedTokenReason}`); - setStatus({ - lifecycle: "blocked", - terminalDisconnect: true, - lastError: unauthorizedTokenReason, - }); + setStatus(channelBlockedPatch(unauthorizedTokenReason)); throw new Error(unauthorizedTokenReason); } ctx.log?.info(`[${account.accountId}] starting provider${telegramBotLabel}`); diff --git a/extensions/telegram/src/polling-status.test.ts b/extensions/telegram/src/polling-status.test.ts index f5e92588a746..47a7317c6725 100644 --- a/extensions/telegram/src/polling-status.test.ts +++ b/extensions/telegram/src/polling-status.test.ts @@ -23,6 +23,7 @@ describe("createTelegramPollingStatusPublisher", () => { }); expect(setStatus).toHaveBeenNthCalledWith(2, { mode: "polling", + running: true, connected: true, lastConnectedAt: 1234, lastEventAt: 1234, diff --git a/extensions/telegram/src/polling-status.ts b/extensions/telegram/src/polling-status.ts index e243d533e180..16f233c11c2e 100644 --- a/extensions/telegram/src/polling-status.ts +++ b/extensions/telegram/src/polling-status.ts @@ -1,7 +1,7 @@ // Telegram plugin module implements polling status behavior. import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contract"; import { - createConnectedChannelStatusPatch, + channelReadyPatch, createTransportActivityStatusPatch, } from "openclaw/plugin-sdk/gateway-runtime"; @@ -19,17 +19,16 @@ export function createTelegramPollingStatusPublisher(setStatus?: TelegramPolling }); }, notePollSuccess(at = Date.now()) { - setStatus?.({ - ...createConnectedChannelStatusPatch(at), - // A successful getUpdates call proves the Telegram HTTP long-poll is alive - // even when the response has no user-visible updates. - ...createTransportActivityStatusPatch(at), - mode: "polling", - lifecycle: "ready", - // Runtime patches merge, so a repaired token must clear the prior terminal auth fact. - terminalDisconnect: undefined, - lastError: null, - }); + setStatus?.( + channelReadyPatch({ + lastConnectedAt: at, + lastEventAt: at, + // A successful getUpdates call proves the Telegram HTTP long-poll is alive + // even when the response has no user-visible updates. + ...createTransportActivityStatusPatch(at), + mode: "polling", + }), + ); }, notePollingRecovery() { setStatus?.({ lifecycle: "recovering" }); diff --git a/extensions/telegram/src/webhook-status.test.ts b/extensions/telegram/src/webhook-status.test.ts index b9bd0162e956..0805fe11e6d0 100644 --- a/extensions/telegram/src/webhook-status.test.ts +++ b/extensions/telegram/src/webhook-status.test.ts @@ -24,6 +24,7 @@ describe("createTelegramWebhookStatusPublisher", () => { }); expect(setStatus).toHaveBeenNthCalledWith(2, { mode: "webhook", + running: true, connected: true, lastConnectedAt: 1234, lastEventAt: 1234, @@ -33,6 +34,7 @@ describe("createTelegramWebhookStatusPublisher", () => { }); expect(setStatus).toHaveBeenNthCalledWith(3, { mode: "webhook", + running: true, connected: true, lastConnectedAt: 2345, lastEventAt: 2345, diff --git a/extensions/telegram/src/webhook-status.ts b/extensions/telegram/src/webhook-status.ts index 564db19e19ff..f65557361452 100644 --- a/extensions/telegram/src/webhook-status.ts +++ b/extensions/telegram/src/webhook-status.ts @@ -1,6 +1,6 @@ // Telegram plugin module implements webhook status behavior. import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contract"; -import { createConnectedChannelStatusPatch } from "openclaw/plugin-sdk/gateway-runtime"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; type TelegramWebhookStatusSink = (patch: Omit) => void; @@ -16,23 +16,22 @@ export function createTelegramWebhookStatusPublisher(setStatus?: TelegramWebhook }); }, noteWebhookAdvertised(at = Date.now()) { - setStatus?.({ - ...createConnectedChannelStatusPatch(at), - mode: "webhook", - lifecycle: "ready", - terminalDisconnect: undefined, - lastError: null, - }); + setStatus?.( + channelReadyPatch({ + lastConnectedAt: at, + lastEventAt: at, + mode: "webhook", + }), + ); }, noteWebhookUpdateReceived(at = Date.now()) { - setStatus?.({ - ...createConnectedChannelStatusPatch(at), - mode: "webhook", - lifecycle: "ready", - // Runtime patches merge, so a repaired token must clear the prior terminal auth fact. - terminalDisconnect: undefined, - lastError: null, - }); + setStatus?.( + channelReadyPatch({ + lastConnectedAt: at, + lastEventAt: at, + mode: "webhook", + }), + ); }, noteWebhookRecovery() { setStatus?.({ lifecycle: "recovering" }); diff --git a/extensions/twitch/src/twitch-client.ts b/extensions/twitch/src/twitch-client.ts index 0023d42c45bd..c41d46cbfdc1 100644 --- a/extensions/twitch/src/twitch-client.ts +++ b/extensions/twitch/src/twitch-client.ts @@ -3,6 +3,7 @@ import { RefreshingAuthProvider, StaticAuthProvider } from "@twurple/auth"; import { ChatClient, LogLevel } from "@twurple/chat"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { chunkTextForOutbound } from "openclaw/plugin-sdk/text-chunking"; import { sliceUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; import { TWITCH_CHAT_MESSAGE_LIMIT } from "./constants.js"; @@ -39,13 +40,7 @@ export class TwitchClientManager { } private publishReady(): void { - this.statusSink?.({ - connected: true, - lifecycle: "ready", - lastConnectedAt: Date.now(), - lastError: null, - terminalDisconnect: undefined, - }); + this.statusSink?.(channelReadyPatch()); } private publishRecovering(lastError: string): void { diff --git a/extensions/whatsapp/src/auto-reply/monitor-state.ts b/extensions/whatsapp/src/auto-reply/monitor-state.ts index c3acd1b84437..8494c2d9395e 100644 --- a/extensions/whatsapp/src/auto-reply/monitor-state.ts +++ b/extensions/whatsapp/src/auto-reply/monitor-state.ts @@ -1,6 +1,7 @@ // Whatsapp plugin module implements monitor state behavior. import { - createConnectedChannelStatusPatch, + channelReadyPatch, + channelStoppedPatch, createTransportActivityStatusPatch, } from "openclaw/plugin-sdk/gateway-runtime"; import type { WebChannelHealthState, WebChannelStatus } from "./types.js"; @@ -52,17 +53,14 @@ export function createWebChannelStatusController(statusSink?: (status: WebChanne emit, snapshot: () => status, noteConnected(at = Date.now()) { - Object.assign(status, createConnectedChannelStatusPatch(at)); + Object.assign(status, channelReadyPatch({ lastConnectedAt: at, lastEventAt: at })); Object.assign(status, createTransportActivityStatusPatch(at)); if (lastDisconnectWasWatchdogRecovery) { status.lastDisconnect = null; status.reconnectAttempts = 0; lastDisconnectWasWatchdogRecovery = false; } - status.lastError = null; status.healthState = "healthy"; - status.lifecycle = "ready"; - status.terminalDisconnect = undefined; emit(); }, noteInbound(at = Date.now()) { @@ -133,13 +131,17 @@ export function createWebChannelStatusController(statusSink?: (status: WebChanne emit(); }, markStopped(at = Date.now()) { - status.running = false; - status.connected = false; - status.lastEventAt = at; - status.terminalDisconnect = status.lifecycle === "blocked"; + const terminalDisconnect = status.lifecycle === "blocked"; if (!isTerminalHealthState(status.healthState)) { + Object.assign(status, channelStoppedPatch({ lastEventAt: at, terminalDisconnect })); status.healthState = "stopped"; - status.lifecycle = "stopped"; + } else { + Object.assign(status, { + running: false, + connected: false, + lastEventAt: at, + terminalDisconnect, + }); } emit(); }, diff --git a/extensions/zalo/src/monitor.lifecycle.test.ts b/extensions/zalo/src/monitor.lifecycle.test.ts index b392cc4df5c3..adab37dd4f32 100644 --- a/extensions/zalo/src/monitor.lifecycle.test.ts +++ b/extensions/zalo/src/monitor.lifecycle.test.ts @@ -142,6 +142,7 @@ describe("monitorZaloProvider lifecycle", () => { await vi.waitFor(() => expect(statusSink).toHaveBeenCalledWith({ + running: true, connected: true, lifecycle: "ready", terminalDisconnect: undefined, @@ -277,6 +278,7 @@ describe("monitorZaloProvider lifecycle", () => { await setWebhookCalled; await settleLifecycleWork(); expect(statusSink).toHaveBeenCalledWith({ + running: true, connected: true, lifecycle: "ready", terminalDisconnect: undefined, @@ -357,7 +359,7 @@ describe("monitorZaloProvider lifecycle", () => { webhookUrl: "https://example.com/hooks/zalo", }); - await expect(run).rejects.toThrow("route replacement denied"); + await expect(run).rejects.toThrow("route reuse denied"); expect(getWebhookInfoMock).not.toHaveBeenCalled(); expect(getUpdatesMock).not.toHaveBeenCalled(); @@ -365,7 +367,7 @@ describe("monitorZaloProvider lifecycle", () => { expect(statusSink).toHaveBeenCalledWith({ connected: false, lifecycle: "recovering", - lastError: expect.stringContaining("route replacement denied"), + lastError: expect.stringContaining("route reuse denied"), }); expect(statusSink).not.toHaveBeenCalledWith(expect.objectContaining({ lifecycle: "ready" })); expect(runtime.log).toHaveBeenCalledWith("[default] Zalo provider stopped mode=polling"); diff --git a/extensions/zalo/src/monitor.polling.media-reply.test-support.ts b/extensions/zalo/src/monitor.polling.media-reply.test-support.ts index 3c5925277c4f..5cc27c3b55bc 100644 --- a/extensions/zalo/src/monitor.polling.media-reply.test-support.ts +++ b/extensions/zalo/src/monitor.polling.media-reply.test-support.ts @@ -555,6 +555,7 @@ describe("Zalo polling media replies", () => { hostedMediaRoutes[0], "active Zalo hosted-media route", ); + expect(hostedMediaRoute).toBe(firstHostedMediaRoute); await writeHostedZaloMediaFixture({ id: "abc123abc123abc123abc123", diff --git a/extensions/zalo/src/monitor.ts b/extensions/zalo/src/monitor.ts index cbb3d2f31eb3..4cd64232d44d 100644 --- a/extensions/zalo/src/monitor.ts +++ b/extensions/zalo/src/monitor.ts @@ -11,6 +11,7 @@ import { resolveStableChannelMessageIngress } from "openclaw/plugin-sdk/channel- import { createMessageReceiptFromOutboundResults } from "openclaw/plugin-sdk/channel-outbound"; import { createChannelPairingController } from "openclaw/plugin-sdk/channel-pairing"; import type { MarkdownTableMode, OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { createLazyRuntimeModule, createLazyRuntimeNamedExport, @@ -141,21 +142,19 @@ const loadZaloWebhookModule = createLazyRuntimeModule(async () => ({ function registerSharedHostedMediaRoute(params: { path: string; - accountId: string; log?: (message: string) => void; }): () => void { const routeKey = canonicalizeWebhookRouteKey(params.path); - // Every account registers the account-agnostic handler so a swapped registry is populated. - // Exact unregister handles stay leased until the final account releases the shared route. + // Every account attempts the account-agnostic route so the first acquire after a registry swap + // repopulates it; exact same-owner conflicts reuse the existing route without replacement. const unregister = registerPluginHttpRoute({ auth: "plugin", match: "prefix", path: params.path, pluginId: "zalo", source: "zalo-hosted-media", - accountId: params.accountId, log: params.log, - replaceExisting: true, + reuseExistingSameOwner: true, throwOnFailure: true, handler: async (req, res) => { const handled = await tryHandleHostedZaloMediaRequest(req, res); @@ -298,13 +297,7 @@ function startPollingLoop(params: ZaloPollingLoopParams) { return undefined; } if (response.ok) { - statusSink?.({ - connected: true, - lifecycle: "ready", - terminalDisconnect: undefined, - lastConnectedAt: Date.now(), - lastError: null, - }); + statusSink?.(channelReadyPatch()); } if (response.ok && response.result) { statusSink?.({ lastInboundAt: Date.now() }); @@ -936,7 +929,6 @@ export async function monitorZaloProvider(options: ZaloMonitorOptions): Promise< if (hostedMediaRoutePath) { const unregisterHostedMediaRoute = registerSharedHostedMediaRoute({ path: hostedMediaRoutePath, - accountId: account.accountId, log: runtime.log, }); stopHandlers.push(unregisterHostedMediaRoute); @@ -1021,13 +1013,7 @@ export async function monitorZaloProvider(options: ZaloMonitorOptions): Promise< { url: effectiveWebhookUrl, secret_token: webhookSecret }, // pragma: allowlist secret fetcher, ); - statusSink?.({ - connected: true, - lifecycle: "ready", - terminalDisconnect: undefined, - lastConnectedAt: Date.now(), - lastError: null, - }); + statusSink?.(channelReadyPatch()); let webhookCleanupPromise: Promise | undefined; cleanupWebhook = async () => { if (!webhookCleanupPromise) { diff --git a/extensions/zalouser/src/monitor.account-scope.test.ts b/extensions/zalouser/src/monitor.account-scope.test.ts index 0cffc34a9def..d35c5181ab68 100644 --- a/extensions/zalouser/src/monitor.account-scope.test.ts +++ b/extensions/zalouser/src/monitor.account-scope.test.ts @@ -179,7 +179,14 @@ describe("zalouser monitor lifecycle", () => { }); try { await vi.waitFor(() => { - expect(statusSink).toHaveBeenCalledWith({ lifecycle: "ready" }); + expect(statusSink).toHaveBeenCalledWith({ + running: true, + connected: true, + lifecycle: "ready", + lastConnectedAt: expect.any(Number), + lastError: null, + terminalDisconnect: undefined, + }); }); } finally { abortController.abort(); diff --git a/extensions/zalouser/src/monitor.group-gating.test.ts b/extensions/zalouser/src/monitor.group-gating.test.ts index 51bc137bcd53..cdfde8ee1d8a 100644 --- a/extensions/zalouser/src/monitor.group-gating.test.ts +++ b/extensions/zalouser/src/monitor.group-gating.test.ts @@ -1,4 +1,5 @@ // Zalouser tests cover monitor.group gating plugin behavior. +import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contract"; import { createChannelMessageReplyPipeline } from "openclaw/plugin-sdk/channel-outbound"; import { beforeEach, describe, expect, it, vi } from "vitest"; import type { OpenClawConfig, PluginRuntime } from "../runtime-api.js"; @@ -310,7 +311,7 @@ async function processMessageThroughMonitor(params: { config: OpenClawConfig; runtime: ReturnType; historyState?: { historyLimit?: number }; - statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void; + statusSink?: (patch: Omit) => void; }): Promise { const messages = params.messages ?? (params.message ? [params.message] : []); const account = params.historyState?.historyLimit diff --git a/extensions/zalouser/src/monitor.ts b/extensions/zalouser/src/monitor.ts index a3266c37df3d..7ba9cadabd8f 100644 --- a/extensions/zalouser/src/monitor.ts +++ b/extensions/zalouser/src/monitor.ts @@ -18,6 +18,7 @@ import { isDangerousNameMatchingEnabled } from "openclaw/plugin-sdk/dangerous-na // Zalouser plugin module implements monitor behavior. import { expectDefined } from "openclaw/plugin-sdk/expect-runtime"; import { createDeferred } from "openclaw/plugin-sdk/extension-shared"; +import { channelReadyPatch } from "openclaw/plugin-sdk/gateway-runtime"; import { DEFAULT_GROUP_HISTORY_LIMIT, type HistoryEntry, @@ -68,11 +69,7 @@ type ZalouserMonitorOptions = { config: OpenClawConfig; runtime: RuntimeEnv; abortSignal: AbortSignal; - statusSink?: (patch: { - lastInboundAt?: number; - lastOutboundAt?: number; - lifecycle?: ChannelAccountSnapshot["lifecycle"]; - }) => void; + statusSink?: (patch: Omit) => void; ingressQueue?: Parameters[0]["queue"]; }; @@ -964,7 +961,7 @@ export async function monitorZalouserProvider( listenerStop(); listenerStop = null; } else if (!abortSignal.aborted) { - statusSink?.({ lifecycle: "ready" }); + statusSink?.(channelReadyPatch()); } if (abortSignal.aborted) { diff --git a/scripts/plugin-sdk-surface-report.mjs b/scripts/plugin-sdk-surface-report.mjs index a2f9e06faa35..f36f8caf960d 100644 --- a/scripts/plugin-sdk-surface-report.mjs +++ b/scripts/plugin-sdk-surface-report.mjs @@ -217,7 +217,8 @@ export function readPluginSdkSurfaceBudgets(env = process.env) { // +1: forwarding-routed approver-restricted native approval capability factory. // +1: shared inbound-event delivery correlation factory for channel plugins. // +1: canonical webhook route identity for plugin-owned target registries. - 4826, + // +3: canonical ready, blocked, and stopped channel lifecycle patch factories. + 4829, env, ), publicFunctionExports: readPluginSdkSurfaceBudgetEnv( @@ -262,7 +263,8 @@ export function readPluginSdkSurfaceBudgets(env = process.env) { // +1: forwarding-routed approver-restricted native approval capability factory. // +1: shared inbound-event delivery correlation factory for channel plugins. // +1: canonical webhook route identity for plugin-owned target registries. - 2903, + // +3: canonical ready, blocked, and stopped channel lifecycle patch factories. + 2906, env, ), publicDeprecatedExports: readPluginSdkSurfaceBudgetEnv( diff --git a/src/channels/plugins/stateful-target-builtins.test.ts b/src/channels/plugins/stateful-target-builtins.test.ts new file mode 100644 index 000000000000..6666e698df97 --- /dev/null +++ b/src/channels/plugins/stateful-target-builtins.test.ts @@ -0,0 +1,36 @@ +// Stateful target builtin tests cover repeatable registration after registry drains. +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + driver: { + id: "acp", + ensureReady: vi.fn(), + ensureSession: vi.fn(), + }, + register: vi.fn(), +})); + +vi.mock("./acp-stateful-target-driver.js", () => ({ + acpStatefulBindingTargetDriver: mocks.driver, +})); + +vi.mock("./stateful-target-drivers.js", () => ({ + registerStatefulBindingTargetDriver: mocks.register, +})); + +import { ensureStatefulTargetBuiltinsRegistered } from "./stateful-target-builtins.js"; + +describe("stateful target builtin registration", () => { + beforeEach(() => { + mocks.register.mockClear(); + }); + + it("re-registers an already-loaded builtin after its registry is drained", async () => { + await ensureStatefulTargetBuiltinsRegistered(); + await ensureStatefulTargetBuiltinsRegistered(); + + expect(mocks.register).toHaveBeenCalledTimes(2); + expect(mocks.register).toHaveBeenNthCalledWith(1, mocks.driver); + expect(mocks.register).toHaveBeenNthCalledWith(2, mocks.driver); + }); +}); diff --git a/src/channels/plugins/stateful-target-builtins.ts b/src/channels/plugins/stateful-target-builtins.ts index ffb0dba0b0d6..82a54472daa1 100644 --- a/src/channels/plugins/stateful-target-builtins.ts +++ b/src/channels/plugins/stateful-target-builtins.ts @@ -6,8 +6,6 @@ import { createLazyRuntimeModule } from "../../shared/lazy-runtime.js"; */ import { registerStatefulBindingTargetDriver } from "./stateful-target-drivers.js"; -let builtinsRegisteredPromise: Promise | null = null; - const loadAcpStatefulTargetDriverModule = createLazyRuntimeModule( () => import("./acp-stateful-target-driver.js"), ); @@ -17,20 +15,12 @@ export function isStatefulTargetBuiltinDriverId(id: string): boolean { } export async function ensureStatefulTargetBuiltinsRegistered(): Promise { - if (builtinsRegisteredPromise) { - await builtinsRegisteredPromise; - return; - } - builtinsRegisteredPromise = (async () => { + try { const { acpStatefulBindingTargetDriver } = await loadAcpStatefulTargetDriverModule(); registerStatefulBindingTargetDriver(acpStatefulBindingTargetDriver); - })(); - try { - await builtinsRegisteredPromise; } catch (error) { - // Retry after failed dynamic import/registration; a rejected singleton would - // otherwise permanently disable later setup or binding attempts. - builtinsRegisteredPromise = null; + // A rejected lazy import is cached; clear it so a later setup or binding attempt can retry. + loadAcpStatefulTargetDriverModule.clear(); throw error; } } diff --git a/src/channels/plugins/stateful-target-drivers.test.ts b/src/channels/plugins/stateful-target-drivers.test.ts new file mode 100644 index 000000000000..235e95b4475c --- /dev/null +++ b/src/channels/plugins/stateful-target-drivers.test.ts @@ -0,0 +1,28 @@ +// Stateful target driver tests cover process-wide registry identity. +import { describe, expect, it } from "vitest"; +import type { StatefulBindingTargetDriver } from "./stateful-target-drivers.js"; + +const moduleUrl = new URL("./stateful-target-drivers.ts", import.meta.url).href; + +describe("stateful target driver registry", () => { + it("shares registrations across duplicate module instances", async () => { + const first = (await import( + `${moduleUrl}?instance=first-${Date.now()}` + )) as typeof import("./stateful-target-drivers.js"); + const second = (await import( + `${moduleUrl}?instance=second-${Date.now()}` + )) as typeof import("./stateful-target-drivers.js"); + const driver: StatefulBindingTargetDriver = { + id: "duplicate-module-test", + ensureReady: async () => ({ ok: true }), + ensureSession: async () => ({ ok: true, sessionKey: "agent:test:shared" }), + }; + + const unregister = first.registerStatefulBindingTargetDriver(driver); + try { + expect(second.getStatefulBindingTargetDriver(driver.id)).toMatchObject({ id: driver.id }); + } finally { + unregister(); + } + }); +}); diff --git a/src/channels/plugins/stateful-target-drivers.ts b/src/channels/plugins/stateful-target-drivers.ts index 96da5ecb7448..f69caecd7eb1 100644 --- a/src/channels/plugins/stateful-target-drivers.ts +++ b/src/channels/plugins/stateful-target-drivers.ts @@ -4,6 +4,7 @@ * Stores lifecycle drivers for binding targets that carry mutable external session state. */ import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { resolveGlobalMap } from "../../shared/global-singleton.js"; import type { ConfiguredBindingResolution, StatefulBindingTargetDescriptor, @@ -41,7 +42,10 @@ export type StatefulBindingTargetDriver = { }) => Promise; }; -const registeredStatefulBindingTargetDrivers = new Map(); +const registeredStatefulBindingTargetDrivers = resolveGlobalMap< + string, + StatefulBindingTargetDriver +>(Symbol.for("openclaw.statefulBindingTargetDrivers"), "plugin-registry"); function listStatefulBindingTargetDrivers(): StatefulBindingTargetDriver[] { return [...registeredStatefulBindingTargetDrivers.values()]; diff --git a/src/gateway/channel-status-patches.test.ts b/src/gateway/channel-status-patches.test.ts index 6107d63519ac..24c4f8369be4 100644 --- a/src/gateway/channel-status-patches.test.ts +++ b/src/gateway/channel-status-patches.test.ts @@ -1,8 +1,11 @@ /** * Channel status patching tests. */ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { + channelBlockedPatch, + channelReadyPatch, + channelStoppedPatch, createConnectedChannelStatusPatch, createTransportActivityStatusPatch, } from "./channel-status-patches.js"; @@ -24,3 +27,40 @@ describe("createTransportActivityStatusPatch", () => { }); }); }); + +describe("channel lifecycle status patches", () => { + it("creates a ready patch that clears retained terminal state", () => { + const now = vi.spyOn(Date, "now").mockReturnValue(1234); + try { + expect(channelReadyPatch({ mode: "polling" })).toEqual({ + running: true, + connected: true, + lifecycle: "ready", + lastConnectedAt: 1234, + lastError: null, + terminalDisconnect: undefined, + mode: "polling", + }); + } finally { + now.mockRestore(); + } + }); + + it("creates a blocked patch with the required error and channel extras", () => { + expect(channelBlockedPatch("invalid token", { connected: false })).toEqual({ + lifecycle: "blocked", + terminalDisconnect: true, + lastError: "invalid token", + connected: false, + }); + }); + + it("creates a stopped patch and applies extras after the shared base", () => { + expect(channelStoppedPatch({ lastStopAt: 1234 })).toEqual({ + running: false, + connected: false, + lifecycle: "stopped", + lastStopAt: 1234, + }); + }); +}); diff --git a/src/gateway/channel-status-patches.ts b/src/gateway/channel-status-patches.ts index e25279efb541..9182ed864cc0 100644 --- a/src/gateway/channel-status-patches.ts +++ b/src/gateway/channel-status-patches.ts @@ -1,5 +1,7 @@ // Channel status patch factories centralize timestamp fields that multiple // runtime paths send into the gateway status store. +import type { ChannelAccountSnapshot } from "../channels/plugins/types.core.js"; + /** Patch emitted when a channel connection is established. */ type ConnectedChannelStatusPatch = { connected: true; @@ -12,6 +14,39 @@ type TransportActivityChannelStatusPatch = { lastTransportActivityAt: number; }; +type ReadyChannelStatusPatch = { + running: true; + connected: true; + lifecycle: "ready"; + lastConnectedAt: number; + lastError: null; + terminalDisconnect: undefined; +}; + +type BlockedChannelStatusPatch = { + lifecycle: "blocked"; + terminalDisconnect: true; + lastError: string; +}; + +type StoppedChannelStatusPatch = { + running: false; + connected: false; + lifecycle: "stopped"; +}; + +type ReadyChannelStatusExtras = Partial< + Omit +> & { + lastConnectedAt?: number; +}; +type BlockedChannelStatusExtras = Partial< + Omit +>; +type StoppedChannelStatusExtras = Partial< + Omit +>; + /** Creates a connected-channel status patch with matching connection/event timestamps. */ export function createConnectedChannelStatusPatch( at: number = Date.now(), @@ -31,3 +66,62 @@ export function createTransportActivityStatusPatch( lastTransportActivityAt: at, }; } + +/** Creates a ready patch that clears any retained terminal-auth verdict. */ +export function channelReadyPatch(): ReadyChannelStatusPatch; +export function channelReadyPatch( + extras: TExtras, +): ReadyChannelStatusPatch & TExtras; +export function channelReadyPatch( + extras: ReadyChannelStatusExtras = {}, +): ReadyChannelStatusPatch & ReadyChannelStatusExtras { + return Object.assign( + { + running: true as const, + connected: true as const, + lifecycle: "ready" as const, + lastConnectedAt: Date.now(), + lastError: null, + terminalDisconnect: undefined, + }, + extras, + ); +} + +/** Creates a terminal blocked patch with a required operator-facing error. */ +export function channelBlockedPatch(lastError: string): BlockedChannelStatusPatch; +export function channelBlockedPatch( + lastError: string, + extras: TExtras, +): BlockedChannelStatusPatch & TExtras; +export function channelBlockedPatch( + lastError: string, + extras: BlockedChannelStatusExtras = {}, +): BlockedChannelStatusPatch & BlockedChannelStatusExtras { + return Object.assign( + { + lifecycle: "blocked" as const, + terminalDisconnect: true as const, + lastError, + }, + extras, + ); +} + +/** Creates the shared patch emitted after a channel account has stopped. */ +export function channelStoppedPatch(): StoppedChannelStatusPatch; +export function channelStoppedPatch( + extras: TExtras, +): StoppedChannelStatusPatch & TExtras; +export function channelStoppedPatch( + extras: StoppedChannelStatusExtras = {}, +): StoppedChannelStatusPatch & StoppedChannelStatusExtras { + return Object.assign( + { + running: false as const, + connected: false as const, + lifecycle: "stopped" as const, + }, + extras, + ); +} diff --git a/src/gateway/server-channels.test.ts b/src/gateway/server-channels.test.ts index 8c840c837e61..cb6c19dc0581 100644 --- a/src/gateway/server-channels.test.ts +++ b/src/gateway/server-channels.test.ts @@ -30,6 +30,7 @@ import { setActiveDegradedSecretOwners, } from "../secrets/runtime-degraded-state.js"; import { evaluateChannelHealth } from "./channel-health-policy.js"; +import { channelReadyPatch, createTransportActivityStatusPatch } from "./channel-status-patches.js"; import { createChannelManager, type ChannelManager } from "./server-channels.js"; const hoisted = vi.hoisted(() => { @@ -927,6 +928,140 @@ describe("server-channels auto restart", () => { expect(lifecycleAtHandoff).toEqual(["starting", "starting"]); }); + it("accepts explicit channel-authored ready recovery within the same task", async () => { + let publishReady: (() => void) | undefined; + let publishStopped: (() => void) | undefined; + let blockedLastStartAt: number | null | undefined; + const startAccount = vi.fn(async (ctx: ChannelGatewayContext) => { + ctx.setStatus({ + accountId: ctx.accountId, + terminalDisconnect: true, + lifecycle: "blocked", + lastError: "relink required", + }); + blockedLastStartAt = ctx.getStatus().lastStartAt; + publishReady = () => ctx.setStatus(channelReadyPatch({ accountId: ctx.accountId })); + publishStopped = () => + ctx.setStatus({ + accountId: ctx.accountId, + running: false, + connected: false, + lifecycle: "stopped", + }); + await new Promise((resolve) => { + ctx.abortSignal.addEventListener("abort", () => resolve(), { once: true }); + }); + }); + installTestRegistry(createTestPlugin({ startAccount })); + const manager = createManager(); + + await manager.startChannels(); + await vi.waitFor(() => expect(publishReady).toBeDefined()); + expect(healthOf(manager.getRuntimeSnapshot().channelAccounts.discord?.default).reason).toBe( + "blocked", + ); + + publishReady?.(); + + const recovered = manager.getRuntimeSnapshot().channelAccounts.discord?.default; + expect(startAccount).toHaveBeenCalledOnce(); + expect(recovered).toMatchObject({ + running: true, + connected: true, + lifecycle: "ready", + terminalDisconnect: undefined, + lastError: null, + lastStartAt: blockedLastStartAt, + }); + expect(healthOf(recovered)).toEqual({ healthy: true, reason: "healthy" }); + + publishStopped?.(); + + const stopped = manager.getRuntimeSnapshot().channelAccounts.discord?.default; + expect(stopped).toMatchObject({ + running: false, + connected: false, + lifecycle: "stopped", + terminalDisconnect: undefined, + }); + expect(healthOf(stopped)).toEqual({ healthy: false, reason: "not-running" }); + }); + + it.each([ + { + name: "ready lifecycle without terminal clear", + patch: { accountId: DEFAULT_ACCOUNT_ID, lifecycle: "ready" } as ChannelAccountSnapshot, + }, + { + name: "terminal clear without ready lifecycle", + patch: { + accountId: DEFAULT_ACCOUNT_ID, + terminalDisconnect: undefined, + } as ChannelAccountSnapshot, + }, + ])("keeps terminal diagnosis sticky for $name", async ({ patch }) => { + let publishIncompleteRecovery: (() => void) | undefined; + const startAccount = vi.fn(async (ctx: ChannelGatewayContext) => { + ctx.setStatus({ + accountId: ctx.accountId, + terminalDisconnect: true, + lifecycle: "blocked", + lastError: "relink required", + }); + publishIncompleteRecovery = () => ctx.setStatus(patch); + await new Promise((resolve) => { + ctx.abortSignal.addEventListener("abort", () => resolve(), { once: true }); + }); + }); + installTestRegistry(createTestPlugin({ startAccount })); + const manager = createManager(); + + await manager.startChannels(); + await vi.waitFor(() => expect(publishIncompleteRecovery).toBeDefined()); + publishIncompleteRecovery?.(); + + expect(manager.getRuntimeSnapshot().channelAccounts.discord?.default).toMatchObject({ + lifecycle: "blocked", + lastError: "relink required", + }); + }); + + it("keeps terminal diagnosis sticky across activity and connected backfill patches", async () => { + let publishDerivedSignals: (() => void) | undefined; + const startAccount = vi.fn(async (ctx: ChannelGatewayContext) => { + ctx.setStatus({ + accountId: ctx.accountId, + terminalDisconnect: true, + lifecycle: "blocked", + lastError: "relink required", + }); + publishDerivedSignals = () => { + ctx.setStatus({ + accountId: ctx.accountId, + ...createTransportActivityStatusPatch(), + }); + ctx.setStatus({ accountId: ctx.accountId, connected: true }); + }; + await new Promise((resolve) => { + ctx.abortSignal.addEventListener("abort", () => resolve(), { once: true }); + }); + }); + installTestRegistry(createTestPlugin({ startAccount })); + const manager = createManager(); + + await manager.startChannels(); + await vi.waitFor(() => expect(publishDerivedSignals).toBeDefined()); + publishDerivedSignals?.(); + + expect(manager.getRuntimeSnapshot().channelAccounts.discord?.default).toMatchObject({ + connected: true, + lifecycle: "blocked", + terminalDisconnect: true, + lastError: "relink required", + lastTransportActivityAt: expect.any(Number), + }); + }); + it("recovers a manually restarted channel from a transient failure after terminal disconnect", async () => { const handoffStates: ChannelAccountSnapshot[] = []; const handoffSignals: AbortSignal[] = []; diff --git a/src/gateway/server-channels.ts b/src/gateway/server-channels.ts index 6042a8ef20eb..55c64baae552 100644 --- a/src/gateway/server-channels.ts +++ b/src/gateway/server-channels.ts @@ -389,12 +389,18 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage ): ChannelAccountSnapshot => { const store = getStore(channelId); const current = getRuntime(channelId, accountId); - // Terminal channel diagnosis survives task cleanup and late status patches. - // Only the gateway-owned start transition proves a new lifecycle began. + const hasExplicitReadyRecovery = + Object.hasOwn(patch, "lifecycle") && + patch.lifecycle === "ready" && + Object.hasOwn(patch, "terminalDisconnect") && + patch.terminalDisconnect === undefined; + // Weaker/derived signals never clear a terminal diagnosis. Gateway-owned starting still + // begins a new lifecycle; a channel-authored explicit ready + terminal clear proves recovery. const lifecycle = current.lifecycle === "blocked" && current.terminalDisconnect === true && - patch.lifecycle !== "starting" + patch.lifecycle !== "starting" && + !hasExplicitReadyRecovery ? "blocked" : (patch.lifecycle ?? (patch.restartPending === true diff --git a/src/plugin-sdk/gateway-runtime.ts b/src/plugin-sdk/gateway-runtime.ts index b040d8b0dc92..0a6f55e74d47 100644 --- a/src/plugin-sdk/gateway-runtime.ts +++ b/src/plugin-sdk/gateway-runtime.ts @@ -40,6 +40,9 @@ export { ErrorCodes, errorShape } from "../../packages/gateway-protocol/src/sche export type { GatewayRequestHandlerOptions } from "../gateway/server-methods/types.js"; export { + channelBlockedPatch, + channelReadyPatch, + channelStoppedPatch, createConnectedChannelStatusPatch, createTransportActivityStatusPatch, } from "../gateway/channel-status-patches.js"; diff --git a/src/plugins/http-registry.test.ts b/src/plugins/http-registry.test.ts index d1fb89fde592..ab509f8a058c 100644 --- a/src/plugins/http-registry.test.ts +++ b/src/plugins/http-registry.test.ts @@ -238,6 +238,65 @@ describe("registerPluginHttpRoute", () => { }); }); + it("reuses an exact same-owner route without replacing its handler", () => { + const { registry, logs, register } = createLoggedRouteHarness(); + const firstHandler = vi.fn(); + const secondHandler = vi.fn(); + const unregisterFirst = register({ + path: "/plugins/shared", + auth: "plugin", + handler: firstHandler, + pluginId: "demo", + source: "shared-route", + }); + + const unregisterSecond = register({ + path: "/PLUGINS//SHARED/", + auth: "plugin", + handler: secondHandler, + pluginId: "demo", + source: "shared-route", + reuseExistingSameOwner: true, + throwOnFailure: true, + }); + + expect(registry.httpRoutes).toHaveLength(1); + expect(registry.httpRoutes[0]?.handler).toBe(firstHandler); + expect(logs.at(-1)).toContain("reusing existing webhook path"); + unregisterSecond(); + expect(registry.httpRoutes).toHaveLength(1); + unregisterFirst(); + expect(registry.httpRoutes).toHaveLength(0); + }); + + it.each([ + { pluginId: "other", source: "shared-route" }, + { pluginId: "demo", source: "other-route" }, + ])("rejects route reuse by $pluginId/$source", ({ pluginId, source }) => { + const { registry, register } = createLoggedRouteHarness(); + const firstHandler = vi.fn(); + register({ + path: "/plugins/shared", + auth: "plugin", + handler: firstHandler, + pluginId: "demo", + source: "shared-route", + }); + + expect(() => + register({ + path: "/plugins/shared", + auth: "plugin", + pluginId, + source, + reuseExistingSameOwner: true, + throwOnFailure: true, + }), + ).toThrow("plugin: route reuse denied"); + expect(registry.httpRoutes).toHaveLength(1); + expect(registry.httpRoutes[0]?.handler).toBe(firstHandler); + }); + it("finds a canonical exact alias behind an earlier prefix overlap", () => { const { registry, register } = createLoggedRouteHarness(); register({ diff --git a/src/plugins/http-registry.ts b/src/plugins/http-registry.ts index bdb681faa33a..c0d034a2ef0a 100644 --- a/src/plugins/http-registry.ts +++ b/src/plugins/http-registry.ts @@ -28,6 +28,8 @@ export function registerPluginHttpRoute(params: { gatewayRuntimeScopeSurface?: PluginHttpRouteRegistration["gatewayRuntimeScopeSurface"]; /** Replace an existing canonical route owned by the same plugin and compatible route source. */ replaceExisting?: boolean; + /** Reuse an existing canonical route only when its nonempty plugin and source owners match. */ + reuseExistingSameOwner?: boolean; /** Throw when the route cannot be registered instead of returning a no-op cleanup. */ throwOnFailure?: boolean; pluginId?: string; @@ -83,24 +85,40 @@ export function registerPluginHttpRoute(params: { `plugin: route conflict at ${normalizedPath} (${routeMatch})${suffix}`, ); } + const requestedOwner = normalizeOptionalString(params.pluginId); + const requestedSource = normalizeOptionalString(params.source); + const mismatchedOwner = canonicalMatches.find( + (route) => + normalizeOptionalString(route.pluginId) !== requestedOwner || + normalizeOptionalString(route.source) !== requestedSource, + ); + if (!params.replaceExisting && params.reuseExistingSameOwner) { + if (requestedOwner !== undefined && requestedSource !== undefined && !mismatchedOwner) { + params.log?.( + `plugin: reusing existing webhook path ${normalizedPath} (${routeMatch}) (${requestedOwner}/${requestedSource})`, + ); + return noopUnregister; + } + const conflictingOwner = mismatchedOwner ?? existing; + return rejectRegistration( + `plugin: route reuse denied for ${normalizedPath} (${routeMatch})${suffix}; owned by ${conflictingOwner.pluginId ?? "unknown-plugin"} (${conflictingOwner.source ?? "unknown-source"})`, + ); + } if (!params.replaceExisting) { return rejectRegistration( `plugin: route conflict at ${normalizedPath} (${routeMatch})${suffix}; owned by ${existing.pluginId ?? "unknown-plugin"} (${existing.source ?? "unknown-source"})`, ); } - const replacementOwner = normalizeOptionalString(params.pluginId); - const replacementSource = normalizeOptionalString(params.source); // Source-less same-plugin replacement shipped before route-source ownership. // Preserve it only when both sides omit source; otherwise require an exact source match. - const mismatchedOwner = canonicalMatches.find( + const incompatibleReplacement = canonicalMatches.find( (route) => - normalizeOptionalString(route.pluginId) !== replacementOwner || - (replacementOwner !== undefined && - normalizeOptionalString(route.source) !== replacementSource), + normalizeOptionalString(route.pluginId) !== requestedOwner || + (requestedOwner !== undefined && normalizeOptionalString(route.source) !== requestedSource), ); - if (mismatchedOwner) { + if (incompatibleReplacement) { return rejectRegistration( - `plugin: route replacement denied for ${normalizedPath} (${routeMatch})${suffix}; owned by ${mismatchedOwner.pluginId ?? "unknown-plugin"} (${mismatchedOwner.source ?? "unknown-source"})`, + `plugin: route replacement denied for ${normalizedPath} (${routeMatch})${suffix}; owned by ${incompatibleReplacement.pluginId ?? "unknown-plugin"} (${incompatibleReplacement.source ?? "unknown-source"})`, ); } const pluginHint = params.pluginId ? ` (${params.pluginId})` : "";