diff --git a/docs/plugins/hooks.md b/docs/plugins/hooks.md index 92590e4399e8..9386718ab65e 100644 --- a/docs/plugins/hooks.md +++ b/docs/plugins/hooks.md @@ -595,6 +595,9 @@ startup and scheduler replacement during config reload. The event reports cron still emits with `enabled: false`, allowing an external projection to clear stale wakes. Use `ctx.getCron?.()` for the exact scheduler instance that completed reconciliation; a later reload does not retarget that callback. +`ctx.abortSignal` owns that same scheduler snapshot. The Gateway aborts it as +soon as a newer scheduler is armed or shutdown starts. Pass it through every +durable side effect and do not accept the snapshot after it aborts. This is a scheduler lifecycle signal, not a plugin-activation signal: a plugin-only hot reload does not replay it. A newly enabled consumer receives its first baseline on the next scheduler replacement or Gateway start. @@ -661,6 +664,7 @@ export function registerCronProjection(api: OpenClawPluginApi, host: ExternalWak let cron: CronReader | undefined; let enabled = false; let hasBaseline = false; + let reconciliationSignal: AbortSignal | undefined; let requestedRevision = 0; let appliedRevision = 0; let worker = Promise.resolve(); @@ -670,9 +674,13 @@ export function registerCronProjection(api: OpenClawPluginApi, host: ExternalWak let retryMs = 1_000; while (!lifecycle.signal.aborted && appliedRevision < requestedRevision) { + const ownerSignal = reconciliationSignal; + if (!ownerSignal || ownerSignal.aborted) { + return; + } const targetRevision = requestedRevision; const attempt = new AbortController(); - const signal = AbortSignal.any([lifecycle.signal, attempt.signal]); + const signal = AbortSignal.any([lifecycle.signal, ownerSignal, attempt.signal]); activeAttempt = attempt; try { @@ -694,7 +702,7 @@ export function registerCronProjection(api: OpenClawPluginApi, host: ExternalWak appliedRevision = targetRevision; retryMs = 1_000; } catch { - if (lifecycle.signal.aborted) { + if (lifecycle.signal.aborted || ownerSignal.aborted) { return; } if (attempt.signal.aborted) { @@ -740,6 +748,7 @@ export function registerCronProjection(api: OpenClawPluginApi, host: ExternalWak cron = reconciledCron; enabled = event.enabled; hasBaseline = true; + reconciliationSignal = ctx.abortSignal; return requestProjection(); }); diff --git a/docs/plugins/sdk-overview.md b/docs/plugins/sdk-overview.md index e56a96cd100f..055d95391cfc 100644 --- a/docs/plugins/sdk-overview.md +++ b/docs/plugins/sdk-overview.md @@ -493,7 +493,7 @@ cover CLI and Gateway-backed install or update paths. - `message_received`: use the typed `threadId` field when you need inbound thread/topic routing. Keep `metadata` for channel-specific extras. - `message_sending`: use typed `replyToId` / `threadId` routing fields before falling back to channel-specific `metadata`. - `gateway_start`: use `ctx.config`, `ctx.workspaceDir`, and `ctx.getCron?.()` for gateway-owned startup state instead of relying on internal `gateway:startup` hooks. Cron may still be loading at this point. -- `cron_reconciled`: rebuild a full external cron projection after startup or scheduler reload. It includes `reason` and the effective `enabled` state, including `enabled: false`, while `ctx.getCron?.()` returns the exact reconciled scheduler. +- `cron_reconciled`: rebuild a full external cron projection after startup or scheduler reload. It includes `reason` and the effective `enabled` state, including `enabled: false`, while `ctx.getCron?.()` returns the exact reconciled scheduler. Pass `ctx.abortSignal` into durable projection work; it aborts when that scheduler snapshot is superseded or the Gateway closes. - `cron_changed`: observe gateway-owned cron lifecycle changes. `scheduled` and `removed` events are post-commit reconciliation hints, not an ordered delta log. A scheduled event's `event.nextRunAtMs` is absent when the job has no next wake; a removed event still carries the deleted job snapshot. External wake schedulers should debounce or coalesce `cron_changed` events, diff --git a/scripts/plugin-sdk-surface-report.mjs b/scripts/plugin-sdk-surface-report.mjs index a38705809e4d..12177e76ce33 100644 --- a/scripts/plugin-sdk-surface-report.mjs +++ b/scripts/plugin-sdk-surface-report.mjs @@ -195,7 +195,7 @@ export function readPluginSdkSurfaceBudgets(env = process.env) { ), publicExports: readPluginSdkSurfaceBudgetEnv( "OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_EXPORTS", - 10505, + 10507, env, ), publicFunctionExports: readPluginSdkSurfaceBudgetEnv( diff --git a/src/gateway/server-cron-reconciled.test.ts b/src/gateway/server-cron-reconciled.test.ts index 28e5cd675804..24b547d29ee4 100644 --- a/src/gateway/server-cron-reconciled.test.ts +++ b/src/gateway/server-cron-reconciled.test.ts @@ -39,6 +39,8 @@ describe("gateway cron reconciliation lifecycle", () => { config, }); expect(ctx?.getCron?.()).toBe(cron); + expect(ctx?.abortSignal).toBeInstanceOf(AbortSignal); + expect(ctx?.abortSignal.aborted).toBe(false); }); it("suppresses a startup completion superseded by reload", async () => { @@ -95,12 +97,14 @@ describe("gateway cron reconciliation lifecycle", () => { expect(runHook).not.toHaveBeenCalled(); }); - it("serializes snapshots so a reload cannot settle before startup", async () => { + it("aborts a superseded snapshot without blocking the replacement", async () => { let releaseStartup: (() => void) | undefined; + let startupSignal: AbortSignal | undefined; const order: string[] = []; - const runHook = vi.fn(async (event) => { + const runHook = vi.fn(async (event, ctx) => { order.push(`${event.reason}:start`); if (event.reason === "startup") { + startupSignal = ctx.abortSignal; await new Promise((resolve) => { releaseStartup = resolve; }); @@ -125,17 +129,49 @@ describe("gateway cron reconciliation lifecycle", () => { config: {} as OpenClawConfig, cronState: createCronState("reload", true), }); - const reloadCompletion = reload.complete(); + expect(startupSignal?.aborted).toBe(true); + await reload.complete(); - await Promise.resolve(); - expect(order).toEqual(["startup:start"]); + expect(order).toEqual(["startup:start", "reload:start", "reload:end"]); if (!releaseStartup) { throw new Error("Expected startup hook to be pending"); } releaseStartup(); - await Promise.all([startupCompletion, reloadCompletion]); + await startupCompletion; - expect(order).toEqual(["startup:start", "startup:end", "reload:start", "reload:end"]); + expect(order).toEqual(["startup:start", "reload:start", "reload:end", "startup:end"]); + }); + + it("aborts an active snapshot when reconciliation is invalidated", async () => { + let releaseHook: (() => void) | undefined; + let activeSignal: AbortSignal | undefined; + const runHook = vi.fn(async (_event, ctx) => { + activeSignal = ctx.abortSignal; + await new Promise((resolve) => { + releaseHook = resolve; + }); + }); + const reconciliation = createGatewayCronReconciliation({ + port: 18789, + workspaceDir: "/tmp/openclaw-workspace", + isClosing: () => false, + runHook, + }); + const armed = reconciliation.arm({ + reason: "startup", + config: {} as OpenClawConfig, + cronState: createCronState("startup", true), + }); + const completion = armed.complete(); + await vi.waitFor(() => expect(runHook).toHaveBeenCalledTimes(1)); + + reconciliation.invalidate(); + expect(activeSignal?.aborted).toBe(true); + if (!releaseHook) { + throw new Error("Expected cron reconciliation hook to be pending"); + } + releaseHook(); + await completion; }); }); diff --git a/src/gateway/server-cron-reconciled.ts b/src/gateway/server-cron-reconciled.ts index 432629d1badc..63e148a55fa4 100644 --- a/src/gateway/server-cron-reconciled.ts +++ b/src/gateway/server-cron-reconciled.ts @@ -2,8 +2,8 @@ // Suppresses stale scheduler completions across reload and shutdown boundaries. import type { OpenClawConfig } from "../config/types.openclaw.js"; import type { + PluginHookCronReconciledContext, PluginHookCronReconciledEvent, - PluginHookGatewayContext, PluginHookGatewayCronService, } from "../plugins/hook-types.js"; import type { GatewayCronState } from "./server-cron.js"; @@ -25,14 +25,26 @@ export function createGatewayCronReconciliation(params: { port: number; workspaceDir: string; isClosing: () => boolean; - runHook: (event: PluginHookCronReconciledEvent, ctx: PluginHookGatewayContext) => Promise; + runHook: ( + event: PluginHookCronReconciledEvent, + ctx: PluginHookCronReconciledContext, + ) => Promise; }): GatewayCronReconciliation { let lifecycleGeneration = 0; - let dispatchTail = Promise.resolve(); + let activeAbortController: AbortController | undefined; + + const supersedeActive = () => { + lifecycleGeneration += 1; + activeAbortController?.abort(); + activeAbortController = undefined; + }; return { arm: ({ reason, config, cronState }) => { - const generation = ++lifecycleGeneration; + supersedeActive(); + const generation = lifecycleGeneration; + const abortController = new AbortController(); + activeAbortController = abortController; const cron = cronState.cron as PluginHookGatewayCronService; const event: PluginHookCronReconciledEvent = { reason, @@ -46,27 +58,25 @@ export function createGatewayCronReconciliation(params: { return; } completed = true; - const dispatch = dispatchTail.then(async () => { - // A newer scheduler or shutdown owns reconciliation now. Dispatching - // this completion would let plugins replace current state with stale data. - if (params.isClosing() || generation !== lifecycleGeneration) { - return; - } - await params.runHook(event, { - port: params.port, - config, - workspaceDir: params.workspaceDir, - getCron: () => cron, - }); + // Each signal owns one exact scheduler snapshot. Do not serialize + // generations: a stuck stale observer must not hide the current state. + if ( + params.isClosing() || + generation !== lifecycleGeneration || + abortController.signal.aborted + ) { + return; + } + await params.runHook(event, { + port: params.port, + config, + workspaceDir: params.workspaceDir, + getCron: () => cron, + abortSignal: abortController.signal, }); - // Preserve lifecycle order even when one plugin callback is slow or fails. - dispatchTail = dispatch.catch(() => {}); - await dispatch; }, }; }, - invalidate: () => { - lifecycleGeneration += 1; - }, + invalidate: supersedeActive, }; } diff --git a/src/plugins/hook-types.ts b/src/plugins/hook-types.ts index a5c65bf980c8..c1e4bb4e416e 100644 --- a/src/plugins/hook-types.ts +++ b/src/plugins/hook-types.ts @@ -862,6 +862,11 @@ export type PluginHookGatewayContext = { getCron?: () => PluginHookGatewayCronService | undefined; }; +export type PluginHookCronReconciledContext = PluginHookGatewayContext & { + /** Aborts when this exact scheduler snapshot is superseded or the Gateway closes. */ + abortSignal: AbortSignal; +}; + export type PluginHookGatewayStartEvent = { port: number; }; @@ -1283,7 +1288,7 @@ export type PluginHookHandlerMap = { | void; cron_reconciled: ( event: PluginHookCronReconciledEvent, - ctx: PluginHookGatewayContext, + ctx: PluginHookCronReconciledContext, ) => Promise | void; cron_changed: ( event: PluginHookCronChangedEvent, diff --git a/src/plugins/hooks.ts b/src/plugins/hooks.ts index 82a1f0eb05a3..929cc9183b6b 100644 --- a/src/plugins/hooks.ts +++ b/src/plugins/hooks.ts @@ -58,6 +58,7 @@ import type { PluginHeartbeatPromptContributionEvent, PluginHeartbeatPromptContributionResult, PluginHookBeforeAgentRunEvent, + PluginHookCronReconciledContext, PluginHookCronReconciledEvent, PluginHookCronChangedEvent, PluginHookGatewayCronDeliveryStatus, @@ -139,6 +140,7 @@ export type { PluginHookBeforeToolCallEvent, PluginHookBeforeToolCallResult, PluginHookBeforeAgentRunEvent, + PluginHookCronReconciledContext, PluginHookCronReconciledEvent, PluginHookAfterToolCallEvent, PluginHookToolResultPersistContext, @@ -1570,7 +1572,7 @@ export function createHookRunner( */ async function runCronReconciled( event: PluginHookCronReconciledEvent, - ctx: PluginHookGatewayContext, + ctx: PluginHookCronReconciledContext, ): Promise { return runVoidHook("cron_reconciled", event, ctx); } diff --git a/src/plugins/wired-hooks-gateway.test.ts b/src/plugins/wired-hooks-gateway.test.ts index c82598a60b6f..142cc5b6362f 100644 --- a/src/plugins/wired-hooks-gateway.test.ts +++ b/src/plugins/wired-hooks-gateway.test.ts @@ -9,6 +9,7 @@ import { describe, expect, it, vi } from "vitest"; import { createHookRunnerWithRegistry } from "./hooks.test-helpers.js"; import type { PluginHookCronChangedEvent, + PluginHookCronReconciledContext, PluginHookCronReconciledEvent, PluginHookGatewayContext, PluginHookGatewayStartEvent, @@ -47,6 +48,10 @@ describe("gateway hook runner methods", () => { workspaceDir: "/tmp/openclaw-workspace", getCron: () => undefined, }; + const cronReconciledCtx: PluginHookCronReconciledContext = { + ...gatewayCtx, + abortSignal: new AbortController().signal, + }; it.each([ { @@ -93,9 +98,9 @@ describe("gateway hook runner methods", () => { const { runner } = createHookRunnerWithRegistry([{ hookName: "cron_reconciled", handler }]); const event: PluginHookCronReconciledEvent = { reason, enabled }; - await runner.runCronReconciled(event, gatewayCtx); + await runner.runCronReconciled(event, cronReconciledCtx); - expect(handler).toHaveBeenCalledWith(event, gatewayCtx); + expect(handler).toHaveBeenCalledWith(event, cronReconciledCtx); }); it("runCronChanged passes scheduled events with the durable wake snapshot", async () => {