[Feat] Let Fast Sessions schedule reminders and recurring checks

.changeset/session-wakeups.md

@@ -0,0 +1,5 @@
+---
+'@roomote/web': minor
+---
+
+Fast Sessions can schedule a message to themselves with the new `manage_wakeups` tool. Ask Fast to "remind me in twenty minutes" or "check every ten minutes whether CI is green" and it creates a wakeup for that Session; when it fires, Fast picks the conversation back up with its history in context, does what was asked, and replies on the surface the Session lives on. One-shot wakeups always reply, recurring ones stay quiet unless there is news and cancel themselves when the monitored condition resolves. Wakeups are scoped to the conversation, need no administrator, are capped at ten per Session, and are cancelled when the Session is archived. Firing is durable: each occurrence is claimed on the row, a delayed queue job is only a hint, and a recovery sweep re-adds lost hints.

apps/bullmq/src/index.ts

@@ -54,6 +54,7 @@ import { startTaskSleepQueue } from './task-sleep-queue';
 import { startAutomationRecommendationsQueue } from './automation-recommendations-queue';
 import { startFastAgentParentEventQueue } from './fast-agent-parent-event-queue';
 import { readBullMqQueueHealth } from './health';
+import { startSessionWakeupQueue } from './session-wakeup-queue';
 import { installBullMqGracefulShutdown } from './graceful-shutdown';
 
 // Deployments roll every service at once while migrations run only ahead
@@ -216,6 +217,11 @@ const {
   worker: fastAgentParentEventWorker,
   queueEvents: fastAgentParentEventQueueEvents,
 } = await startFastAgentParentEventQueue();
+const {
+  queue: sessionWakeupQueue,
+  worker: sessionWakeupWorker,
+  queueEvents: sessionWakeupQueueEvents,
+} = await startSessionWakeupQueue();
 
 const serverAdapter = new HonoAdapter(serveStatic);
 
@@ -256,6 +262,7 @@ createBullBoard({
       readOnlyMode: false,
     }),
     new BullMQAdapter(fastAgentParentEventQueue, { readOnlyMode: false }),
+    new BullMQAdapter(sessionWakeupQueue, { readOnlyMode: false }),
   ],
   serverAdapter,
 });
