fix(cron): abort superseded reconciliation hooks (#104368)

* fix(cron): abort superseded reconciliation hooks

* fix(ci): align plugin SDK surface budget

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();
   });
 

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,

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(

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<RunHook>(async (event) => {
+    const runHook = vi.fn<RunHook>(async (event, ctx) => {
       order.push(`${event.reason}:start`);
       if (event.reason === "startup") {
+        startupSignal = ctx.abortSignal;
         await new Promise<void>((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<RunHook>(async (_event, ctx) => {
+      activeSignal = ctx.abortSignal;
+      await new Promise<void>((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;
   });
 });
 

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<void>;
+  runHook: (
+    event: PluginHookCronReconciledEvent,
+    ctx: PluginHookCronReconciledContext,
+  ) => Promise<void>;
 }): 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,
   };
 }

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> | void;
   cron_changed: (
     event: PluginHookCronChangedEvent,

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<void> {
     return runVoidHook("cron_reconciled", event, ctx);
   }

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 () => {