Fire long-horizon agent reminders via the job queue

packages/daemon/src/app.ts

@@ -74,11 +74,16 @@ import {
 import { createSkillValidateHandler } from './lib/job-handlers/skill-validate.handler';
 import {
   JOB_QUEUE_CLEANUP,
+  LONG_HORIZON_AGENT_REMINDER_FIRE,
   MEMORY_CONSOLIDATION,
   SKILL_VALIDATE,
   TASK_SCHEDULE_FIRE,
 } from './lib/job-queue-constants';
 import { handleTaskScheduleFire } from './lib/job-handlers/task-schedule-fire.handler';
+import {
+  enqueueLongHorizonAgentReminderScanIfMissing,
+  handleLongHorizonAgentReminderFire,
+} from './lib/job-handlers/long-horizon-agent-reminder-fire.handler';
 import { TaskScheduleRepository } from './storage/repositories/task-schedule-repository';
 import { SpaceRepository } from './storage/repositories/space-repository';
 import { SpaceTaskRepository } from './storage/repositories/space-task-repository';
@@ -932,6 +937,17 @@ export async function createDaemonApp(options: CreateDaemonAppOptions): Promise<
       });
     });
 
+    // Register longHorizonAgentReminder.fire scanner — fires due LH agent
+    // reminders and delivers them to the owning agent session.
+    const lhAgentReminderRepo = new SpaceLongHorizonAgentRepository(db.getDatabase());
+    jobProcessor.register(LONG_HORIZON_AGENT_REMINDER_FIRE, async (job) => {
+      return handleLongHorizonAgentReminderFire(job, {
+        reminderRepo: lhAgentReminderRepo,
+        jobQueue,
+        deliver: (args) => spaceRuntimeService.deliverLongHorizonAgentReminder(args),
+      });
+    });
+
     // Startup resilience: re-seed active schedules whose pending jobs are missing.
     // This handles crash recovery in two cases:
     //   1) Schedule has a pendingJobId, but the underlying job is gone OR has reached
@@ -1026,6 +1042,8 @@ export async function createDaemonApp(options: CreateDaemonAppOptions): Promise<
     }
     enqueueMemoryConsolidationIfMissing(jobQueue, Date.now());
     logInfo('[Daemon] Ensured initial memory_consolidation job');
+    enqueueLongHorizonAgentReminderScanIfMissing(jobQueue, Date.now());
+    logInfo('[Daemon] Ensured initial longHorizonAgentReminder.fire scan job');
 
     // Start job queue processor last (after all handler registrations)
     jobProcessor.start();

packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts

