Merge decision paging repairs into retention stack

Co-Authored-By: Paperclip <noreply@paperclip.ing>

* codex/sleep-13-grok-dispatch:
  fix(server): retain database precision in effect recovery cursor
  fix(server): preserve expiry paging timestamp precision

server/src/__tests__/decision-effect-recovery-work.test.ts

@@ -223,6 +223,24 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
     expect(await recovery.nextEffectRecoveryAt()).not.toBeNull();
   });
 
+  it("pages past invalid claims using database timestamp precision", async () => {
+    vi.stubEnv("PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE", "1");
+    const invalid = await create(), valid = await create();
+    await claim(invalid.id); await claim(valid.id, new Date(Date.now() - 120_000));
+    await db.execute(sql`update decisions set decided_by_user_id = null,
+      updated_at = date_trunc('milliseconds', now() - interval '3 minutes') + interval '456 microseconds'
+      where id = ${invalid.id}`);
+    const recovery = service();
+    expect(await recovery.recoverEffects()).toEqual({ resumed: 0 });
+    expect(await recovery.recoverEffects()).toEqual({ resumed: 1 });
+    expect(await comments()).toHaveLength(1);
+    expect((await recovery.get(invalid.id))?.executionStatus).toBe("running");
+    expect(await recovery.nextEffectRecoveryAt()).not.toBeNull();
+    expect(await recovery.recoverEffects()).toEqual({ resumed: 0 });
+    expect(await recovery.recoverEffects()).toEqual({ resumed: 0 });
+    expect(await comments()).toHaveLength(1);
+  });
+
   it("advances past malformed claims and retries them on a later pass", async () => {
     vi.stubEnv("PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE", "1");
     const invalid = await create(), valid = await create();

server/src/__tests__/decision-expiry-work.test.ts

@@ -244,4 +244,24 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
     await start(); await quiet(); expect(wake).toHaveBeenCalledTimes(2);
   });
 
+
+  it("pages past a full failed batch using database timestamp precision", async () => {
+    const seed = await create(); await db.delete(decisions).where(eq(decisions.id, seed.id));
+    const ids = Array.from({ length: 101 }, (_, i) => `20000000-0000-4000-8000-${String(i).padStart(12, "0")}`);
+    await db.insert(decisions).values(ids.map(id => ({ ...seed, id })));
+    await db.execute(sql`update decisions set expires_at = date_trunc('milliseconds', now() - interval '1 minute') + interval '456 microseconds'`);
+    await db.execute(sql`create function decision_expiry_fail() returns trigger language plpgsql as $$ begin
+      if NEW.id <> '20000000-0000-4000-8000-000000000100' and NEW.status = 'expired' then raise exception 'expiry unavailable'; end if;
+      return NEW; end $$`);
+    await db.execute(sql`create trigger decision_expiry_fail before update on decisions for each row execute function decision_expiry_fail()`);
+    const recovery = service();
+    await expect(recovery.sweepTtlExpiry()).rejects.toThrow("Decision expiry failed");
+    expect(await recovery.sweepTtlExpiry()).toEqual({ expired: 1 });
+    expect((await service().get(ids.at(-1)!))?.status).toBe("expired");
+    await db.execute(sql`drop trigger decision_expiry_fail on decisions`);
+    expect(await recovery.sweepTtlExpiry()).toEqual({ expired: 100 });
+    expect(await recovery.sweepTtlExpiry()).toEqual({ expired: 0 });
+    expect(wake).toHaveBeenCalledTimes(101);
+  });
+
 });

server/src/services/decision-effect-recovery.ts

@@ -14,14 +14,14 @@ function recoveryGraceMs() {
 }
 
 export function decisionEffectRecoveryService(db: Db, resume: (id: string) => Promise<boolean>) {
-  let cursor: { updatedAt: Date; id: string } | undefined;
+  let cursor: { updatedAt: string; id: string } | undefined;
 
   async function sweep(now = new Date(), limit = decisionSweepBatchSize(), signal?: AbortSignal) {
     let resumed = 0;
     if (signal?.aborted) return { resumed };
-    const rows = await db.select({ id: decisions.id, updatedAt: decisions.updatedAt }).from(decisions)
+    const rows = await db.select({ id: decisions.id, updatedAt: sql<string>`${decisions.updatedAt}::text` }).from(decisions)
       .where(and(running, lte(decisions.updatedAt, new Date(now.getTime() - recoveryGraceMs())), cursor
-        ? sql`(${decisions.updatedAt}, ${decisions.id}) > (${cursor.updatedAt.toISOString()}::timestamptz, ${cursor.id}::uuid)` : undefined))
+        ? sql`(${decisions.updatedAt}, ${decisions.id}) > (${cursor.updatedAt}::timestamptz, ${cursor.id}::uuid)` : undefined))
       .orderBy(asc(decisions.updatedAt), asc(decisions.id)).limit(limit);
     const errors: unknown[] = [];
     for (const row of rows) {

server/src/services/decision-expiry.ts

@@ -1,4 +1,4 @@
-import { and, asc, eq, lte, sql } from "drizzle-orm";
+import { and, asc, eq, getTableColumns, lte, sql } from "drizzle-orm";
 import { decisions, type Db } from "@paperclipai/db";
 import { logActivity } from "./activity-log.js";
 import { writeDecisionChange } from "./decision-work-notifications.js";
@@ -8,7 +8,7 @@ type ExpiryReason = "ttl" | "target_gone" | "target_completed";
 const open = sql`${decisions.status} = 'open'`;
 
 export function decisionExpiryService(db: Db, deliverContinuation: (id: string) => Promise<void>) {
-  let recoveryCursor: { expiresAt: Date; id: string } | undefined;
+  let recoveryCursor: { expiresAt: string; id: string } | undefined;
   async function expire(decision: Decision, reason: ExpiryReason, now: Date) {
     const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({
       status: "expired", updatedAt: now,
@@ -24,9 +24,10 @@ export function decisionExpiryService(db: Db, deliverContinuation: (id: string)
     return 1;
   }
 
-  async function due(now: Date, limit: number, cursor?: { expiresAt: Date; id: string }) {
-    return db.select().from(decisions).where(and(open, lte(decisions.expiresAt, now), cursor
-      ? sql`(${decisions.expiresAt}, ${decisions.id}) > (${cursor.expiresAt.toISOString()}::timestamptz, ${cursor.id}::uuid)` : undefined))
+  async function due(now: Date, limit: number, cursor?: { expiresAt: string; id: string }) {
+    return db.select({ ...getTableColumns(decisions), cursorExpiresAt: sql<string>`${decisions.expiresAt}::text` })
+      .from(decisions).where(and(open, lte(decisions.expiresAt, now), cursor
+      ? sql`(${decisions.expiresAt}, ${decisions.id}) > (${cursor.expiresAt}::timestamptz, ${cursor.id}::uuid)` : undefined))
       .orderBy(asc(decisions.expiresAt), asc(decisions.id)).limit(limit);
   }
 
@@ -38,7 +39,7 @@ export function decisionExpiryService(db: Db, deliverContinuation: (id: string)
     for (const row of rows) {
       if (signal?.aborted) return { expired };
       try { expired += await expire(row, "ttl", now); } catch (error) { errors.push(error); }
-      recoveryCursor = { expiresAt: row.expiresAt, id: row.id };
+      recoveryCursor = { expiresAt: row.cursorExpiresAt, id: row.id };
     }
     if (rows.length < 100) recoveryCursor = undefined;
     if (errors.length === 1) throw errors[0];