From 42be965eace1b02dbfd6587b4030fa222d752ab8 Mon Sep 17 00:00:00 2001 From: Chris Eckert Date: Sat, 18 Jul 2026 01:33:06 -0700 Subject: [PATCH] improve(mattermost): restart only the changed account on config reload (#99312) * improve(gateway): restart only the changed account on channel config reload When a config change is scoped entirely to one channel account (channels.{kind}.accounts.{accountId}[.*]), restart just that account instead of the whole channel. Wholesale channel restarts disconnect every account on the channel; on gateways running many accounts, each account-scoped config write (adding an account, changing one account's settings) briefly dropped every other account's connection. The reload plan gains a restartChannelAccounts bucket populated only when all of a channel's changed paths are account-scoped; any channel-global path falls back to the existing wholesale restart, and a channel scheduled for wholesale restart drops its per-account entries so each (channel, account) pair restarts at most once. The executor runs per-account restarts through the existing stopChannel/startChannel accountId parameter. * fix(gateway): preserve account reload admission * fix(gateway): scope surgical reloads to isolated plugins * test(gateway): avoid preactivating reload snapshot * docs(channels): define account reload isolation contract * fix(gateway): preflight scoped account reloads * fix(gateway): re-drain live scoped reload targets * docs: refresh generated docs map * fix(mattermost): align scoped reload with durable ingress --------- Co-authored-by: Peter Steinberger --- docs/docs_map.md | 1 + docs/plugins/sdk-channel-plugins.md | 26 ++ extensions/mattermost/src/channel.setup.ts | 9 +- extensions/mattermost/src/channel.test.ts | 58 +++ extensions/mattermost/src/channel.ts | 9 +- .../src/mattermost/monitor-ingress.test.ts | 117 +++++- src/channels/plugins/types.plugin.ts | 12 +- src/cli/config-cli.ts | 2 +- src/gateway/config-reload-plan.ts | 120 +++++- src/gateway/config-reload-recovery.ts | 1 + src/gateway/config-reload.test.ts | 140 ++++++- src/gateway/config-reload.ts | 4 +- src/gateway/server-aux-handlers.test.ts | 26 ++ src/gateway/server-aux-handlers.ts | 11 +- .../server-methods/config-write-flow.ts | 5 +- src/gateway/server-reload-handlers.test.ts | 372 +++++++++++++++++- src/gateway/server-reload-handlers.ts | 115 +++++- 17 files changed, 979 insertions(+), 49 deletions(-) diff --git a/docs/docs_map.md b/docs/docs_map.md index f610c76b1a93..99a23ae118b8 100644 --- a/docs/docs_map.md +++ b/docs/docs_map.md @@ -7385,6 +7385,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`. - H2: Message adapter - H3: Inbound ingress (experimental) - H3: Durable ingress and replay dedupe + - H4: Account-scoped restart contract - H3: Typing indicators - H3: Media source params - H3: Native payload shaping diff --git a/docs/plugins/sdk-channel-plugins.md b/docs/plugins/sdk-channel-plugins.md index aa73aec0fd82..74de79020a78 100644 --- a/docs/plugins/sdk-channel-plugins.md +++ b/docs/plugins/sdk-channel-plugins.md @@ -116,6 +116,32 @@ guard when adopting the drain and size `completedTtlMs`/`completedMaxEntries` to cover the old guard window instead. Non-dedupe protections (age fences, outbound echo caches) are unrelated to this rule and stay. +#### Account-scoped restart contract + +Channel config changes restart the whole channel by default. A multi-account +channel may set `reload.accountScopedRestart: true` only when configuration +resolution reads channel-wide shared fields plus the selected account, never a +sibling account, and the Gateway can stop and start one `(channel, accountId)` +runtime without replacing sibling runtimes. + +The scoped path applies only to changes under +`channels..accounts..*`. Changes to shared channel +fields, `accounts.default`, removed or unresolvable accounts, and mixed changes +that can affect inheritance are promoted to a whole-channel restart. Plugins +that do not opt in always use the whole-channel path. + +For channels using the durable ingress drain, the account monitor's stop path +must first settle all accepted transport admissions, then dispose and await its +drain. Starting the account opens the same account-keyed queue, whose initial +drain recovers undispatched durable rows. Do not add a second reload-specific +replay pass; queue recovery is the canonical restart path. + +Treat this flag as a capability claim, not a performance preference. Contract +tests should prove that adding and editing one named account leaves a sibling's +resolved config unchanged, stopping one account settles only that account's +monitor and drain, and a fresh monitor recovers that account's rows exactly +once. If any guarantee cannot be proved, omit the flag. + ### Typing indicators If your channel supports typing indicators outside inbound replies, expose diff --git a/extensions/mattermost/src/channel.setup.ts b/extensions/mattermost/src/channel.setup.ts index f2fff701bb93..6a6616bd8ca2 100644 --- a/extensions/mattermost/src/channel.setup.ts +++ b/extensions/mattermost/src/channel.setup.ts @@ -24,7 +24,14 @@ export const mattermostSetupPlugin: ChannelPlugin = { media: true, nativeCommands: true, }, - reload: { configPrefixes: ["channels.mattermost"] }, + reload: { + configPrefixes: ["channels.mattermost"], + /** + * accounts.default is promoted; named resolution merges only channel-wide fields + * plus the selected account. Runtime monitor, debounce, and ingress use accountId. + */ + accountScopedRestart: true, + }, configSchema: MattermostChannelConfigSchema, config: { ...mattermostConfigAdapter, diff --git a/extensions/mattermost/src/channel.test.ts b/extensions/mattermost/src/channel.test.ts index 2983c846131d..55487f2cf71d 100644 --- a/extensions/mattermost/src/channel.test.ts +++ b/extensions/mattermost/src/channel.test.ts @@ -165,6 +165,64 @@ describe("mattermostPlugin", () => { }); }); + it("opts into account-scoped config restarts", () => { + expect(mattermostPlugin.reload).toMatchObject({ accountScopedRestart: true }); + }); + + it("keeps sibling resolution stable across named-account additions and edits", () => { + const before: OpenClawConfig = { + channels: { + mattermost: { + replyToMode: "first", + accounts: { + beta: { + baseUrl: "https://beta.example.com", + chatmode: "onmessage", + }, + }, + }, + }, + }; + const afterAdd: OpenClawConfig = { + channels: { + mattermost: { + replyToMode: "first", + accounts: { + alpha: { + baseUrl: "https://alpha.example.com", + chatmode: "oncall", + }, + beta: { + baseUrl: "https://beta.example.com", + chatmode: "onmessage", + }, + }, + }, + }, + }; + const afterEdit: OpenClawConfig = { + channels: { + mattermost: { + replyToMode: "first", + accounts: { + alpha: { + baseUrl: "https://alpha-new.example.com", + chatmode: "onchar", + }, + beta: { + baseUrl: "https://beta.example.com", + chatmode: "onmessage", + }, + }, + }, + }, + }; + + const expectedBeta = mattermostPlugin.config.resolveAccount(before, "beta"); + expect(mattermostPlugin.config.resolveAccount(afterAdd, "beta")).toEqual(expectedBeta); + expect(mattermostPlugin.config.resolveAccount(afterEdit, "beta")).toEqual(expectedBeta); + }); + describe("messaging", () => { it("keeps @username targets", () => { const normalize = requireMattermostNormalizeTarget(); diff --git a/extensions/mattermost/src/channel.ts b/extensions/mattermost/src/channel.ts index 0ae6a4e943f4..99568d9a6cce 100644 --- a/extensions/mattermost/src/channel.ts +++ b/extensions/mattermost/src/channel.ts @@ -798,7 +798,14 @@ export const mattermostPlugin: ChannelPlugin = create streaming: { blockStreamingCoalesceDefaults: { minChars: 1500, idleMs: 1000 }, }, - reload: { configPrefixes: ["channels.mattermost"] }, + reload: { + configPrefixes: ["channels.mattermost"], + /** + * accounts.default is promoted; named resolution merges only channel-wide fields + * plus the selected account. Monitor debounce and durable ingress use accountId. + */ + accountScopedRestart: true, + }, configSchema: MattermostChannelConfigSchema, config: { ...mattermostConfigAdapter, diff --git a/extensions/mattermost/src/mattermost/monitor-ingress.test.ts b/extensions/mattermost/src/mattermost/monitor-ingress.test.ts index edc2f74bd147..53c64c4cd0b0 100644 --- a/extensions/mattermost/src/mattermost/monitor-ingress.test.ts +++ b/extensions/mattermost/src/mattermost/monitor-ingress.test.ts @@ -38,9 +38,13 @@ function postedEvent(params?: { }); } -function startMonitor(queue: MattermostIngressQueue, dispatch: MattermostIngressDispatch) { +function startMonitor( + queue: MattermostIngressQueue, + dispatch: MattermostIngressDispatch, + accountId = "default", +) { return createMattermostIngressMonitor({ - accountId: "default", + accountId, queue, dispatch, runtime: { error: vi.fn(), log: vi.fn() }, @@ -49,22 +53,37 @@ function startMonitor(queue: MattermostIngressQueue, dispatch: MattermostIngress }); } -async function withQueue(fn: (queue: MattermostIngressQueue) => Promise): Promise { - const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-mattermost-ingress-")); - const stateDir = await fs.realpath(created); - const queue = createChannelIngressQueueForTests({ +function createQueue(stateDir: string, accountId: string): MattermostIngressQueue { + return createChannelIngressQueueForTests({ channelId: "mattermost", - accountId: "default", + accountId, stateDir, }); +} + +async function withStateDir(fn: (stateDir: string) => Promise): Promise { + const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-mattermost-ingress-")); + const stateDir = await fs.realpath(created); try { - return await fn(queue); + return await fn(stateDir); } finally { closeOpenClawStateDatabaseForTest(); await fs.rm(stateDir, { recursive: true, force: true }); } } +async function withQueue(fn: (queue: MattermostIngressQueue) => Promise): Promise { + return await withStateDir(async (stateDir) => await fn(createQueue(stateDir, "default"))); +} + +function createDeferred(): { promise: Promise; resolve: () => void } { + let resolvePromise = () => {}; + const promise = new Promise((resolve) => { + resolvePromise = resolve; + }); + return { promise, resolve: resolvePromise }; +} + function testLifecycle() { const calls = { adopted: vi.fn(async () => {}), @@ -131,6 +150,88 @@ describe("Mattermost durable ingress", () => { }); }); + it("settles only the restarted account and recovers its durable row exactly once", async () => { + await withStateDir(async (stateDir) => { + const queueA = createQueue(stateDir, "account-a"); + const queueB = createQueue(stateDir, "account-b"); + const dispatchA = vi.fn(async (_post, _payload, lifecycle) => { + lifecycle.onDeferred(); + return { kind: "deferred" } as const; + }); + const dispatchBBeforeRestart = vi.fn(async () => undefined); + const monitorA = startMonitor(queueA, dispatchA, "account-a"); + const monitorBBeforeRestart = startMonitor(queueB, dispatchBBeforeRestart, "account-b"); + const admissionStored = createDeferred(); + const releaseAdmission = createDeferred(); + const enqueueB = queueB.enqueue.bind(queueB); + queueB.enqueue = async (...args) => { + const result = await enqueueB(...args); + admissionStored.resolve(); + await releaseAdmission.promise; + return result; + }; + let admittingB: Promise | undefined; + let stoppingB: Promise | undefined; + + try { + await Promise.all([monitorA.waitForIdle(), monitorBBeforeRestart.waitForIdle()]); + await monitorA.receive(postedEvent({ postId: "post-account-a" })); + await monitorA.waitForIdle(); + const claimsABefore = await queueA.listClaims(); + expect(claimsABefore).toHaveLength(1); + expect(dispatchA).toHaveBeenCalledTimes(1); + + admittingB = monitorBBeforeRestart.receive(postedEvent({ postId: "post-account-b" })); + await admissionStored.promise; + let stopSettled = false; + stoppingB = monitorBBeforeRestart.stop().then(() => { + stopSettled = true; + }); + await Promise.resolve(); + + // stop() must wait for the serialized append admission. Account A's + // drain claim and debounce handoff stay owned by its live monitor. + expect(stopSettled).toBe(false); + expect(await queueA.listClaims()).toEqual(claimsABefore); + expect(dispatchA).toHaveBeenCalledTimes(1); + + releaseAdmission.resolve(); + await Promise.all([admittingB, stoppingB]); + expect(dispatchBBeforeRestart).not.toHaveBeenCalled(); + expect(await queueB.listPending({ limit: "all" })).toEqual([ + expect.objectContaining({ id: "post-account-b" }), + ]); + expect(await queueA.listClaims()).toEqual(claimsABefore); + + const dispatchBAfterRestart = vi.fn( + async (_post, _payload, lifecycle) => { + await lifecycle.onAdopted(); + }, + ); + const monitorBAfterRestart = startMonitor(queueB, dispatchBAfterRestart, "account-b"); + try { + await monitorBAfterRestart.waitForIdle(); + expect(dispatchBAfterRestart).toHaveBeenCalledTimes(1); + expect(dispatchBAfterRestart.mock.calls[0]?.[0].id).toBe("post-account-b"); + + await monitorBAfterRestart.receive(postedEvent({ postId: "post-account-b" })); + await monitorBAfterRestart.waitForIdle(); + expect(dispatchBAfterRestart).toHaveBeenCalledTimes(1); + expect(await queueA.listClaims()).toEqual(claimsABefore); + expect(dispatchA).toHaveBeenCalledTimes(1); + } finally { + await monitorBAfterRestart.stop(); + } + } finally { + releaseAdmission.resolve(); + await Promise.allSettled( + [admittingB, stoppingB].filter((task): task is Promise => task !== undefined), + ); + await Promise.allSettled([monitorA.stop(), monitorBBeforeRestart.stop()]); + } + }); + }); + it("retains completion so a duplicate post id cannot dispatch twice", async () => { await withQueue(async (queue) => { const dispatch = vi.fn(async (_post, _payload, lifecycle) => { diff --git a/src/channels/plugins/types.plugin.ts b/src/channels/plugins/types.plugin.ts index d7cec93ba9df..da3e389ed95d 100644 --- a/src/channels/plugins/types.plugin.ts +++ b/src/channels/plugins/types.plugin.ts @@ -65,7 +65,17 @@ export type ChannelPlugin; configSchema?: ChannelConfigSchema; diff --git a/src/cli/config-cli.ts b/src/cli/config-cli.ts index 5281f3423d97..fda6841cc3ba 100644 --- a/src/cli/config-cli.ts +++ b/src/cli/config-cli.ts @@ -1051,7 +1051,7 @@ function configApplyHintForPaths(paths: string[], afterConfig: OpenClawConfig): if (paths.some(isPluginEntryConfigPath)) { return RESTART_HINT; } - const plan = buildGatewayReloadPlan(paths); + const plan = buildGatewayReloadPlan(paths, { candidateConfig: afterConfig }); if (plan.restartGateway) { return RESTART_HINT; } diff --git a/src/gateway/config-reload-plan.ts b/src/gateway/config-reload-plan.ts index 405cc3510a4c..1df9ed27d206 100644 --- a/src/gateway/config-reload-plan.ts +++ b/src/gateway/config-reload-plan.ts @@ -1,11 +1,17 @@ // Gateway config reload planner. // Maps changed config paths to hot-reload actions, no-ops, or full restarts. -import { type ChannelId, listChannelPlugins } from "../channels/plugins/index.js"; +import { + type ChannelId, + type ChannelPlugin, + listChannelPlugins, +} from "../channels/plugins/index.js"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; import { getActivePluginChannelRegistryVersion, getActivePluginHttpRouteRegistry, getActivePluginHttpRouteRegistryVersion, } from "../plugins/runtime.js"; +import { DEFAULT_ACCOUNT_ID } from "../routing/account-id.js"; import { isPlainObject } from "../utils.js"; export type ChannelKind = ChannelId; @@ -23,6 +29,8 @@ export type GatewayReloadPlan = { reloadPlugins: boolean; restartChannels: Set; disposeMcpRuntimes: boolean; + /** Account targets; absent means no targeted restarts for hand-built plans. */ + restartChannelAccounts?: Map>; noopPaths: string[]; }; @@ -30,6 +38,7 @@ type ReloadRule = { prefix: string; kind: "restart" | "hot" | "none"; actions?: ReloadAction[]; + accountScopedPlugin?: ChannelPlugin; }; type ConfigReloadMetadata = { @@ -44,11 +53,14 @@ type ReloadAction = | "restart-health-monitor" | "reload-plugins" | "dispose-mcp-runtimes" + | `restart-channel-account:${ChannelId}` | `restart-channel:${ChannelId}`; type GatewayReloadPlanOptions = { noopPaths?: Iterable; forceChangedPaths?: Iterable; + /** Candidate config used to reject removed, unknown, or unresolvable account targets. */ + candidateConfig?: OpenClawConfig; }; const PLUGIN_INSTALL_TIMESTAMP_KEYS = ["installedAt", "resolvedAt"] as const; @@ -179,15 +191,22 @@ function listReloadRules(): ReloadRule[] { return cachedReloadRules; } // Channel docking: plugins contribute hot reload/no-op prefixes here. - const channelReloadRules: ReloadRule[] = listChannelPlugins().flatMap((plugin) => - (plugin.reload?.configPrefixes ?? []) - .map( - (prefix): ReloadRule => ({ + const channelReloadRules: ReloadRule[] = listChannelPlugins().flatMap((plugin) => { + const restartAction = plugin.reload?.accountScopedRestart + ? (`restart-channel-account:${plugin.id}` as ReloadAction) + : (`restart-channel:${plugin.id}` as ReloadAction); + return (plugin.reload?.configPrefixes ?? []) + .map((prefix): ReloadRule => { + const rule: ReloadRule = { prefix, kind: "hot", - actions: [`restart-channel:${plugin.id}` as ReloadAction], - }), - ) + actions: [restartAction], + }; + if (plugin.reload?.accountScopedRestart) { + rule.accountScopedPlugin = plugin; + } + return rule; + }) .concat( (plugin.reload?.noopPrefixes ?? []).map( (prefix): ReloadRule => ({ @@ -195,8 +214,8 @@ function listReloadRules(): ReloadRule[] { kind: "none", }), ), - ), - ); + ); + }); const channelPluginStateRules: ReloadRule[] = listChannelPlugins().flatMap((plugin) => [ { prefix: `plugins.entries.${plugin.id}`, @@ -338,12 +357,53 @@ export function listPluginInstallWholeRecordPaths( ); } +function extractAccountIdFromPath(channel: ChannelId, path: string): string | null { + const prefix = `channels.${channel}.accounts.`; + if (!path.startsWith(prefix)) { + return null; + } + const rest = path.slice(prefix.length); + if (rest.length === 0) { + return null; + } + const dotIdx = rest.indexOf("."); + const id = dotIdx === -1 ? rest : rest.slice(0, dotIdx); + if (id.length === 0) { + return null; + } + // Default config is the inheritance base, so it can change every account. + if (id === DEFAULT_ACCOUNT_ID) { + return null; + } + return id; +} + +function isResolvableChannelAccount(params: { + plugin: ChannelPlugin | undefined; + accountId: string; + config: OpenClawConfig; +}): boolean { + if (!params.plugin) { + return false; + } + try { + if (!params.plugin.config.listAccountIds(params.config).includes(params.accountId)) { + return false; + } + params.plugin.config.resolveAccount(params.config, params.accountId); + return true; + } catch { + return false; + } +} + export function buildGatewayReloadPlan( changedPaths: string[], options: GatewayReloadPlanOptions = {}, ): GatewayReloadPlan { const noopPaths = new Set(options.noopPaths); const forceChangedPaths = new Set(options.forceChangedPaths); + const restartChannelAccounts = new Map>(); const plan: GatewayReloadPlan = { changedPaths, restartGateway: false, @@ -357,10 +417,41 @@ export function buildGatewayReloadPlan( reloadPlugins: false, restartChannels: new Set(), disposeMcpRuntimes: false, + restartChannelAccounts, noopPaths: [], }; - const applyAction = (action: ReloadAction) => { + const applyAction = ( + action: ReloadAction, + originatingPath: string, + accountScopedPlugin?: ChannelPlugin, + ) => { + if (action.startsWith("restart-channel-account:")) { + const channel = action.slice("restart-channel-account:".length) as ChannelId; + const accountId = extractAccountIdFromPath(channel, originatingPath); + if (accountId !== null) { + if ( + options.candidateConfig && + !isResolvableChannelAccount({ + plugin: accountScopedPlugin, + accountId, + config: options.candidateConfig, + }) + ) { + plan.restartChannels.add(channel); + return; + } + let set = restartChannelAccounts.get(channel); + if (!set) { + set = new Set(); + restartChannelAccounts.set(channel, set); + } + set.add(accountId); + return; + } + plan.restartChannels.add(channel); + return; + } if (action.startsWith("restart-channel:")) { const channel = action.slice("restart-channel:".length) as ChannelId; plan.restartChannels.add(channel); @@ -418,10 +509,15 @@ export function buildGatewayReloadPlan( } plan.hotReasons.push(path); for (const action of rule.actions ?? []) { - applyAction(action); + applyAction(action, path, rule.accountScopedPlugin); } } + // A wholesale restart covers its account targets and must run only once. + for (const channel of plan.restartChannels) { + restartChannelAccounts.delete(channel); + } + if (plan.restartGmailWatcher) { plan.reloadHooks = true; } diff --git a/src/gateway/config-reload-recovery.ts b/src/gateway/config-reload-recovery.ts index b6fc51d58669..e3949c770043 100644 --- a/src/gateway/config-reload-recovery.ts +++ b/src/gateway/config-reload-recovery.ts @@ -24,6 +24,7 @@ export function reloadPlanNeedsRecovery(plan: GatewayReloadPlan): boolean { plan.restartGmailWatcher || plan.reloadPlugins || plan.restartChannels.size > 0 || + (plan.restartChannelAccounts?.size ?? 0) > 0 || shouldRefreshContextWindowCache(plan) ); } diff --git a/src/gateway/config-reload.test.ts b/src/gateway/config-reload.test.ts index 29c5c6c087cf..e58b90d96f2e 100644 --- a/src/gateway/config-reload.test.ts +++ b/src/gateway/config-reload.test.ts @@ -14,6 +14,7 @@ import type { PluginInstallRecord } from "../config/types.plugins.js"; import { pinActivePluginChannelRegistry, pinActivePluginHttpRouteRegistry, + releasePinnedPluginChannelRegistry, resetPluginRuntimeStateForTest, setActivePluginRegistry, } from "../plugins/runtime.js"; @@ -28,7 +29,11 @@ import { } from "../skills/runtime/refresh-state.js"; import { createTestRegistry } from "../test-utils/channel-plugins.js"; import { diffConfigPaths, diffGatewayReloadPaths } from "./config-diff.js"; -import { buildGatewayReloadPlan, resolveConfigReloadMetadata } from "./config-reload-plan.js"; +import { + buildGatewayReloadPlan, + type ChannelKind, + resolveConfigReloadMetadata, +} from "./config-reload-plan.js"; import { resolveGatewayReloadSettings } from "./config-reload-settings.js"; import { type GatewayConfigReloadTransactionOwnership, @@ -179,9 +184,26 @@ describe("buildGatewayReloadPlan", () => { noopPrefixes: ["channels.whatsapp"], }, }; + const mattermostPlugin: ChannelPlugin = { + id: "mattermost", + meta: { + id: "mattermost", + label: "Mattermost", + selectionLabel: "Mattermost", + docsPath: "/channels/mattermost", + blurb: "test", + }, + capabilities: { chatTypes: ["direct"] }, + config: { + listAccountIds: (cfg) => Object.keys(cfg.channels?.mattermost?.accounts ?? {}), + resolveAccount: () => ({}), + }, + reload: { configPrefixes: ["channels.mattermost"], accountScopedRestart: true }, + }; const registry = createTestRegistry([ { pluginId: "telegram", plugin: telegramPlugin, source: "test" }, { pluginId: "whatsapp", plugin: whatsappPlugin, source: "test" }, + { pluginId: "mattermost", plugin: mattermostPlugin, source: "test" }, ]); registry.reloads = [ { @@ -373,6 +395,75 @@ describe("buildGatewayReloadPlan", () => { expect(plan.restartChannels).toEqual(new Set(["telegram"])); }); + const mattermostAccountConfig = { + channels: { + mattermost: { + accounts: { + alpha: { enabled: true }, + beta: { enabled: true }, + }, + }, + }, + } as OpenClawConfig; + + it.each([ + { + label: "targets changed named accounts", + paths: [ + "channels.mattermost.accounts.alpha.enabled", + "channels.mattermost.accounts.beta.commands", + ], + expectedChannels: new Set(), + expectedAccounts: new Map>([ + ["mattermost", new Set(["alpha", "beta"])], + ]), + }, + { + label: "promotes accounts.default changes", + paths: ["channels.mattermost.accounts.default.commands"], + expectedChannels: new Set(["mattermost"]), + expectedAccounts: new Map>(), + }, + { + label: "promotes channel-global changes", + paths: ["channels.mattermost.botToken"], + expectedChannels: new Set(["mattermost"]), + expectedAccounts: new Map>(), + }, + { + label: "promotes unlisted account changes", + paths: ["channels.mattermost.accounts.removed.enabled"], + expectedChannels: new Set(["mattermost"]), + expectedAccounts: new Map>(), + }, + { + label: "lets an unlisted account replace earlier scoped targets", + paths: [ + "channels.mattermost.accounts.alpha.enabled", + "channels.mattermost.accounts.removed.enabled", + ], + expectedChannels: new Set(["mattermost"]), + expectedAccounts: new Map>(), + }, + { + label: "lets a mixed global change replace scoped targets", + paths: ["channels.mattermost.accounts.alpha.enabled", "channels.mattermost.botToken"], + expectedChannels: new Set(["mattermost"]), + expectedAccounts: new Map>(), + }, + { + label: "keeps non-opted-in channels wholesale", + paths: ["channels.telegram.accounts.alpha.enabled"], + expectedChannels: new Set(["telegram"]), + expectedAccounts: new Map>(), + }, + ])("$label", ({ paths, expectedChannels, expectedAccounts }) => { + const plan = buildGatewayReloadPlan(paths, { candidateConfig: mattermostAccountConfig }); + + expect(plan.restartChannels).toEqual(expectedChannels); + expect(plan.restartChannelAccounts).toEqual(expectedAccounts); + }); + it("restarts every channel whose config prefix matches", () => { const plan = buildGatewayReloadPlan(["web.enabled", "channels.telegram.botToken"]); @@ -1661,6 +1752,53 @@ describe("startGatewayConfigReloader", () => { await harness.reloader.stop(); }); + it("runs account-scoped channel changes through hot reload", async () => { + const channelRegistry = createTestRegistry([ + { + pluginId: "mattermost", + plugin: { + id: "mattermost", + meta: { + id: "mattermost", + label: "Mattermost", + selectionLabel: "Mattermost", + docsPath: "/channels/mattermost", + blurb: "test", + }, + capabilities: { chatTypes: ["direct"] }, + config: { listAccountIds: () => ["default", "alpha"], resolveAccount: () => ({}) }, + reload: { configPrefixes: ["channels.mattermost"], accountScopedRestart: true }, + } satisfies ChannelPlugin, + source: "test", + }, + ]); + const initialConfig = { + gateway: { reload: { debounceMs: 0 } }, + channels: { mattermost: { accounts: { alpha: { enabled: false } } } }, + } as OpenClawConfig; + const nextConfig = { + gateway: { reload: { debounceMs: 0 } }, + channels: { mattermost: { accounts: { alpha: { enabled: true } } } }, + } as OpenClawConfig; + const harness = createReloaderHarness( + vi.fn(async () => makeSnapshot({ config: nextConfig, hash: "account-reload" })), + { initialConfig }, + ); + + pinActivePluginChannelRegistry(channelRegistry); + try { + harness.watcher.emit("change"); + await vi.runAllTimersAsync(); + + const [plan] = getOnlyHotReloadCall(harness); + expect(plan.restartChannelAccounts).toEqual(new Map([["mattermost", new Set(["alpha"])]])); + expect(harness.onNoopConfigCommit).not.toHaveBeenCalled(); + } finally { + releasePinnedPluginChannelRegistry(channelRegistry); + await harness.reloader.stop(); + } + }); + it("plans one immutable runtime override snapshot per candidate", async () => { const initialConfig: OpenClawConfig = { gateway: { reload: { debounceMs: 0 } }, diff --git a/src/gateway/config-reload.ts b/src/gateway/config-reload.ts index a0b1f74a71f0..ef2066295775 100644 --- a/src/gateway/config-reload.ts +++ b/src/gateway/config-reload.ts @@ -82,7 +82,8 @@ function isNoopReloadPlan(plan: GatewayReloadPlan): boolean { !plan.restartHealthMonitor && !plan.reloadPlugins && !plan.disposeMcpRuntimes && - plan.restartChannels.size === 0 + plan.restartChannels.size === 0 && + (plan.restartChannelAccounts?.size ?? 0) === 0 ); } @@ -530,6 +531,7 @@ export function startGatewayConfigReloader(opts: { const plan = buildGatewayReloadPlan(changedPaths, { noopPaths: pluginInstallTimestampNoopPaths, forceChangedPaths: pluginInstallWholeRecordPaths, + candidateConfig: nextConfig, }); if (nextSettings.mode === "off") { opts.log.info("config reload disabled (gateway.reload.mode=off)"); diff --git a/src/gateway/server-aux-handlers.test.ts b/src/gateway/server-aux-handlers.test.ts index 3ec195c5d640..3338ec57d61d 100644 --- a/src/gateway/server-aux-handlers.test.ts +++ b/src/gateway/server-aux-handlers.test.ts @@ -47,6 +47,7 @@ function createReloadPlan(overrides?: Partial): GatewayReload restartHealthMonitor: overrides?.restartHealthMonitor ?? false, reloadPlugins: overrides?.reloadPlugins ?? false, restartChannels: overrides?.restartChannels ?? new Set(), + restartChannelAccounts: overrides?.restartChannelAccounts, disposeMcpRuntimes: overrides?.disposeMcpRuntimes ?? false, noopPaths: overrides?.noopPaths ?? [], }; @@ -304,6 +305,31 @@ describe("gateway aux handlers", () => { expect(respond).toHaveBeenCalledWith(true, { ok: true, warningCount: 0 }); }); + it("restarts the whole channel when a secret change is scoped to one account", async () => { + // secrets.reload has no per-account restart path — account-scoped plan + // entries must still produce a channel restart so rotated credentials + // are applied. + const buildReloadPlan = () => + createReloadPlan({ + restartChannels: new Set(), + restartChannelAccounts: new Map([["slack", new Set(["ops"])]]), + }); + activateSnapshot(slackConfig("old-slack-secret")); + const prepared = createSnapshot(slackConfig("new-slack-secret")); + const activateRuntimeSecrets = vi.fn().mockResolvedValue(prepared); + const { reload, respond, startChannel, stopChannel } = + createSecretsReloadHarnessWithChannelMocks({ + activateRuntimeSecrets, + buildReloadPlan, + }); + + await reload(); + + expect(stopChannel.mock.calls.map(([ch]) => ch)).toEqual(["slack"]); + expect(startChannel.mock.calls.map(([ch]) => ch)).toEqual(["slack"]); + expect(respond).toHaveBeenCalledWith(true, { ok: true, warningCount: 0 }); + }); + it("coalesces concurrent secrets.reload calls so channels are not restarted twice", async () => { const buildReloadPlan = buildRestartChannelsPlan("slack"); activateSnapshot(slackConfig("old-slack-secret")); diff --git a/src/gateway/server-aux-handlers.ts b/src/gateway/server-aux-handlers.ts index fae206dcfae9..f3b4d5d5ee2a 100644 --- a/src/gateway/server-aux-handlers.ts +++ b/src/gateway/server-aux-handlers.ts @@ -361,8 +361,15 @@ export function createGatewayAuxHandlers(params: { expectedGeneration: nextSharedGatewaySessionGeneration, }); } - if (plan.restartChannels.size > 0) { - const restartChannels = [...plan.restartChannels]; + // Account-scoped changes restart their whole channel here: + // secrets.reload has no per-account restart path, and a missed + // restart would leave rotated credentials unapplied. + const channelsToRestart = new Set([ + ...plan.restartChannels, + ...(plan.restartChannelAccounts?.keys() ?? []), + ]); + if (channelsToRestart.size > 0) { + const restartChannels = [...channelsToRestart]; if ( isTruthyEnvValue(process.env.OPENCLAW_SKIP_CHANNELS) || isTruthyEnvValue(process.env.OPENCLAW_SKIP_PROVIDERS) diff --git a/src/gateway/server-methods/config-write-flow.ts b/src/gateway/server-methods/config-write-flow.ts index 76c3c172af99..821533852a9c 100644 --- a/src/gateway/server-methods/config-write-flow.ts +++ b/src/gateway/server-methods/config-write-flow.ts @@ -141,7 +141,8 @@ function isNoopConfigReloadPlan(plan: ReturnType) !plan.restartHealthMonitor && !plan.reloadPlugins && !plan.disposeMcpRuntimes && - plan.restartChannels.size === 0 + plan.restartChannels.size === 0 && + (plan.restartChannelAccounts?.size ?? 0) === 0 ); } @@ -150,7 +151,7 @@ function resolveConfigRestartRequirement(params: { nextConfig: OpenClawConfig; }): { requiresRestart: boolean; scheduleDirectRestart: boolean } { const reloadSettings = resolveGatewayReloadSettings(params.nextConfig); - const plan = buildGatewayReloadPlan(params.changedPaths); + const plan = buildGatewayReloadPlan(params.changedPaths, { candidateConfig: params.nextConfig }); if (isNoopConfigReloadPlan(plan)) { return { requiresRestart: false, scheduleDirectRestart: false }; } diff --git a/src/gateway/server-reload-handlers.test.ts b/src/gateway/server-reload-handlers.test.ts index ecc62a14763d..266cfd491249 100644 --- a/src/gateway/server-reload-handlers.test.ts +++ b/src/gateway/server-reload-handlers.test.ts @@ -332,12 +332,15 @@ function createDeferredVoid() { function createReloadHandlersForTest( logReload = { info: vi.fn(), warn: vi.fn() }, channels?: { - start: (channel: ChannelKind) => Promise; - stop: (channel: ChannelKind) => Promise; + start: ReloadHandlerParams["startChannel"]; + stop: ReloadHandlerParams["stopChannel"]; }, reloadPlugins?: Parameters[0]["reloadPlugins"], stopPostReadySidecars = vi.fn(), recovery: boolean | NonNullable = true, + options?: { + getChannelAutostartSuppression?: ReloadHandlerParams["getChannelAutostartSuppression"]; + }, ) { const cron = { start: vi.fn(async () => {}), stop: vi.fn() }; const stopExitWatchers = vi.fn(); @@ -362,6 +365,7 @@ function createReloadHandlersForTest( }); const cronReconciliation = createTestCronReconciliation(); const logCron = { error: vi.fn() }; + const logChannels = { info: vi.fn(), error: vi.fn() }; const handlers = createGatewayReloadHandlers({ deps: {} as never, broadcast: vi.fn(), @@ -369,6 +373,7 @@ function createReloadHandlersForTest( setState, startChannel: channels?.start ?? vi.fn(async () => {}), stopChannel: channels?.stop ?? vi.fn(async () => {}), + getChannelAutostartSuppression: options?.getChannelAutostartSuppression, stopPostReadySidecars, reloadPlugins: reloadPlugins ?? @@ -379,7 +384,7 @@ function createReloadHandlersForTest( }), ), logHooks: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, - logChannels: { info: vi.fn(), error: vi.fn() }, + logChannels, logCron, logReload, cronReconciliation, @@ -397,6 +402,7 @@ function createReloadHandlersForTest( cron, cronReconciliation, heartbeatRunner, + logChannels, logCron, setState, stopExitWatchers, @@ -3114,6 +3120,366 @@ describe("gateway channel hot reload handlers", () => { } } + function createAccountReloadPlan( + accountIds: string[], + overrides: Partial = {}, + ): GatewayReloadPlan { + return { + ...createChannelReloadPlan([]), + changedPaths: accountIds.map((accountId) => `channels.discord.accounts.${accountId}`), + restartChannelAccounts: new Map([["discord", new Set(accountIds)]]), + ...overrides, + }; + } + + async function withDiscordAccountResolver( + listAccountIds: () => string[], + run: () => Promise, + resolveAccount: (cfg: OpenClawConfig, accountId?: string | null) => unknown = () => ({}), + ) { + const registry = createTestRegistry([ + { + pluginId: "discord", + plugin: { + ...createChannelTestPluginBase({ + id: "discord", + config: { listAccountIds, resolveAccount }, + }), + }, + source: "test", + }, + ]); + pinActivePluginChannelRegistry(registry); + try { + await run(); + } finally { + releasePinnedPluginChannelRegistry(registry); + } + } + + async function withDiscordAccounts(accountIds: string[], run: () => Promise) { + await withDiscordAccountResolver(() => accountIds, run); + } + + it("restarts only the changed account", async () => { + const events: string[] = []; + const startRootCounts: number[] = []; + const accountStopSettled = createDeferredVoid(); + const channels = { + stop: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`stop:${channel}:${accountId}`); + await accountStopSettled.promise; + }), + start: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`start:${channel}:${accountId}`); + startRootCounts.push(getActiveGatewayRootWorkCount({ excludeCurrent: true })); + }), + }; + const { applyHotReload } = createReloadHandlersForTest(undefined, channels); + const root = tryBeginGatewayRootWorkAdmission(); + expect(root).not.toBeNull(); + let reload: Promise | undefined; + + try { + await root?.run(async () => { + await withChannelReloadsEnabled(async () => { + await withDiscordAccounts(["default", "alpha", "beta"], async () => { + reload = applyHotReload(createAccountReloadPlan(["alpha"]), {}); + await waitForFast(() => expect(events).toEqual(["stop:discord:alpha"])); + expect(channels.start).not.toHaveBeenCalled(); + + accountStopSettled.resolve(); + await reload; + }); + }); + }); + } finally { + accountStopSettled.resolve(); + await reload?.catch(() => {}); + root?.release(); + } + + expect(events).toEqual(["stop:discord:alpha", "start:discord:alpha"]); + expect(startRootCounts).toEqual([1]); + expect(channels.stop).toHaveBeenCalledOnce(); + expect(channels.start).toHaveBeenCalledOnce(); + }); + + it("continues targeted restarts after an account failure", async () => { + const events: string[] = []; + const channels = { + stop: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`stop:${channel}:${accountId}`); + if (accountId === "alpha") { + throw new Error("stop failed"); + } + }), + start: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`start:${channel}:${accountId}`); + }), + }; + const requestRecoveryRestart = vi.fn(() => ({ status: "emitted" as const })); + const { applyHotReload } = createReloadHandlersForTest( + undefined, + channels, + undefined, + undefined, + requestRecoveryRestart, + ); + + await withChannelReloadsEnabled(async () => { + await withDiscordAccounts(["default", "alpha", "beta"], async () => { + await applyHotReload(createAccountReloadPlan(["alpha", "beta"]), {}); + }); + }); + + expect(events).toEqual(["stop:discord:alpha", "stop:discord:beta", "start:discord:beta"]); + expect(requestRecoveryRestart).toHaveBeenCalledOnce(); + }); + + it("promotes unlisted accounts to a wholesale restart", async () => { + const events: string[] = []; + const channels = { + stop: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`stop:${channel}:${accountId}`); + }), + start: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`start:${channel}:${accountId}`); + }), + }; + const { applyHotReload } = createReloadHandlersForTest(undefined, channels); + + await withChannelReloadsEnabled(async () => { + await withDiscordAccounts(["default", "alpha"], async () => { + await applyHotReload(createAccountReloadPlan(["removed-account"]), {}); + }); + }); + + expect(events).toEqual(["stop:discord:undefined", "start:discord:undefined"]); + }); + + it("promotes unresolvable accounts to a wholesale restart before stopping any account", async () => { + const events: string[] = []; + const channels = { + stop: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`stop:${channel}:${accountId}`); + }), + start: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`start:${channel}:${accountId}`); + }), + }; + const { applyHotReload, logChannels } = createReloadHandlersForTest(undefined, channels); + + await withChannelReloadsEnabled(async () => { + await withDiscordAccountResolver( + () => ["default", "alpha", "beta"], + async () => { + await applyHotReload(createAccountReloadPlan(["alpha", "beta"]), {}); + }, + (_cfg, accountId) => { + if (accountId === "beta") { + throw new Error("account resolution failed"); + } + return {}; + }, + ); + }); + + expect(events).toEqual(["stop:discord:undefined", "start:discord:undefined"]); + expect(logChannels.info).toHaveBeenCalledWith( + "promoting discord account reload to whole-channel restart after account resolution failed: account resolution failed", + ); + }); + + it("requests recovery when account enumeration fails after config commit", async () => { + const channels = { + stop: vi.fn(async () => {}), + start: vi.fn(async () => {}), + }; + const requestRecoveryRestart = vi.fn(() => ({ status: "emitted" as const })); + const { applyHotReload } = createReloadHandlersForTest( + undefined, + channels, + undefined, + undefined, + requestRecoveryRestart, + ); + + await withChannelReloadsEnabled(async () => { + await withDiscordAccountResolver( + () => { + throw new Error("account enumeration failed"); + }, + async () => { + await applyHotReload(createAccountReloadPlan(["alpha"]), {}); + }, + ); + }); + + expect(channels.stop).not.toHaveBeenCalled(); + expect(channels.start).not.toHaveBeenCalled(); + expect(requestRecoveryRestart).toHaveBeenCalledOnce(); + }); + + it("skips per-account restarts for channels already queued for wholesale restart", async () => { + const events: string[] = []; + const channels = { + stop: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`stop:${channel}:${accountId}`); + }), + start: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`start:${channel}:${accountId}`); + }), + }; + const { applyHotReload } = createReloadHandlersForTest(undefined, channels); + + await withChannelReloadsEnabled(async () => { + await withDiscordAccounts(["default", "alpha"], async () => { + await applyHotReload( + createAccountReloadPlan(["alpha"], { restartChannels: new Set(["discord"]) }), + {}, + ); + }); + }); + + expect(events).toEqual(["stop:discord:undefined", "start:discord:undefined"]); + }); + + it("aggregates targeted and wholesale stop failures into one suppressed recovery request", async () => { + const events: string[] = []; + const channels = { + stop: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`stop:${channel}:${accountId}`); + throw new Error("stop failed"); + }), + start: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`start:${channel}:${accountId}`); + }), + }; + const requestRecoveryRestart = vi.fn(() => ({ status: "emitted" as const })); + const { applyHotReload } = createReloadHandlersForTest( + undefined, + channels, + undefined, + undefined, + requestRecoveryRestart, + { + getChannelAutostartSuppression: () => ({ + reason: "crash-loop-breaker", + message: "safe mode", + }), + }, + ); + + await withChannelReloadsEnabled(async () => { + await withDiscordAccounts(["default", "alpha"], async () => { + await applyHotReload( + createAccountReloadPlan(["alpha"], { + restartChannels: new Set(["telegram"]), + }), + {}, + ); + }); + }); + + expect(events).toEqual(["stop:discord:alpha", "stop:telegram:undefined"]); + expect(requestRecoveryRestart).toHaveBeenCalledOnce(); + }); + + it("stops account targets without restarting them while autostart is suppressed", async () => { + const events: string[] = []; + const channels = { + stop: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`stop:${channel}:${accountId}`); + }), + start: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`start:${channel}:${accountId}`); + }), + }; + const { applyHotReload } = createReloadHandlersForTest( + undefined, + channels, + undefined, + undefined, + true, + { + getChannelAutostartSuppression: () => ({ + reason: "crash-loop-breaker", + message: "safe mode", + }), + }, + ); + + await withChannelReloadsEnabled(async () => { + await withDiscordAccounts(["default", "alpha"], async () => { + await applyHotReload(createAccountReloadPlan(["alpha"]), {}); + }); + }); + + expect(events).toEqual(["stop:discord:alpha"]); + }); + + it("rechecks agent work admitted after plugin reload leaves the channel running", async () => { + const events: string[] = []; + const channels = { + stop: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`stop:${channel}:${accountId}`); + }), + start: vi.fn(async (channel: ChannelKind, accountId?: string) => { + events.push(`start:${channel}:${accountId}`); + }), + }; + const reloadPlugins = vi.fn(async (params): Promise => { + await params.beforeReplace(new Set()); + hoisted.activeEmbeddedRunCount.value = 1; + return { + restartChannels: new Set(), + activeChannels: new Set(["discord"]), + }; + }); + const logReload = { info: vi.fn(), warn: vi.fn() }; + const { applyHotReload } = createReloadHandlersForTest(logReload, channels, reloadPlugins); + vi.useFakeTimers(); + let reload: Promise | undefined; + + try { + await withChannelReloadsEnabled(async () => { + await withDiscordAccounts(["default", "alpha"], async () => { + reload = applyHotReload(createAccountReloadPlan(["alpha"], { reloadPlugins: true }), {}); + await vi.advanceTimersByTimeAsync(0); + expect(events).toEqual([]); + expect(logReload.warn).toHaveBeenCalledWith(expect.stringContaining("(discord)")); + + hoisted.activeEmbeddedRunCount.value = 0; + await vi.advanceTimersByTimeAsync(500); + await reload; + }); + }); + } finally { + hoisted.activeEmbeddedRunCount.value = 0; + await vi.advanceTimersByTimeAsync(500).catch(() => {}); + vi.useRealTimers(); + await reload?.catch(() => {}); + } + + expect(events).toEqual(["stop:discord:alpha", "start:discord:alpha"]); + expect(reloadPlugins).toHaveBeenCalledOnce(); + }); + + it("requires a recovery owner for targeted account reloads", async () => { + const { applyHotReload } = createReloadHandlersForTest( + undefined, + undefined, + undefined, + undefined, + false, + ); + + await expect(applyHotReload(createAccountReloadPlan(["alpha"]), {})).rejects.toThrow( + "config reload requires a managed gateway restart owner for irreversible hot reload", + ); + }); + it("refuses channel restarts while crash-loop safe mode suppresses autostart", async () => { const logChannels = { info: vi.fn(), error: vi.fn() }; const channels = { diff --git a/src/gateway/server-reload-handlers.ts b/src/gateway/server-reload-handlers.ts index fc1bd1ec0f4f..182e9364935c 100644 --- a/src/gateway/server-reload-handlers.ts +++ b/src/gateway/server-reload-handlers.ts @@ -10,6 +10,7 @@ import { warmCurrentProviderAuthStateOffMainThread, } from "../agents/model-provider-auth.js"; import { getTotalPendingReplies } from "../auto-reply/reply/dispatcher-registry.js"; +import { getChannelPlugin } from "../channels/plugins/index.js"; import type { CliDeps } from "../cli/deps.types.js"; import { isRestartEnabled } from "../config/commands.flags.js"; import { getConfigValueAtPath } from "../config/config-paths.js"; @@ -562,6 +563,8 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) resetDirectoryCache(); const channelsToRestart = new Set(plan.restartChannels); + const restartChannelAccounts = + plan.restartChannelAccounts ?? new Map>(); const channelsStoppedBeforePluginReload = new Set(); let activePluginChannelsAfterReload: ReadonlySet | null = null; let pluginReloadAborted = false; @@ -578,6 +581,8 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) const shouldSkipChannelRestart = isTruthyEnvValue(candidateEnv.OPENCLAW_SKIP_CHANNELS) || isTruthyEnvValue(candidateEnv.OPENCLAW_SKIP_PROVIDERS); + const channelReloadTargets = () => + new Set([...channelsToRestart, ...restartChannelAccounts.keys()]); const getChannelAutostartSuppression = () => params.getChannelAutostartSuppression?.() ?? null; const logSuppressedChannelRestart = ( channels: ReadonlySet, @@ -757,16 +762,11 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) for (const channel of channels) { channelsToRestart.add(channel); } - if (channelsToRestart.size === 0 || shouldSkipChannelRestart) { + const targets = channelReloadTargets(); + if (targets.size === 0 || shouldSkipChannelRestart) { return; } - if ( - await waitForActiveWorkBeforeChannelReload( - channelsToRestart, - nextConfig, - isTransactionCurrent, - ) - ) { + if (await waitForActiveWorkBeforeChannelReload(targets, nextConfig, isTransactionCurrent)) { params.logChannels.info( "channel reload before plugin replace cancelled by config supersession or restart", ); @@ -870,9 +870,15 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) } } - if (!plan.reloadPlugins && channelsToRestart.size > 0 && !shouldSkipChannelRestart) { + const channelTargets = channelReloadTargets(); + const hasLiveChannelTargets = [...channelTargets].some( + (channel) => !channelsStoppedBeforePluginReload.has(channel), + ); + // Plugin replacement can admit new agent work while an account monitor stays live. + // Recheck that work here; durable ingress replay remains owned by the fresh monitor drain. + if (!pluginReloadAborted && hasLiveChannelTargets && !shouldSkipChannelRestart) { pluginReloadAborted = await waitForActiveWorkBeforeChannelReload( - channelsToRestart, + channelTargets, nextConfig, isTransactionCurrent, ); @@ -946,7 +952,48 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) } } - if (channelsToRestart.size > 0) { + // Suppressed and normal reloads share fallback selection so stale account + // ids always reach the wholesale path that evicts their old runtime. + const collectChannelAccountTargets = (): Array<[ChannelKind, string]> => { + const targets: Array<[ChannelKind, string]> = []; + for (const [channel, accountIds] of restartChannelAccounts) { + if ( + channelsToRestart.has(channel) || + (plan.reloadPlugins && activePluginChannelsAfterReload?.has(channel) === false) + ) { + continue; + } + const plugin = getChannelPlugin(channel); + let listedAccountIds: Set; + try { + listedAccountIds = new Set(plugin?.config.listAccountIds(nextConfig) ?? []); + } catch (err) { + scheduleRecoveryRestart(`channel account enumeration (${channel})`, err); + continue; + } + if ([...accountIds].some((accountId) => !listedAccountIds.has(accountId))) { + channelsToRestart.add(channel); + continue; + } + try { + for (const accountId of accountIds) { + plugin?.config.resolveAccount(nextConfig, accountId); + } + } catch (err) { + params.logChannels.info( + `promoting ${channel} account reload to whole-channel restart after account resolution failed: ${formatErrorMessage(err)}`, + ); + channelsToRestart.add(channel); + continue; + } + for (const accountId of accountIds) { + targets.push([channel, accountId]); + } + } + return targets; + }; + + if (channelsToRestart.size > 0 || restartChannelAccounts.size > 0) { if (shouldSkipChannelRestart) { params.logChannels.info( "skipping channel reload (OPENCLAW_SKIP_CHANNELS=1 or OPENCLAW_SKIP_PROVIDERS=1)", @@ -956,6 +1003,21 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) if (cancelledByRestart) { params.logChannels.info("channel restart cancelled by in-process restart"); } else { + const accountStops = collectChannelAccountTargets(); + const accountStopFailures: string[] = []; + for (const [channel, accountId] of accountStops) { + try { + params.logChannels.info( + `stopping ${channel} account ${accountId} before suppressed hot reload`, + ); + await params.stopChannel(channel, accountId, { manual: false }); + } catch (err) { + accountStopFailures.push(`${channel}[${accountId}]`); + params.logChannels.error( + `failed to stop ${channel} account ${accountId} during suppressed hot reload: ${formatErrorMessage(err)}`, + ); + } + } const stopFailures = await collectChannelOperationFailures({ channels: channelsToRestart, run: async (channel) => { @@ -976,16 +1038,36 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) ); }, }); - if (stopFailures.length > 0) { - scheduleRecoveryRestart(`channel stop (${stopFailures.join(", ")})`); + const allStopFailures = [...accountStopFailures, ...stopFailures]; + if (allStopFailures.length > 0) { + scheduleRecoveryRestart(`channel stop (${allStopFailures.join(", ")})`); } - logSuppressedChannelRestart(channelsToRestart, "channel restart during hot reload"); + logSuppressedChannelRestart(channelReloadTargets(), "channel restart during hot reload"); } } else { const cancelledByRestart = pluginReloadAborted; if (cancelledByRestart) { params.logChannels.info("channel restart cancelled by in-process restart"); } else { + const accountRestarts = collectChannelAccountTargets(); + const accountRestartFailures: string[] = []; + for (const [channel, accountId] of accountRestarts) { + try { + params.logChannels.info(`restarting ${channel} account ${accountId}`); + await params.stopChannel(channel, accountId, { manual: false }); + if (isLifecycleReloadAborted()) { + continue; + } + await runOutsideGatewayRootWorkAdmission(() => + params.startChannel(channel, accountId), + ); + } catch (err) { + accountRestartFailures.push(`${channel}[${accountId}]`); + params.logChannels.error( + `failed to restart ${channel} account ${accountId} during hot reload: ${formatErrorMessage(err)}`, + ); + } + } const restartChannel = async (name: ChannelKind) => { if (plan.reloadPlugins && activePluginChannelsAfterReload?.has(name) === false) { return; @@ -1008,8 +1090,9 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) ); }, }); - if (restartFailures.length > 0) { - scheduleRecoveryRestart(`channel restart (${restartFailures.join(", ")})`); + const allRestartFailures = [...accountRestartFailures, ...restartFailures]; + if (allRestartFailures.length > 0) { + scheduleRecoveryRestart(`channel restart (${allRestartFailures.join(", ")})`); } } }