fix(scheduler): Dispatch scheduled tasks through agent runtime

Scheduled task heartbeats now dispatch the stored task text through the public agent dispatch API with explicit source, destination, and metadata instead of compiling a scheduler-specific prompt. Core dispatch handling renders generic dispatch context and relies on normal final-answer delivery for API dispatches.

Add focused scheduler eval coverage for schedule creation and due task execution, plus an ast-grep guard against opaque dispatch context blobs.

Fixes GH-611
Co-Authored-By: GPT-5 Codex <noreply@openai.com>

Author: dcramer

ast-grep/rules/no-plugin-dispatch-context.yml

@@ -0,0 +1,9 @@
+id: no-plugin-dispatch-context
+language: TypeScript
+severity: error
+message: Plugin agent dispatch must use typed source/destination/metadata, not an opaque context prompt blob.
+files:
+  - packages/**/*.ts
+  - packages/**/*.tsx
+rule:
+  pattern: "$CTX.agent.dispatch({ $$$BEFORE, context: $VALUE, $$$AFTER })"

packages/junior-evals/README.md

@@ -78,7 +78,7 @@ These knobs work by overriding services on the eval-local runtime instance. They
 
 Tool replay:
 
-- `webFetch` and `webSearch` are wrapped with `vitest-evals/replay` in the eval harness. `pnpm evals` uses `auto` replay mode; use `pnpm evals:record` to force fresh recordings under `.vitest-evals/recordings`.
+- `webFetch` and `webSearch` are wrapped with `vitest-evals/replay` in the eval harness. Use `pnpm evals:record` to force fresh recordings under `.vitest-evals/recordings`.
 - Keep committed recordings minimal and source-specific. Regenerate them from the evals that need replay, then review for stale exploratory fetches and secret-like values before committing.
 
 ## Running

packages/junior-evals/evals/behavior-harness.ts