@@ -0,0 +1,198 @@
+/**
+ * Job handler for the longHorizonAgentReminder.fire scanner queue.
+ *
+ * Periodically scans for due long-horizon agent reminders and fires each:
+ *   1. Query active reminders with `next_run_at <= now` whose owning agent is
+ *      active (the partial `idx_space_lh_agent_reminders_due` index serves the
+ *      reminder-side predicate; the join skips paused/archived owners).
+ *   2. For each reminder: re-check it is still active, deliver a formatted
+ *      reminder message to the owning LH agent session, then advance the
+ *      reminder. Cron reminders get `next_run_at` recomputed from the
+ *      expression (status stays 'active'); one-shot `'at'` reminders flip to
+ *      `status='fired'`.
+ *   3. Self-schedule the next scan so the scanner keeps running every
+ *      `NEXT_SCAN_DELAY_MS`.
+ *
+ * The scanner mirrors the self-scheduling pattern from
+ * memory-consolidation.handler.ts and the guard/advance shape from
+ * task-schedule-fire.handler.ts.
+ *
+ * Double-fire safety: each reminder is delivered then immediately advanced via
+ * a compare-and-swap (`status='active' AND next_run_at=<the value we read>`).
+ * A retried scan that re-selects the same reminder finds it already advanced
+ * (or fired) and the CAS no-ops, so no second message is delivered. The job
+ * processor is single-threaded and the daemon holds an exclusive DB lock, so
+ * there is no cross-tick concurrency; the CAS is the belt-and-suspenders fence
+ * for crash-retry races.
+ *
+ * Ordering note: delivery happens *before* the advance so a one-shot reminder
+ * is never marked `fired` without having actually been delivered. If the
+ * process crashes between deliver and advance for one reminder, that single
+ * reminder may double-fire on the next scan — acceptable for a nag, and far
+ * better than silently losing a one-shot.
+ */
+
+import type { SpaceLongHorizonAgentReminder } from '@hyperneo/shared';
+import { LONG_HORIZON_AGENT_REMINDER_FIRE } from '../job-queue-constants';
+import { Logger } from '../logger';
+import { getNextRunAt } from '../space/schedule/cron-utils';
+import type { SpaceLongHorizonAgentRepository } from '../../storage/repositories/space-long-horizon-agent-repository';
+import type { JobQueueRepository, Job } from '../../storage/repositories/job-queue-repository';
+
+const log = new Logger('lh-agent-reminder-fire-handler');
+
+/** How often the scanner re-runs. A near-future reminder fires within one window. */
+const NEXT_SCAN_DELAY_MS = 30_000;
+
+export interface LongHorizonAgentReminderFireResult extends Record<string, unknown> {
+  scanned: number;
+  fired: number;
+  skipped: number;
+  failed: number;
+  nextScanAt: number;
+}
+
+export interface LongHorizonAgentReminderFireDeps {
+  reminderRepo: SpaceLongHorizonAgentRepository;
+  jobQueue: JobQueueRepository;
+  /**
+   * Deliver the reminder body to the owning LH agent session. Mirrors
+   * SpaceRuntimeService.deliverLongHorizonAgentReminder (which in turn mirrors
+   * deliverLongHorizonExternalEvent): re-checks the agent is present, in-space,
+   * and `status === 'active'`, then ensures the session and injects the
+   * message. Returns `delivered:false` (no throw) for a missing/paused/
+   * archived agent so the scanner treats it as a skip.
+   */
+  deliver: (args: {
+    spaceId: string;
+    agentId: string;
+    message: string;
+    idempotencyKey: string;
+  }) => Promise<{ delivered: boolean }>;
+}
+
+export async function handleLongHorizonAgentReminderFire(
+  _job: Job,
+  deps: LongHorizonAgentReminderFireDeps
+): Promise<LongHorizonAgentReminderFireResult> {
+  const { reminderRepo, jobQueue, deliver } = deps;
+  const now = Date.now();
+  const nextScanAt = now + NEXT_SCAN_DELAY_MS;
+  // Schedule the next scan first so the chain survives a crash mid-scan.
+  enqueueLongHorizonAgentReminderScanIfMissing(jobQueue, nextScanAt);
+
+  const due = reminderRepo.listDueReminders(now);
+  let fired = 0;
+  let skipped = 0;
+  let failed = 0;
+
+  for (const reminder of due) {
+    const outcome = await fireReminder(reminder, reminderRepo, deliver, now);
+    if (outcome === 'fired') fired++;
+    else if (outcome === 'skipped') skipped++;
+    else failed++;
+  }
+
+  if (due.length > 0) {
+    log.debug('lh-agent-reminder-fire: scan complete', {
+      scanned: due.length,
+      fired,
+      skipped,
+      failed,
+    });
+  }
+
+  return { scanned: due.length, fired, skipped, failed, nextScanAt };
+}
+
+async function fireReminder(
+  reminder: SpaceLongHorizonAgentReminder,
+  reminderRepo: SpaceLongHorizonAgentRepository,
+  deliver: LongHorizonAgentReminderFireDeps['deliver'],
+  now: number
+): Promise<'fired' | 'skipped' | 'failed'> {
+  // Re-read: the reminder may have been paused/cancelled/fired between the
+  // due-scan and now. A null next_run_at means it is no longer schedulable.
+  const fresh = reminderRepo.getReminder(reminder.id);
+  if (!fresh || fresh.status !== 'active' || fresh.nextRunAt === null) {
+    return 'skipped';
+  }
+
+  const message = formatReminderMessage(fresh);
+  // Deterministic per-fire idempotency key: scoped to this reminder and the
+  // occurrence we are firing (its current next_run_at). A re-delivery of the
+  // same occurrence reuses the key.
+  const idempotencyKey = `reminder:${reminder.id}:${reminder.nextRunAt}`;
+
+  let delivered: boolean;
+  try {
+    const result = await deliver({
+      spaceId: reminder.spaceId,
+      agentId: reminder.agentId,
+      message,
+      idempotencyKey,
+    });
+    delivered = result.delivered;
+  } catch (err) {
+    log.warn('lh-agent-reminder-fire: delivery threw', {
+      reminderId: reminder.id,
+      error: err instanceof Error ? err.message : err,
+    });
+    return 'failed';
+  }
+
+  // Delivery returned false (agent missing/paused/archived, or session could
+  // not be ensured). Do NOT advance: a paused agent's reminder stays due and
+  // fires when the agent is active again (the due-query filters paused agents
+  // out, so there is no re-selection spam while it stays paused).
+  if (!delivered) return 'skipped';
+
+  // Compute the post-fire state: cron → recompute next run (stay active);
+  // 'at' → terminal 'fired' with no future run. If a cron expression yields no
+  // future occurrence, treat it as terminal too.
+  const nextRunAt =
+    fresh.triggerType === 'cron' && fresh.cronExpression
+      ? getNextRunAt(fresh.cronExpression, fresh.timezone, now)
+      : null;
+  const updates =
+    nextRunAt === null
+      ? { status: 'fired' as const, nextRunAt: null, lastFiredAt: now }
+      : { status: 'active' as const, nextRunAt, lastFiredAt: now };
+
+  const applied = reminderRepo.advanceReminderAfterFire(fresh.id, fresh.nextRunAt, updates);
+  if (!applied) {
+    // The reminder changed under us (paused/rescheduled/fired by another path)
+    // between our read and the CAS. We already delivered — leave the row as the
+    // winning mutation left it; nothing more to do.
+    log.debug('lh-agent-reminder-fire: advance CAS missed', { reminderId: reminder.id });
+  }
+  return 'fired';
+}
+
+/**
+ * Enqueue the next reminder-fire scan if none is already pending. Called at
+ * startup and at the start of every scan so the scanner keeps recurring.
+ */
+export function enqueueLongHorizonAgentReminderScanIfMissing(
+  jobQueue: JobQueueRepository,
+  runAt = Date.now()
+): void {
+  const pending = jobQueue.listJobs({
+    queue: LONG_HORIZON_AGENT_REMINDER_FIRE,
+    status: 'pending',
+    limit: 1,
+  });
+  if (pending.length === 0) {
+    jobQueue.enqueue({ queue: LONG_HORIZON_AGENT_REMINDER_FIRE, payload: {}, runAt });
+  }
+}
+
+function formatReminderMessage(reminder: SpaceLongHorizonAgentReminder): string {
+  // `title` carries the reminder text (the create_agent_reminder tool stores
+  // the caller's `message` there); `body` is optional detail.
+  const parts = [`## Scheduled Reminder\n\n${reminder.title}`];
+  if (reminder.body.trim()) {
+    parts.push(`\n${reminder.body}`);
+  }
+  return parts.join('\n');
+}

