ref(dispatch): route dispatches through conversation work

packages/docs/src/content/docs/extend/scheduler-plugin.md

@@ -80,7 +80,7 @@ For recurring or non-reminder scheduled work, Junior should show the proposed ta
 ## Failure modes
 
 - No due tasks run: confirm `/api/internal/heartbeat` is called every minute and the route secret matches the configured bearer token.
-- Tasks list but never complete: check scheduler and dispatch logs for missing Slack destination fields or stale dispatch recovery errors.
+- Tasks list but never complete: check scheduler logs for missing Slack destination fields, then check conversation work logs for mailbox append, lease, or delivery failures.
 - Unexpected timezone: set `JUNIOR_TIMEZONE` to the deployment default, or include the timezone in the user's schedule request.
 
 ## Next step

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

@@ -57,14 +57,19 @@ import {
 } from "@sentry/junior-scheduler";
 import { createMemoryPlugin } from "@sentry/junior-memory";
 import { runPluginHeartbeats } from "@/chat/agent-dispatch/heartbeat";
-import { runAgentDispatchSlice } from "@/chat/agent-dispatch/runner";
+import {
+  buildDispatchRoutingContext,
+  createAgentDispatchConversationWorker,
+  createAgentDispatchWorkRouter,
+} from "@/chat/agent-dispatch/work";
 import {
   ConversationTurnLifecycleService,
   type ConversationTurnLifecycle,
 } from "@/chat/conversations/turn-lifecycle";
-import { verifyDispatchCallbackRequest } from "@/chat/agent-dispatch/signing";
-import { getDispatchRecord } from "@/chat/agent-dispatch/store";
-import type { DispatchCallback } from "@/chat/agent-dispatch/types";
+import {
+  getDispatchInputMessageIds,
+  getDispatchRecord,
+} from "@/chat/agent-dispatch/store";
 import { ingestResourceEvent } from "@/chat/resource-events/ingest";
 import { createResourceEventSubscription } from "@/chat/resource-events/store";
 import { getStateAdapter } from "@/chat/state/adapter";