@@ -45,7 +45,16 @@ import {
 import { generateAssistantReply } from "@/chat/respond";
 import type { JuniorDatabase } from "@/chat/sql/db";
 import { juniorSqlSchema } from "@/chat/sql/schema";
-import { schedulerPlugin } from "@sentry/junior-scheduler";
+import {
+  createSchedulerSqlStore,
+  schedulerPlugin,
+  type ScheduledTask,
+} from "@sentry/junior-scheduler";
+import { runPluginHeartbeats } from "@/chat/agent-dispatch/heartbeat";
+import { runAgentDispatchSlice } from "@/chat/agent-dispatch/runner";
+import { verifyDispatchCallbackRequest } from "@/chat/agent-dispatch/signing";
+import { getDispatchRecord } from "@/chat/agent-dispatch/store";
+import type { DispatchCallback } from "@/chat/agent-dispatch/types";
 import { getStateAdapter } from "@/chat/state/adapter";
 import { resetSkillDiscoveryCache } from "@/chat/skills";
 import { createWebFetchTool } from "@/chat/tools/web/fetch-tool";
@@ -130,11 +139,22 @@ interface AssistantContextChangedEvent extends EvalBaseEvent {
   user_id?: string;
 }
 
+interface ScheduledTaskDueEvent extends EvalBaseEvent {
+  type: "scheduled_task_due";
+  now_ms?: number;
+  recurrence?: "daily" | "weekly" | "monthly" | "yearly";
+  schedule?: string;
+  schedule_kind?: "one_off" | "recurring";
+  task_text: string;
+  timezone?: string;
+}
+
 export type EvalEvent =
   | MentionEvent
   | SubscribedMessageEvent
   | AssistantThreadStartedEvent
-  | AssistantContextChangedEvent;
+  | AssistantContextChangedEvent
+  | ScheduledTaskDueEvent;
 
 interface SubscribedDecisionFixture {
   reason: string;
@@ -1276,7 +1296,7 @@ async function setupHarnessEnvironment(
       scenario.events.flatMap((event) =>
         "message" in event
           ? [event.message.author?.user_id?.trim() || "U-test"]
-          : event.user_id
+          : "user_id" in event && event.user_id
             ? [event.user_id]
             : [],
       ),
@@ -1569,12 +1589,21 @@ function buildRuntimeServices(
 async function processEvents(args: {
   scenario: EvalScenario;
   env: HarnessEnvironment;
+  generateAssistantReply: typeof generateAssistantReply;
   slackRuntime: ReturnType<typeof createSlackRuntime>;
   getThreadRecord: (fixture: EvalEventThreadFixture) => EvalThreadRecord;
   readyQueueDeliveries: QueueDelivery[];
+  schedulerDb: PluginDb;
 }): Promise<void> {
-  const { scenario, env, slackRuntime, getThreadRecord, readyQueueDeliveries } =
-    args;
+  const {
+    scenario,
+    env,
+    generateAssistantReply,
+    slackRuntime,
+    getThreadRecord,
+    readyQueueDeliveries,
+    schedulerDb,
+  } = args;
 
   const consumedOauthStates = new Set<string>();
   const consumedMcpAuthSessions = new Set<string>();
@@ -1650,9 +1679,113 @@ async function processEvents(args: {
     await slackRuntime.handleAssistantContextChanged(lifecycleEvent);
   };
 
+  const runScheduledTaskDue = async (
+    event: ScheduledTaskDueEvent,
+  ): Promise<void> => {
+    const { thread } = getThreadRecord(event.thread);
+    const nowMs = event.now_ms ?? Date.parse("2026-05-26T12:00:00.000Z");
+    const scheduleKind = event.schedule_kind ?? "one_off";
+    const taskId = `eval_schedule_${thread.channelId}_${nowMs}`;
+    const task: ScheduledTask = {
+      id: taskId,
+      createdAtMs: nowMs - 60_000,
+      createdBy: { slackUserId: "U-test", userName: "testuser" },
+      destination: createEvalDestination(
+        thread,
+      ) as ScheduledTask["destination"],
+      nextRunAtMs: nowMs,
+      schedule: {
+        description:
+          event.schedule ??
+          (scheduleKind === "recurring" ? "Weekly at noon" : "Once now"),
+        kind: scheduleKind,
+        timezone: event.timezone ?? "UTC",
+        ...(scheduleKind === "recurring"
+          ? {
+              recurrence: {
+                frequency: event.recurrence ?? "weekly",
+                interval: 1,
+                startDate: new Date(nowMs).toISOString().slice(0, 10),
+                time: { hour: 12, minute: 0 },
+              },
+            }
+          : {}),
+      },
+      status: "active",
+      task: { text: event.task_text },
+      updatedAtMs: nowMs - 60_000,
+    };
+    const schedulerStore = createSchedulerSqlStore(schedulerDb);
+    await schedulerStore.saveTask(task);
+
+    const callbacks: DispatchCallback[] = [];
+    const expectedCallbackUrl = new URL(
+      "/api/internal/agent-dispatch",
+      process.env.JUNIOR_BASE_URL,
+    ).href;
+    const originalFetch = globalThis.fetch;
+    globalThis.fetch = (async (input, init) => {
+      const url =
+        typeof input === "string"
+          ? input
+          : input instanceof URL
+            ? input.href
+            : input.url;
+      if (new URL(url).href === expectedCallbackUrl) {
+        const callback = await verifyDispatchCallbackRequest(
+          new Request(input, init),
+        );
+        if (!callback) {
+          return new Response("Unauthorized", { status: 401 });
+        }
+        callbacks.push(callback);
+        return new Response("Accepted", { status: 202 });
+      }
+      return await originalFetch(input, init);
+    }) as typeof fetch;
+    try {
+      await runPluginHeartbeats({ nowMs });
+    } finally {
+      globalThis.fetch = originalFetch;
+    }
+    if (callbacks.length === 0) {
+      throw new Error(
+        "Scheduled eval task did not enqueue a dispatch callback.",
+      );
+    }
+
+    const dispatchedRuns = (await schedulerStore.listIncompleteRuns()).filter(
+      (run) => run.taskId === taskId && run.dispatchId,
+    );
+    if (dispatchedRuns.length === 0) {
+      const runs = (await schedulerStore.listIncompleteRuns()).filter(
+        (run) => run.taskId === taskId,
+      );
+      const savedTask = await schedulerStore.getTask(taskId);
+      throw new Error(
+        `Scheduled eval task did not create a dispatch: ${JSON.stringify({ runs, savedTask })}`,
+      );
+    }
+    for (const run of dispatchedRuns) {
+      const dispatch = await getDispatchRecord(run.dispatchId!);
+      if (!dispatch) {
+        throw new Error("Scheduled eval dispatch record was not found.");
+      }
+      const callback = callbacks.find(
+        (candidate) => candidate.id === dispatch.id,
+      );
+      if (!callback) {
+        throw new Error("Scheduled eval dispatch callback was not captured.");
+      }
+      await runAgentDispatchSlice(callback, { generateAssistantReply });
+    }
+  };
+
   for (const event of scenario.events) {
     if (event.type === "new_mention" || event.type === "subscribed_message") {
       enqueueEvent(event);
+    } else if (event.type === "scheduled_task_due") {
+      await runScheduledTaskDue(event);
     } else {
       await runLifecycleEvent(event);
     }
@@ -1803,6 +1936,11 @@ export async function runEvalScenario(
       threadRecordsById,
       observations,
     );
+    const generateEvalAssistantReply =
+      services.replyExecutor?.generateAssistantReply;
+    if (!generateEvalAssistantReply) {
+      throw new Error("Eval reply executor was not configured.");
+    }
 
     const slackRuntime = createSlackRuntime({
       getSlackAdapter: () => slackAdapter as any,
@@ -1812,9 +1950,11 @@ export async function runEvalScenario(
     await processEvents({
       scenario,
       env,
+      generateAssistantReply: generateEvalAssistantReply,
       slackRuntime,
       getThreadRecord,
       readyQueueDeliveries,
+      schedulerDb: schedulerSql.db,
     });
 
     return collectResults(

packages/junior-evals/evals/core/scheduler.eval.ts

@@ -1,72 +1,87 @@
 import { describeEval } from "vitest-evals";
-import { mention, rubric, slackEvals } from "../helpers";
+import { mention, rubric, scheduledTaskDue, slackEvals } from "../helpers";
 
 describeEval("Scheduler", slackEvals, (it) => {
-  it("when asked for a simple one-off reminder, create it without asking for confirmation", async ({
+  it("when asked for a specific one-off reminder, preserve the future work in the schedule", async ({
     run,
   }) => {
     await run({
-      events: [mention("@bot remind me in 1 minute to wash my hands")],
+      events: [
+        mention(
+          "@bot remind me in 2 minutes to tell the channel standup moved",
+        ),
+      ],
       criteria: rubric({
         contract:
-          "A simple one-off reminder request is scheduled immediately for the active Slack context.",
+          "A one-off reminder request is scheduled with the future reminder work preserved as the task.",
         pass: [
-          "The reply confirms that a one-off reminder to wash hands was scheduled.",
-          "The schedule creation omits recurrence.",
-          "The reply does not ask the user to confirm first.",
+          "The observed slackScheduleCreateTask tool call has schedule_kind=one_off.",
+          "The observed slackScheduleCreateTask tool call omits recurrence.",
+          "The observed slackScheduleCreateTask task is the reminder work to perform later, not instructions for how to create or manage a schedule.",
         ],
         fail: [
-          "Do not ask the user to confirm the reminder before creating it.",
-          "Do not ask the user to provide a channel ID.",
-          "Do not describe the reminder as a recurring schedule.",
+          "Do not store task text that tells Junior to schedule a reminder later.",
+          "Do not ask the user to confirm before creating this clear reminder.",
         ],
       }),
     });
   });
 
-  it("when asked for a terse one-off reminder, create it without recurrence", async ({
+  it("when a one-off reminder becomes due, deliver the reminder outcome", async ({
     run,
   }) => {
     await run({
-      events: [mention("@bot remind me to drink water in 1m")],
+      events: [
+        scheduledTaskDue("Post this reminder: Standup moved to 10:30 today.", {
+          schedule: "Once at noon UTC",
+          schedule_kind: "one_off",
+        }),
+      ],
       criteria: rubric({
         contract:
-          "A terse one-off reminder request is scheduled immediately for the active Slack context.",
+          "A due one-off scheduled task is executed now and posts the requested reminder outcome to the destination channel.",
         pass: [
-          "The reply confirms that a one-off reminder to drink water was scheduled.",
-          "The schedule creation omits recurrence.",
-          "The reply does not ask the user to retry with a different one-time format.",
+          "The channel_posts output contains a Slack channel message saying standup moved to 10:30 today.",
+          "The delivered message is the reminder content itself, not a schedule creation confirmation.",
+          "The delivered message does not ask for clarification or confirmation.",
         ],
         fail: [
-          "Do not reject the request as an invalid one-off task format.",
-          "Do not ask the user to confirm the reminder before creating it.",
-          "Do not describe the reminder as a recurring schedule.",
+          "Do not use webSearch, webFetch, bash, callMcpTool, sandbox, or Slack history tools for this reminder-only task.",
+          "Do not say that a reminder was scheduled or will be scheduled.",
+          "Do not omit the 10:30 standup update.",
+          "Do not ask the user what to do with the reminder.",
         ],
       }),
     });
   });
 
-  it("when asked to schedule clear recurring work, create it without confirmation", async ({
+  it("when a recurring scheduled task becomes due, deliver that occurrence", async ({
     run,
   }) => {
     await run({
       events: [
-        mention(
-          "@bot schedule this every Monday at 9am Pacific: check open GitHub issues about the scheduler and post a short digest here.",
+        scheduledTaskDue(
+          "Post this reminder: Submit timesheets by 5pm today.",
+          {
+            recurrence: "weekly",
+            schedule: "Weekly on Monday at noon UTC",
+            schedule_kind: "recurring",
+          },
         ),
       ],
       criteria: rubric({
         contract:
-          "A clear future or recurring task request is normalized and scheduled immediately for the active Slack context.",
+          "A due recurring scheduled task is executed for the current occurrence and posts the requested reminder outcome to the destination channel.",
         pass: [
-          "The created task describes checking scheduler-related GitHub issues, not creating a schedule.",
-          "The schedule creation sets recurrence=weekly.",
-          "The reply confirms the recurring schedule was created for Monday at 9am Pacific.",
+          "The channel_posts output contains a Slack channel message reminding people to submit timesheets by 5pm today.",
+          "The delivered message treats this as the current due occurrence.",
+          "The delivered message is not just a confirmation that a recurring task exists.",
         ],
         fail: [
-          "Do not ask the user to confirm before creating the clear recurring task.",
-          "Do not ask the user to provide a channel ID.",
-          "Do not only give instructions for how the user can set up an external cron.",
+          "Do not use webSearch, webFetch, bash, callMcpTool, sandbox, or Slack history tools for this reminder-only task.",
+          "Do not say only that a weekly reminder was scheduled.",
+          "Do not omit the timesheets by 5pm content.",
+          "Do not ask the user to confirm the recurring task before posting.",
         ],
       }),
     });

packages/junior-evals/evals/helpers.ts

@@ -518,6 +518,36 @@ export function threadMessage(
   };
 }
 
+/** Builds an event for a scheduled task becoming due and dispatching output. */
+export function scheduledTaskDue(
+  taskText: string,
+  opts?: {
+    now_ms?: number;
+    recurrence?: "daily" | "weekly" | "monthly" | "yearly";
+    schedule?: string;
+    schedule_kind?: "one_off" | "recurring";
+    thread?: ThreadOverrides;
+    timezone?: string;
+  },
+) {
+  const seq = nextId();
+  return {
+    type: "scheduled_task_due" as const,
+    thread: {
+      id: `thread-${seq}`,
+      channel_id: `C${seq}`,
+      thread_ts: `17000000.${seq}`,
+      ...opts?.thread,
+    },
+    task_text: taskText,
+    ...(opts?.now_ms ? { now_ms: opts.now_ms } : {}),
+    ...(opts?.recurrence ? { recurrence: opts.recurrence } : {}),
+    ...(opts?.schedule ? { schedule: opts.schedule } : {}),
+    ...(opts?.schedule_kind ? { schedule_kind: opts.schedule_kind } : {}),
+    ...(opts?.timezone ? { timezone: opts.timezone } : {}),
+  };
+}
+
 /** Builds an assistant thread lifecycle start event for a harnessed Slack eval. */
 export function threadStart(opts?: {
   thread?: ThreadOverrides;

packages/junior-evals/package.json

@@ -5,8 +5,8 @@
   "type": "module",
   "scripts": {
     "test": "vitest run",
-    "evals": "JUNIOR_STATE_ADAPTER=memory VITEST_EVALS_REPLAY_MODE=auto pnpm exec vitest run -c vitest.evals.config.ts",
-    "evals:record": "JUNIOR_STATE_ADAPTER=memory VITEST_EVALS_REPLAY_MODE=record pnpm exec vitest run -c vitest.evals.config.ts"
+    "evals": "pnpm exec vitest run -c vitest.evals.config.ts",
+    "evals:record": "VITEST_EVALS_REPLAY_MODE=record pnpm exec vitest run -c vitest.evals.config.ts"
   },
   "devDependencies": {
     "@sentry/junior": "workspace:*",