fix(scheduling): re-land #11259 scheduler runtime (tz/cron/event-bridge) clobbered by #11271 (#11413)

plugins/plugin-scheduling/src/scheduled-task/after-task-chain.test.ts

@@ -1,18 +1,21 @@
 /**
- * A9 — `trigger.kind: "after_task"` chain-after-terminal coverage.
+ * `trigger.kind: "after_task"` — chain-after-terminal contract.
  *
- * The spine schema accepts `trigger: { kind: "after_task"; taskId; outcome }`
- * with `outcome: TerminalState` (`completed | skipped | expired | failed |
- * dismissed`). The runner does NOT auto-fire from `after_task` triggers — the
- * scheduler tick is the entry point, and the in-memory test fixture has no
- * tick. These tests therefore lock down the **structural acceptance** plus
- * the schedule-time semantics: the child task is persisted with the
- * `after_task` trigger pointing at the parent, and remains in `scheduled`
- * even after the parent reaches the recorded terminal outcome.
+ * CONTRACT CHANGE (#10721/#10723 trigger-realness wave): the previous
+ * revision of this file pinned that the runner does NOT auto-fire `after_task`
+ * children, making the trigger kind contract larp — schema-accepted, never
+ * fired by anything. The trigger is live in the SCHEDULED_TASKS action (users
+ * create chains from chat without editing the parent) and is NOT redundant
+ * with `pipeline.on*`: pipeline refs are declared on the PARENT and only
+ * propagate completed/skipped/failed, while `after_task` is declared on the
+ * CHILD and covers all five terminal outcomes. The runner now fires matching
+ * children on the parent's terminal transition (`settleTerminal` /
+ * `fireAfterTaskChildren` in runner.ts), race-safe via the store's atomic
+ * fire claim.
  *
- * If the runner gains an `after_task` evaluator, the
- * "child does not auto-fire" assertion below is the contract that needs to
- * change first — making this file the canonical seam.
+ * These tests lock the new contract: structural acceptance, auto-fire on the
+ * matching outcome, no fire on mismatched outcome/parent, and the documented
+ * global-pause exception (pause suppresses chaining).
  */
 
 import { describe, expect, it } from "vitest";
@@ -42,7 +45,7 @@ import {
 import { createInMemoryScheduledTaskLogStore } from "./state-log.js";
 import type { GlobalPauseView, ScheduledTask, TerminalState } from "./types.js";
 
-function makeRunner(): {
+function makeRunner(opts: { pauseActive?: boolean } = {}): {
   runner: ScheduledTaskRunnerHandle;
   setNow: (iso: string) => void;
 } {
@@ -65,7 +68,7 @@ function makeRunner(): {
     consolidation: createConsolidationRegistry(),
     ownerFacts: () => ({}),
     globalPause: {
-      current: async () => ({ active: false }),
+      current: async () => ({ active: opts.pauseActive === true }),
     } as GlobalPauseView,
     activity: { hasSignalSince: () => false },
     subjectStore: { wasUpdatedSince: () => false },
@@ -122,6 +125,15 @@ async function forceParentTerminal(
   }
 }
 
+async function getTask(
+  runner: ScheduledTaskRunnerHandle,
+  taskId: string,
+): Promise<ScheduledTask> {
+  const found = (await runner.list()).find((t) => t.taskId === taskId);
+  if (!found) throw new Error(`task ${taskId} not found`);
+  return found;
+}
+
 const TERMINAL_OUTCOMES: TerminalState[] = [
   "completed",
   "skipped",
@@ -150,9 +162,11 @@ describe("ScheduledTaskRunner — after_task trigger structural acceptance (A9)"
       expect(child.trigger.outcome).toBe(outcome);
     });
   }
+});
 