@@ -1930,7 +1935,6 @@ async function processEvents(args: {
   scenario: EvalScenario;
   env: HarnessEnvironment;
   agentRunner: AgentRunner;
-  turnLifecycle: ConversationTurnLifecycle;
   getSlackAdapter: () => FakeSlackAdapter;
   conversationWorkQueue: ConversationWorkQueueTestAdapter;
   slackRuntime: ReturnType<typeof createSlackRuntime>;
@@ -1944,7 +1948,6 @@ async function processEvents(args: {
     scenario,
     env,
     agentRunner,
-    turnLifecycle,
     getSlackAdapter,
     conversationWorkQueue,
     slackRuntime,
@@ -2020,6 +2023,55 @@ async function processEvents(args: {
   };
 
   const drainQueuedConversationWork = async (): Promise<void> => {
+    const slackWorker = createSlackConversationWorker({
+      getSlackAdapter: () => getSlackAdapter() as unknown as SlackAdapter,
+      resumeAwaitingContinuation: async (conversationId) =>
+        await resumeAwaitingSlackContinuation(conversationId, {
+          agentRunner,
+          scheduleAgentContinue: async (request) => {
+            await scheduleAgentContinue(request, {
+              queue: conversationWorkQueue,
+              state: env.stateAdapter,
+            });
+          },
+          scheduleSessionCompletedPluginTasks: async (params) => {
+            await scheduleSessionCompletedPluginTasks(params, {
+              send: async (message) => {
+                await processEvalPluginTask(message);
+              },
+            });
+          },
+        }),
+      runtime: workerRuntime,
+      state: env.stateAdapter,
+    });
+    const dispatchWorker = createAgentDispatchConversationWorker({
+      resumeTurn: async (dispatch, hooks) => {
+        await resumeAwaitingSlackContinuation(
+          `agent-dispatch:${dispatch.id}`,
+          {
+            agentRunner,
+            inputMessageIds: getDispatchInputMessageIds(dispatch.id),
+            routingContext: buildDispatchRoutingContext(dispatch),
+            scheduleAgentContinue: async (request) => {
+              await scheduleAgentContinue(request, {
+                queue: conversationWorkQueue,
+                state: env.stateAdapter,
+              });
+            },
+            scheduleSessionCompletedPluginTasks: async (params) => {
+              await scheduleSessionCompletedPluginTasks(params, {
+                send: async (message) => {
+                  await processEvalPluginTask(message);
+                },
+              });
+            },
+          },
+          { shouldYield: hooks.shouldYield },
+        );
+      },
+      runTurn: workerRuntime.runDispatchTurn,
+    });
     let processed = 0;
     while (conversationWorkQueue.hasQueuedMessages()) {
       processed += 1;
@@ -2030,27 +2082,9 @@ async function processEvents(args: {
         conversationWorkQueue.takeMessage(),
         {
           queue: conversationWorkQueue,
-          run: createSlackConversationWorker({
-            getSlackAdapter: () => getSlackAdapter() as unknown as SlackAdapter,
-            resumeAwaitingContinuation: async (conversationId) =>
-              await resumeAwaitingSlackContinuation(conversationId, {
-                agentRunner,
-                scheduleAgentContinue: async (request) => {
-                  await scheduleAgentContinue(request, {
-                    queue: conversationWorkQueue,
-                    state: env.stateAdapter,
-                  });
-                },
-                scheduleSessionCompletedPluginTasks: async (params) => {
-                  await scheduleSessionCompletedPluginTasks(params, {
-                    send: async (message) => {
-                      await processEvalPluginTask(message);
-                    },
-                  });
-                },
-              }),
-            runtime: workerRuntime,
-            state: env.stateAdapter,
+          run: createAgentDispatchWorkRouter({
+            dispatchWorker,
+            fallbackWorker: slackWorker,
           }),
           state: env.stateAdapter,
         },
@@ -2172,41 +2206,10 @@ async function processEvents(args: {
     );
     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.",
-      );
-    }
+    await runPluginHeartbeats({
+      conversationWorkQueue,
+      nowMs,
+    });
 
     const dispatchedRuns = (await schedulerStore.listIncompleteRuns()).filter(
       (run) => run.taskId === taskId && run.dispatchId,
@@ -2225,14 +2228,8 @@ async function processEvents(args: {
       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, { agentRunner, turnLifecycle });
     }
+    await drainQueuedConversationWork();
   };
 
   const runGitHubWebhook = async (event: GitHubWebhookEvent): Promise<void> => {
@@ -2531,7 +2528,6 @@ export async function runEvalScenario(
       scenario,
       env,
       agentRunner: evalAgentRunner,
-      turnLifecycle,
       getSlackAdapter: () => slackAdapter,
       conversationWorkQueue,
       slackRuntime,

packages/junior/src/app.ts

@@ -687,7 +687,9 @@ export async function createApp(options?: JuniorAppOptions): Promise<Hono> {
   });
 
   app.post("/api/internal/agent-dispatch", (c) => {
-    return agentDispatchPOST(c.req.raw, waitUntil, { agentRunner });
+    return agentDispatchPOST(c.req.raw, waitUntil, {
+      conversationWorkQueue: getVercelConversationWorkQueue(),
+    });
   });
 
   let agentContinuePOST:

packages/junior/src/chat/README.md

@@ -30,6 +30,8 @@ mailbox-backed provider.
 - `ingress/`: source parsing, classification, and routing.
 - `task-execution/`: mailbox, queue, lease, worker, and recovery.
 - `runtime/`: turn orchestration and provider-neutral delivery callbacks.
+- `agent-dispatch/`: plugin dispatch authority, mailbox adaptation, and
+  plugin-facing outcome projection.
 - `agent/` and `pi/`: model execution and Pi state conversion.
 - `services/`: consumer-owned domain decisions.
 - `state/` and `conversations/`: persistence by concern.

packages/junior/src/chat/agent-dispatch/README.md

@@ -0,0 +1,59 @@
+# Agent Dispatch
+
+This feature turns an idempotent plugin dispatch into normal conversation work.
+It owns dispatch authority and the plugin-facing status projection. The
+conversation worker owns leases, retries, execution slices, and recovery; the
+shared turn runtime owns agent execution and durable turn outcomes; the Slack
+adapter owns destination delivery.
+
+## Durable Facts
+
+- The dispatch record is the plugin-owned request and status projection.
+- The conversation mailbox contains a deferred inbound message that identifies
+  the dispatch. Queue payloads only wake the conversation.
+- The session record is authoritative for the active turn, resume state,
+  accepted delivery receipt, and explicit terminal dispatch outcome.
+- A dispatch uses one isolated conversation and one stable turn across all runs
+  and execution slices.
+
+The mailbox carries no credential authority. Every run rebuilds actor,
+credential subject, source, destination, and plugin metadata from the dispatch
+record.
+
+## Lifecycle
+
+1. Plugin heartbeat code creates the dispatch idempotently.
+2. The dispatch is indexed before its mailbox append so heartbeat recovery can
+   repair a crash between those writes.
+3. The conversation worker validates dispatch identity and destination before
+   starting or resuming the turn.
+4. Durable `running` and `awaiting_resume` sessions resume even when the
+   original mailbox item is redelivered.
+5. Explicit session outcomes project to `completed`, `blocked`, or `failed`.
+   A failed session without an explicit dispatch outcome remains retryable; the
+   queue's final attempt owns terminal failure.
+6. An accepted destination message is a delivery fence and must never be
+   generated again during projection recovery.
+
+Dispatch state transitions use a short dispatch lock while the conversation
+lease is already held. Dispatch code never waits for a conversation lease while
+holding that lock.
+
+## Boundaries
+
+- `context.ts` exposes plugin-scoped create/get capabilities.
+- `store.ts` owns dispatch records, projections, and mailbox-append recovery
+  receipts.
+- `work.ts` adapts dispatches to conversation work and projects durable turn
+  results.
+- `heartbeat.ts` repairs incomplete mailbox appends before running plugin
+  heartbeat hooks.
+- `slack/dispatch-turn.ts` owns Slack message/thread adaptation and receives
+  only dispatch-owned contracts.
+
+The legacy signed callback route only appends conversation work. It cannot
+execute a turn or reclaim lifecycle ownership.
+
+Representative behavior coverage lives in
+`tests/integration/agent-dispatch-work.test.ts`; local transition and authority
+contracts live in `tests/component/agent-dispatch-worker.test.ts`.

packages/junior/src/chat/agent-dispatch/context.ts

@@ -9,38 +9,21 @@ import {
 import { getDb } from "@/chat/db";
 import { createPluginLogger } from "@/chat/plugins/logging";
 import { createPluginState } from "@/chat/plugins/state";
+import type { ConversationWorkQueue } from "@/chat/task-execution/queue";
 import {
   createOrGetDispatch,
   getPluginDispatchProjection,
   isTerminalDispatchStatus,
 } from "./store";
-import { scheduleDispatchCallback } from "./signing";
-import type {
-  BoundDispatchOptions,
-  DispatchRecord,
-  SlackDispatchOptions,
-} from "./types";
+import { enqueueAgentDispatch } from "./work";
+import type { BoundDispatchOptions, SlackDispatchOptions } from "./types";
 import {
   validateDispatchOptions,
   verifyDispatchCredentialSubjectAccess,
 } from "./validation";
 
 const MAX_DISPATCHES_PER_HEARTBEAT = 25;
 
-function shouldScheduleDispatch(
-  record: DispatchRecord,
-  nowMs: number,
-): boolean {
-  if (isTerminalDispatchStatus(record.status)) {
-    return false;
-  }
-  return (
-    record.status !== "running" ||
-    typeof record.leaseExpiresAtMs !== "number" ||
-    record.leaseExpiresAtMs <= nowMs
-  );
-}
-
 function bindDispatchCredentialSubject(
   options: SlackDispatchOptions,
   plugin: string,
@@ -76,6 +59,7 @@ function bindDispatchCredentialSubject(
 
 /** Build the plugin-scoped heartbeat context that gates durable dispatch access. */
 export function createHeartbeatContext(args: {
+  conversationWorkQueue: ConversationWorkQueue;
   nowMs: number;
   plugin: string | PluginRegistration;
 }): HeartbeatHookContext {
@@ -108,10 +92,10 @@ export function createHeartbeatContext(args: {
           nowMs: args.nowMs,
         });
         dispatchCount += 1;
-        if (shouldScheduleDispatch(result.record, args.nowMs)) {
-          await scheduleDispatchCallback({
-            id: result.record.id,
-            expectedVersion: result.record.version,
+        if (!isTerminalDispatchStatus(result.record.status)) {
+          await enqueueAgentDispatch(result.record, {
+            queue: args.conversationWorkQueue,
+            nowMs: args.nowMs,
           });
         }
         return {