fix(notify): preserve only relevant pending wakes

src/index.ts

@@ -253,6 +253,8 @@ export default function (pi: ExtensionAPI) {
       trigger: entry.trigger,
       timestamp: Date.now(),
       expiresAt: entry.expiresAt,
+      controllerCreatedAt: entry.createdAt,
+      fireCount: entry.fireCount,
       readOnly: entry.readOnly,
       recurring: entry.recurring,
       persistent: entry.recurring,

src/notification-reducer.ts

@@ -1,14 +1,19 @@
-import type { DynamicLoopState, LoopEntry, OrchestrationState, WorkflowRunState } from "./types.js";
+import type { DynamicLoopState, LoopEntry, OrchestrationState, Trigger, WorkflowRunState } from "./types.js";
 
 type ReducerSource = "tool" | "command" | "scheduler" | "eventbus" | "monitor" | "session" | "coordinator" | "system";
 
 export interface ReducerNotification {
   key: string;
   loopId: string;
+  prompt: string;
   message: string;
   timestamp: number;
   expiresAt?: number;
-  trigger: unknown;
+  controllerCreatedAt?: number;
+  fireCount?: number;
+  fireLimitReached?: boolean;
+  workflowStateFireLimitReached?: boolean;
+  trigger: Trigger | string;
   recurring?: boolean;
   persistent?: boolean;
   autoTask?: boolean;

src/runtime/notification-runtime.ts

@@ -33,6 +33,8 @@ export interface LoopFireEvent {
   trigger: Trigger | string;
   timestamp: number;
   expiresAt?: number;
+  controllerCreatedAt?: number;
+  fireCount?: number;
   readOnly?: boolean;
   recurring?: boolean;
   persistent?: boolean;
@@ -347,16 +349,27 @@ export function createNotificationRuntime(options: NotificationRuntimeOptions):
     };
   }
 
-  function workflowNotificationIsCurrent(notification: ReducerNotification): boolean {
+  function refreshWorkflowNotification(notification: ReducerNotification): ReducerNotification | undefined {
     const queued = notification.workflow;
-    if (!queued || !getLoop || notification.controllerStatus === undefined) return true;
+    if (!queued || !getLoop || notification.controllerStatus === undefined) return notification;
     const current = getLoop(notification.loopId);
-    if (!current?.workflow) return false;
-    if (notification.controllerStatus !== undefined && current.status !== notification.controllerStatus) return false;
-    return (current.workflow.definitionRevision ?? 1) === (queued.definitionRevision ?? 1)
-      && current.workflow.currentState === queued.currentState
-      && current.workflow.transitionSeq === queued.transitionSeq
-      && current.workflow.activeExecution?.id === queued.activeExecution?.id;
+    if (!current?.workflow || current.status !== notification.controllerStatus) return undefined;
+    if (current.workflow.currentState !== queued.currentState
+      || current.workflow.transitionSeq !== queued.transitionSeq
+      || current.workflow.activeExecution?.id !== queued.activeExecution?.id) return undefined;
+    if ((current.workflow.definitionRevision ?? 1) === (queued.definitionRevision ?? 1)) return notification;
+    const refreshed = { ...notification, workflow: current.workflow };
+    return { ...refreshed, message: buildLoopFireMessage(refreshed) };
+  }
+
+  function loopNotificationIsCurrent(notification: ReducerNotification): boolean {
+    if (!getLoop || notification.controllerStatus === undefined) return true;
+    const current = getLoop(notification.loopId);
+    if (!current) return notification.recurring === false
+      || (notification.fireLimitReached === true && !notification.workflow && !notification.orchestration);
+    if (notification.controllerCreatedAt !== undefined && current.createdAt !== notification.controllerCreatedAt) return false;
+    if (notification.fireCount !== undefined && (current.fireCount ?? 0) !== notification.fireCount) return false;
+    return current.status === notification.controllerStatus;
   }
 
   function orchestrationNotificationIsCurrent(notification: ReducerNotification): boolean {
@@ -368,7 +381,8 @@ export function createNotificationRuntime(options: NotificationRuntimeOptions):
     return current.orchestration.pendingWake?.sequence === sequence;
   }
 
-  async function deliverNotification(notification: ReducerNotification): Promise<boolean> {
+  async function deliverNotification(queuedNotification: ReducerNotification): Promise<boolean> {
+    let notification = queuedNotification;
     const deliveryGeneration = notification.sessionGeneration ?? sessionGeneration;
     if (notification.autoTask) {
       const pending = await hasPendingTasks();
@@ -391,14 +405,32 @@ export function createNotificationRuntime(options: NotificationRuntimeOptions):
       debug?.(`loop:fire #${notification.loopId} — expiry boundary passed before delivery, dropping wake`);
       return false;
     }
-    if (!workflowNotificationIsCurrent(notification)) {
+    if (!loopNotificationIsCurrent(notification)) {
+      debug?.(`loop:fire #${notification.loopId} — controller changed before delivery, dropping wake`);
+      return false;
+    }
+    const refreshed = refreshWorkflowNotification(notification);
+    if (!refreshed) {
       debug?.(`loop:fire #${notification.loopId} — workflow execution changed before delivery, dropping wake`);
       return false;
     }
+    notification = refreshed;
     if (!orchestrationNotificationIsCurrent(notification)) {
       debug?.(`loop:fire #${notification.loopId} — orchestration wake changed before delivery, dropping wake`);
       return false;
     }
+    if (notificationState.agentRunning) {
+      debug?.(`loop:fire #${notification.loopId} — runtime became busy before delivery, retaining wake`);
+      applyNotificationEvent({
+        type: "NOTIFICATION_QUEUED",
+        at: Date.now(),
+        source: "system",
+        entityType: "notification",
+        entityId: notification.key,
+        payload: { notification },
+      });
+      return false;
+    }
     syncRuntimeState({ agentRunning: true });
     pi.sendMessage({
       customType: "pi-loop",

test/notification-coordinator.test.ts

@@ -11,6 +11,7 @@ function notification(overrides: Partial<ReducerNotification> = {}): ReducerNoti
   return {
     key: "loop:1",
     loopId: "1",
+    prompt: "hello",
     message: "hello",
     timestamp: 100,
     trigger: { type: "cron", schedule: "*/5 * * * *" },

test/notification-runtime.test.ts

@@ -1,10 +1,161 @@
 import { describe, expect, it, vi } from "vitest";
 import { createOrchestrationState } from "../src/orchestration-reducer.js";
 import { createNotificationRuntime } from "../src/runtime/notification-runtime.js";
+import { LoopStore } from "../src/store.js";
 import type { LoopEntry } from "../src/types.js";
 import { createMockPi } from "./helpers/mock-pi.js";
 
 describe("notification runtime session boundary", () => {
+  it.each(["delete", "pause", "complete"] as const)("AUD-04: explicit %s invalidates a buffered loop wake", async (action) => {
+    vi.useFakeTimers();
+    vi.setSystemTime(1_000);
+    try {
+      const { pi, sentMessages } = createMockPi();
+      const store = new LoopStore();
+      const entry = store.create(action === "complete" ? { type: "dynamic" } : { type: "cron", schedule: "* * * * *" }, "Cancelled work", {
+        recurring: true,
+        ...(action === "complete" ? { dynamic: { goal: "Cancelled work", awaitingUpdate: false } } : {}),
+      });
+      const runtime = createNotificationRuntime({
+        pi,
+        hasPendingTasks: async () => 0,
+        cleanDoneTasks: async () => {},
+        getHasPendingMessages: () => false,
+        getLoop: (id) => store.get(id),
+      });
+      runtime.syncRuntimeState({ agentRunning: true });
+      store.fire(entry.id);
+      await runtime.queueOrDeliverNotification({
+        loopId: entry.id,
+        prompt: entry.prompt,
+        trigger: entry.trigger,
+        timestamp: 1_000,
+        expiresAt: entry.expiresAt,
+        recurring: true,
+        dynamic: entry.dynamic,
+        controllerStatus: "active",
+      });
+      expect(sentMessages).toEqual([]);
+      if (action === "pause") store.pause(entry.id);
+      else if (action === "complete") {
+        const current = store.get(entry.id)!;
+        expect(store.stopDynamic(entry.id, "completed", {
+          status: current.status, iteration: current.dynamic!.iteration, updatedAt: current.updatedAt,
+        })).toBe(true);
+      } else store.delete(entry.id);
+      runtime.syncRuntimeState({ agentRunning: false, hasPendingMessages: false });
+      await runtime.flushPendingNotifications({ ignorePendingMessages: true });
+
+      expect(sentMessages).toEqual([]);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("drops a buffered wake after the controller fires again", async () => {
+    const { pi, sentMessages } = createMockPi();
+    const store = new LoopStore();
+    const entry = store.create({ type: "cron", schedule: "* * * * *" }, "Old work", {
+      recurring: true,
+      maxFires: 3,
+    });
+    const first = store.fire(entry.id)!;
+    const runtime = createNotificationRuntime({
+      pi,
+      hasPendingTasks: async () => 0,
+      cleanDoneTasks: async () => {},
+      getHasPendingMessages: () => false,
+      getLoop: (id) => store.get(id),
+    });
+    runtime.syncRuntimeState({ agentRunning: true });
+    await runtime.queueOrDeliverNotification({
+      loopId: first.id,
+      prompt: first.prompt,
+      trigger: first.trigger,
+      timestamp: first.updatedAt,
+      recurring: true,
+      controllerStatus: first.status,
+      fireCount: first.fireCount,
+    });
+    store.fire(entry.id);
+    runtime.syncRuntimeState({ agentRunning: false, hasPendingMessages: false });
+    await runtime.flushPendingNotifications({ ignorePendingMessages: true });
+
+    expect(sentMessages).toEqual([]);
+  });
+
+  it.each(["one-shot", "fire-cap"] as const)("delivers a legitimate %s final fire after controller cleanup", async (kind) => {
+    const { pi, sentMessages } = createMockPi();
+    const store = new LoopStore();
+    const entry = store.create({ type: "cron", schedule: "* * * * *" }, "Final work", {
+      recurring: kind === "fire-cap",
+      ...(kind === "fire-cap" ? { maxFires: 1 } : {}),
+    });
+    const fired = store.fire(entry.id)!;
+    if (kind === "one-shot") store.delete(entry.id);
+    const runtime = createNotificationRuntime({
+      pi,
+      hasPendingTasks: async () => 0,
+      cleanDoneTasks: async () => {},
+      getHasPendingMessages: () => false,
+      getLoop: (id) => store.get(id),
+    });
+    runtime.syncRuntimeState({ agentRunning: true });
+    await runtime.queueOrDeliverNotification({
+      loopId: fired.id,
+      prompt: fired.prompt,
+      trigger: fired.trigger,
+      timestamp: fired.updatedAt,
+      recurring: fired.recurring,
+      controllerStatus: fired.status,
+      fireLimitReached: kind === "fire-cap",
+    });
+    runtime.syncRuntimeState({ agentRunning: false, hasPendingMessages: false });
+    await runtime.flushPendingNotifications({ ignorePendingMessages: true });
+
+    expect(sentMessages).toHaveLength(1);
+    expect(sentMessages[0]?.message.content).toContain("Final work");
+  });
+
+  it("AUD-05: a task lookup finishing while busy retains the wake for exactly one idle delivery", async () => {
+    const { pi, sentMessages } = createMockPi();
+    let entered!: () => void;
+    let release!: (count: number) => void;
+    const lookupEntered = new Promise<void>((resolve) => { entered = resolve; });
+    const lookupReply = new Promise<number>((resolve) => { release = resolve; });
+    const runtime = createNotificationRuntime({
+      pi,
+      hasPendingTasks: () => { entered(); return lookupReply; },
+      cleanDoneTasks: async () => {},
+      getHasPendingMessages: () => false,
+    });
+    runtime.syncRuntimeState({ agentRunning: false, hasPendingMessages: false });
+    const delivering = runtime.queueOrDeliverNotification({
+      loopId: "1",
+      prompt: "Adopt pending work",
+      trigger: { type: "cron", schedule: "* * * * *" },
+      timestamp: 1_000,
+      autoTask: true,
+      recurring: true,
+    });
+    await lookupEntered;
+    runtime.syncRuntimeState({ agentRunning: true });
+    release(1);
+    await delivering;
+    const deliveriesWhileBusy = sentMessages.length;
+
+    runtime.syncRuntimeState({ agentRunning: false, hasPendingMessages: false });
+    await runtime.flushPendingNotifications({ ignorePendingMessages: true });
+    const deliveriesAfterIdle = sentMessages.length;
+    runtime.syncRuntimeState({ agentRunning: false, hasPendingMessages: false });
+    await runtime.flushPendingNotifications({ ignorePendingMessages: true });
+
+    expect.soft(deliveriesWhileBusy).toBe(0);
+    expect.soft(deliveriesAfterIdle - deliveriesWhileBusy).toBe(1);
+    expect.soft(sentMessages).toHaveLength(1);
+    expect.soft(sentMessages[0]?.message.content).toContain("Adopt pending work");
+  });
+
   it("delivers an explicit recurring-loop expiry notification", async () => {
     const { pi, sentMessages } = createMockPi();
     const runtime = createNotificationRuntime({

test/workflow-task-integration.test.ts

@@ -121,6 +121,40 @@ describe("embedded workflow execution integration", () => {
     return { ...harness, ctx, taskPath, loopPath };
   }
 
+  it("AUD-03: revising future work preserves the pending current execution wake at the new revision", async () => {
+    vi.setSystemTime(new Date("2026-01-01T00:00:00Z"));
+    const h = await setup();
+    try {
+      await h.emitExtension("agent_start", null, h.ctx);
+      await h.toolMap.get("WorkflowCreate")!.execute!("create", {
+        goal: "Implement requirements",
+        definition: JSON.stringify(ORIGINAL_REQUIREMENTS_WORKFLOW),
+      });
+      const before = new LoopStore(h.loopPath).get("1")!;
+      expect(before.dynamic?.awaitingUpdate).toBe(true);
+      expect(h.sentMessages).toEqual([]);
+      const revised = await h.toolMap.get("WorkflowRevise")!.execute!("revise", {
+        id: "1",
+        expectedRevision: 1,
+        expectedState: "investigate",
+        expectedTransitionSeq: 0,
+        reason: "Clarify future implementation.",
+        changes: [{ op: "revise_state", stateId: "implement", prompt: "Implement the clarified requirement." }],
+      });
+      expect(revised.content[0].text).toContain("revision 1 → 2");
+      expect(new LoopStore(h.loopPath).get("1")?.workflow?.activeExecution).toEqual(before.workflow?.activeExecution);
+      await h.emitExtension("agent_end", null, h.ctx);
+      await vi.advanceTimersByTimeAsync(30_000);
+
+      expect(h.sentMessages).toHaveLength(1);
+      expect(h.sentMessages[0]?.message.content).toContain("Determine the implementation constraints.");
+      expect(h.sentMessages[0]?.message.content).toContain("Definition revision: 2");
+      expect(new LoopStore(h.loopPath).get("1")?.workflow).toMatchObject({ currentState: "investigate", definitionRevision: 2 });
+    } finally {
+      await h.emitExtension("session_shutdown", null, h.ctx);
+    }
+  });
+
   it("AUD-06: the default budget permits an ordinary phase followed by one cadence fire and completion", async () => {
     vi.setSystemTime(new Date("2026-01-01T00:00:00Z"));
     const h = await setup();