@@ -448,6 +455,9 @@ installBullMqGracefulShutdown({
     await pullRequestMergeabilityCheckQueue.close();
     await fastAgentParentEventQueueEvents.close();
     await fastAgentParentEventQueue.close();
+    await sessionWakeupWorker.close();
+    await sessionWakeupQueueEvents.close();
+    await sessionWakeupQueue.close();
     await discordGatewaySupervisor.stop();
     await closeRedis();
   },

apps/bullmq/src/session-wakeup-queue.ts

@@ -0,0 +1,87 @@
+import { Queue, QueueEvents, Worker, type Job } from 'bullmq';
+
+import {
+  SESSION_WAKEUP_FIRE_JOB_NAME,
+  SESSION_WAKEUP_QUEUE_NAME,
+  fireSessionWakeup,
+  recoverPendingSessionWakeups,
+  type SessionWakeupFireJob,
+} from '@roomote/sdk/server';
+
+import { getRedis } from './redis';
+
+const RECOVERY_JOB_NAME = 'recover-due';
+const RECOVERY_SCHEDULER_ID = 'session-wakeup-recovery';
+const RECOVERY_INTERVAL_MS = 60_000;
+
+type SessionWakeupQueueJob = SessionWakeupFireJob | { recovery: true };
+
+async function processJob(job: Job<SessionWakeupQueueJob>) {
+  if (job.name === RECOVERY_JOB_NAME || 'recovery' in job.data) {
+    const recovered = await recoverPendingSessionWakeups();
+    if (recovered > 0) {
+      console.log(
+        `[SessionWakeupQueue] Re-added ${recovered} due wakeup hint(s).`,
+      );
+    }
+    return;
+  }
+  if (job.name !== SESSION_WAKEUP_FIRE_JOB_NAME) return;
+  const result = await fireSessionWakeup(job.data);
+  if (result.outcome === 'skipped') {
+    console.log(
+      `[SessionWakeupQueue] Skipped wakeup ${job.data.wakeupId}: ${result.reason}`,
+    );
+  }
+}
+
+/**
+ * Fires session wakeups. Every job is a hint for a `session_wakeups` row:
+ * the row's `next_run_at` decides whether anything happens, and a recovery
+ * sweep re-adds hints for due rows so lost jobs cannot strand a wakeup.
+ */
+export async function startSessionWakeupQueue() {
+  const connection = getRedis();
+  const queue = new Queue<SessionWakeupQueueJob>(SESSION_WAKEUP_QUEUE_NAME, {
+    connection,
+    defaultJobOptions: {
+      attempts: 3,
+      backoff: { type: 'exponential', delay: 2_000 },
+      removeOnComplete: true,
+      removeOnFail: true,
+    },
+  });
+
+  await queue.upsertJobScheduler(
+    RECOVERY_SCHEDULER_ID,
+    { every: RECOVERY_INTERVAL_MS },
+    { name: RECOVERY_JOB_NAME, data: { recovery: true } },
+  );
+  await recoverPendingSessionWakeups();
+
+  const worker = new Worker<SessionWakeupQueueJob>(
+    SESSION_WAKEUP_QUEUE_NAME,
+    processJob,
+    // Firing only admits an event into the conversation's durable inbox; the
+    // turn itself runs on the Fast parent event worker.
+    { connection, concurrency: 10, autorun: true },
+  );
+
+  worker.on('failed', (job, error) =>
+    console.error(
+      `[SessionWakeupQueue] job ${job?.id} failed: ${error.message}`,
+    ),
+  );
+  worker.on('error', (error) =>
+    console.error('[SessionWakeupQueue] worker error:', error),
+  );
+
+  const queueEvents = new QueueEvents(SESSION_WAKEUP_QUEUE_NAME, {
+    connection,
+  });
+  queueEvents.on('failed', ({ jobId, failedReason }) =>
+    console.error(`[SessionWakeupQueue] job ${jobId} failed: ${failedReason}`),
+  );
+
+  return { queue, worker, queueEvents };
+}

apps/docs/fast-sessions.mdx

@@ -145,6 +145,31 @@ In shared Slack and Discord conversations, Fast can stay silent when linked
 participants are talking to each other. A direct reply to Roomote, a mention,
 or a direct message still requires a response.
 
+## Reminders and monitors
+
+A Session can schedule a message to itself. Ask Fast to "remind me in twenty
+minutes", "check every ten minutes whether CI is green", or "every weekday at
+nine, summarize the open pull requests", and it creates a wakeup for that
+Session. When the wakeup fires, Fast picks the conversation back up with its
+full history in context, does what was asked, and replies on the same surface
+the Session lives on: the transcript in the dashboard, or the originating Slack,
+Discord, Teams, or Telegram thread.
+
+A one-shot wakeup is a reminder and always replies. A recurring wakeup is a
+monitor and stays quiet unless it has something to report, such as a change, a
+result, a blocker, or a decision you need to make. When the monitored condition
+resolves, Fast tells you and cancels the wakeup itself. You can also ask Fast to
+list or cancel wakeups at any time; archiving a Session cancels all of its
+wakeups.
+
+Wakeups belong to the conversation and do not require an administrator. A
+Session may hold up to ten active wakeups, intervals range from one minute to
+seven days, and recurring wakeups can be bounded by a run count or an end time.
+Intervals under five minutes must carry one of those bounds. A recurring wakeup
+whose turns fail five times in a row is retired. Deployment-wide recurring work
+that should run outside a conversation or report to a channel is an
+[automation](/automations) instead.
+
 ## Session and execution access
 
 Any signed-in deployment user with a Session link can view its timeline and

apps/web/src/app/(sandbox)/task/[taskId]/messages/acp/tool-presentation.ts

@@ -104,6 +104,7 @@ const COMMUNICATION_TOOL_NAMES = new Set([
 ]);
 const TOOL_ICON_OVERRIDES: Readonly<Partial<Record<string, ToolIconKey>>> = {
   manage_custom_automations: 'task',
+  manage_wakeups: 'task',
   get_about_me: 'roomote',
   describe_video: 'video',
   manage_goal: 'target',