Cron: route reminders by session namespace

src/infra/heartbeat-runner.returns-default-unset.test.ts

@@ -721,6 +721,81 @@ describe("runHeartbeatOnce", () => {
     }
   });
 
+  it("runs heartbeats in forced session key overrides passed at call time", async () => {
+    const tmpDir = await createCaseDir("hb-forced-session-override");
+    const storePath = path.join(tmpDir, "sessions.json");
+    const replySpy = vi.spyOn(replyModule, "getReplyFromConfig");
+    try {
+      const cfg: OpenClawConfig = {
+        agents: {
+          defaults: {
+            workspace: tmpDir,
+            heartbeat: {
+              every: "5m",
+              target: "last",
+            },
+          },
+        },
+        channels: { whatsapp: { allowFrom: ["*"] } },
+        session: { store: storePath },
+      };
+      const mainSessionKey = resolveMainSessionKey(cfg);
+      const agentId = resolveAgentIdFromSessionKey(mainSessionKey);
+      const forcedSessionKey = buildAgentPeerSessionKey({
+        agentId,
+        channel: "whatsapp",
+        peerKind: "dm",
+        peerId: "+15559990000",
+      });
+
+      await fs.writeFile(
+        storePath,
+        JSON.stringify({
+          [mainSessionKey]: {
+            sessionId: "sid-main",
+            updatedAt: Date.now(),
+            lastChannel: "whatsapp",
+            lastTo: "+1555",
+          },
+          [forcedSessionKey]: {
+            sessionId: "sid-forced",
+            updatedAt: Date.now() + 10_000,
+            lastChannel: "whatsapp",
+            lastTo: "+15559990000",
+          },
+        }),
+      );
+
+      replySpy.mockResolvedValue([{ text: "Forced alert" }]);
+      const sendWhatsApp = vi.fn().mockResolvedValue({
+        messageId: "m1",
+        toJid: "jid",
+      });
+
+      await runHeartbeatOnce({
+        cfg,
+        sessionKey: forcedSessionKey,
+        deps: {
+          sendWhatsApp,
+          getQueueSize: () => 0,
+          nowMs: () => 0,
+          webAuthExists: async () => true,
+          hasActiveWebListener: () => true,
+        },
+      });
+
+      expect(sendWhatsApp).toHaveBeenCalledTimes(1);
+      expect(sendWhatsApp).toHaveBeenCalledWith("+15559990000", "Forced alert", expect.any(Object));
+      expect(replySpy).toHaveBeenCalledWith(
+        expect.objectContaining({ SessionKey: forcedSessionKey }),
+        expect.objectContaining({ isHeartbeat: true }),
+        cfg,
+      );
+    } finally {
+      replySpy.mockRestore();
+    }
+  });
+
   it("suppresses duplicate heartbeat payloads within 24h", async () => {
     const tmpDir = await createCaseDir("hb-dup-suppress");
     const storePath = path.join(tmpDir, "sessions.json");

src/infra/heartbeat-runner.scheduler.test.ts

@@ -1,6 +1,7 @@
 import { afterEach, describe, expect, it, vi } from "vitest";
 import type { OpenClawConfig } from "../config/config.js";
 import { startHeartbeatRunner } from "./heartbeat-runner.js";
+import { requestHeartbeatNow, resetHeartbeatWakeStateForTests } from "./heartbeat-wake.js";
 
 describe("startHeartbeatRunner", () => {
   function startDefaultRunner(runOnce: (typeof startHeartbeatRunner)[0]["runOnce"]) {
@@ -13,6 +14,7 @@ describe("startHeartbeatRunner", () => {
   }
 
   afterEach(() => {
+    resetHeartbeatWakeStateForTests();
     vi.useRealTimers();
     vi.restoreAllMocks();
   });
@@ -162,4 +164,42 @@ describe("startHeartbeatRunner", () => {
 
     runner.stop();
   });
+
+  it("routes targeted wake requests to the requested agent/session", async () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date(0));
+
+    const runSpy = vi.fn().mockResolvedValue({ status: "ran", durationMs: 1 });
+    const runner = startHeartbeatRunner({
+      cfg: {
+        agents: {
+          defaults: { heartbeat: { every: "30m" } },
+          list: [
+            { id: "main", heartbeat: { every: "30m" } },
+            { id: "ops", heartbeat: { every: "15m" } },
+          ],
+        },
+      } as OpenClawConfig,
+      runOnce: runSpy,
+    });
+
+    requestHeartbeatNow({
+      reason: "cron:job-123",
+      agentId: "ops",
+      sessionKey: "agent:ops:discord:channel:alerts",
+      coalesceMs: 0,
+    });
+    await vi.advanceTimersByTimeAsync(1);
+
+    expect(runSpy).toHaveBeenCalledTimes(1);
+    expect(runSpy).toHaveBeenCalledWith(
+      expect.objectContaining({
+        agentId: "ops",
+        reason: "cron:job-123",
+        sessionKey: "agent:ops:discord:channel:alerts",
+      }),
+    );
+
+    runner.stop();
+  });
 });

src/infra/heartbeat-runner.ts

@@ -259,6 +259,7 @@ function resolveHeartbeatSession(
   cfg: OpenClawConfig,
   agentId?: string,
   heartbeat?: HeartbeatConfig,
+  forcedSessionKey?: string,
 ) {
   const sessionCfg = cfg.session;
   const scope = sessionCfg?.scope ?? "per-sender";
@@ -276,6 +277,31 @@ function resolveHeartbeatSession(
     return { sessionKey: mainSessionKey, storePath, store, entry: mainEntry };
   }
 
+  const forced = forcedSessionKey?.trim();
+  if (forced) {
+    const forcedCandidate = toAgentStoreSessionKey({
+      agentId: resolvedAgentId,
+      requestKey: forced,
+      mainKey: cfg.session?.mainKey,
+    });
+    const forcedCanonical = canonicalizeMainSessionAlias({
+      cfg,
+      agentId: resolvedAgentId,
+      sessionKey: forcedCandidate,
+    });
+    if (forcedCanonical !== "global") {
+      const sessionAgentId = resolveAgentIdFromSessionKey(forcedCanonical);
+      if (sessionAgentId === normalizeAgentId(resolvedAgentId)) {
+        return {
+          sessionKey: forcedCanonical,
+          storePath,
+          store,
+          entry: store[forcedCanonical],
+        };
+      }
+    }
+  }
+
   const trimmed = heartbeat?.session?.trim() ?? "";
   if (!trimmed) {
     return { sessionKey: mainSessionKey, storePath, store, entry: mainEntry };
@@ -437,6 +463,7 @@ function normalizeHeartbeatReply(
 export async function runHeartbeatOnce(opts: {
   cfg?: OpenClawConfig;
   agentId?: string;
+  sessionKey?: string;
   heartbeat?: HeartbeatConfig;
   reason?: string;
   deps?: HeartbeatDeps;
@@ -493,7 +520,12 @@ export async function runHeartbeatOnce(opts: {
     // The LLM prompt says "if it exists" so this is expected behavior.
   }
 
-  const { entry, sessionKey, storePath } = resolveHeartbeatSession(cfg, agentId, heartbeat);
+  const { entry, sessionKey, storePath } = resolveHeartbeatSession(
+    cfg,
+    agentId,
+    heartbeat,
+    opts.sessionKey,
+  );
   const previousUpdatedAt = entry?.updatedAt;
   const delivery = resolveHeartbeatDeliveryTarget({ cfg, entry, heartbeat });
   const heartbeatAccountId = heartbeat?.accountId?.trim();
@@ -969,11 +1001,45 @@ export function startHeartbeatRunner(opts: {
     }
 
     const reason = params?.reason;
+    const requestedAgentId = params?.agentId ? normalizeAgentId(params.agentId) : undefined;
+    const requestedSessionKey = params?.sessionKey?.trim() || undefined;
     const isInterval = reason === "interval";
     const startedAt = Date.now();
     const now = startedAt;
     let ran = false;
 
+    if (requestedSessionKey || requestedAgentId) {
+      const targetAgentId = requestedAgentId ?? resolveAgentIdFromSessionKey(requestedSessionKey);
+      const targetAgent = state.agents.get(targetAgentId);
+      if (!targetAgent) {
+        scheduleNext();
+        return { status: "skipped", reason: "disabled" };
+      }
+      try {
+        const res = await runOnce({
+          cfg: state.cfg,
+          agentId: targetAgent.agentId,
+          heartbeat: targetAgent.heartbeat,
+          reason,
+          sessionKey: requestedSessionKey,
+          deps: { runtime: state.runtime },
+        });
+        if (res.status !== "skipped" || res.reason !== "disabled") {
+          advanceAgentSchedule(targetAgent, now);
+        }
+        scheduleNext();
+        return res.status === "ran" ? { status: "ran", durationMs: Date.now() - startedAt } : res;
+      } catch (err) {
+        const errMsg = formatErrorMessage(err);
+        log.error(`heartbeat runner: targeted runOnce threw unexpectedly: ${errMsg}`, {
+          error: errMsg,
+        });
+        advanceAgentSchedule(targetAgent, now);
+        scheduleNext();
+        return { status: "failed", reason: errMsg };
+      }
+    }
+
     for (const agent of state.agents.values()) {
       if (isInterval && now < agent.nextDueMs) {
         continue;
@@ -1016,7 +1082,12 @@ export function startHeartbeatRunner(opts: {
     return { status: "skipped", reason: isInterval ? "not-due" : "disabled" };
   };
 
-  const wakeHandler: HeartbeatWakeHandler = async (params) => run({ reason: params.reason });
+  const wakeHandler: HeartbeatWakeHandler = async (params) =>
+    run({
+      reason: params.reason,
+      agentId: params.agentId,
+      sessionKey: params.sessionKey,
+    });
   const disposeWakeHandler = setHeartbeatWakeHandler(wakeHandler);
   updateConfig(state.cfg);
 

src/infra/heartbeat-wake.test.ts

@@ -247,4 +247,36 @@ describe("heartbeat-wake", () => {
     expect(handler).toHaveBeenCalledWith({ reason: "manual" });
     expect(hasPendingHeartbeatWake()).toBe(false);
   });
+
+  it("forwards wake target fields and preserves them across retries", async () => {
+    vi.useFakeTimers();
+    const handler = vi
+      .fn()
+      .mockResolvedValueOnce({ status: "skipped", reason: "requests-in-flight" })
+      .mockResolvedValueOnce({ status: "ran", durationMs: 1 });
+    setHeartbeatWakeHandler(handler);
+
+    requestHeartbeatNow({
+      reason: "cron:job-1",
+      agentId: "ops",
+      sessionKey: "agent:ops:discord:channel:alerts",
+      coalesceMs: 0,
+    });
+
+    await vi.advanceTimersByTimeAsync(1);
+    expect(handler).toHaveBeenCalledTimes(1);
+    expect(handler.mock.calls[0]?.[0]).toEqual({
+      reason: "cron:job-1",
+      agentId: "ops",
+      sessionKey: "agent:ops:discord:channel:alerts",
+    });
+
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(handler).toHaveBeenCalledTimes(2);
+    expect(handler.mock.calls[1]?.[0]).toEqual({
+      reason: "cron:job-1",
+      agentId: "ops",
+      sessionKey: "agent:ops:discord:channel:alerts",
+    });
+  });
 });

src/infra/heartbeat-wake.ts

@@ -3,13 +3,19 @@ export type HeartbeatRunResult =
   | { status: "skipped"; reason: string }
   | { status: "failed"; reason: string };
 
-export type HeartbeatWakeHandler = (opts: { reason?: string }) => Promise<HeartbeatRunResult>;
+export type HeartbeatWakeHandler = (opts: {
+  reason?: string;
+  agentId?: string;
+  sessionKey?: string;
+}) => Promise<HeartbeatRunResult>;
 
 type WakeTimerKind = "normal" | "retry";
 type PendingWakeReason = {
   reason: string;
   priority: number;
   requestedAt: number;
+  agentId?: string;
+  sessionKey?: string;
 };
 
 let handler: HeartbeatWakeHandler | null = null;
@@ -56,12 +62,25 @@ function normalizeWakeReason(reason?: string): string {
   return trimmed.length > 0 ? trimmed : "requested";
 }
 
-function queuePendingWakeReason(reason?: string, requestedAt = Date.now()) {
-  const normalizedReason = normalizeWakeReason(reason);
+function normalizeWakeTarget(value?: string): string | undefined {
+  const trimmed = typeof value === "string" ? value.trim() : "";
+  return trimmed || undefined;
+}
+
+function queuePendingWakeReason(params?: {
+  reason?: string;
+  requestedAt?: number;
+  agentId?: string;
+  sessionKey?: string;
+}) {
+  const requestedAt = params?.requestedAt ?? Date.now();
+  const normalizedReason = normalizeWakeReason(params?.reason);
   const next: PendingWakeReason = {
     reason: normalizedReason,
     priority: resolveReasonPriority(normalizedReason),
     requestedAt,
+    agentId: normalizeWakeTarget(params?.agentId),
+    sessionKey: normalizeWakeTarget(params?.sessionKey),
   };
   if (!pendingWake) {
     pendingWake = next;
@@ -113,18 +132,33 @@ function schedule(coalesceMs: number, kind: WakeTimerKind = "normal") {
     }
 
     const reason = pendingWake?.reason;
+    const agentId = pendingWake?.agentId;
+    const sessionKey = pendingWake?.sessionKey;
     pendingWake = null;
     running = true;
     try {
-      const res = await active({ reason: reason ?? undefined });
+      const wakeOpts = {
+        reason: reason ?? undefined,
+        ...(agentId ? { agentId } : {}),
+        ...(sessionKey ? { sessionKey } : {}),
+      };
+      const res = await active(wakeOpts);
       if (res.status === "skipped" && res.reason === "requests-in-flight") {
         // The main lane is busy; retry soon.
-        queuePendingWakeReason(reason ?? "retry");
+        queuePendingWakeReason({
+          reason: reason ?? "retry",
+          agentId,
+          sessionKey,
+        });
         schedule(DEFAULT_RETRY_MS, "retry");
       }
     } catch {
       // Error is already logged by the heartbeat runner; schedule a retry.
-      queuePendingWakeReason(reason ?? "retry");
+      queuePendingWakeReason({
+        reason: reason ?? "retry",
+        agentId,
+        sessionKey,
+      });
       schedule(DEFAULT_RETRY_MS, "retry");
     } finally {
       running = false;
@@ -178,8 +212,17 @@ export function setHeartbeatWakeHandler(next: HeartbeatWakeHandler | null): () =
   };
 }
 
-export function requestHeartbeatNow(opts?: { reason?: string; coalesceMs?: number }) {
-  queuePendingWakeReason(opts?.reason);
+export function requestHeartbeatNow(opts?: {
+  reason?: string;
+  coalesceMs?: number;
+  agentId?: string;
+  sessionKey?: string;
+}) {
+  queuePendingWakeReason({
+    reason: opts?.reason,
+    agentId: opts?.agentId,
+    sessionKey: opts?.sessionKey,
+  });
   schedule(opts?.coalesceMs ?? DEFAULT_COALESCE_MS, "normal");
 }