From ca93b1e941de606a7fbaeefb77e4f6b1d8ad22e4 Mon Sep 17 00:00:00 2001 From: Marc Liu Date: Sun, 26 Jul 2026 15:34:31 -0400 Subject: [PATCH 1/9] feat: fire long-horizon agent reminders via the job queue Wire reminder firing into the job-queue machinery so due LH agent reminders wake the owning agent session on schedule. Reminders were previously write-only records. - Add listDueReminders (active reminders, next_run_at<=now, active owner) and advanceReminderAfterFire (CAS advance) to the LH agent repository. - Add a self-scheduling scanner handler (longHorizonAgentReminder.fire) mirroring task-schedule-fire's guard/deliver/advance shape plus memory-consolidation's self-scheduling. Per reminder: deliver, then CAS-advance (cron recomputes next_run_at and stays active; one-shot 'at' flips to fired). A paused agent's reminder is filtered at the query and skipped at delivery without advancing. - Expose SpaceRuntimeService.deliverLongHorizonAgentReminder, delegating to the existing ensure-session + inject path with agent-active guards. - Register the handler and seed the initial scan in app.ts. --- packages/daemon/src/app.ts | 18 ++ ...ong-horizon-agent-reminder-fire.handler.ts | 198 ++++++++++++ .../daemon/src/lib/job-queue-constants.ts | 5 + .../space/runtime/space-runtime-service.ts | 18 ++ .../space-long-horizon-agent-repository.ts | 55 ++++ ...orizon-agent-reminder-fire.handler.test.ts | 300 ++++++++++++++++++ ...pace-long-horizon-agent-repository.test.ts | 147 +++++++++ 7 files changed, 741 insertions(+) create mode 100644 packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts create mode 100644 packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts diff --git a/packages/daemon/src/app.ts b/packages/daemon/src/app.ts index b3a4d7b56b..b4180c8dfe 100644 --- a/packages/daemon/src/app.ts +++ b/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'; @@ -937,6 +942,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 @@ -1031,6 +1047,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(); diff --git a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts new file mode 100644 index 0000000000..a72e6d8383 --- /dev/null +++ b/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=`). + * 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 { + 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 { + 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'); +} diff --git a/packages/daemon/src/lib/job-queue-constants.ts b/packages/daemon/src/lib/job-queue-constants.ts index 8c3ad449a9..0f42d8516a 100644 --- a/packages/daemon/src/lib/job-queue-constants.ts +++ b/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 diff --git a/packages/daemon/src/lib/space/runtime/space-runtime-service.ts b/packages/daemon/src/lib/space/runtime/space-runtime-service.ts index 9780795839..35b20f95e4 100644 --- a/packages/daemon/src/lib/space/runtime/space-runtime-service.ts +++ b/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 diff --git a/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts b/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts index 06341b4c7d..3218da0212 100644 --- a/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts +++ b/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[]; + 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 { diff --git a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts new file mode 100644 index 0000000000..8219e657cb --- /dev/null +++ b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts @@ -0,0 +1,300 @@ +/** + * Tests for the longHorizonAgentReminder.fire scanner handler. + */ + +import { afterEach, beforeEach, describe, expect, it } from 'bun:test'; +import { Database } from 'bun:sqlite'; +import { + enqueueLongHorizonAgentReminderScanIfMissing, + handleLongHorizonAgentReminderFire, +} from '../../../../src/lib/job-handlers/long-horizon-agent-reminder-fire.handler'; +import { LONG_HORIZON_AGENT_REMINDER_FIRE } from '../../../../src/lib/job-queue-constants'; +import type { Job } from '../../../../src/storage/repositories/job-queue-repository'; +import { JobQueueRepository } from '../../../../src/storage/repositories/job-queue-repository'; +import { SpaceLongHorizonAgentRepository } from '../../../../src/storage/repositories/space-long-horizon-agent-repository'; +import { SpaceRepository } from '../../../../src/storage/repositories/space-repository'; +import { createSpaceTables } from '../../helpers/space-test-db'; + +interface DeliverArgs { + spaceId: string; + agentId: string; + message: string; + idempotencyKey: string; +} + +function makeJob(): Job { + return { + id: 'job-1', + queue: LONG_HORIZON_AGENT_REMINDER_FIRE, + status: 'processing', + payload: {}, + result: null, + error: null, + priority: 0, + maxRetries: 3, + retryCount: 0, + runAt: Date.now(), + createdAt: Date.now(), + startedAt: Date.now(), + completedAt: null, + }; +} + +function recordingDeliver() { + const calls: DeliverArgs[] = []; + const fn = async (args: DeliverArgs): Promise<{ delivered: boolean }> => { + calls.push(args); + return { delivered: true }; + }; + return { calls, fn }; +} + +describe('handleLongHorizonAgentReminderFire', () => { + let db: Database; + let reminderRepo: SpaceLongHorizonAgentRepository; + let jobQueue: JobQueueRepository; + let spaceId: string; + let agentId: string; + + beforeEach(() => { + db = new Database(':memory:'); + createSpaceTables(db); + // job_queue is not created by createSpaceTables. + db.exec(` + CREATE TABLE IF NOT EXISTS job_queue ( + id TEXT PRIMARY KEY, + queue TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending' + CHECK(status IN ('pending', 'processing', 'completed', 'failed', 'dead')), + payload TEXT NOT NULL DEFAULT '{}', + result TEXT, + error TEXT, + priority INTEGER NOT NULL DEFAULT 0, + max_retries INTEGER NOT NULL DEFAULT 3, + retry_count INTEGER NOT NULL DEFAULT 0, + run_at INTEGER NOT NULL, + created_at INTEGER NOT NULL, + started_at INTEGER, + completed_at INTEGER + ); + CREATE INDEX IF NOT EXISTS idx_job_queue_dequeue ON job_queue(queue, status, priority DESC, run_at ASC); + `); + + const spaceRepo = new SpaceRepository(db as never); + reminderRepo = new SpaceLongHorizonAgentRepository(db); + jobQueue = new JobQueueRepository(db as never); + + const space = spaceRepo.createSpace({ + slug: 'test', + workspacePath: '/workspace/test', + name: 'Test', + description: 'Test space', + }); + spaceId = space.id; + const agent = reminderRepo.create({ spaceId, handle: 'steward', displayName: 'Steward' }); + agentId = agent.id; + }); + + afterEach(() => { + db.close(); + }); + + function makeDeps(deliver: (args: DeliverArgs) => Promise<{ delivered: boolean }>) { + return { reminderRepo, jobQueue, deliver }; + } + + it('fires a due one-shot reminder, delivers the body, and marks it fired', async () => { + const now = Date.now(); + const reminder = reminderRepo.createReminder({ + spaceId, + agentId, + title: 'Review goals', + body: 'Check their statuses.', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + + const deliver = recordingDeliver(); + const result = await handleLongHorizonAgentReminderFire(makeJob(), makeDeps(deliver.fn)); + + expect(result.fired).toBe(1); + expect(result.scanned).toBe(1); + expect(deliver.calls).toHaveLength(1); + expect(deliver.calls[0].agentId).toBe(agentId); + expect(deliver.calls[0].message).toContain('Review goals'); + expect(deliver.calls[0].message).toContain('Check their statuses.'); + expect(deliver.calls[0].idempotencyKey).toBe(`reminder:${reminder.id}:${now - 1000}`); + + const after = reminderRepo.getReminder(reminder.id)!; + expect(after.status).toBe('fired'); + expect(after.nextRunAt).toBeNull(); + expect(after.lastFiredAt).not.toBeNull(); + }); + + it('advances a cron reminder next_run_at and re-fires when it next comes due', async () => { + const before = Date.now(); + const reminder = reminderRepo.createReminder({ + spaceId, + agentId, + title: 'Weekly review', + triggerType: 'cron', + cronExpression: '0 9 * * 1', + nextRunAt: before - 1000, + }); + + const deliver = recordingDeliver(); + const r1 = await handleLongHorizonAgentReminderFire(makeJob(), makeDeps(deliver.fn)); + expect(r1.fired).toBe(1); + + const after1 = reminderRepo.getReminder(reminder.id)!; + expect(after1.status).toBe('active'); + expect(after1.nextRunAt).not.toBeNull(); + expect(after1.nextRunAt!).toBeGreaterThan(before); + + // next_run_at is now in the future -> not due, does not re-fire. + deliver.calls.length = 0; + const r2 = await handleLongHorizonAgentReminderFire(makeJob(), makeDeps(deliver.fn)); + expect(r2.fired).toBe(0); + expect(deliver.calls).toHaveLength(0); + + // Simulate time passing: next_run_at laps into the past -> fires again. + db.prepare('UPDATE space_long_horizon_agent_reminders SET next_run_at = ? WHERE id = ?').run( + Date.now() - 1000, + reminder.id + ); + const r3 = await handleLongHorizonAgentReminderFire(makeJob(), makeDeps(deliver.fn)); + expect(r3.fired).toBe(1); + expect(deliver.calls).toHaveLength(1); + }); + + it('skips a paused agent reminder without delivering', async () => { + const now = Date.now(); + const pausedAgent = reminderRepo.create({ + spaceId, + handle: 'paused', + displayName: 'Paused', + status: 'paused', + }); + reminderRepo.createReminder({ + spaceId, + agentId: pausedAgent.id, + title: 'should not fire', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + + const deliver = recordingDeliver(); + const result = await handleLongHorizonAgentReminderFire(makeJob(), makeDeps(deliver.fn)); + + // The due-query filters paused owners out, so it is never selected. + expect(result.scanned).toBe(0); + expect(result.fired).toBe(0); + expect(deliver.calls).toHaveLength(0); + }); + + it('does not advance when delivery reports not delivered', async () => { + const now = Date.now(); + const reminder = reminderRepo.createReminder({ + spaceId, + agentId, + title: 'maybe paused', + triggerType: 'cron', + cronExpression: '0 9 * * 1', + nextRunAt: now - 1000, + }); + const noDeliver = async (): Promise<{ delivered: boolean }> => ({ delivered: false }); + + const result = await handleLongHorizonAgentReminderFire(makeJob(), makeDeps(noDeliver)); + + expect(result.skipped).toBe(1); + expect(result.fired).toBe(0); + // Stays active and still due — fires when the agent is active again. + const after = reminderRepo.getReminder(reminder.id)!; + expect(after.status).toBe('active'); + expect(after.nextRunAt).toBe(now - 1000); + expect(after.lastFiredAt).toBeNull(); + }); + + it('does not double-fire when a previous scan already advanced the reminder', async () => { + const now = Date.now(); + const reminder = reminderRepo.createReminder({ + spaceId, + agentId, + title: 'race', + triggerType: 'cron', + cronExpression: '0 9 * * 1', + nextRunAt: now - 1000, + }); + + const deliver = recordingDeliver(); + await handleLongHorizonAgentReminderFire(makeJob(), makeDeps(deliver.fn)); + expect(deliver.calls).toHaveLength(1); + const advancedNext = reminderRepo.getReminder(reminder.id)!.nextRunAt!; + expect(advancedNext).toBeGreaterThan(now); + + // A retried scan sees the reminder no longer due (next_run_at moved forward). + deliver.calls.length = 0; + const r2 = await handleLongHorizonAgentReminderFire(makeJob(), makeDeps(deliver.fn)); + expect(r2.fired).toBe(0); + expect(deliver.calls).toHaveLength(0); + }); + + it('self-schedules the next scan job', async () => { + const deliver = recordingDeliver(); + const result = await handleLongHorizonAgentReminderFire(makeJob(), makeDeps(deliver.fn)); + expect(result.nextScanAt).toBeGreaterThan(Date.now()); + const pending = jobQueue.listJobs({ + queue: LONG_HORIZON_AGENT_REMINDER_FIRE, + status: 'pending', + limit: 5, + }); + expect(pending.length).toBeGreaterThanOrEqual(1); + }); +}); + +describe('enqueueLongHorizonAgentReminderScanIfMissing', () => { + let db: Database; + let jobQueue: JobQueueRepository; + + beforeEach(() => { + db = new Database(':memory:'); + createSpaceTables(db); + db.exec(` + CREATE TABLE IF NOT EXISTS job_queue ( + id TEXT PRIMARY KEY, + queue TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending' + CHECK(status IN ('pending', 'processing', 'completed', 'failed', 'dead')), + payload TEXT NOT NULL DEFAULT '{}', + result TEXT, + error TEXT, + priority INTEGER NOT NULL DEFAULT 0, + max_retries INTEGER NOT NULL DEFAULT 3, + retry_count INTEGER NOT NULL DEFAULT 0, + run_at INTEGER NOT NULL, + created_at INTEGER NOT NULL, + started_at INTEGER, + completed_at INTEGER + ); + CREATE INDEX IF NOT EXISTS idx_job_queue_dequeue ON job_queue(queue, status, priority DESC, run_at ASC); + `); + jobQueue = new JobQueueRepository(db as never); + }); + + afterEach(() => { + db.close(); + }); + + it('is idempotent — does not enqueue a second pending scan', () => { + enqueueLongHorizonAgentReminderScanIfMissing(jobQueue, Date.now() + 1000); + enqueueLongHorizonAgentReminderScanIfMissing(jobQueue, Date.now() + 2000); + const pending = jobQueue.listJobs({ + queue: LONG_HORIZON_AGENT_REMINDER_FIRE, + status: 'pending', + limit: 5, + }); + expect(pending).toHaveLength(1); + }); +}); diff --git a/packages/daemon/tests/unit/4-space-storage/storage/space-long-horizon-agent-repository.test.ts b/packages/daemon/tests/unit/4-space-storage/storage/space-long-horizon-agent-repository.test.ts index d7b0947731..fb328154d7 100644 --- a/packages/daemon/tests/unit/4-space-storage/storage/space-long-horizon-agent-repository.test.ts +++ b/packages/daemon/tests/unit/4-space-storage/storage/space-long-horizon-agent-repository.test.ts @@ -276,4 +276,151 @@ describe('SpaceLongHorizonAgentRepository', () => { ); expect(repo.listSubscriptions(agent.id)).toEqual([]); }); + + test('listDueReminders returns only active due reminders for active agents', () => { + const now = 10_000_000; + const activeAgent = repo.create({ + spaceId: 'space-1', + handle: 'active-agent', + displayName: 'Active', + }); + const pausedAgent = repo.create({ + spaceId: 'space-1', + handle: 'paused-agent', + displayName: 'Paused', + status: 'paused', + }); + + const dueReminder = repo.createReminder({ + spaceId: 'space-1', + agentId: activeAgent.id, + title: 'due', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + // Future next_run_at -> not due. + repo.createReminder({ + spaceId: 'space-1', + agentId: activeAgent.id, + title: 'future', + triggerType: 'at', + runAt: now + 60_000, + nextRunAt: now + 60_000, + }); + // Reminder status paused -> excluded. + repo.createReminder({ + spaceId: 'space-1', + agentId: activeAgent.id, + title: 'paused-reminder', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + status: 'paused', + }); + // Due but owner paused -> excluded by the agent-status join. + repo.createReminder({ + spaceId: 'space-1', + agentId: pausedAgent.id, + title: 'paused-owner', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + // Null next_run_at -> not schedulable, excluded. + repo.createReminder({ + spaceId: 'space-1', + agentId: activeAgent.id, + title: 'no-next', + triggerType: 'cron', + cronExpression: '0 9 * * 1', + }); + + const due = repo.listDueReminders(now); + expect(due.map((r) => r.id)).toEqual([dueReminder.id]); + }); + + test('advanceReminderAfterFire advances cron, fires one-shot, and honors the CAS', () => { + const now = 20_000_000; + const agent = repo.ensureCoordinator('space-1'); + + // cron -> caller-supplied next run, status stays active + const cron = repo.createReminder({ + spaceId: 'space-1', + agentId: agent.id, + title: 'cron', + triggerType: 'cron', + cronExpression: '0 9 * * 1', + nextRunAt: now - 1000, + }); + const futureNext = now + 60_000; + expect( + repo.advanceReminderAfterFire(cron.id, now - 1000, { + status: 'active', + nextRunAt: futureNext, + lastFiredAt: now, + }) + ).toBe(true); + const cronAfter = repo.getReminder(cron.id)!; + expect(cronAfter.status).toBe('active'); + expect(cronAfter.nextRunAt).toBe(futureNext); + expect(cronAfter.lastFiredAt).toBe(now); + + // one-shot 'at' -> terminal fired, no future run + const oneShot = repo.createReminder({ + spaceId: 'space-1', + agentId: agent.id, + title: 'one-shot', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + expect( + repo.advanceReminderAfterFire(oneShot.id, now - 1000, { + status: 'fired', + nextRunAt: null, + lastFiredAt: now, + }) + ).toBe(true); + const atAfter = repo.getReminder(oneShot.id)!; + expect(atAfter.status).toBe('fired'); + expect(atAfter.nextRunAt).toBeNull(); + + // CAS miss: expected next_run_at wrong -> no change + const live = repo.createReminder({ + spaceId: 'space-1', + agentId: agent.id, + title: 'live', + triggerType: 'cron', + cronExpression: '0 9 * * 1', + nextRunAt: now - 500, + }); + expect( + repo.advanceReminderAfterFire(live.id, now - 1, { + status: 'active', + nextRunAt: now + 60_000, + lastFiredAt: now, + }) + ).toBe(false); + expect(repo.getReminder(live.id)!.nextRunAt).toBe(now - 500); + + // CAS miss: reminder no longer active -> status guard rejects + const fired = repo.createReminder({ + spaceId: 'space-1', + agentId: agent.id, + title: 'already-fired', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + status: 'fired', + }); + expect( + repo.advanceReminderAfterFire(fired.id, now - 1000, { + status: 'fired', + nextRunAt: null, + lastFiredAt: now, + }) + ).toBe(false); + expect(repo.getReminder(fired.id)!.lastFiredAt).toBeNull(); + }); }); From f2117948f8096c09f5eea3bcd07411084b15d926 Mon Sep 17 00:00:00 2001 From: Marc Liu Date: Sun, 26 Jul 2026 15:51:45 -0400 Subject: [PATCH 2/9] fix: seed nextRunAt in createReminder RPC so cron reminders fire MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review on #2262. P1 (blocking): spaceLongHorizonAgent.createReminder accepted triggerType 'at'|'cron' but never passed nextRunAt, so the column defaulted to NULL and the scanner's due-query (next_run_at IS NOT NULL) never selected these rows. Since cron reminders can only be created via this RPC (the MCP create_agent_reminder hardcodes 'at'), the cron loop never worked end-to-end. Now seeds nextRunAt: 'at' -> runAt (required), 'cron' -> getNextRunAt after isValidCronExpression validation. P2: softened the handler idempotency-key comment — the SDK does not dedupe by message uuid, so the key is a label/tracing handle, not a hard dedupe fence; the CAS advance is the real double-fire guard. P3: build the idempotency key from fresh.nextRunAt (consistent with the CAS) and clarified the agent-state skip vs reminder-status guard. Adds RPC-path tests: a cron reminder created via the RPC is selected by listDueReminders when due; 'at' seeds nextRunAt=runAt; missing runAt and invalid cron are rejected. --- ...ong-horizon-agent-reminder-fire.handler.ts | 26 ++--- .../space-long-horizon-agent-handlers.ts | 24 +++++ .../space-long-horizon-agent-handlers.test.ts | 98 ++++++++++++++++++- 3 files changed, 135 insertions(+), 13 deletions(-) diff --git a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts index a72e6d8383..47adb2c967 100644 --- a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts +++ b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts @@ -119,32 +119,36 @@ async function fireReminder( } 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}`; + // Per-fire idempotency key derived from `fresh` (the same row the CAS below + // keys on). This labels/traces the delivery and is passed as the SDK message + // uuid — it is NOT a hard dedupe fence, since the SDK does not dedupe by + // message uuid. The real double-fire guard is the CAS advance below: once + // next_run_at moves forward, a retried scan no longer sees this occurrence. + const idempotencyKey = `reminder:${fresh.id}:${fresh.nextRunAt}`; let delivered: boolean; try { const result = await deliver({ - spaceId: reminder.spaceId, - agentId: reminder.agentId, + spaceId: fresh.spaceId, + agentId: fresh.agentId, message, idempotencyKey, }); delivered = result.delivered; } catch (err) { log.warn('lh-agent-reminder-fire: delivery threw', { - reminderId: reminder.id, + reminderId: fresh.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). + // Delivery returned false: the owning AGENT is missing/paused/disabled/ + // archived, or its session could not be ensured. (The reminder's own + // paused/cancelled status is handled by the active-status guard above — this + // branch is specifically an agent-state skip.) Do NOT advance: the 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); diff --git a/packages/daemon/src/lib/rpc-handlers/space-long-horizon-agent-handlers.ts b/packages/daemon/src/lib/rpc-handlers/space-long-horizon-agent-handlers.ts index cf5e40ff17..6af82751f1 100644 --- a/packages/daemon/src/lib/rpc-handlers/space-long-horizon-agent-handlers.ts +++ b/packages/daemon/src/lib/rpc-handlers/space-long-horizon-agent-handlers.ts @@ -24,6 +24,7 @@ import { validateSource, } from '../external-events/topic-validator'; import { getLongHorizonAgentTemplates } from '../space/agents/long-horizon-agent-templates'; +import { getNextRunAt, isValidCronExpression } from '../space/schedule/cron-utils'; import type { SpaceAgentManager } from '../space/managers/space-agent-manager'; import type { SpaceManager } from '../space/managers/space-manager'; import type { SpaceRuntimeService } from '../space/runtime/space-runtime-service'; @@ -501,6 +502,28 @@ export function setupSpaceLongHorizonAgentHandlers( if (!params.agentId) throw new Error('agentId is required'); if (!params.title) throw new Error('title is required'); if (!params.triggerType) throw new Error('triggerType is required'); + // Seed nextRunAt so the reminder-fire scanner can select the row — its + // due-query filters on `next_run_at IS NOT NULL`. Without this the column + // defaults to NULL and the reminder never fires. + let nextRunAt: number | null = null; + if (params.triggerType === 'at') { + if (typeof params.runAt !== 'number') { + throw new Error('runAt is required for triggerType "at"'); + } + nextRunAt = params.runAt; + } else { + const expression = params.cronExpression; + if (!expression) throw new Error('cronExpression is required for triggerType "cron"'); + if (!isValidCronExpression(expression)) { + throw new Error(`Invalid cron expression: ${expression}`); + } + const timezone = params.timezone ?? 'UTC'; + const firstRunAt = getNextRunAt(expression, timezone); + if (firstRunAt === null) { + throw new Error(`Invalid timezone or cron expression for reminder: ${timezone}`); + } + nextRunAt = firstRunAt; + } const reminder = repo.createReminder({ spaceId: params.spaceId, agentId: params.agentId, @@ -510,6 +533,7 @@ export function setupSpaceLongHorizonAgentHandlers( runAt: params.runAt, cronExpression: params.cronExpression, timezone: params.timezone, + nextRunAt, }); return { reminder }; }); diff --git a/packages/daemon/tests/unit/2-handlers/rpc-handlers/space-long-horizon-agent-handlers.test.ts b/packages/daemon/tests/unit/2-handlers/rpc-handlers/space-long-horizon-agent-handlers.test.ts index 31cb6cc31d..c508467741 100644 --- a/packages/daemon/tests/unit/2-handlers/rpc-handlers/space-long-horizon-agent-handlers.test.ts +++ b/packages/daemon/tests/unit/2-handlers/rpc-handlers/space-long-horizon-agent-handlers.test.ts @@ -1,8 +1,10 @@ -import { beforeEach, describe, expect, it, mock } from 'bun:test'; +import { afterEach, beforeEach, describe, expect, it, mock } from 'bun:test'; +import { Database } from 'bun:sqlite'; import type { MessageHub } from '@hyperneo/shared'; import { setupSpaceLongHorizonAgentHandlers } from '../../../../src/lib/rpc-handlers/space-long-horizon-agent-handlers'; import type { SpaceManager } from '../../../../src/lib/space/managers/space-manager'; -import type { SpaceLongHorizonAgentRepository } from '../../../../src/storage/repositories/space-long-horizon-agent-repository'; +import { SpaceLongHorizonAgentRepository } from '../../../../src/storage/repositories/space-long-horizon-agent-repository'; +import { createSpaceTables } from '../../helpers/space-test-db'; type RequestHandler = (data: unknown, context: unknown) => Promise; @@ -1066,3 +1068,95 @@ describe('Space long-horizon agent handlers', () => { }); }); }); + +// Drives the createReminder RPC with a REAL repository so we can assert the +// seeded next_run_at is visible to the reminder-fire scanner's due-query. +describe('spaceLongHorizonAgent.createReminder — nextRunAt seeding', () => { + let db: Database; + let repo: SpaceLongHorizonAgentRepository; + let handlers: Map; + let agentId: string; + + beforeEach(() => { + db = new Database(':memory:'); + createSpaceTables(db); + db.prepare( + `INSERT INTO spaces ( + id, slug, workspace_path, name, description, background_context, instructions, + allowed_models, session_ids, status, paused, stopped, autonomy_level, + max_concurrent_tasks, created_at, updated_at + ) VALUES (?, ?, ?, '', '', '', '', '[]', '[]', 'active', 0, 0, 1, 1, ?, ?)` + ).run('space-1', 'space-1', '/tmp/space-1', 1, 1); + repo = new SpaceLongHorizonAgentRepository(db); + const agent = repo.create({ spaceId: 'space-1', handle: 'steward', displayName: 'Steward' }); + agentId = agent.id; + + const hubData = createMockMessageHub(); + handlers = hubData.handlers; + setupSpaceLongHorizonAgentHandlers(hubData.hub, createMockSpaceManager(), repo); + }); + + afterEach(() => { + db.close(); + }); + + it('seeds nextRunAt for a cron reminder so the scanner selects it when due', async () => { + const result = await call<{ reminder: { id: string; nextRunAt: number | null } }>( + handlers, + 'spaceLongHorizonAgent.createReminder', + { + spaceId: 'space-1', + agentId, + title: 'Weekly review', + triggerType: 'cron', + cronExpression: '0 9 * * 1', + } + ); + expect(result.reminder.nextRunAt).not.toBeNull(); + const nextRunAt = result.reminder.nextRunAt as number; + // Next occurrence is in the future -> not due yet. + expect(repo.listDueReminders(Date.now()).map((r) => r.id)).toEqual([]); + // When the scheduled time arrives, the scanner selects it. + expect(repo.listDueReminders(nextRunAt + 1000).map((r) => r.id)).toEqual([result.reminder.id]); + }); + + it('seeds nextRunAt = runAt for a one-shot "at" reminder and is immediately due', async () => { + const runAt = Date.now() - 1000; + const result = await call<{ reminder: { id: string; nextRunAt: number | null } }>( + handlers, + 'spaceLongHorizonAgent.createReminder', + { + spaceId: 'space-1', + agentId, + title: 'Once', + triggerType: 'at', + runAt, + } + ); + expect(result.reminder.nextRunAt).toBe(runAt); + expect(repo.listDueReminders(Date.now()).map((r) => r.id)).toEqual([result.reminder.id]); + }); + + it('requires runAt for triggerType "at"', async () => { + await expect( + call(handlers, 'spaceLongHorizonAgent.createReminder', { + spaceId: 'space-1', + agentId, + title: 'no-runat', + triggerType: 'at', + }) + ).rejects.toThrow('runAt is required for triggerType "at"'); + }); + + it('rejects an invalid cron expression', async () => { + await expect( + call(handlers, 'spaceLongHorizonAgent.createReminder', { + spaceId: 'space-1', + agentId, + title: 'bad-cron', + triggerType: 'cron', + cronExpression: 'not a cron', + }) + ).rejects.toThrow('Invalid cron expression'); + }); +}); From c7429b3f1d32784bd31579ceb0d3f0fb6382759f Mon Sep 17 00:00:00 2001 From: Marc Liu Date: Sun, 26 Jul 2026 16:03:23 -0400 Subject: [PATCH 3/9] fix: harden reminder scanner (concurrency, space lifecycle, starvation) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the remaining inline review threads on #2262. P1 — overlapping scans double-deliver: the job processor runs with maxConcurrent=5 and the scanner pre-enqueues its successor, so two scans can overlap and both deliver the same reminder before either CAS- advances. Add an in-process per-reminder lock (mirrors SpaceRuntimeService.longTermAgentFlushes) so concurrent scans serialize on a reminder; the loser re-reads the advanced row and skips. Strengthen the guard to skip when fresh.nextRunAt > now. P1 — space lifecycle: a due reminder fired even when the owning space was paused/stopped/archived, letting the scanner recreate a stopped space's LH session (violating the stop contract). Filter space state in listDueReminders (join spaces: active, not paused, not stopped) and recheck in the handler before delivery, mirroring task-schedule-fire. P2 — queue starvation: a batch of always-failing reminders could fill every page and starve later healthy ones. listDueReminders now takes excludeIds; the scan pages through due reminders, excluding IDs already attempted this tick, until drained (per-scan cap with a log if hit). Adds tests: overlapping-scan lock delivers once; paused/stopped space skip (query + handler recheck); excludeIds pagination past poison rows. --- packages/daemon/src/app.ts | 2 + ...ong-horizon-agent-reminder-fire.handler.ts | 163 +++++++++++++----- .../space-long-horizon-agent-repository.ts | 26 ++- ...orizon-agent-reminder-fire.handler.test.ts | 73 +++++++- ...pace-long-horizon-agent-repository.test.ts | 88 ++++++++++ 5 files changed, 305 insertions(+), 47 deletions(-) diff --git a/packages/daemon/src/app.ts b/packages/daemon/src/app.ts index b4180c8dfe..aea1253592 100644 --- a/packages/daemon/src/app.ts +++ b/packages/daemon/src/app.ts @@ -945,9 +945,11 @@ 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()); + const lhAgentReminderSpaceRepo = new SpaceRepository(db.getDatabase()); jobProcessor.register(LONG_HORIZON_AGENT_REMINDER_FIRE, async (job) => { return handleLongHorizonAgentReminderFire(job, { reminderRepo: lhAgentReminderRepo, + spaceRepo: lhAgentReminderSpaceRepo, jobQueue, deliver: (args) => spaceRuntimeService.deliverLongHorizonAgentReminder(args), }); diff --git a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts index 47adb2c967..ca569860b3 100644 --- a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts +++ b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts @@ -3,33 +3,41 @@ * * 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'`. + * active AND whose owning space is active/not-paused/not-stopped (the + * partial `idx_space_lh_agent_reminders_due` index serves the reminder- + * side predicate; the agent/space joins skip non-deliverable states). + * 2. For each reminder: re-check it is still active and due, re-check the + * space lifecycle, 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. + * 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=`). - * 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. + * Concurrency / double-delivery safety: the job processor runs with + * `maxConcurrent` > 1, and this scanner pre-enqueues its own successor at the + * start of each run (for crash resilience). Two scan jobs can therefore overlap + * — a slow scan plus its due successor, or a reclaimed stale scan plus its + * successor on restart — and both could select the same reminder. Because all + * scan jobs run in a single daemon process (exclusive DB lock), an in-process + * per-reminder lock (`reminderFireLocks`, same pattern as + * SpaceRuntimeService.longTermAgentFlushes) serializes them: the second scan to + * reach a reminder re-reads it after the first has advanced/ fired it and skips. + * The compare-and-swap advance is a belt-and-suspenders fence for any path that + * bypasses the lock. * * 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. + * reminder may be re-delivered on the next scan — acceptable for a nag, and + * far better than silently losing a one-shot. + * + * Starvation safety: the scan pages through due reminders, excluding IDs it has + * already attempted this tick, so a batch of poison (always-failing) reminders + * cannot indefinitely block later, healthy ones. */ import type { SpaceLongHorizonAgentReminder } from '@hyperneo/shared'; @@ -37,12 +45,25 @@ 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 { SpaceRepository } from '../../storage/repositories/space-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; +/** Due-reminders fetched per query page within a single scan. */ +const PAGE_SIZE = 100; +/** Safety cap on reminders processed per scan (defends against pathological volume). */ +const MAX_PER_SCAN = 5000; + +/** + * In-process per-reminder locks. Serializes overlapping scans (concurrent + * processor slots, or a reclaimed stale scan + its successor) so two scans + * never deliver the same reminder occurrence. Module-level because all scan + * jobs share one daemon process. + */ +const reminderFireLocks = new Map>(); export interface LongHorizonAgentReminderFireResult extends Record { scanned: number; @@ -54,6 +75,7 @@ export interface LongHorizonAgentReminderFireResult extends Record { - const { reminderRepo, jobQueue, deliver } = deps; + const { reminderRepo, spaceRepo, 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 scanned = 0; let fired = 0; let skipped = 0; let failed = 0; + const attempted = new Set(); - for (const reminder of due) { - const outcome = await fireReminder(reminder, reminderRepo, deliver, now); - if (outcome === 'fired') fired++; - else if (outcome === 'skipped') skipped++; - else failed++; + // Page through due reminders, excluding ones already attempted this scan so a + // poison batch cannot starve later, healthy reminders. Loop until a page is + // short (drained) or the per-scan cap is reached. + while (scanned < MAX_PER_SCAN) { + const due = reminderRepo.listDueReminders(now, PAGE_SIZE, Array.from(attempted)); + if (due.length === 0) break; + for (const reminder of due) { + attempted.add(reminder.id); + scanned++; + let outcome: 'fired' | 'skipped' | 'failed'; + try { + outcome = await fireReminderSerialized(reminder, reminderRepo, spaceRepo, deliver, now); + } catch (err) { + outcome = 'failed'; + log.warn('lh-agent-reminder-fire: error firing reminder', { + reminderId: reminder.id, + error: err instanceof Error ? err.message : err, + }); + } + if (outcome === 'fired') fired++; + else if (outcome === 'skipped') skipped++; + else failed++; + } + if (due.length < PAGE_SIZE) break; // drained } - if (due.length > 0) { - log.debug('lh-agent-reminder-fire: scan complete', { - scanned: due.length, - fired, - skipped, - failed, + if (scanned >= MAX_PER_SCAN) { + // Not silent: if the cap is hit, some due reminders were deferred to a + // later scan. + log.warn('lh-agent-reminder-fire: per-scan cap reached; remaining due reminders deferred', { + scanned, }); } + if (scanned > 0) { + log.debug('lh-agent-reminder-fire: scan complete', { scanned, fired, skipped, failed }); + } - return { scanned: due.length, fired, skipped, failed, nextScanAt }; + return { scanned, fired, skipped, failed, nextScanAt }; +} + +/** + * Run fireReminder under a per-reminder in-process lock so overlapping scans + * serialize on the same reminder. The loser re-reads the row after the winner + * has advanced/fired it and skips. Mirrors longTermAgentFlushes. + */ +async function fireReminderSerialized( + reminder: SpaceLongHorizonAgentReminder, + reminderRepo: SpaceLongHorizonAgentRepository, + spaceRepo: SpaceRepository, + deliver: LongHorizonAgentReminderFireDeps['deliver'], + now: number +): Promise<'fired' | 'skipped' | 'failed'> { + const lockKey = reminder.id; + const previous = reminderFireLocks.get(lockKey) ?? Promise.resolve(); + let outcome: 'fired' | 'skipped' | 'failed' = 'failed'; + const current = previous + .catch(() => {}) + .then(async () => { + outcome = await fireReminder(reminder, reminderRepo, spaceRepo, deliver, now); + }); + reminderFireLocks.set(lockKey, current); + try { + await current; + } finally { + if (reminderFireLocks.get(lockKey) === current) reminderFireLocks.delete(lockKey); + } + return outcome; } async function fireReminder( reminder: SpaceLongHorizonAgentReminder, reminderRepo: SpaceLongHorizonAgentRepository, + spaceRepo: SpaceRepository, 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. + // Re-read: the reminder may have been paused/cancelled/fired/advanced between + // the due-scan and now (e.g. by a serialized peer scan). A null or future + // next_run_at means it is no longer schedulable for this tick. const fresh = reminderRepo.getReminder(reminder.id); - if (!fresh || fresh.status !== 'active' || fresh.nextRunAt === null) { + if (!fresh || fresh.status !== 'active' || fresh.nextRunAt === null || fresh.nextRunAt > now) { + return 'skipped'; + } + + // Space lifecycle guard: never fire (and never let delivery recreate a + // session) for a paused/stopped/archived space — mirrors task-schedule-fire's + // space contract. The due-query already filters this, but a space can be + // stopped between select and fire. + const space = spaceRepo.getSpace(fresh.spaceId); + if (!space || space.status !== 'active' || space.paused || space.stopped) { return 'skipped'; } @@ -122,8 +206,9 @@ async function fireReminder( // Per-fire idempotency key derived from `fresh` (the same row the CAS below // keys on). This labels/traces the delivery and is passed as the SDK message // uuid — it is NOT a hard dedupe fence, since the SDK does not dedupe by - // message uuid. The real double-fire guard is the CAS advance below: once - // next_run_at moves forward, a retried scan no longer sees this occurrence. + // message uuid. The real double-delivery guard is the per-reminder lock plus + // the CAS advance: once next_run_at moves forward, a peer/retried scan no + // longer sees this occurrence as due. const idempotencyKey = `reminder:${fresh.id}:${fresh.nextRunAt}`; let delivered: boolean; @@ -168,7 +253,7 @@ async function fireReminder( // 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 }); + log.debug('lh-agent-reminder-fire: advance CAS missed', { reminderId: fresh.id }); } return 'fired'; } diff --git a/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts b/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts index 3218da0212..ecc080dcbb 100644 --- a/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts +++ b/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts @@ -296,22 +296,36 @@ export class SpaceLongHorizonAgentRepository { /** * 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. + * status is 'active' AND whose owning space is active and not paused/stopped. + * Powers the reminder-fire scanner job — the partial index + * `idx_space_lh_agent_reminders_due` (status='active') serves the reminder- + * side predicate; the agent/space joins skip reminders whose owner or space + * is not in a deliverable lifecycle state. + * + * `excludeIds` lets a single scan page past reminders it has already + * attempted this tick, so a batch of poison (always-failing) reminders + * cannot indefinitely block later, healthy ones. */ - listDueReminders(now: number, limit = 100): SpaceLongHorizonAgentReminder[] { + listDueReminders( + now: number, + limit = 100, + excludeIds: string[] = [] + ): SpaceLongHorizonAgentReminder[] { + const excludeClause = + excludeIds.length > 0 ? `AND r.id NOT IN (${excludeIds.map(() => '?').join(',')})` : ''; 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 + INNER JOIN spaces s ON s.id = r.space_id WHERE r.status = 'active' AND r.next_run_at IS NOT NULL AND r.next_run_at <= ? AND a.status = 'active' + AND s.status = 'active' AND s.paused = 0 AND s.stopped = 0 + ${excludeClause} ORDER BY r.next_run_at ASC LIMIT ?` ) - .all(now, limit) as Record[]; + .all(now, ...excludeIds, limit) as Record[]; return rows.map(rowToReminder); } diff --git a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts index 8219e657cb..c83a8b8dab 100644 --- a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts +++ b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts @@ -52,6 +52,7 @@ function recordingDeliver() { describe('handleLongHorizonAgentReminderFire', () => { let db: Database; let reminderRepo: SpaceLongHorizonAgentRepository; + let spaceRepo: SpaceRepository; let jobQueue: JobQueueRepository; let spaceId: string; let agentId: string; @@ -80,7 +81,7 @@ describe('handleLongHorizonAgentReminderFire', () => { CREATE INDEX IF NOT EXISTS idx_job_queue_dequeue ON job_queue(queue, status, priority DESC, run_at ASC); `); - const spaceRepo = new SpaceRepository(db as never); + spaceRepo = new SpaceRepository(db as never); reminderRepo = new SpaceLongHorizonAgentRepository(db); jobQueue = new JobQueueRepository(db as never); @@ -100,7 +101,7 @@ describe('handleLongHorizonAgentReminderFire', () => { }); function makeDeps(deliver: (args: DeliverArgs) => Promise<{ delivered: boolean }>) { - return { reminderRepo, jobQueue, deliver }; + return { reminderRepo, spaceRepo, jobQueue, deliver }; } it('fires a due one-shot reminder, delivers the body, and marks it fired', async () => { @@ -241,6 +242,74 @@ describe('handleLongHorizonAgentReminderFire', () => { expect(deliver.calls).toHaveLength(0); }); + it('does not double-deliver when two scans overlap (in-process lock)', async () => { + const now = Date.now(); + reminderRepo.createReminder({ + spaceId, + agentId, + title: 'contended', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + + const deliver = recordingDeliver(); + // Two scan jobs running concurrently (the processor has >1 slot). + const [, r2] = await Promise.all([ + handleLongHorizonAgentReminderFire(makeJob(), makeDeps(deliver.fn)), + handleLongHorizonAgentReminderFire(makeJob(), makeDeps(deliver.fn)), + ]); + + // Only one scan delivers; the loser re-reads the advanced row and skips. + expect(deliver.calls).toHaveLength(1); + expect(r2.fired + r2.skipped).toBeGreaterThanOrEqual(1); + }); + + it('skips a reminder whose owning space is paused', async () => { + const now = Date.now(); + reminderRepo.createReminder({ + spaceId, + agentId, + title: 'paused-space', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + // Pause the space after creation. + db.prepare('UPDATE spaces SET paused = 1 WHERE id = ?').run(spaceId); + + const deliver = recordingDeliver(); + const result = await handleLongHorizonAgentReminderFire(makeJob(), makeDeps(deliver.fn)); + + // Filtered at the due-query (not even selected) and skipped at delivery. + expect(result.scanned).toBe(0); + expect(result.fired).toBe(0); + expect(deliver.calls).toHaveLength(0); + }); + + it('skips when the space is stopped between select and fire', async () => { + const now = Date.now(); + reminderRepo.createReminder({ + spaceId, + agentId, + title: 'stop-mid-scan', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + + // The deliver hook stops the space right before injection, exercising the + // handler-level space recheck (the due-query already passed). + const stopAndDeny = async (): Promise<{ delivered: boolean }> => { + db.prepare('UPDATE spaces SET stopped = 1 WHERE id = ?').run(spaceId); + return { delivered: false }; + }; + const result = await handleLongHorizonAgentReminderFire(makeJob(), makeDeps(stopAndDeny)); + + expect(result.fired).toBe(0); + expect(result.skipped).toBe(1); + }); + it('self-schedules the next scan job', async () => { const deliver = recordingDeliver(); const result = await handleLongHorizonAgentReminderFire(makeJob(), makeDeps(deliver.fn)); diff --git a/packages/daemon/tests/unit/4-space-storage/storage/space-long-horizon-agent-repository.test.ts b/packages/daemon/tests/unit/4-space-storage/storage/space-long-horizon-agent-repository.test.ts index fb328154d7..2f7f1e806b 100644 --- a/packages/daemon/tests/unit/4-space-storage/storage/space-long-horizon-agent-repository.test.ts +++ b/packages/daemon/tests/unit/4-space-storage/storage/space-long-horizon-agent-repository.test.ts @@ -340,6 +340,94 @@ describe('SpaceLongHorizonAgentRepository', () => { expect(due.map((r) => r.id)).toEqual([dueReminder.id]); }); + test('listDueReminders excludes reminders for paused or stopped spaces', () => { + const now = 30_000_000; + // A second space, paused; and a third, stopped. + db.prepare( + `INSERT INTO spaces ( + id, slug, workspace_path, name, description, background_context, instructions, + allowed_models, session_ids, status, paused, stopped, autonomy_level, + max_concurrent_tasks, created_at, updated_at + ) VALUES (?, ?, ?, '', '', '', '', '[]', '[]', 'active', 1, 0, 1, 1, ?, ?)` + ).run('space-paused', 'space-paused', '/tmp/space-paused', 1, 1); + db.prepare( + `INSERT INTO spaces ( + id, slug, workspace_path, name, description, background_context, instructions, + allowed_models, session_ids, status, paused, stopped, autonomy_level, + max_concurrent_tasks, created_at, updated_at + ) VALUES (?, ?, ?, '', '', '', '', '[]', '[]', 'active', 0, 1, 1, 1, ?, ?)` + ).run('space-stopped', 'space-stopped', '/tmp/space-stopped', 1, 1); + + const activeAgent = repo.create({ spaceId: 'space-1', handle: 'a', displayName: 'A' }); + const pausedSpaceAgent = repo.create({ + spaceId: 'space-paused', + handle: 'p', + displayName: 'P', + }); + const stoppedSpaceAgent = repo.create({ + spaceId: 'space-stopped', + handle: 's', + displayName: 'S', + }); + + const activeReminder = repo.createReminder({ + spaceId: 'space-1', + agentId: activeAgent.id, + title: 'active-space', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + repo.createReminder({ + spaceId: 'space-paused', + agentId: pausedSpaceAgent.id, + title: 'paused-space', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + repo.createReminder({ + spaceId: 'space-stopped', + agentId: stoppedSpaceAgent.id, + title: 'stopped-space', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + + const due = repo.listDueReminders(now); + expect(due.map((r) => r.id)).toEqual([activeReminder.id]); + }); + + test('listDueReminders pages past excluded ids so poison batches cannot starve later rows', () => { + const now = 40_000_000; + const agent = repo.ensureCoordinator('space-1'); + const ids: string[] = []; + for (let i = 0; i < 3; i++) { + const r = repo.createReminder({ + spaceId: 'space-1', + agentId: agent.id, + title: `r${i}`, + triggerType: 'at', + runAt: now - 1000 - i, + nextRunAt: now - 1000 - i, + }); + ids.push(r.id); + } + + // First page returns the two earliest. + const page1 = repo.listDueReminders(now, 2); + expect(page1.map((r) => r.id)).toEqual([ids[2], ids[1]]); + + // Second page excludes the first page's ids and returns the remaining one. + const page2 = repo.listDueReminders(now, 2, [ids[2], ids[1]]); + expect(page2.map((r) => r.id)).toEqual([ids[0]]); + + // Once all are excluded, nothing remains. + const page3 = repo.listDueReminders(now, 2, ids); + expect(page3).toEqual([]); + }); + test('advanceReminderAfterFire advances cron, fires one-shot, and honors the CAS', () => { const now = 20_000_000; const agent = repo.ensureCoordinator('space-1'); From 2f83caa8285c5561c45ea60d8caae549867ef7df Mon Sep 17 00:00:00 2001 From: Marc Liu Date: Sun, 26 Jul 2026 16:10:58 -0400 Subject: [PATCH 4/9] fix: backfill legacy reminders, advance from fire time, bound scan MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the second round of inline review on #2262. P1 — legacy reminders never fire: active reminders created via the RPC before the nextRunAt seed fix retain a NULL next_run_at, which the due-query excludes. Add a startup backfill (idempotent; only touches NULL rows) that seeds next_run_at from run_at ('at') or the next cron occurrence ('cron'), plus the supporting repo methods. P2 — cron advance from stale now: a slow scan could deliver a reminder after the scan-wide `now` was captured, persisting a next_run_at that is already overdue and causing an immediate re-fire. Compute the next occurrence and last_fired_at from a fresh timestamp at advance time. P2 — unbounded scan stalls shutdown: each awaited delivery can block on the message-queue timeout and jobProcessor.stop() awaits in-flight jobs with no timeout, so a poison batch could stall graceful shutdown for minutes. Add a per-scan wall-clock budget (20s, < NEXT_SCAN_DELAY_MS) with a warn-log when hit. Adds a backfill unit test (seeds NULL rows for 'at' and cron; idempotent). --- packages/daemon/src/app.ts | 15 ++++ ...ong-horizon-agent-reminder-fire.handler.ts | 65 ++++++++++++++-- .../space-long-horizon-agent-repository.ts | 29 ++++++++ ...orizon-agent-reminder-fire.handler.test.ts | 74 +++++++++++++++++++ 4 files changed, 176 insertions(+), 7 deletions(-) diff --git a/packages/daemon/src/app.ts b/packages/daemon/src/app.ts index aea1253592..10b8034146 100644 --- a/packages/daemon/src/app.ts +++ b/packages/daemon/src/app.ts @@ -81,6 +81,7 @@ import { } from './lib/job-queue-constants'; import { handleTaskScheduleFire } from './lib/job-handlers/task-schedule-fire.handler'; import { + backfillLongHorizonAgentReminderNextRunAt, enqueueLongHorizonAgentReminderScanIfMissing, handleLongHorizonAgentReminderFire, } from './lib/job-handlers/long-horizon-agent-reminder-fire.handler'; @@ -1051,6 +1052,20 @@ export async function createDaemonApp(options: CreateDaemonAppOptions): Promise< logInfo('[Daemon] Ensured initial memory_consolidation job'); enqueueLongHorizonAgentReminderScanIfMissing(jobQueue, Date.now()); logInfo('[Daemon] Ensured initial longHorizonAgentReminder.fire scan job'); + // Backfill next_run_at for reminders created before the scanner shipped + // (their create paths now seed it). Idempotent; non-fatal on error. + if (process.env.NODE_ENV !== 'test') { + try { + const backfilled = backfillLongHorizonAgentReminderNextRunAt(lhAgentReminderRepo); + if (backfilled > 0) { + logInfo('[Daemon] Backfilled next_run_at for LH agent reminders', { + count: backfilled, + }); + } + } catch (err) { + logError('[Daemon] LH agent reminder backfill failed (non-fatal):', err); + } + } // Start job queue processor last (after all handler registrations) jobProcessor.start(); diff --git a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts index ca569860b3..84105d434d 100644 --- a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts +++ b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts @@ -56,6 +56,14 @@ const NEXT_SCAN_DELAY_MS = 30_000; const PAGE_SIZE = 100; /** Safety cap on reminders processed per scan (defends against pathological volume). */ const MAX_PER_SCAN = 5000; +/** + * Per-scan wall-clock budget. A single slow delivery (e.g. a session that never + * drains its message queue) can block on the message-queue timeout, and the job + * processor's stop() awaits in-flight jobs with no timeout — so without a bound + * a poison batch could stall graceful shutdown for minutes. Kept below + * NEXT_SCAN_DELAY_MS so a scan normally finishes before its successor is due. + */ +const SCAN_DEADLINE_MS = 20_000; /** * In-process per-reminder locks. Serializes overlapping scans (concurrent @@ -108,11 +116,18 @@ export async function handleLongHorizonAgentReminderFire( let skipped = 0; let failed = 0; const attempted = new Set(); + const scanStartedAt = now; // Page through due reminders, excluding ones already attempted this scan so a // poison batch cannot starve later, healthy reminders. Loop until a page is - // short (drained) or the per-scan cap is reached. + // short (drained), the per-scan cap is reached, or the wall-clock budget + // expires (bounds shutdown latency — see SCAN_DEADLINE_MS). + let deadlineHit = false; while (scanned < MAX_PER_SCAN) { + if (Date.now() - scanStartedAt > SCAN_DEADLINE_MS) { + deadlineHit = true; + break; + } const due = reminderRepo.listDueReminders(now, PAGE_SIZE, Array.from(attempted)); if (due.length === 0) break; for (const reminder of due) { @@ -142,6 +157,12 @@ export async function handleLongHorizonAgentReminderFire( scanned, }); } + if (deadlineHit) { + log.warn('lh-agent-reminder-fire: scan deadline reached; remaining due reminders deferred', { + scanned, + budgetMs: SCAN_DEADLINE_MS, + }); + } if (scanned > 0) { log.debug('lh-agent-reminder-fire: scan complete', { scanned, fired, skipped, failed }); } @@ -236,17 +257,21 @@ async function fireReminder( // 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. + // Compute the post-fire state from a FRESH timestamp (not the scan-wide + // `now`): a slow scan may deliver this reminder well after the scan started, + // and computing the cron's next occurrence from a stale `now` could persist a + // next_run_at that is already overdue → an immediate re-fire. 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. + const firedAt = Date.now(); const nextRunAt = fresh.triggerType === 'cron' && fresh.cronExpression - ? getNextRunAt(fresh.cronExpression, fresh.timezone, now) + ? getNextRunAt(fresh.cronExpression, fresh.timezone, firedAt) : null; const updates = nextRunAt === null - ? { status: 'fired' as const, nextRunAt: null, lastFiredAt: now } - : { status: 'active' as const, nextRunAt, lastFiredAt: now }; + ? { status: 'fired' as const, nextRunAt: null, lastFiredAt: firedAt } + : { status: 'active' as const, nextRunAt, lastFiredAt: firedAt }; const applied = reminderRepo.advanceReminderAfterFire(fresh.id, fresh.nextRunAt, updates); if (!applied) { @@ -276,6 +301,32 @@ export function enqueueLongHorizonAgentReminderScanIfMissing( } } +/** + * Backfill `next_run_at` for active reminders that pre-date the scanner (their + * create paths now seed it, so only rows created before this feature shipped + * have a NULL value). cron → next occurrence from the expression; 'at' → run_at + * (or now if unknown). Idempotent — only touches rows with a NULL next_run_at. + * Returns the number of rows backfilled. + */ +export function backfillLongHorizonAgentReminderNextRunAt( + reminderRepo: SpaceLongHorizonAgentRepository +): number { + const stale = reminderRepo.listActiveRemindersWithNullNextRunAt(); + if (stale.length === 0) return 0; + const now = Date.now(); + let count = 0; + for (const reminder of stale) { + const nextRunAt = + reminder.triggerType === 'cron' && reminder.cronExpression + ? getNextRunAt(reminder.cronExpression, reminder.timezone || 'UTC', now) + : (reminder.runAt ?? now); + if (nextRunAt === null) continue; // unparseable cron — leave for an operator to fix + reminderRepo.setReminderNextRunAt(reminder.id, nextRunAt); + count++; + } + return count; +} + 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. diff --git a/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts b/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts index ecc080dcbb..ff7c76b034 100644 --- a/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts +++ b/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts @@ -363,6 +363,35 @@ export class SpaceLongHorizonAgentRepository { return result.changes > 0; } + /** + * Active reminders (for active agents) with a NULL `next_run_at` — rows + * created before the scanner existed (the create paths now seed it). Used by + * the startup backfill so pre-existing reminders become schedulable. + */ + listActiveRemindersWithNullNextRunAt(): 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 NULL AND a.status = 'active'` + ) + .all() as Record[]; + return rows.map(rowToReminder); + } + + /** + * Set `next_run_at` unconditionally (backfill path). Distinct from + * `advanceReminderAfterFire` (a CAS) because these rows have a NULL + * `next_run_at` the CAS cannot match. + */ + setReminderNextRunAt(id: string, nextRunAt: number): void { + this.db + .prepare( + `UPDATE space_long_horizon_agent_reminders SET next_run_at = ?, updated_at = ? WHERE id = ?` + ) + .run(nextRunAt, Date.now(), id); + } + createSubscription( params: CreateSpaceLongHorizonAgentSubscriptionParams ): SpaceLongHorizonAgentEventSubscription { diff --git a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts index c83a8b8dab..5e12d92ecc 100644 --- a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts +++ b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts @@ -5,6 +5,7 @@ import { afterEach, beforeEach, describe, expect, it } from 'bun:test'; import { Database } from 'bun:sqlite'; import { + backfillLongHorizonAgentReminderNextRunAt, enqueueLongHorizonAgentReminderScanIfMissing, handleLongHorizonAgentReminderFire, } from '../../../../src/lib/job-handlers/long-horizon-agent-reminder-fire.handler'; @@ -367,3 +368,76 @@ describe('enqueueLongHorizonAgentReminderScanIfMissing', () => { expect(pending).toHaveLength(1); }); }); + +describe('backfillLongHorizonAgentReminderNextRunAt', () => { + let db: Database; + let reminderRepo: SpaceLongHorizonAgentRepository; + let spaceId: string; + let agentId: string; + + beforeEach(() => { + db = new Database(':memory:'); + createSpaceTables(db); + const spaceRepo = new SpaceRepository(db as never); + reminderRepo = new SpaceLongHorizonAgentRepository(db); + const space = spaceRepo.createSpace({ + slug: 'test', + workspacePath: '/workspace/test', + name: 'Test', + description: 'Test space', + }); + spaceId = space.id; + agentId = reminderRepo.create({ spaceId, handle: 'steward', displayName: 'Steward' }).id; + }); + + afterEach(() => { + db.close(); + }); + + it('seeds next_run_at for pre-existing reminders with a NULL value', () => { + const past = Date.now() - 60_000; + // 'at' reminder created before the seed fix (NULL next_run_at). + const atReminder = reminderRepo.createReminder({ + spaceId, + agentId, + title: 'legacy-at', + triggerType: 'at', + runAt: past, + }); + // cron reminder created before the seed fix (NULL next_run_at). + const cronReminder = reminderRepo.createReminder({ + spaceId, + agentId, + title: 'legacy-cron', + triggerType: 'cron', + cronExpression: '0 9 * * 1', + }); + expect(atReminder.nextRunAt).toBeNull(); + expect(cronReminder.nextRunAt).toBeNull(); + + const count = backfillLongHorizonAgentReminderNextRunAt(reminderRepo); + + expect(count).toBe(2); + const atAfter = reminderRepo.getReminder(atReminder.id)!; + const cronAfter = reminderRepo.getReminder(cronReminder.id)!; + expect(atAfter.nextRunAt).toBe(past); // 'at' falls back to run_at + expect(cronAfter.nextRunAt).not.toBeNull(); + expect(cronAfter.nextRunAt!).toBeGreaterThan(Date.now()); // next cron occurrence + + // Both are now schedulable (the 'at' one is immediately due). + const due = reminderRepo.listDueReminders(past + 1000).map((r) => r.id); + expect(due).toContain(atReminder.id); + }); + + it('is idempotent — no-op when no NULL rows remain', () => { + reminderRepo.createReminder({ + spaceId, + agentId, + title: 'already-seeded', + triggerType: 'at', + runAt: Date.now() - 1000, + nextRunAt: Date.now() - 1000, + }); + expect(backfillLongHorizonAgentReminderNextRunAt(reminderRepo)).toBe(0); + }); +}); From 8e8cbe4371cf4c27e829d8a04eb52b11b125d4c9 Mon Sep 17 00:00:00 2001 From: Marc Liu Date: Sun, 26 Jul 2026 16:45:01 -0400 Subject: [PATCH 5/9] fix: per-reminder deadline, persisted-message guard, backfill all agents MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the round-3 review (#4782543292) on #2262. P2 — deadline checked only between pages: a page of 100 stuck sessions (each blocking ~30s on enqueueWithId) could run ~50min on one worker before the between-page check tripped. Move the SCAN_DEADLINE_MS check inside the inner per-reminder loop (labeled break) so it trips before each delivery. P2 — persisted-message amplification on failed delivery: saveUserMessage runs before enqueueWithId, which can time out and throw, leaving a durable sdk_messages row while the handler reports failure and retries. Each retry would insert another row (new PK, same uuid; no unique constraint) and flood the agent on replay. Added a pre-delivery idempotency guard (isOccurrencePersisted dep, wired via getMessageByStatusAndUuid): if the occurrence's uuid is already in sdk_messages, skip re-injection and advance — exactly one row per occurrence, replayed once, and the reminder advances to break the retry loop. Chose skip+advance over delete-on-failure so a legitimate persisted reminder is never lost on a transient failure. P3 — backfill omitted paused-agent reminders: dropped the agent-status filter from listActiveRemindersWithNullNextRunAt so reminders owned by paused/disabled agents get next_run_at seeded and fire on reactivation (the due-query gates firing on agent state at fire time). Adds tests: persisted-occurrence guard skips injection + advances; backfill covers paused-agent reminders. --- packages/daemon/src/app.ts | 14 +++ ...ong-horizon-agent-reminder-fire.handler.ts | 107 ++++++++++++------ .../space-long-horizon-agent-repository.ts | 14 ++- ...orizon-agent-reminder-fire.handler.test.ts | 56 ++++++++- 4 files changed, 150 insertions(+), 41 deletions(-) diff --git a/packages/daemon/src/app.ts b/packages/daemon/src/app.ts index 10b8034146..f72814db03 100644 --- a/packages/daemon/src/app.ts +++ b/packages/daemon/src/app.ts @@ -85,6 +85,7 @@ import { enqueueLongHorizonAgentReminderScanIfMissing, handleLongHorizonAgentReminderFire, } from './lib/job-handlers/long-horizon-agent-reminder-fire.handler'; +import { longTermAgentSessionId } from './lib/space/long-term-agent-session'; import { TaskScheduleRepository } from './storage/repositories/task-schedule-repository'; import { SpaceRepository } from './storage/repositories/space-repository'; import { SpaceTaskRepository } from './storage/repositories/space-task-repository'; @@ -953,6 +954,19 @@ export async function createDaemonApp(options: CreateDaemonAppOptions): Promise< spaceRepo: lhAgentReminderSpaceRepo, jobQueue, deliver: (args) => spaceRuntimeService.deliverLongHorizonAgentReminder(args), + // Guards against duplicate persisted messages on retry: saveUserMessage + // runs before enqueueWithId, so a timed-out delivery leaves a durable + // row that would otherwise be re-inserted each retry. Checks the + // occurrence's uuid across the enqueued/consumed send statuses. + isOccurrencePersisted: (spaceId, agentId, idempotencyKey) => { + const sessionId = longTermAgentSessionId(spaceId, agentId); + const messageDb = reactiveDb?.db; + if (!messageDb) return false; + return ( + messageDb.getMessageByStatusAndUuid(sessionId, 'enqueued', idempotencyKey) != null || + messageDb.getMessageByStatusAndUuid(sessionId, 'consumed', idempotencyKey) != null + ); + }, }); }); diff --git a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts index 84105d434d..2460076bf1 100644 --- a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts +++ b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts @@ -99,13 +99,24 @@ export interface LongHorizonAgentReminderFireDeps { message: string; idempotencyKey: string; }) => Promise<{ delivered: boolean }>; + /** + * Optional pre-delivery guard against duplicate persisted messages. A prior + * attempt may have already persisted the occurrence's sdk_messages row — + * `saveUserMessage` runs before `enqueueWithId`, which can time out and throw, + * leaving the row durable while the handler reports failure and retries. + * Re-injecting would insert a duplicate row (new PK, same uuid; no unique + * constraint) and flood the agent on replay. When this returns true the + * scanner skips delivery and advances — the persisted row is replayed by the + * session. Returns false (treat as not-persisted) when unset. + */ + isOccurrencePersisted?: (spaceId: string, agentId: string, idempotencyKey: string) => boolean; } export async function handleLongHorizonAgentReminderFire( _job: Job, deps: LongHorizonAgentReminderFireDeps ): Promise { - const { reminderRepo, spaceRepo, jobQueue, deliver } = deps; + const { reminderRepo, spaceRepo, jobQueue, deliver, isOccurrencePersisted } = 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. @@ -123,19 +134,29 @@ export async function handleLongHorizonAgentReminderFire( // short (drained), the per-scan cap is reached, or the wall-clock budget // expires (bounds shutdown latency — see SCAN_DEADLINE_MS). let deadlineHit = false; - while (scanned < MAX_PER_SCAN) { - if (Date.now() - scanStartedAt > SCAN_DEADLINE_MS) { - deadlineHit = true; - break; - } + scanLoop: while (scanned < MAX_PER_SCAN) { const due = reminderRepo.listDueReminders(now, PAGE_SIZE, Array.from(attempted)); if (due.length === 0) break; for (const reminder of due) { + // Check the budget before EACH delivery, not just between pages: a single + // stuck delivery can block for the message-queue timeout (~30s), so a + // page of 100 could otherwise run for many minutes on one worker. + if (Date.now() - scanStartedAt > SCAN_DEADLINE_MS) { + deadlineHit = true; + break scanLoop; + } attempted.add(reminder.id); scanned++; let outcome: 'fired' | 'skipped' | 'failed'; try { - outcome = await fireReminderSerialized(reminder, reminderRepo, spaceRepo, deliver, now); + outcome = await fireReminderSerialized( + reminder, + reminderRepo, + spaceRepo, + deliver, + isOccurrencePersisted, + now + ); } catch (err) { outcome = 'failed'; log.warn('lh-agent-reminder-fire: error firing reminder', { @@ -180,6 +201,7 @@ async function fireReminderSerialized( reminderRepo: SpaceLongHorizonAgentRepository, spaceRepo: SpaceRepository, deliver: LongHorizonAgentReminderFireDeps['deliver'], + isOccurrencePersisted: LongHorizonAgentReminderFireDeps['isOccurrencePersisted'], now: number ): Promise<'fired' | 'skipped' | 'failed'> { const lockKey = reminder.id; @@ -188,7 +210,14 @@ async function fireReminderSerialized( const current = previous .catch(() => {}) .then(async () => { - outcome = await fireReminder(reminder, reminderRepo, spaceRepo, deliver, now); + outcome = await fireReminder( + reminder, + reminderRepo, + spaceRepo, + deliver, + isOccurrencePersisted, + now + ); }); reminderFireLocks.set(lockKey, current); try { @@ -204,6 +233,7 @@ async function fireReminder( reminderRepo: SpaceLongHorizonAgentRepository, spaceRepo: SpaceRepository, deliver: LongHorizonAgentReminderFireDeps['deliver'], + isOccurrencePersisted: LongHorizonAgentReminderFireDeps['isOccurrencePersisted'], now: number ): Promise<'fired' | 'skipped' | 'failed'> { // Re-read: the reminder may have been paused/cancelled/fired/advanced between @@ -225,37 +255,48 @@ async function fireReminder( const message = formatReminderMessage(fresh); // Per-fire idempotency key derived from `fresh` (the same row the CAS below - // keys on). This labels/traces the delivery and is passed as the SDK message - // uuid — it is NOT a hard dedupe fence, since the SDK does not dedupe by - // message uuid. The real double-delivery guard is the per-reminder lock plus - // the CAS advance: once next_run_at moves forward, a peer/retried scan no - // longer sees this occurrence as due. + // keys on). Passed as the SDK message uuid. const idempotencyKey = `reminder:${fresh.id}:${fresh.nextRunAt}`; + // Pre-delivery idempotency: a prior attempt at this occurrence may already + // have persisted its sdk_messages row — `saveUserMessage` runs before + // `enqueueWithId`, which can time out and throw, leaving the row durable + // while we report failure and retry. Re-injecting would insert a duplicate + // row (new PK, same uuid) and flood the agent on replay. If it's already + // persisted, skip the inject and treat the occurrence as delivered — the row + // is replayed/delivered by the session — then advance. let delivered: boolean; - try { - const result = await deliver({ - spaceId: fresh.spaceId, - agentId: fresh.agentId, - message, - idempotencyKey, - }); - delivered = result.delivered; - } catch (err) { - log.warn('lh-agent-reminder-fire: delivery threw', { + if (isOccurrencePersisted?.(fresh.spaceId, fresh.agentId, idempotencyKey)) { + log.debug('lh-agent-reminder-fire: occurrence already persisted, advancing', { reminderId: fresh.id, - error: err instanceof Error ? err.message : err, }); - return 'failed'; - } + delivered = true; + } else { + try { + const result = await deliver({ + spaceId: fresh.spaceId, + agentId: fresh.agentId, + message, + idempotencyKey, + }); + delivered = result.delivered; + } catch (err) { + log.warn('lh-agent-reminder-fire: delivery threw', { + reminderId: fresh.id, + error: err instanceof Error ? err.message : err, + }); + return 'failed'; + } - // Delivery returned false: the owning AGENT is missing/paused/disabled/ - // archived, or its session could not be ensured. (The reminder's own - // paused/cancelled status is handled by the active-status guard above — this - // branch is specifically an agent-state skip.) Do NOT advance: the 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'; + // Delivery returned false: the owning AGENT is missing/paused/disabled/ + // archived, or its session could not be ensured. (The reminder's own + // paused/cancelled status is handled by the active-status guard above — + // this branch is specifically an agent-state skip.) Do NOT advance: the + // 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 from a FRESH timestamp (not the scan-wide // `now`): a slow scan may deliver this reminder well after the scan started, diff --git a/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts b/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts index ff7c76b034..da2027c595 100644 --- a/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts +++ b/packages/daemon/src/storage/repositories/space-long-horizon-agent-repository.ts @@ -364,16 +364,18 @@ export class SpaceLongHorizonAgentRepository { } /** - * Active reminders (for active agents) with a NULL `next_run_at` — rows - * created before the scanner existed (the create paths now seed it). Used by - * the startup backfill so pre-existing reminders become schedulable. + * Active reminders with a NULL `next_run_at` — rows created before the + * scanner existed (the create paths now seed it). Used by the startup + * backfill so pre-existing reminders become schedulable. Intentionally does + * NOT filter on agent status: a paused/disabled agent's reminder still needs + * its `next_run_at` seeded so it fires when the agent returns to active; the + * due-query gates firing on agent/space state at fire time. */ listActiveRemindersWithNullNextRunAt(): 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 NULL AND a.status = 'active'` + `SELECT * FROM space_long_horizon_agent_reminders + WHERE status = 'active' AND next_run_at IS NULL` ) .all() as Record[]; return rows.map(rowToReminder); diff --git a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts index 5e12d92ecc..aed3fdb9d2 100644 --- a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts +++ b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts @@ -101,8 +101,11 @@ describe('handleLongHorizonAgentReminderFire', () => { db.close(); }); - function makeDeps(deliver: (args: DeliverArgs) => Promise<{ delivered: boolean }>) { - return { reminderRepo, spaceRepo, jobQueue, deliver }; + function makeDeps( + deliver: (args: DeliverArgs) => Promise<{ delivered: boolean }>, + isOccurrencePersisted?: (spaceId: string, agentId: string, idempotencyKey: string) => boolean + ) { + return { reminderRepo, spaceRepo, jobQueue, deliver, isOccurrencePersisted }; } it('fires a due one-shot reminder, delivers the body, and marks it fired', async () => { @@ -266,6 +269,33 @@ describe('handleLongHorizonAgentReminderFire', () => { expect(r2.fired + r2.skipped).toBeGreaterThanOrEqual(1); }); + it('advances without re-injecting when the occurrence is already persisted', async () => { + const now = Date.now(); + const reminder = reminderRepo.createReminder({ + spaceId, + agentId, + title: 'already-persisted', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + + const deliver = recordingDeliver(); + // Simulate a prior attempt that persisted the message then threw on enqueue. + const result = await handleLongHorizonAgentReminderFire( + makeJob(), + makeDeps(deliver.fn, () => true) + ); + + // Did not re-inject, but did advance (the persisted row is replayed by the + // session) — so no duplicate message and no infinite retry. + expect(deliver.calls).toHaveLength(0); + expect(result.fired).toBe(1); + const after = reminderRepo.getReminder(reminder.id)!; + expect(after.status).toBe('fired'); + expect(after.nextRunAt).toBeNull(); + }); + it('skips a reminder whose owning space is paused', async () => { const now = Date.now(); reminderRepo.createReminder({ @@ -440,4 +470,26 @@ describe('backfillLongHorizonAgentReminderNextRunAt', () => { }); expect(backfillLongHorizonAgentReminderNextRunAt(reminderRepo)).toBe(0); }); + + it('backfills paused-agent reminders too (gated at fire time, not backfill)', () => { + const pausedAgent = reminderRepo.create({ + spaceId, + handle: 'paused', + displayName: 'Paused', + status: 'paused', + }); + const reminder = reminderRepo.createReminder({ + spaceId, + agentId: pausedAgent.id, + title: 'paused-owner', + triggerType: 'at', + runAt: Date.now() - 1000, + }); + expect(reminder.nextRunAt).toBeNull(); + + const count = backfillLongHorizonAgentReminderNextRunAt(reminderRepo); + + expect(count).toBe(1); + expect(reminderRepo.getReminder(reminder.id)!.nextRunAt).not.toBeNull(); + }); }); From 023a13255cf91f87bc1a0aa8a1a8b204473a8c0d Mon Sep 17 00:00:00 2001 From: Marc Liu Date: Sun, 26 Jul 2026 17:17:43 -0400 Subject: [PATCH 6/9] fix: tri-state occurrence guard; lifecycle recheck before inject MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the round-4 review (#4782600533) on #2262. P2 #1 — persisted-message guard violated the deliver-before-advance invariant. isOccurrencePersisted treated a persisted row (enqueued OR consumed) as delivered and advanced, but "persisted as enqueued" is not "delivered": if enqueueWithId timed out on a stuck SDK, the row sits enqueued (drained only on restart/turn-end the SDK never reaches) yet a one-shot was advanced to status='fired'. Replaced with a tri-state getOccurrenceDeliveryState: 'consumed' -> advance (delivered; invariant holds); 'enqueued' -> skip WITHOUT advancing and WITHOUT re-injecting (leave due; the SDK drains the row and the scanner re-checks next tick until consumed); 'absent' -> deliver as usual. Preserves both the invariant and the no-amplification fix. P2 #2 — lifecycle TOCTOU before inject. deliverLongHorizonExternalEvent checked the agent active but not space paused/stopped after the ensureLongHorizonAgentSession await and before inject; a space paused/stopped during session prep still got injected. Added a recheck of space (active/not-paused/not-stopped) and agent (active) right before injectLongTermAgentMessage, returning {delivered:false} otherwise. Closes the TOCTOU at the inject boundary for both reminders and external events. Tests: 'consumed' advances without re-inject; 'enqueued' defers (skips, no advance, no re-inject); handler treats the inject-path delivered:false as a skip. --- packages/daemon/src/app.ts | 24 ++--- ...ong-horizon-agent-reminder-fire.handler.ts | 89 ++++++++++++------- .../space/runtime/space-runtime-service.ts | 21 ++++- ...orizon-agent-reminder-fire.handler.test.ts | 49 ++++++++-- 4 files changed, 131 insertions(+), 52 deletions(-) diff --git a/packages/daemon/src/app.ts b/packages/daemon/src/app.ts index f72814db03..391ea27583 100644 --- a/packages/daemon/src/app.ts +++ b/packages/daemon/src/app.ts @@ -954,18 +954,22 @@ export async function createDaemonApp(options: CreateDaemonAppOptions): Promise< spaceRepo: lhAgentReminderSpaceRepo, jobQueue, deliver: (args) => spaceRuntimeService.deliverLongHorizonAgentReminder(args), - // Guards against duplicate persisted messages on retry: saveUserMessage - // runs before enqueueWithId, so a timed-out delivery leaves a durable - // row that would otherwise be re-inserted each retry. Checks the - // occurrence's uuid across the enqueued/consumed send statuses. - isOccurrencePersisted: (spaceId, agentId, idempotencyKey) => { + // Tri-state guard against duplicate persisted messages on retry: + // saveUserMessage runs before enqueueWithId, so a timed-out delivery + // leaves a durable row. 'consumed' → advance; 'enqueued' → defer (skip, + // don't re-inject or advance); 'absent' → deliver. Probes the + // occurrence's uuid across the consumed/enqueued send statuses. + getOccurrenceDeliveryState: (spaceId, agentId, idempotencyKey) => { const sessionId = longTermAgentSessionId(spaceId, agentId); const messageDb = reactiveDb?.db; - if (!messageDb) return false; - return ( - messageDb.getMessageByStatusAndUuid(sessionId, 'enqueued', idempotencyKey) != null || - messageDb.getMessageByStatusAndUuid(sessionId, 'consumed', idempotencyKey) != null - ); + if (!messageDb) return 'absent'; + if (messageDb.getMessageByStatusAndUuid(sessionId, 'consumed', idempotencyKey) != null) { + return 'consumed'; + } + if (messageDb.getMessageByStatusAndUuid(sessionId, 'enqueued', idempotencyKey) != null) { + return 'enqueued'; + } + return 'absent'; }, }); }); diff --git a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts index 2460076bf1..2aeac9faeb 100644 --- a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts +++ b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts @@ -33,7 +33,11 @@ * is never marked `fired` without having actually been delivered. If the * process crashes between deliver and advance for one reminder, that single * reminder may be re-delivered on the next scan — acceptable for a nag, and - * far better than silently losing a one-shot. + * far better than silently losing a one-shot. The persisted-occurrence guard + * (getOccurrenceDeliveryState) respects this: a prior attempt whose row is + * 'consumed' advances (it WAS delivered); a row still 'enqueued' (persisted + * but not consumed by a stuck SDK) is deferred — skipped without advancing or + * re-injecting — so the invariant holds and no duplicate row is inserted. * * Starvation safety: the scan pages through due reminders, excluding IDs it has * already attempted this tick, so a batch of poison (always-failing) reminders @@ -81,6 +85,9 @@ export interface LongHorizonAgentReminderFireResult extends Record Promise<{ delivered: boolean }>; /** - * Optional pre-delivery guard against duplicate persisted messages. A prior - * attempt may have already persisted the occurrence's sdk_messages row — - * `saveUserMessage` runs before `enqueueWithId`, which can time out and throw, - * leaving the row durable while the handler reports failure and retries. - * Re-injecting would insert a duplicate row (new PK, same uuid; no unique - * constraint) and flood the agent on replay. When this returns true the - * scanner skips delivery and advances — the persisted row is replayed by the - * session. Returns false (treat as not-persisted) when unset. + * Optional pre-delivery guard against duplicate persisted messages AND the + * advance-before-delivery trap. A prior attempt may have already persisted + * the occurrence's sdk_messages row — `saveUserMessage` runs before + * `enqueueWithId`, which can time out and throw, leaving the row durable + * while the handler reports failure and retries. Re-injecting would insert a + * duplicate row (new PK, same uuid; no unique constraint) and flood the agent + * on replay. Tri-state, mirroring the row's send_status: + * - 'consumed': the SDK consumed it (delivered) → advance; the deliver- + * before-advance invariant holds. + * - 'enqueued': persisted but not yet consumed (e.g. a stuck SDK that timed + * out on enqueue) → skip WITHOUT advancing and WITHOUT re-injecting; the + * SDK drains the row via its normal turn-end/startup paths and the + * scanner re-selects next tick until consumed. Avoids both amplification + * and marking a one-shot 'fired' before delivery. + * - 'absent': no prior row → deliver as usual. Also the effective value + * when this dep is unset. */ - isOccurrencePersisted?: (spaceId: string, agentId: string, idempotencyKey: string) => boolean; + getOccurrenceDeliveryState?: ( + spaceId: string, + agentId: string, + idempotencyKey: string + ) => ReminderOccurrenceDeliveryState; } export async function handleLongHorizonAgentReminderFire( _job: Job, deps: LongHorizonAgentReminderFireDeps ): Promise { - const { reminderRepo, spaceRepo, jobQueue, deliver, isOccurrencePersisted } = deps; + const { reminderRepo, spaceRepo, jobQueue, deliver, getOccurrenceDeliveryState } = 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. @@ -154,7 +173,7 @@ export async function handleLongHorizonAgentReminderFire( reminderRepo, spaceRepo, deliver, - isOccurrencePersisted, + getOccurrenceDeliveryState, now ); } catch (err) { @@ -201,7 +220,7 @@ async function fireReminderSerialized( reminderRepo: SpaceLongHorizonAgentRepository, spaceRepo: SpaceRepository, deliver: LongHorizonAgentReminderFireDeps['deliver'], - isOccurrencePersisted: LongHorizonAgentReminderFireDeps['isOccurrencePersisted'], + getOccurrenceDeliveryState: LongHorizonAgentReminderFireDeps['getOccurrenceDeliveryState'], now: number ): Promise<'fired' | 'skipped' | 'failed'> { const lockKey = reminder.id; @@ -215,7 +234,7 @@ async function fireReminderSerialized( reminderRepo, spaceRepo, deliver, - isOccurrencePersisted, + getOccurrenceDeliveryState, now ); }); @@ -233,7 +252,7 @@ async function fireReminder( reminderRepo: SpaceLongHorizonAgentRepository, spaceRepo: SpaceRepository, deliver: LongHorizonAgentReminderFireDeps['deliver'], - isOccurrencePersisted: LongHorizonAgentReminderFireDeps['isOccurrencePersisted'], + getOccurrenceDeliveryState: LongHorizonAgentReminderFireDeps['getOccurrenceDeliveryState'], now: number ): Promise<'fired' | 'skipped' | 'failed'> { // Re-read: the reminder may have been paused/cancelled/fired/advanced between @@ -258,16 +277,27 @@ async function fireReminder( // keys on). Passed as the SDK message uuid. const idempotencyKey = `reminder:${fresh.id}:${fresh.nextRunAt}`; - // Pre-delivery idempotency: a prior attempt at this occurrence may already - // have persisted its sdk_messages row — `saveUserMessage` runs before - // `enqueueWithId`, which can time out and throw, leaving the row durable - // while we report failure and retry. Re-injecting would insert a duplicate - // row (new PK, same uuid) and flood the agent on replay. If it's already - // persisted, skip the inject and treat the occurrence as delivered — the row - // is replayed/delivered by the session — then advance. + // Pre-delivery idempotency against a prior attempt that already persisted + // this occurrence's sdk_messages row (saveUserMessage runs before + // enqueueWithId, which can time out and throw). Tri-state: + // 'enqueued' — persisted but not yet consumed (e.g. a stuck SDK). Skip + // WITHOUT advancing and WITHOUT re-injecting: re-injecting would + // duplicate the row, and advancing would mark a one-shot 'fired' before + // delivery (violating the deliver-before-advance invariant). The SDK + // drains the row via its normal paths; the scanner re-selects next tick + // and re-checks until consumed. + // 'consumed' — already delivered; advance without re-injecting. + // 'absent' — no prior row; deliver as usual. + const occurrenceState = + getOccurrenceDeliveryState?.(fresh.spaceId, fresh.agentId, idempotencyKey) ?? 'absent'; + if (occurrenceState === 'enqueued') { + log.debug('lh-agent-reminder-fire: occurrence enqueued, deferring', { reminderId: fresh.id }); + return 'skipped'; + } + let delivered: boolean; - if (isOccurrencePersisted?.(fresh.spaceId, fresh.agentId, idempotencyKey)) { - log.debug('lh-agent-reminder-fire: occurrence already persisted, advancing', { + if (occurrenceState === 'consumed') { + log.debug('lh-agent-reminder-fire: occurrence already consumed, advancing', { reminderId: fresh.id, }); delivered = true; @@ -289,12 +319,11 @@ async function fireReminder( } // Delivery returned false: the owning AGENT is missing/paused/disabled/ - // archived, or its session could not be ensured. (The reminder's own - // paused/cancelled status is handled by the active-status guard above — - // this branch is specifically an agent-state skip.) Do NOT advance: the - // 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. + // archived, or its session could not be ensured, or the space was + // paused/stopped during delivery (the inject-path lifecycle recheck). + // (The reminder's own paused/cancelled status is handled by the active- + // status guard above.) Do NOT advance: the reminder stays due and fires + // when deliverable again. if (!delivered) return 'skipped'; } diff --git a/packages/daemon/src/lib/space/runtime/space-runtime-service.ts b/packages/daemon/src/lib/space/runtime/space-runtime-service.ts index 35b20f95e4..01044b315b 100644 --- a/packages/daemon/src/lib/space/runtime/space-runtime-service.ts +++ b/packages/daemon/src/lib/space/runtime/space-runtime-service.ts @@ -374,11 +374,24 @@ export class SpaceRuntimeService { return { delivered: false }; } const session = await this.ensureLongHorizonAgentSession(args.spaceId, args.agentId); - if (session) { - await this.injectLongTermAgentMessage(session, args.message, args.idempotencyKey); - return { delivered: true }; + if (!session) return { delivered: false }; + // Re-check lifecycle AFTER the ensureSession await and BEFORE injecting: + // the space/agent may have been paused/stopped/archived during session + // prep, and ensureLongHorizonAgentSession only checks that the space + // exists — not that it is active/not-paused/not-stopped. Don't inject (or + // let a reminder recreate a session) into a non-deliverable space. Mirrors + // task-schedule-fire's space contract and closes the TOCTOU the scanner's + // pre-await precheck leaves open. + const space = await this.config.spaceManager.getSpace(args.spaceId); + if (!space || space.status !== 'active' || space.paused || space.stopped) { + return { delivered: false }; + } + const freshAgent = this.config.longHorizonAgentRepo?.getById(args.agentId); + if (!freshAgent || freshAgent.status !== 'active') { + return { delivered: false }; } - return { delivered: false }; + await this.injectLongTermAgentMessage(session, args.message, args.idempotencyKey); + return { delivered: true }; } /** diff --git a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts index aed3fdb9d2..370d300378 100644 --- a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts +++ b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts @@ -8,6 +8,7 @@ import { backfillLongHorizonAgentReminderNextRunAt, enqueueLongHorizonAgentReminderScanIfMissing, handleLongHorizonAgentReminderFire, + type ReminderOccurrenceDeliveryState, } from '../../../../src/lib/job-handlers/long-horizon-agent-reminder-fire.handler'; import { LONG_HORIZON_AGENT_REMINDER_FIRE } from '../../../../src/lib/job-queue-constants'; import type { Job } from '../../../../src/storage/repositories/job-queue-repository'; @@ -103,9 +104,13 @@ describe('handleLongHorizonAgentReminderFire', () => { function makeDeps( deliver: (args: DeliverArgs) => Promise<{ delivered: boolean }>, - isOccurrencePersisted?: (spaceId: string, agentId: string, idempotencyKey: string) => boolean + getOccurrenceDeliveryState?: ( + spaceId: string, + agentId: string, + idempotencyKey: string + ) => ReminderOccurrenceDeliveryState ) { - return { reminderRepo, spaceRepo, jobQueue, deliver, isOccurrencePersisted }; + return { reminderRepo, spaceRepo, jobQueue, deliver, getOccurrenceDeliveryState }; } it('fires a due one-shot reminder, delivers the body, and marks it fired', async () => { @@ -269,26 +274,25 @@ describe('handleLongHorizonAgentReminderFire', () => { expect(r2.fired + r2.skipped).toBeGreaterThanOrEqual(1); }); - it('advances without re-injecting when the occurrence is already persisted', async () => { + it('advances without re-injecting when the occurrence is already consumed', async () => { const now = Date.now(); const reminder = reminderRepo.createReminder({ spaceId, agentId, - title: 'already-persisted', + title: 'already-consumed', triggerType: 'at', runAt: now - 1000, nextRunAt: now - 1000, }); const deliver = recordingDeliver(); - // Simulate a prior attempt that persisted the message then threw on enqueue. + // Simulate a prior attempt whose message the SDK already consumed. const result = await handleLongHorizonAgentReminderFire( makeJob(), - makeDeps(deliver.fn, () => true) + makeDeps(deliver.fn, () => 'consumed') ); - // Did not re-inject, but did advance (the persisted row is replayed by the - // session) — so no duplicate message and no infinite retry. + // Did not re-inject, but did advance (the occurrence was delivered). expect(deliver.calls).toHaveLength(0); expect(result.fired).toBe(1); const after = reminderRepo.getReminder(reminder.id)!; @@ -296,6 +300,35 @@ describe('handleLongHorizonAgentReminderFire', () => { expect(after.nextRunAt).toBeNull(); }); + it('defers (skips without advancing) when the occurrence is enqueued but not consumed', async () => { + const now = Date.now(); + const reminder = reminderRepo.createReminder({ + spaceId, + agentId, + title: 'stuck-enqueued', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + + const deliver = recordingDeliver(); + // Prior attempt persisted the row but the SDK hasn't consumed it (e.g. a + // stuck session that timed out on enqueue). + const result = await handleLongHorizonAgentReminderFire( + makeJob(), + makeDeps(deliver.fn, () => 'enqueued') + ); + + // No re-inject (no duplicate) AND no advance (one-shot not marked fired + // before delivery). Stays due for the next scan to re-check. + expect(deliver.calls).toHaveLength(0); + expect(result.fired).toBe(0); + expect(result.skipped).toBe(1); + const after = reminderRepo.getReminder(reminder.id)!; + expect(after.status).toBe('active'); + expect(after.nextRunAt).toBe(now - 1000); + }); + it('skips a reminder whose owning space is paused', async () => { const now = Date.now(); reminderRepo.createReminder({ From e7e610394577302c1b75586afd3c5549755c2af9 Mon Sep 17 00:00:00 2001 From: Marc Liu Date: Sun, 26 Jul 2026 17:44:52 -0400 Subject: [PATCH 7/9] fix: bound delivery timeout; pre-await lifecycle gate; reconcile invariant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the round-5 review (#4782639447) on #2262. P1 (blocking) — stuck SDK delivery deadlocked the job processor. enqueueWithId's 30s timer only rejects messages still in the in-memory queue; once the SDK shifts the message the timer no-ops and the promise waits on onSent (run after yield message), so a stuck SDK that never resumes its for-await leaves enqueueWithId — and thus deliver — unsettled forever. The handler awaited this inside the per-reminder lock, with the scan deadline checked only before the lock, so one hung delivery pinned a slot; successive pre-enqueued scans (30s apart) each took a slot on the same reminder and, with maxConcurrent=5, saturated all slots and stalled unrelated queues. Wrap await deliver(...) in a Promise.race against a ~35s deadline (just past the 30s in-queue timeout); on timeout return 'failed' so the lock + slot release and the reminder retries next scan, each attempt bounded. Timeout is injectable (deliveryTimeoutMs) for tests. P2 — ensureLongHorizonAgentSession side-effects fired before the lifecycle gate. Session creation, config refresh + resetQuery, agent-row update, metadata write, and MCP attach all run inside ensureLongHorizonAgentSession, which the post-await recheck didn't cover. Added the same space/agent lifecycle gate BEFORE await ensureLongHorizonAgentSession; kept the post-await recheck for the during-await race. P2 — 'consumed' is not a reliable delivery signal. acknowledgeOldestQueued- UserOnTurnEnd (turn-end cleanup) marks un-yielded enqueued user rows consumed, so advancing on 'consumed' could fire a one-shot the model never yielded, re-breaking the round-4 invariant. Rather than touch the shared SDK message-handling path, reconciled code with doc: for a nag, 'consumed' is treated as "handed to the SDK pipeline" (the message is in the session history for the next turn), and the header + guard docs now state that honestly, including the cleanup-consumed residual. Defer-on-enqueued is preserved. Also refactored fireReminderSerialized/fireReminder to take the deps bag (simpler than widening param lists). Test: a never-settling deliver is bounded by the timeout and returns 'failed'. --- ...ong-horizon-agent-reminder-fire.handler.ts | 126 ++++++++++++------ .../space/runtime/space-runtime-service.ts | 21 ++- ...orizon-agent-reminder-fire.handler.test.ts | 42 +++++- 3 files changed, 138 insertions(+), 51 deletions(-) diff --git a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts index 2aeac9faeb..f4b746451c 100644 --- a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts +++ b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts @@ -29,15 +29,21 @@ * The compare-and-swap advance is a belt-and-suspenders fence for any path that * bypasses the lock. * - * 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 be re-delivered on the next scan — acceptable for a nag, and - * far better than silently losing a one-shot. The persisted-occurrence guard - * (getOccurrenceDeliveryState) respects this: a prior attempt whose row is - * 'consumed' advances (it WAS delivered); a row still 'enqueued' (persisted - * but not consumed by a stuck SDK) is deferred — skipped without advancing or - * re-injecting — so the invariant holds and no duplicate row is inserted. + * Delivery / advance contract: a reminder advances once its message has been + * injected (deliver returned delivered:true) OR is already in the SDK pipeline + * (the occurrence's sdk_messages row is send_status 'consumed'). It is + * deferred — left due, not advanced, not re-injected — while a prior attempt's + * row is still 'enqueued' (persisted but not consumed, e.g. a stuck SDK), so no + * duplicate row is inserted and the row is drained by the SDK's turn_end + * auto-send path. NB: for a nag, 'consumed' is treated as "handed to the SDK + * pipeline" rather than a strict "the model yielded it this turn" — turn-end + * cleanup (acknowledgeOldestQueuedUserOnTurnEnd) can mark an un-yielded user + * row consumed, so a one-shot may advance with the message in the session + * history (visible to the next turn) but not yet yielded to the model. This is + * the deliberate, documented bar for a periodic nag; if a stricter guarantee + * is ever needed, split the cleanup-consumed transition or key off a yield + * receipt. If the process crashes between deliver and advance, that single + * reminder may be re-delivered on the next scan — acceptable for a nag. * * Starvation safety: the scan pages through due reminders, excluding IDs it has * already attempted this tick, so a batch of poison (always-failing) reminders @@ -68,6 +74,19 @@ const MAX_PER_SCAN = 5000; * NEXT_SCAN_DELAY_MS so a scan normally finishes before its successor is due. */ const SCAN_DEADLINE_MS = 20_000; +/** + * Per-delivery wall-clock bound. enqueueWithId's 30s timer only rejects + * messages still in the in-memory queue — once the SDK shifts the message the + * timer no-ops and the promise waits on `onSent` (run after `yield message`), + * so a stuck SDK that never resumes its `for await` leaves enqueueWithId — and + * thus `deliver` — unsettled forever. Racing against a deadline just past the + * 30s in-queue timeout guarantees the per-reminder lock and the job slot are + * released, so a single hung delivery can't pin the scanner (and, across + * successive pre-enqueued scans, saturate maxConcurrent and stall unrelated + * queues). On timeout the occurrence is treated as failed and retried next + * scan; the orphaned deliver promise settles harmlessly if/when the SDK wakes. + */ +const DELIVERY_TIMEOUT_MS = 35_000; /** * In-process per-reminder locks. Serializes overlapping scans (concurrent @@ -114,8 +133,11 @@ export interface LongHorizonAgentReminderFireDeps { * while the handler reports failure and retries. Re-injecting would insert a * duplicate row (new PK, same uuid; no unique constraint) and flood the agent * on replay. Tri-state, mirroring the row's send_status: - * - 'consumed': the SDK consumed it (delivered) → advance; the deliver- - * before-advance invariant holds. + * - 'consumed': the occurrence's message is in the SDK pipeline (the row + * reached send_status 'consumed' via a genuine SDK yield OR turn-end + * cleanup of un-yielded rows) → advance without re-injecting. For a nag + * this is the "handed to the SDK" bar, not a strict yield guarantee (see + * the Delivery / advance contract in the header). * - 'enqueued': persisted but not yet consumed (e.g. a stuck SDK that timed * out on enqueue) → skip WITHOUT advancing and WITHOUT re-injecting; the * SDK drains the row via its normal turn-end/startup paths and the @@ -129,13 +151,19 @@ export interface LongHorizonAgentReminderFireDeps { agentId: string, idempotencyKey: string ) => ReminderOccurrenceDeliveryState; + /** + * Per-delivery timeout override (mainly for tests); defaults to + * DELIVERY_TIMEOUT_MS. Bounds a stuck SDK delivery so it can't pin the lock + * or the job slot. + */ + deliveryTimeoutMs?: number; } export async function handleLongHorizonAgentReminderFire( _job: Job, deps: LongHorizonAgentReminderFireDeps ): Promise { - const { reminderRepo, spaceRepo, jobQueue, deliver, getOccurrenceDeliveryState } = deps; + const { reminderRepo, jobQueue } = 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. @@ -168,14 +196,7 @@ export async function handleLongHorizonAgentReminderFire( scanned++; let outcome: 'fired' | 'skipped' | 'failed'; try { - outcome = await fireReminderSerialized( - reminder, - reminderRepo, - spaceRepo, - deliver, - getOccurrenceDeliveryState, - now - ); + outcome = await fireReminderSerialized(reminder, deps, now); } catch (err) { outcome = 'failed'; log.warn('lh-agent-reminder-fire: error firing reminder', { @@ -217,10 +238,7 @@ export async function handleLongHorizonAgentReminderFire( */ async function fireReminderSerialized( reminder: SpaceLongHorizonAgentReminder, - reminderRepo: SpaceLongHorizonAgentRepository, - spaceRepo: SpaceRepository, - deliver: LongHorizonAgentReminderFireDeps['deliver'], - getOccurrenceDeliveryState: LongHorizonAgentReminderFireDeps['getOccurrenceDeliveryState'], + deps: LongHorizonAgentReminderFireDeps, now: number ): Promise<'fired' | 'skipped' | 'failed'> { const lockKey = reminder.id; @@ -229,14 +247,7 @@ async function fireReminderSerialized( const current = previous .catch(() => {}) .then(async () => { - outcome = await fireReminder( - reminder, - reminderRepo, - spaceRepo, - deliver, - getOccurrenceDeliveryState, - now - ); + outcome = await fireReminder(reminder, deps, now); }); reminderFireLocks.set(lockKey, current); try { @@ -249,12 +260,10 @@ async function fireReminderSerialized( async function fireReminder( reminder: SpaceLongHorizonAgentReminder, - reminderRepo: SpaceLongHorizonAgentRepository, - spaceRepo: SpaceRepository, - deliver: LongHorizonAgentReminderFireDeps['deliver'], - getOccurrenceDeliveryState: LongHorizonAgentReminderFireDeps['getOccurrenceDeliveryState'], + deps: LongHorizonAgentReminderFireDeps, now: number ): Promise<'fired' | 'skipped' | 'failed'> { + const { reminderRepo, spaceRepo, deliver, getOccurrenceDeliveryState } = deps; // Re-read: the reminder may have been paused/cancelled/fired/advanced between // the due-scan and now (e.g. by a serialized peer scan). A null or future // next_run_at means it is no longer schedulable for this tick. @@ -286,7 +295,7 @@ async function fireReminder( // delivery (violating the deliver-before-advance invariant). The SDK // drains the row via its normal paths; the scanner re-selects next tick // and re-checks until consumed. - // 'consumed' — already delivered; advance without re-injecting. + // 'consumed' — message is in the SDK pipeline; advance without re-injecting. // 'absent' — no prior row; deliver as usual. const occurrenceState = getOccurrenceDeliveryState?.(fresh.spaceId, fresh.agentId, idempotencyKey) ?? 'absent'; @@ -303,15 +312,23 @@ async function fireReminder( delivered = true; } else { try { - const result = await deliver({ - spaceId: fresh.spaceId, - agentId: fresh.agentId, - message, - idempotencyKey, - }); + // Bound the delivery: a stuck SDK can leave enqueueWithId (and thus + // deliver) unsettled forever. On timeout, treat as failed so the lock + // and job slot release; the reminder retries next scan. See + // DELIVERY_TIMEOUT_MS. + const result = await withTimeout( + deliver({ + spaceId: fresh.spaceId, + agentId: fresh.agentId, + message, + idempotencyKey, + }), + deps.deliveryTimeoutMs ?? DELIVERY_TIMEOUT_MS, + 'reminder delivery' + ); delivered = result.delivered; } catch (err) { - log.warn('lh-agent-reminder-fire: delivery threw', { + log.warn('lh-agent-reminder-fire: delivery threw or timed out', { reminderId: fresh.id, error: err instanceof Error ? err.message : err, }); @@ -406,3 +423,24 @@ function formatReminderMessage(reminder: SpaceLongHorizonAgentReminder): string } return parts.join('\n'); } + +/** + * Race a promise against a deadline. On timeout, rejects with a descriptive + * error; the underlying promise is left to settle on its own (its result is + * ignored). Used to bound delivery so a stuck SDK can't pin the scanner. + */ +function withTimeout(promise: Promise, ms: number, scope: string): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error(`${scope} timed out after ${ms}ms`)), ms); + promise.then( + (value) => { + clearTimeout(timer); + resolve(value); + }, + (err) => { + clearTimeout(timer); + reject(err); + } + ); + }); +} diff --git a/packages/daemon/src/lib/space/runtime/space-runtime-service.ts b/packages/daemon/src/lib/space/runtime/space-runtime-service.ts index 01044b315b..cb153c5a51 100644 --- a/packages/daemon/src/lib/space/runtime/space-runtime-service.ts +++ b/packages/daemon/src/lib/space/runtime/space-runtime-service.ts @@ -373,15 +373,26 @@ export class SpaceRuntimeService { if (!agent || agent.spaceId !== args.spaceId || agent.status !== 'active') { return { delivered: false }; } + // Pre-await lifecycle gate: ensureLongHorizonAgentSession performs + // side effects (session creation, config refresh + resetQuery, agent-row + // update, metadata write, MCP attach) that must NOT run for a space that + // is no longer active / is paused / is stopped. Bail before any of that. + const spaceBefore = await this.config.spaceManager.getSpace(args.spaceId); + if ( + !spaceBefore || + spaceBefore.status !== 'active' || + spaceBefore.paused || + spaceBefore.stopped + ) { + return { delivered: false }; + } const session = await this.ensureLongHorizonAgentSession(args.spaceId, args.agentId); if (!session) return { delivered: false }; // Re-check lifecycle AFTER the ensureSession await and BEFORE injecting: // the space/agent may have been paused/stopped/archived during session - // prep, and ensureLongHorizonAgentSession only checks that the space - // exists — not that it is active/not-paused/not-stopped. Don't inject (or - // let a reminder recreate a session) into a non-deliverable space. Mirrors - // task-schedule-fire's space contract and closes the TOCTOU the scanner's - // pre-await precheck leaves open. + // prep (which itself only checks the space exists). Don't inject into a + // non-deliverable space. Mirrors task-schedule-fire's space contract and + // closes the TOCTOU the scanner's pre-await precheck leaves open. const space = await this.config.spaceManager.getSpace(args.spaceId); if (!space || space.status !== 'active' || space.paused || space.stopped) { return { delivered: false }; diff --git a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts index 370d300378..04db9dfc85 100644 --- a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts +++ b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts @@ -108,9 +108,17 @@ describe('handleLongHorizonAgentReminderFire', () => { spaceId: string, agentId: string, idempotencyKey: string - ) => ReminderOccurrenceDeliveryState + ) => ReminderOccurrenceDeliveryState, + deliveryTimeoutMs?: number ) { - return { reminderRepo, spaceRepo, jobQueue, deliver, getOccurrenceDeliveryState }; + return { + reminderRepo, + spaceRepo, + jobQueue, + deliver, + getOccurrenceDeliveryState, + deliveryTimeoutMs, + }; } it('fires a due one-shot reminder, delivers the body, and marks it fired', async () => { @@ -300,6 +308,36 @@ describe('handleLongHorizonAgentReminderFire', () => { expect(after.nextRunAt).toBeNull(); }); + it('bounds a stuck delivery with a per-call timeout (no slot deadlock)', async () => { + const now = Date.now(); + const reminder = reminderRepo.createReminder({ + spaceId, + agentId, + title: 'stuck-sdk', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + // A deliver that never settles — simulates a stuck SDK whose enqueueWithId + // never resolves (onSent never fires after a wedged `for await`). + const neverDeliver = (): Promise<{ delivered: boolean }> => new Promise(() => {}); + const start = Date.now(); + const result = await handleLongHorizonAgentReminderFire( + makeJob(), + makeDeps(neverDeliver, undefined, 50) + ); + + // Bounded by the 50ms timeout, not the 35s default — the lock and job slot + // release, so the scanner can't be pinned. + expect(Date.now() - start).toBeLessThan(2000); + expect(result.failed).toBe(1); + expect(result.fired).toBe(0); + // Not advanced; stays due for retry next scan. + const after = reminderRepo.getReminder(reminder.id)!; + expect(after.status).toBe('active'); + expect(after.nextRunAt).toBe(now - 1000); + }); + it('defers (skips without advancing) when the occurrence is enqueued but not consumed', async () => { const now = Date.now(); const reminder = reminderRepo.createReminder({ From 8dd875a3f2097b791290f56b22407c11fdcf1e98 Mon Sep 17 00:00:00 2001 From: Marc Liu Date: Sun, 26 Jul 2026 18:37:55 -0400 Subject: [PATCH 8/9] fix: in-flight delivery tracker; reminder-only lifecycle gate; backfill skip MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the round-6 review (#4782738280) P1s and P2-backfill. P1 — amplification when delivery times out before saveUserMessage. deliver can hang in ensureLongHorizonAgentSession (resetQuery) or injectLongTermAgentMessage's ensureQueryStarted — both BEFORE saveUserMessage. The round-5 withTimeout released the lock on timeout, but with no row persisted the next scan's occurrence probe reads 'absent' and starts a parallel deliver; on recovery each in-flight attempt persists the same UUID (no uniqueness) -> duplicate messages. Added an in-process reminderDeliveriesInFlight map: while a reminder's delivery promise is unsettled, the scanner skips it (no second deliver) and clears the entry only when the delivery actually settles. This also breaks the cross-scan starvation (a repeatedly-timing-out earliest reminder is skipped on subsequent scans instead of being re-selected first every scan and eating the whole budget). P1 (5Zpf) — external-event regression from the shared lifecycle gate. deliverLongHorizonExternalEvent's gate made external events to a paused space retry-and-lose (terminal after bounded retries). Scoped the gate to reminders via a gateSpaceLifecycle option (default false); external events are unchanged (the runtime's requeuePersistedPendingDeliveries reconstructs LH deliveries on resume, so no loss). P2 — backfill fired previously-inert malformed rows. cron-without- expression and at-without-runAt fell to `runAt ?? now`, firing them immediately on upgrade. Now skips unschedulable rows (leaves NULL) — operator can repair or cancel. Tests: in-flight delivery is not stacked (no amplification); backfill skips cron-without-expression and at-without-runAt. --- ...ong-horizon-agent-reminder-fire.handler.ts | 60 ++++++++++--- .../space/runtime/space-runtime-service.ts | 86 +++++++++++-------- ...orizon-agent-reminder-fire.handler.test.ts | 55 ++++++++++++ 3 files changed, 154 insertions(+), 47 deletions(-) diff --git a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts index f4b746451c..5f55e312ab 100644 --- a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts +++ b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts @@ -96,6 +96,18 @@ const DELIVERY_TIMEOUT_MS = 35_000; */ const reminderFireLocks = new Map>(); +/** + * Outstanding delivery promises for reminders whose `deliver` call has not yet + * settled. A delivery that times out before persisting (e.g. ensureQueryStarted + * stalls before saveUserMessage) leaves no sdk_messages row, so the occurrence + * probe reads 'absent' and a later scan would start ANOTHER delivery — stacking + * overlapping deliveries (and, for the earliest stuck reminder, starving later + * ones every scan). While a delivery for a reminder is pending here, the + * scanner skips it (counts as 'skipped', not re-delivered). Entries clear when + * the underlying delivery promise settles. + */ +const reminderDeliveriesInFlight = new Map>(); + export interface LongHorizonAgentReminderFireResult extends Record { scanned: number; fired: number; @@ -311,18 +323,38 @@ async function fireReminder( }); delivered = true; } else { + // A prior delivery that timed out before persisting (no row → 'absent') + // may still be pending in the background. Don't stack another — skip and + // let the scanner retry once it settles (or, if the SDK is wedged, the + // in-flight entry clears on process restart). This keeps a stuck earliest + // reminder from being retried every scan and starving later ones. + if (reminderDeliveriesInFlight.has(fresh.id)) { + log.debug('lh-agent-reminder-fire: delivery already in flight, deferring', { + reminderId: fresh.id, + }); + return 'skipped'; + } + const delivery = deliver({ + spaceId: fresh.spaceId, + agentId: fresh.agentId, + message, + idempotencyKey, + }); + reminderDeliveriesInFlight.set(fresh.id, delivery); + delivery.finally(() => { + if (reminderDeliveriesInFlight.get(fresh.id) === delivery) { + reminderDeliveriesInFlight.delete(fresh.id); + } + }); try { // Bound the delivery: a stuck SDK can leave enqueueWithId (and thus // deliver) unsettled forever. On timeout, treat as failed so the lock - // and job slot release; the reminder retries next scan. See + // and job slot release; the reminder retries next scan. The underlying + // delivery promise stays in reminderDeliveriesInFlight until it actually + // settles, preventing a re-delivery in the meantime. See // DELIVERY_TIMEOUT_MS. const result = await withTimeout( - deliver({ - spaceId: fresh.spaceId, - agentId: fresh.agentId, - message, - idempotencyKey, - }), + delivery, deps.deliveryTimeoutMs ?? DELIVERY_TIMEOUT_MS, 'reminder delivery' ); @@ -403,11 +435,17 @@ export function backfillLongHorizonAgentReminderNextRunAt( const now = Date.now(); let count = 0; for (const reminder of stale) { + // Only seed schedulable reminders. A malformed legacy row (cron with no + // expression, or 'at' with no run_at) is left NULL rather than defaulted + // to `now` — defaulting would immediately fire + mark fired a reminder + // that was previously inert. An operator can repair or cancel it. const nextRunAt = - reminder.triggerType === 'cron' && reminder.cronExpression - ? getNextRunAt(reminder.cronExpression, reminder.timezone || 'UTC', now) - : (reminder.runAt ?? now); - if (nextRunAt === null) continue; // unparseable cron — leave for an operator to fix + reminder.triggerType === 'cron' + ? reminder.cronExpression + ? getNextRunAt(reminder.cronExpression, reminder.timezone || 'UTC', now) + : null + : (reminder.runAt ?? null); + if (nextRunAt === null) continue; reminderRepo.setReminderNextRunAt(reminder.id, nextRunAt); count++; } diff --git a/packages/daemon/src/lib/space/runtime/space-runtime-service.ts b/packages/daemon/src/lib/space/runtime/space-runtime-service.ts index cb153c5a51..5e9e24bb84 100644 --- a/packages/daemon/src/lib/space/runtime/space-runtime-service.ts +++ b/packages/daemon/src/lib/space/runtime/space-runtime-service.ts @@ -363,43 +363,50 @@ export class SpaceRuntimeService { ); } - private async deliverLongHorizonExternalEvent(args: { - spaceId: string; - agentId: string; - message: string; - idempotencyKey: string; - }): Promise<{ delivered: boolean }> { + private async deliverLongHorizonExternalEvent( + args: { + spaceId: string; + agentId: string; + message: string; + idempotencyKey: string; + }, + options: { gateSpaceLifecycle?: boolean } = {} + ): Promise<{ delivered: boolean }> { + const gateLifecycle = options.gateSpaceLifecycle === true; const agent = this.config.longHorizonAgentRepo?.getById(args.agentId); if (!agent || agent.spaceId !== args.spaceId || agent.status !== 'active') { return { delivered: false }; } - // Pre-await lifecycle gate: ensureLongHorizonAgentSession performs - // side effects (session creation, config refresh + resetQuery, agent-row - // update, metadata write, MCP attach) that must NOT run for a space that - // is no longer active / is paused / is stopped. Bail before any of that. - const spaceBefore = await this.config.spaceManager.getSpace(args.spaceId); - if ( - !spaceBefore || - spaceBefore.status !== 'active' || - spaceBefore.paused || - spaceBefore.stopped - ) { - return { delivered: false }; + if (gateLifecycle) { + // Pre-await lifecycle gate (reminders only): ensureLongHorizonAgentSession + // performs side effects (session creation, config refresh + resetQuery, + // agent-row update, metadata write, MCP attach) that must NOT run for a + // space that is no longer active / is paused / is stopped. Bail first. + const spaceBefore = await this.config.spaceManager.getSpace(args.spaceId); + if ( + !spaceBefore || + spaceBefore.status !== 'active' || + spaceBefore.paused || + spaceBefore.stopped + ) { + return { delivered: false }; + } } const session = await this.ensureLongHorizonAgentSession(args.spaceId, args.agentId); if (!session) return { delivered: false }; - // Re-check lifecycle AFTER the ensureSession await and BEFORE injecting: - // the space/agent may have been paused/stopped/archived during session - // prep (which itself only checks the space exists). Don't inject into a - // non-deliverable space. Mirrors task-schedule-fire's space contract and - // closes the TOCTOU the scanner's pre-await precheck leaves open. - const space = await this.config.spaceManager.getSpace(args.spaceId); - if (!space || space.status !== 'active' || space.paused || space.stopped) { - return { delivered: false }; - } - const freshAgent = this.config.longHorizonAgentRepo?.getById(args.agentId); - if (!freshAgent || freshAgent.status !== 'active') { - return { delivered: false }; + if (gateLifecycle) { + // Re-check lifecycle AFTER the ensureSession await and BEFORE injecting + // (reminders only): the space/agent may have been paused/stopped/ + // archived during session prep. Don't inject into a non-deliverable + // space. Mirrors task-schedule-fire's space contract. + const space = await this.config.spaceManager.getSpace(args.spaceId); + if (!space || space.status !== 'active' || space.paused || space.stopped) { + return { delivered: false }; + } + const freshAgent = this.config.longHorizonAgentRepo?.getById(args.agentId); + if (!freshAgent || freshAgent.status !== 'active') { + return { delivered: false }; + } } await this.injectLongTermAgentMessage(session, args.message, args.idempotencyKey); return { delivered: true }; @@ -408,11 +415,18 @@ export class SpaceRuntimeService { /** * 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. + * Public entry point for the reminder-fire job handler. Same ensure-session + + * inject path as external-event delivery, but additionally gates on the space + * lifecycle (active / not paused / not stopped) both before and after the + * ensureSession await — a paused/stopped space never has its session + * recreated or a reminder injected; the scanner treats {delivered:false} as a + * skip and retries next scan. + * + * The lifecycle gate is intentionally reminder-only. External-event delivery + * (deliverLongHorizonExternalEvent without gateSpaceLifecycle) is wrapped in a + * bounded retry loop that terminally fails after repeated delivered:false, so + * gating it would drop events for a paused space. Reminders retry harmlessly + * next scan, so they can afford to fail fast on lifecycle. */ async deliverLongHorizonAgentReminder(args: { spaceId: string; @@ -420,7 +434,7 @@ export class SpaceRuntimeService { message: string; idempotencyKey: string; }): Promise<{ delivered: boolean }> { - return this.deliverLongHorizonExternalEvent(args); + return this.deliverLongHorizonExternalEvent(args, { gateSpaceLifecycle: true }); } private async deliverToLongTermAgent( diff --git a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts index 04db9dfc85..df3af5add9 100644 --- a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts +++ b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts @@ -338,6 +338,36 @@ describe('handleLongHorizonAgentReminderFire', () => { expect(after.nextRunAt).toBe(now - 1000); }); + it('does not stack a second delivery while one is in flight (no amplification)', async () => { + const now = Date.now(); + const reminder = reminderRepo.createReminder({ + spaceId, + agentId, + title: 'inflight', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + // First scan: deliver never settles (stuck before persisting); the handler + // times out (50ms) and returns failed, but the delivery stays registered. + const neverDeliver = (): Promise<{ delivered: boolean }> => new Promise(() => {}); + const r1 = await handleLongHorizonAgentReminderFire( + makeJob(), + makeDeps(neverDeliver, undefined, 50) + ); + expect(r1.failed).toBe(1); + + // Second scan: the prior delivery is still in flight -> skip without + // starting another deliver (no stacking / amplification). + const deliver2 = recordingDeliver(); + const r2 = await handleLongHorizonAgentReminderFire( + makeJob(), + makeDeps(deliver2.fn, undefined, 50) + ); + expect(r2.skipped).toBe(1); + expect(deliver2.calls).toHaveLength(0); + }); + it('defers (skips without advancing) when the occurrence is enqueued but not consumed', async () => { const now = Date.now(); const reminder = reminderRepo.createReminder({ @@ -563,4 +593,29 @@ describe('backfillLongHorizonAgentReminderNextRunAt', () => { expect(count).toBe(1); expect(reminderRepo.getReminder(reminder.id)!.nextRunAt).not.toBeNull(); }); + + it('skips unschedulable reminders instead of firing them immediately', () => { + // Legacy cron reminder with no expression, and a legacy 'at' with no + // run_at — both unschedulable. Must be left NULL, not defaulted to `now`. + const cronNoExpr = reminderRepo.createReminder({ + spaceId, + agentId, + title: 'cron-no-expr', + triggerType: 'cron', + }); + const atNoRunAt = reminderRepo.createReminder({ + spaceId, + agentId, + title: 'at-no-runat', + triggerType: 'at', + }); + expect(cronNoExpr.nextRunAt).toBeNull(); + expect(atNoRunAt.nextRunAt).toBeNull(); + + const count = backfillLongHorizonAgentReminderNextRunAt(reminderRepo); + + expect(count).toBe(0); + expect(reminderRepo.getReminder(cronNoExpr.id)!.nextRunAt).toBeNull(); + expect(reminderRepo.getReminder(atNoRunAt.id)!.nextRunAt).toBeNull(); + }); }); From da3da1069b4209956c9bb7751be1a3adc7452872 Mon Sep 17 00:00:00 2001 From: Marc Liu Date: Mon, 27 Jul 2026 12:35:19 -0400 Subject: [PATCH 9/9] fix: stop in-flight cleanup crashing the daemon on a rejecting deliver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P0 (round-7 review #4782757807). The round-6 in-flight tracker cleared its map entry via `delivery.finally(cb)` as a bare fire-and-forget. `.finally` returns a promise that re-throws when the original rejects, so a routine recoverable delivery failure (deliver rejects) surfaced as an unhandled rejection — which the daemon treats as fatal (process.exit(1) in main.ts, active in dev+prod). The handler already catches the rejection as 'failed' via the withTimeout await, but the orphaned `.finally` promise had no handler. Replaced with `delivery.then(clearInFlight, clearInFlight)`, which runs the cleanup on both fulfillment and rejection and resolves either way — no unhandled rejection. The reviewer's suggested one-liner. Adds a test that exercises a rejecting deliver (the path the green suite missed — prior tests used never-settling or resolving delivers) and asserts the handler returns 'failed' with no unhandledRejection event. Verified the test fails against the `.finally` version before re-applying. --- ...ong-horizon-agent-reminder-fire.handler.ts | 11 +++++-- ...orizon-agent-reminder-fire.handler.test.ts | 33 +++++++++++++++++++ 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts index 5f55e312ab..877baf7a87 100644 --- a/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts +++ b/packages/daemon/src/lib/job-handlers/long-horizon-agent-reminder-fire.handler.ts @@ -341,11 +341,18 @@ async function fireReminder( idempotencyKey, }); reminderDeliveriesInFlight.set(fresh.id, delivery); - delivery.finally(() => { + // Clear the in-flight entry when `delivery` settles (fulfilled OR rejected), + // and attach a rejection handler so a rejecting `deliver` doesn't surface as + // an unhandled rejection — the daemon treats those as fatal (process.exit(1) + // in main.ts). `.then(clear, clear)` resolves on both paths; `.finally` + // would re-throw the rejection and crash the process on a recoverable error + // the handler already catches as 'failed' below. + const clearInFlight = () => { if (reminderDeliveriesInFlight.get(fresh.id) === delivery) { reminderDeliveriesInFlight.delete(fresh.id); } - }); + }; + delivery.then(clearInFlight, clearInFlight); try { // Bound the delivery: a stuck SDK can leave enqueueWithId (and thus // deliver) unsettled forever. On timeout, treat as failed so the lock diff --git a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts index df3af5add9..d985899d21 100644 --- a/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts +++ b/packages/daemon/tests/unit/2-handlers/job-handlers/long-horizon-agent-reminder-fire.handler.test.ts @@ -368,6 +368,39 @@ describe('handleLongHorizonAgentReminderFire', () => { expect(deliver2.calls).toHaveLength(0); }); + it('a rejecting deliver returns failed without an unhandled rejection (no daemon crash)', async () => { + const now = Date.now(); + const reminder = reminderRepo.createReminder({ + spaceId, + agentId, + title: 'rejects', + triggerType: 'at', + runAt: now - 1000, + nextRunAt: now - 1000, + }); + // The daemon treats unhandled rejections as fatal (process.exit(1) in + // main.ts). The in-flight cleanup must attach a rejection handler so a + // rejecting deliver doesn't surface as one. + const rejections: unknown[] = []; + const onUnhandled = (reason: unknown) => rejections.push(reason); + process.on('unhandledRejection', onUnhandled); + try { + const rejectDeliver = async (): Promise<{ delivered: boolean }> => { + throw new Error('deliver boom'); + }; + const result = await handleLongHorizonAgentReminderFire( + makeJob(), + makeDeps(rejectDeliver, undefined, 5000) + ); + expect(result.failed).toBe(1); + // Drain the microtask/macrotask queue so any unhandled rejection surfaces. + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(rejections).toEqual([]); + } finally { + process.off('unhandledRejection', onUnhandled); + } + }); + it('defers (skips without advancing) when the occurrence is enqueued but not consumed', async () => { const now = Date.now(); const reminder = reminderRepo.createReminder({