packages/daemon/src/lib/job-queue-constants.ts

@@ -15,6 +15,11 @@ export const GOAL_AUTOMATION_EXECUTE = 'goalAutomation.execute';
 // ─── Task schedule ────────────────────────────────────────────────────────────
 export const TASK_SCHEDULE_FIRE = 'taskSchedule.fire';
 
+// ─── Long-horizon agent reminders ─────────────────────────────────────────────
+// Recurring scanner that fires due long-horizon agent reminders and delivers
+// them to the owning agent session. Self-schedules like memory_consolidation.
+export const LONG_HORIZON_AGENT_REMINDER_FIRE = 'longHorizonAgentReminder.fire';
+
 // ─── Space workflow run artifact sync queues ──────────────────────────────────
 // Background jobs that populate the workflow_run_artifact_cache table with
 // git-derived data (gate artifacts, commit log, per-file diffs). Running these

packages/daemon/src/lib/space/runtime/space-runtime-service.ts

@@ -381,6 +381,24 @@ export class SpaceRuntimeService {
     return { delivered: false };
   }
 
+  /**
+   * Deliver a long-horizon agent reminder to the owning agent's session.
+   *
+   * Public entry point for the reminder-fire job handler. Delegates to the same
+   * ensure-session + inject path as external-event delivery, with identical
+   * guards: agent present, belongs to the space, and `status === 'active'`.
+   * Returns `delivered:false` (no throw) when the agent is missing, paused,
+   * disabled, or archived — the scanner treats that as a skip, not a failure.
+   */
+  async deliverLongHorizonAgentReminder(args: {
+    spaceId: string;
+    agentId: string;
+    message: string;
+    idempotencyKey: string;
+  }): Promise<{ delivered: boolean }> {
+    return this.deliverLongHorizonExternalEvent(args);
+  }
+
   private async deliverToLongTermAgent(
     actor: ActorRef,
     message: MessageRecord

packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts

@@ -294,6 +294,61 @@ export class SpaceLongHorizonAgentRepository {
     return rows.map(rowToReminder);
   }
 
+  /**
+   * Active reminders that are due (`next_run_at <= now`) for agents whose
+   * status is 'active'. Powers the reminder-fire scanner job — the partial
+   * index `idx_space_lh_agent_reminders_due` (status='active') serves the
+   * reminder-side predicate; the agent-status join skips reminders whose owner
+   * is paused/disabled/archived.
+   */
+  listDueReminders(now: number, limit = 100): SpaceLongHorizonAgentReminder[] {
+    const rows = this.db
+      .prepare(
+        `SELECT r.* FROM space_long_horizon_agent_reminders r
+           INNER JOIN space_long_horizon_agents a ON a.id = r.agent_id
+           WHERE r.status = 'active' AND r.next_run_at IS NOT NULL AND r.next_run_at <= ?
+             AND a.status = 'active'
+           ORDER BY r.next_run_at ASC
+           LIMIT ?`
+      )
+      .all(now, limit) as Record<string, unknown>[];
+    return rows.map(rowToReminder);
+  }
+
+  /**
+   * Compare-and-swap advance of a reminder after it has fired. Only applies if
+   * the reminder is still 'active' and its `next_run_at` still matches the
+   * value the scanner read — a concurrent pause/reschedule/cancel wins and the
+   * caller sees `applied=false`. For cron reminders `nextRunAt` is recomputed
+   * from the expression (status stays 'active'); for one-shot 'at' reminders
+   * the caller passes `status='fired'` and a null `nextRunAt`.
+   */
+  advanceReminderAfterFire(
+    id: string,
+    expectedNextRunAt: number,
+    updates: {
+      status: SpaceLongHorizonAgentReminder['status'];
+      nextRunAt: number | null;
+      lastFiredAt: number;
+    }
+  ): boolean {
+    const result = this.db
+      .prepare(
+        `UPDATE space_long_horizon_agent_reminders
+            SET status = ?, next_run_at = ?, last_fired_at = ?, updated_at = ?
+            WHERE id = ? AND status = 'active' AND next_run_at = ?`
+      )
+      .run(
+        updates.status,
+        updates.nextRunAt,
+        updates.lastFiredAt,
+        Date.now(),
+        id,
+        expectedNextRunAt
+      );
+    return result.changes > 0;
+  }
+
   createSubscription(
     params: CreateSpaceLongHorizonAgentSubscriptionParams
   ): SpaceLongHorizonAgentEventSubscription {