+describe("ScheduledTaskRunner — after_task children fire on the parent's terminal transition", () => {
   for (const outcome of TERMINAL_OUTCOMES) {
-    it(`child does NOT auto-fire when the parent reaches ${outcome} (no scheduler-tick in fixture)`, async () => {
+    it(`auto-fires the child when the parent reaches ${outcome}`, async () => {
       const { runner } = makeRunner();
       const parent = await runner.schedule(baseInput());
       const child = await runner.schedule(
@@ -164,18 +178,139 @@ describe("ScheduledTaskRunner — after_task trigger structural acceptance (A9)"
 
       await forceParentTerminal(runner, parent.taskId, outcome);
 
-      const reloadedParent = (await runner.list()).find(
-        (t) => t.taskId === parent.taskId,
-      );
-      expect(reloadedParent?.state.status).toBe(outcome);
-
-      const reloadedChild = (await runner.list()).find(
-        (t) => t.taskId === child.taskId,
-      );
-      expect(reloadedChild?.state.status).toBe("scheduled");
-      expect(reloadedChild?.state.firedAt).toBeUndefined();
+      expect((await getTask(runner, parent.taskId)).state.status).toBe(outcome);
+      const reloadedChild = await getTask(runner, child.taskId);
+      expect(reloadedChild.state.status).toBe("fired");
+      expect(reloadedChild.state.firedAt).toBe("2026-05-09T12:00:00.000Z");
     });
   }
+
+  it("does not fire a child whose recorded outcome mismatches the transition", async () => {
+    const { runner } = makeRunner();
+    const parent = await runner.schedule(baseInput());
+    const onFailure = await runner.schedule(
+      baseInput({
+        promptInstructions: "chain after failed",
+        trigger: {
+          kind: "after_task",
+          taskId: parent.taskId,
+          outcome: "failed",
+        },
+      }),
+    );
+
+    await runner.apply(parent.taskId, "complete", { reason: "test" });
+
+    expect((await getTask(runner, onFailure.taskId)).state.status).toBe(
+      "scheduled",
+    );
+  });
+
+  it("does not fire a child pointing at a different parent", async () => {
+    const { runner } = makeRunner();
+    const parent = await runner.schedule(baseInput());
+    const otherParent = await runner.schedule(baseInput());
+    const child = await runner.schedule(
+      baseInput({
+        trigger: {
+          kind: "after_task",
+          taskId: otherParent.taskId,
+          outcome: "completed",
+        },
+      }),
+    );
+
+    await runner.apply(parent.taskId, "complete", { reason: "test" });
+
+    expect((await getTask(runner, child.taskId)).state.status).toBe(
+      "scheduled",
+    );
+  });
+
+  it("fires every child chained on the same parent + outcome", async () => {
+    const { runner } = makeRunner();
+    const parent = await runner.schedule(baseInput());
+    const first = await runner.schedule(
+      baseInput({
+        trigger: {
+          kind: "after_task",
+          taskId: parent.taskId,
+          outcome: "completed",
+        },
+      }),
+    );
+    const second = await runner.schedule(
+      baseInput({
+        trigger: {
+          kind: "after_task",
+          taskId: parent.taskId,
+          outcome: "completed",
+        },
+      }),
+    );
+
+    await runner.apply(parent.taskId, "complete", { reason: "test" });
+
+    expect((await getTask(runner, first.taskId)).state.status).toBe("fired");
+    expect((await getTask(runner, second.taskId)).state.status).toBe("fired");
+  });
+
+  it("chains transitively: grandchild fires when the fired child later completes", async () => {
+    const { runner } = makeRunner();
+    const parent = await runner.schedule(baseInput());
+    const child = await runner.schedule(
+      baseInput({
+        trigger: {
+          kind: "after_task",
+          taskId: parent.taskId,
+          outcome: "completed",
+        },
+      }),
+    );
+    const grandchild = await runner.schedule(
+      baseInput({
+        trigger: {
+          kind: "after_task",
+          taskId: child.taskId,
+          outcome: "completed",
+        },
+      }),
+    );
+
+    await runner.apply(parent.taskId, "complete", { reason: "test" });
+    expect((await getTask(runner, child.taskId)).state.status).toBe("fired");
+    expect((await getTask(runner, grandchild.taskId)).state.status).toBe(
+      "scheduled",
+    );
+
+    await runner.apply(child.taskId, "complete", { reason: "test" });
+    expect((await getTask(runner, grandchild.taskId)).state.status).toBe(
+      "fired",
+    );
+  });
+
+  it("global-pause skip does NOT chain: pause suppresses proactive behavior", async () => {
+    const { runner } = makeRunner({ pauseActive: true });
+    const parent = await runner.schedule(
+      baseInput({ respectsGlobalPause: true }),
+    );
+    const child = await runner.schedule(
+      baseInput({
+        trigger: {
+          kind: "after_task",
+          taskId: parent.taskId,
+          outcome: "skipped",
+        },
+      }),
+    );
+
+    const result = await runner.fireWithResult(parent.taskId);
+    expect(result.kind).toBe("skipped");
+    expect((await getTask(runner, parent.taskId)).state.status).toBe("skipped");
+    expect((await getTask(runner, child.taskId)).state.status).toBe(
+      "scheduled",
+    );
+  });
 });
 
 describe("ScheduledTaskRunner — after_task trigger fire path is verb-driven (A9)", () => {

plugins/plugin-scheduling/src/scheduled-task/due.test.ts

@@ -160,3 +160,59 @@ describe("ScheduledTask due evaluation", () => {
     expect(pendingPromptRoomIdForTask(fired)).toBe("room-1");
   });
 });
+
+describe("owner_local cron tz resolution", () => {
+  // 2026-05-10 is inside US daylight time: America/Denver = UTC-6.
+  const trigger = {
+    kind: "cron",
+    expression: "0 9 * * *",
+    tz: "owner_local",
+  } as const;
+
+  it("is NOT due at 09:05 UTC when the owner is in America/Denver (03:05 local)", async () => {
+    const decision = await isScheduledTaskDue(
+      task({
+        trigger,
+        metadata: { createdAtIso: "2026-05-10T08:00:00.000Z" },
+      }),
+      {
+        now: new Date("2026-05-10T09:05:00.000Z"),
+        ownerFacts: { timezone: "America/Denver" },
+      },
+    );
+    expect(decision).toMatchObject({ due: false, reason: "cron_pending" });
+  });
+
+  it("is due at the Denver hour (15:00 UTC) — not the UTC hour", async () => {
+    const decision = await isScheduledTaskDue(
+      task({
+        trigger,
+        metadata: { createdAtIso: "2026-05-10T14:00:00.000Z" },
+      }),
+      {
+        now: new Date("2026-05-10T15:05:00.000Z"),
+        ownerFacts: { timezone: "America/Denver" },
+      },
+    );
+    expect(decision).toMatchObject({
+      due: true,
+      reason: "cron_due",
+      occurrenceAtIso: "2026-05-10T15:00:00.000Z",
+    });
+  });
+
+  it("resolves owner_local to UTC when the owner has no timezone on file", async () => {
+    const decision = await isScheduledTaskDue(
+      task({
+        trigger,
+        metadata: { createdAtIso: "2026-05-10T08:00:00.000Z" },
+      }),
+      { now: new Date("2026-05-10T09:05:00.000Z"), ownerFacts: {} },
+    );
+    expect(decision).toMatchObject({
+      due: true,
+      reason: "cron_due",
+      occurrenceAtIso: "2026-05-10T09:00:00.000Z",
+    });
+  });
+});

plugins/plugin-scheduling/src/scheduled-task/due.ts

@@ -1,6 +1,7 @@
 import { computeNextCronRunAtMs } from "@elizaos/core";
 
 import type { AnchorRegistry } from "../anchors/anchor-registry.js";
+import { resolveTriggerTz } from "./trigger-tz.js";
 import type {
   OwnerFactsView,
   ScheduledTask,
@@ -232,6 +233,7 @@ function cronDue(
   task: ScheduledTask,
   trigger: Extract<ScheduledTaskTrigger, { kind: "cron" }>,
   nowMs: number,
+  ownerFacts: OwnerFactsView | undefined,
 ): ScheduledTaskDueDecision {
   const baseMs =
     parseIsoMs(task.state.firedAt) ??
@@ -246,7 +248,11 @@ function cronDue(
   if (baseMs > nowMs) {
     return { due: false, reason: "cron_pending" };
   }
-  const nextMs = computeNextCronRunAtMs(trigger.expression, baseMs, trigger.tz);
+  const nextMs = computeNextCronRunAtMs(
+    trigger.expression,
+    baseMs,
+    resolveTriggerTz(trigger.tz, ownerFacts),
+  );
   if (nextMs === null) return { due: false, reason: "cron_invalid" };
   return nextMs <= nowMs
     ? {
@@ -443,7 +449,7 @@ export async function isScheduledTaskDue(
     case "interval":
       return intervalDue(task, task.trigger, nowMs);
     case "cron":
-      return cronDue(task, task.trigger, nowMs);
+      return cronDue(task, task.trigger, nowMs, context.ownerFacts);
     case "relative_to_anchor":
       return relativeAnchorDue(task, task.trigger, context, nowMs);
     case "during_window":