Scheduling Work with Pi
durable-harness/pi runs Pi's agent loop with each session in a durable log, so a scheduled task can come back to the conversation that created it. The tool stores the session ID in the schedule's payload, and the consumer sends the prompt to that session when the schedule fires.
Record the user when the session starts. Tools read it from io.ctx.config, and the session ID from io.ctx.runId:
const { sessionId } = await agent.sessions.start({ config: { userId, timezone } });Then add the tool. With replay: 'never', a call interrupted by a crash isn't run again, so it can't create a second schedule. The model is told the call may have run instead:
import type { HostTool } from 'durable-harness';
import { pi } from 'durable-harness/pi';
import { Schedules } from '@vercel/schedules';
const scheduleTask: HostTool<{ prompt: string; cron?: string; at?: string }> = {
name: 'schedule_task',
description: 'Schedule work to run later in this conversation',
replay: 'never',
inputSchema: {
type: 'object',
properties: {
prompt: { type: 'string', description: 'What to do when the task fires' },
cron: { type: 'string', description: 'Cron expression, for recurring tasks' },
at: { type: 'string', description: 'Local time YYYY-MM-DDTHH:mm, for one-time tasks' },
},
required: ['prompt'],
},
execute: async ({ prompt, cron, at }, io) => {
const { userId, timezone } = io.ctx.config as { userId: string; timezone: string };
await Schedules.create({
name: `task-${io.stepId}`,
namespace: userId,
expression: cron ? { type: 'cron', cron } : { type: 'single', at: at! },
timezone,
target: { topic: 'agent-tasks' },
payload: { sessionId: io.ctx.runId, prompt },
});
return 'Scheduled.';
},
};
export const agent = pi({
model: 'anthropic/claude-sonnet-5.5',
tools: [scheduleTask],
// Required for `send` with an `identity`, which the consumer below uses to skip redeliveries.
durability: 'strict',
});When the schedule fires, send the prompt to the session it came from. The session's own queue route runs the turn. Pass the run's executionId as the identity, so a redelivered message returns the original submission instead of starting another turn:
import { handleCallback } from '@vercel/queue';
import type { ScheduleQueueMessage } from '@vercel/schedules';
import { agent } from '@/lib/agent';
type TaskMessage = ScheduleQueueMessage<{ sessionId: string; prompt: string }>;
export const POST = handleCallback<TaskMessage>(async ({ executionId, payload }) => {
if (!payload) return;
await agent.sessions.send(payload.sessionId, `Scheduled task: ${payload.prompt}`, 'prompt', {
identity: `task-${executionId}`,
});
});{
"functions": {
"app/api/agent-tasks/route.ts": {
"experimentalTriggers": [{ "type": "queue/v2beta", "topic": "agent-tasks" }]
}
}
}The scheduled turn is part of the session, so it streams to anyone following the session, and shows up in its history.
Was this helpful?