diff --git a/doc/idle-decision-archive-delivery.md b/doc/idle-decision-archive-delivery.md new file mode 100644 index 00000000000..4378bb00847 --- /dev/null +++ b/doc/idle-decision-archive-delivery.md @@ -0,0 +1,127 @@ +# Decision-archive notification delivery + +Auto-archive already writes `decision_archive_notification_outbox` in the same +transaction that archives its retention row. The old retention sweep called +notification delivery on every maintenance tick, including an empty outbox. +Delivery is now its own coordinator worker. Retention scanning remains a separate +migration; this PR does not change what is archived or when. + +## Ownership and producer audit + +`decision-retention.ts` owns eligibility, current-version checks, source authority, +origin resolution, and archive writes. Its sole outbox producer is `autoArchive`. +It now awaits the delivery notification immediately before inserting the outbox +row in the existing transaction. Caller-owned transactions and savepoints retain +outer commit/rollback ownership through the shared database signal mechanism. +Manual archive/revive and reviewed-manifest archive do not create these +notifications today; this change does not add notifications to those actions. + +`decision-archive-delivery.ts` owns claim recovery, selection, grouping, delivery, +and settlement. `decisionRetentionService.deliverNotifications` remains a +compatible delegate. Its notification callback and batch type retain their +existing contract. The scheduler only sees the delivery operation and next +work deadline. Claim and acknowledgement writes belong to this worker; they do +not broadcast self-generated wakes. No table is added. A concurrent partial index supports the pending delivery +order on the existing outbox. + +The server registers the worker in enabled-heartbeat startup housekeeping and +removes notification delivery from `runRetentionSweep`. Registration scans saved +work before the retention pass can enqueue more. The disabled-heartbeat branch +continues not to dispatch agent notifications. A service without a callback +leaves outbox rows intact and does not attempt notification delivery. + +## Work and deadlines + +Pending outbox rows are due for delivery. Each pass claims up to 500 rows and +groups them by company and origin agent, preserving company scope and the +existing notification payload. Failed or remaining pending rows retain a +30-second retry. Never-attempted rows sort first, then the oldest attempts, so +one failing batch cannot repeatedly hide later rows. + +A delivering row retains its existing five-minute claim deadline. Startup +reconstructs that deadline from `lastAttemptAt`; a legacy missing timestamp is +immediately recoverable. A worker waiting only for a future claim deadline does +not repeatedly query until that deadline. Settlement and failure reset require +the same claim timestamp, so a delayed old result cannot settle or release a +replacement claim. Delivered counts include only rows actually settled. + +An empty outbox or retained delivered history has no worker deadline and no +recurring SQL. The coordinator coalesces writes and prevents overlapping passes. +The shared deadline worker preserves first enqueue during drain, changes during +delivery, and failed retries across drain/resume. Coordinator shutdown joins +in-flight work. These guarantees do not replace the durable rows. + +## Recovery and limits + +A failed enqueue rolls back both the archive mutation and its outbox insertion. +An uncertain transaction retains the shared intent until its outcome is known. +A notification failure returns the same rows to pending. A process stopping +while delivering leaves the claim for deadline recovery on restart. + +Delivery remains at least once. A wake accepted before a lost acknowledgement +can be retried. This change does not promise exactly-once agent wakeups or add +new deduplication storage. The five-minute stale threshold is unchanged. + +Supported producers share the process-local database signal owner. Independent +SQL or another active server needs a separate wake contract. The existing Cloud +sleep proof still blocks this outbox conservatively; no external-wake coverage +or sleep guard is relaxed. Worker quiet does not prove actual database or host +suspension. Retention scanning and decision continuations remain separate duties. + +## Verification + +The original migration selection passed 54 tests: 15 real-PostgreSQL retention/delivery cases +and 39 startup cases. Twelve new cases cover configured idle, retained delivered +history, real enqueue, restart, exact stale-claim timing, legacy missing timestamps, +first enqueue during drain, failed-delivery retry through drain, outer commit and +rollback, enqueue failure rollback, writes during delivery, more than 500 rows, +and stale-result fencing. Quiet assertions observe no reads or writes across an +hour of controlled time. Workspace typecheck and build pass. + +The first 52-case selection passed. Adding fault/concurrency tests then exposed +two fixture errors: duplicate globally unique issue identifiers across two seeded +companies, and asserting quiet before rollback reconciliation had run. Seeded +identifiers now use their company prefix; quiet checks advance the retained retry +and await authoritative settlement. All 54 cases passed on rerun. No production +hold, recovery deadline, or timeout was weakened. Full repository CI and fresh +review are required before this PR is ready. + + +## Review repairs: startup, claim results and backlog selection + +Startup regressions now exercise `startServer` itself. Enabled startup must +register `decisionArchiveDelivery` and invoke the delivery owner; calling the +captured heartbeat interval still performs retention reads without another +notification delivery. Calling the registered worker invokes delivery again. +The disabled-heartbeat case explicitly preserves the existing no-notification +contract. An initial expanded test incorrectly expected that disabled mode to +register the worker and failed; inspection confirmed the documented disabled +contract, so the test now asserts it rather than enabling new background work. + +The real stale-claim regression first returned `notifiedAgents: 1` after the +replacement claim prevented any settlement. The counter now increments only +for a group that actually settles rows, while the existing timestamp fence and +at-least-once delivery contract remain intact. Failure reset uses one timestamp +for `lastAttemptAt` and `updatedAt`. + +Generated migration `0387_boring_professor_monster.sql` creates a concurrent +partial index on `(last_attempt_at NULLS FIRST, created_at, id)` for pending rows. +The existing top-level migration executor handles concurrent creation and +invalid-index recovery. No history migration or existing journal entry changes; +the generated snapshots retain the five newest versions. + +A real migration and query-plan check used a separate database with 50,000 +outbox rows, including 30,000 pending. The existing claim-selection query returned +500 rows in both cases. Without the new index PostgreSQL sorted the pending +backlog with a top-N heapsort and touched 944 shared buffers; with the index it +used an index-only scan without sorting and touched 505 buffers. The baseline +index removal was rolled back inside that dedicated database, and the migrated +index was confirmed valid afterwards. These local measurements establish the +access path, not production latency or cost savings. + +All 58 cases across real PostgreSQL delivery/retention, startup and the shared +deadline worker pass. Generated snapshot drift, migration numbering/safety, +server types and build pass. The earlier stale-counter failure and invalid +scheduler-disabled expectation are retained above. The existing 500-row batch +limit and thirty-second retry for remaining pending work are unchanged. Fresh +current-head CI, Greptile and Ponytail remain required before merge. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 25d3ebf70bc..2345ab9bb90 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -15,7 +15,7 @@ Do not mark an item done from a timer removal or a green check alone. | 01 | Custom-image setup expiry | Session writes, existing expiry, startup/retry/drain | [#17443](https://github.com/paperclipai/paperclip/pull/17443), merged after CI and passing reviews | | 02 | Remaining chat maintenance | Provider runtime discovery; voice reconciliation; milestones; wake notices; interaction continuations; terminal confirmations; GitHub recovery; Teams maintenance, separately by owner | Pending | | 03 | Run and task recovery | Orphan runs and active leases; shutdown workspace settlement; scheduled retries; queued runs; stranded assignments; resolved dependencies; task watchdogs; silent-run monitoring; stale locks | Pending | -| 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Pending | +| 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Retention pending; notification delivery implemented in slice 08, PR pending | | 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Pending | | 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | Pending | | 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Secret proposals: [#17446](https://github.com/paperclipai/paperclip/pull/17446), merged; gateway credentials: [#17450](https://github.com/paperclipai/paperclip/pull/17450), merged; pending tool-review expiry: [#17453](https://github.com/paperclipai/paperclip/pull/17453), merged; approved recovery: [#17454](https://github.com/paperclipai/paperclip/pull/17454), merged after current-head CI and both 5/5 reviews; interrupted claims: [#17458](https://github.com/paperclipai/paperclip/pull/17458), merged after current-head CI and both 5/5 reviews; projection: [#17468](https://github.com/paperclipai/paperclip/pull/17468), interaction-link index verified and awaiting fresh CI/review | @@ -335,3 +335,27 @@ gateway/startup/deadline selection passes 222 cases; server types/build pass. The initial invalid one-wake-per-approval assertion and corrected isolated-hint measurement are retained in the behavior document. Current-head CI/reviews remain required before this slice can merge. + + +## Slice 08: decision-archive notification delivery + +Projection repair is #17468. Its bounded master integration passes local +acceptance and awaits fresh checks and review. Slice 08 builds on that branch +and separates archive notification delivery from retention scanning. +The existing auto-archive transaction is the sole outbox producer. A committed +enqueue wakes delivery; pending rows retain retries and delivering rows retain +their saved five-minute claim deadline. Delivered history is quiet. + +The implementation preserves the delivery API and payload. It fences stale +completion by claim timestamp and orders unattempted/oldest-attempted pending +rows so failed batches do not starve later work. No table is added. Generated +migration `0387_boring_professor_monster.sql` adds the concurrent partial index +for pending delivery order. See [archive delivery](../idle-decision-archive-delivery.md). + +All 58 selected tests pass: 15 real-PostgreSQL retention/delivery cases, 42 startup +cases and the shared deadline worker case. Migration checks, generated snapshot +drift, server typecheck and build pass. The initial fixture failures, stale-claim +counter regression and corrected disabled-mode expectation are recorded in the +behavior document. Full current-head CI and both reviews remain required. +Retention eligibility/scanning, decision expiry, effect recovery, and decision +continuation delivery remain independent unfinished reasons in the full ledger. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 342256ac462..1d98d929d7a 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -505,6 +505,8 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0382_narrow_the_executioner.sql", "0385_open_madame_web.sql", "0386_steady_epoch.sql", + "0387_boring_professor_monster.sql", + "0384_funny_adam_warlock.sql", "0383_tool_review_company_expiry.sql", ]); diff --git a/packages/db/src/migrations/0387_boring_professor_monster.sql b/packages/db/src/migrations/0387_boring_professor_monster.sql new file mode 100644 index 00000000000..9398f9e44bc --- /dev/null +++ b/packages/db/src/migrations/0387_boring_professor_monster.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "decision_archive_notification_outbox_delivery_idx" ON "decision_archive_notification_outbox" USING btree ("last_attempt_at" NULLS FIRST,"created_at","id") WHERE "decision_archive_notification_outbox"."status" = 'pending'; diff --git a/packages/db/src/migrations/meta/0382_snapshot.json b/packages/db/src/migrations/meta/0387_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0382_snapshot.json rename to packages/db/src/migrations/meta/0387_snapshot.json index 6c27dd9aa22..7fd3f6cbe93 100644 --- a/packages/db/src/migrations/meta/0382_snapshot.json +++ b/packages/db/src/migrations/meta/0387_snapshot.json @@ -1,6 +1,6 @@ { - "id": "ecec138e-5312-4ad6-b051-bd4ee9ed4e34", - "prevId": "2e93deb7-f24c-42d1-8c8e-30b21cff90bc", + "id": "de27fdf8-430a-443a-9502-725f92c48892", + "prevId": "da3229b5-4eab-4d8b-ba6b-d756e6c42430", "version": "7", "dialect": "postgresql", "tables": { @@ -22110,6 +22110,34 @@ "concurrently": false, "method": "btree", "with": {} + }, + "decision_archive_notification_outbox_delivery_idx": { + "name": "decision_archive_notification_outbox_delivery_idx", + "columns": [ + { + "expression": "last_attempt_at", + "isExpression": false, + "asc": true, + "nulls": "first" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"decision_archive_notification_outbox\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -52834,6 +52862,66 @@ "method": "btree", "with": {} }, + "tool_action_requests_company_pending_expiry_idx": { + "name": "tool_action_requests_company_pending_expiry_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "tool_action_requests_executing_deadline_idx": { + "name": "tool_action_requests_executing_deadline_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'executing'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "tool_action_requests_approved_idx": { + "name": "tool_action_requests_approved_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'approved'", + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_action_requests_invocation_idx": { "name": "tool_action_requests_invocation_idx", "columns": [ @@ -52849,6 +52937,27 @@ "method": "btree", "with": {} }, + "tool_action_requests_company_interaction_idx": { + "name": "tool_action_requests_company_interaction_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "interaction_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_action_requests_issue_idx": { "name": "tool_action_requests_issue_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 578d8027ed9..4e9bf240449 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2696,6 +2696,13 @@ "when": 1793135146893, "tag": "0386_steady_epoch", "breakpoints": true + }, + { + "idx": 387, + "version": "7", + "when": 1793136115698, + "tag": "0387_boring_professor_monster", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/decision_queues.ts b/packages/db/src/schema/decision_queues.ts index 667c67f8210..12215c34652 100644 --- a/packages/db/src/schema/decision_queues.ts +++ b/packages/db/src/schema/decision_queues.ts @@ -248,6 +248,9 @@ export const decisionArchiveNotificationOutbox = pgTable( table.originAgentId, ), pendingIdx: index("decision_archive_notification_outbox_pending_idx").on(table.status, table.createdAt), + pendingDeliveryIdx: index("decision_archive_notification_outbox_delivery_idx") + .on(table.lastAttemptAt.asc().nullsFirst(), table.createdAt, table.id) + .where(sql`${table.status} = 'pending'`), statusCheck: check( "decision_archive_notification_outbox_status_check", sql`${table.status} IN ('pending', 'delivering', 'delivered')`, diff --git a/server/src/__tests__/decision-retention-service.test.ts b/server/src/__tests__/decision-retention-service.test.ts index 07be9a27859..d03796a5b3e 100644 --- a/server/src/__tests__/decision-retention-service.test.ts +++ b/server/src/__tests__/decision-retention-service.test.ts @@ -1,5 +1,10 @@ +import { sql } from "drizzle-orm"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { registerDecisionArchiveDeliveryWork } from "../services/decision-archive-delivery-work.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; import { randomUUID } from "node:crypto"; -import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { activityLog, agents, @@ -13,19 +18,23 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro import { attentionService } from "../services/attention.js"; import { decisionRetentionService } from "../services/decision-retention.js"; -const support = await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_RETENTION_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describePg = support.supported ? describe : describe.skip; describePg("decision retention", () => { let tempDb: Awaited> | null = null; let db: ReturnType; + let stopWorker: (() => Promise) | undefined; + beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decision-retention-"); - db = createDb(tempDb.connectionString); + if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decision-retention-"); + db = createDb(externalUrl ?? tempDb!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); }, 30_000); afterEach(async () => { + await stopWorker?.(); stopWorker = undefined; await db.delete(decisionArchiveNotificationOutbox); await db.delete(decisionRetention); await db.delete(activityLog); @@ -34,7 +43,7 @@ describePg("decision retention", () => { await db.delete(companies); }); - afterAll(async () => tempDb?.cleanup()); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); async function seedReviewItems(count: number) { const companyId = randomUUID(); @@ -61,7 +70,7 @@ describePg("decision retention", () => { await db.insert(issues).values(issueIds.map((id, index) => ({ id, companyId, - identifier: `RET-${index + 1}`, + identifier: `R${companyId.slice(0, 6)}-${index + 1}`, title: `Old review ${index + 1}`, status: "in_review", priority: "medium", @@ -90,6 +99,180 @@ describePg("decision retention", () => { responsibleUserId: "board-user", }; + describe("archive delivery worker", () => { + let coordinator: ReturnType | undefined; + let fixture: Awaited>; + let service: ReturnType; + let draining: boolean; + const errors: unknown[] = []; + const notify = vi.fn(async (_batch: unknown) => {}); + + beforeEach(async () => { + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + fixture = await seedReviewItems(1); draining = false; errors.length = 0; notify.mockReset(); + service = decisionRetentionService(db, { notifyOriginAgent: notify }); + stopWorker = async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.execute(sql`drop trigger if exists archive_delivery_fail on decision_archive_notification_outbox`); + await db.execute(sql`drop function if exists archive_delivery_fail()`); + expect(errors).toEqual([]); + }; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, + canReconcile: () => true, onError: error => errors.push(error) }); + await registerDecisionArchiveDeliveryWork(coordinator, service, () => !draining).ready; + } + async function archive(source = db) { + const feed = await attentionService(source).list(fixture.companyId, { limit: 100 }); + return decisionRetentionService(source).autoArchive({ companyId: fixture.companyId, items: feed.items }); + } + async function settle(ms = 60_000) { + await vi.advanceTimersByTimeAsync(ms); + if (!draining) await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { timeout: 5000 }); + } + async function quiet() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + const reads = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute"), writes = vi.spyOn(db, "update"); + await settle(60 * 60_000); + expect(reads).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); executes.mockRestore(); writes.mockRestore(); + } + async function delivered() { + const rows = await db.select().from(decisionArchiveNotificationOutbox); + expect(rows.length).toBeGreaterThan(0); expect(rows.every(row => row.status === "delivered")).toBe(true); + } + + it("is quiet with configuration, wakes from real auto-archive, and leaves delivered history quiet", async () => { + await start(); await quiet(); + expect(await archive()).toBe(1); await settle(); await delivered(); await quiet(); + expect(notify).toHaveBeenCalledTimes(1); + expect(notify.mock.calls[0][0]).toMatchObject({ companyId: fixture.companyId, agentId: fixture.agentId, + items: [expect.objectContaining({ issueId: fixture.issueIds[0], archiveVersion: 1 })] }); + }); + + it("reconstructs pending delivery after restart", async () => { + await archive(); await start(); await quiet(); await delivered(); expect(notify).toHaveBeenCalledTimes(1); + }); + + it("waits until an existing delivery claim expires", async () => { + await archive(); const attemptedAt = new Date(); + await db.update(decisionArchiveNotificationOutbox).set({ status: "delivering", lastAttemptAt: attemptedAt, attemptCount: 1 }); + await start(); expect(notify).not.toHaveBeenCalled(); + expect(coordinator!.nextWakeAt()).toBe(attemptedAt.getTime() + 5 * 60_000); + const reads = vi.spyOn(db, "select"); await settle(4 * 60_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await settle(2 * 60_000); await delivered(); await quiet(); expect(notify).toHaveBeenCalledTimes(1); + }); + + it("recovers a legacy delivering row with no attempt timestamp", async () => { + await archive(); await db.update(decisionArchiveNotificationOutbox).set({ status: "delivering" }); + await start(); await delivered(); await quiet(); + }); + + it("retains first enqueue during drain without passive SQL", async () => { + await start(); await quiet(); draining = true; + await archive(); await settle(); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"); + await settle(60 * 60_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); expect(notify).not.toHaveBeenCalled(); + draining = false; await settle(); await delivered(); await quiet(); + }); + + it("retries failed delivery through drain and then becomes quiet", async () => { + await archive(); notify.mockRejectedValueOnce(new Error("wake unavailable")); + await start(); expect(notify).toHaveBeenCalledTimes(1); expect(coordinator!.nextWakeAt()).not.toBeNull(); + draining = true; await settle(60 * 60_000); expect(notify).toHaveBeenCalledTimes(1); + draining = false; + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); await delivered(); }, { timeout: 5000 }); + await quiet(); expect(notify).toHaveBeenCalledTimes(2); + }); + + it.each([false, true])("waits for the archive's outer transaction (rollback %s)", async rollback => { + await start(); await quiet(); + let entered!: () => void, release!: () => void; + const ready = new Promise(resolve => { entered = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const aborted = new Error("fixture rollback"); + const running = db.transaction(async tx => { + await archive(tx as unknown as typeof db); entered(); await gate; if (rollback) throw aborted; + }).catch(error => { if (error !== aborted) throw error; }); + await ready; + try { await vi.advanceTimersByTimeAsync(60_000); expect(notify).not.toHaveBeenCalled(); } + finally { release(); await running; } + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); expect(idleWorkSnapshot().active).toBe(0); + if (rollback) expect(await db.select().from(decisionArchiveNotificationOutbox)).toEqual([]); + else await delivered(); + }, { timeout: 5000 }); + await quiet(); expect(notify).toHaveBeenCalledTimes(rollback ? 0 : 1); + }); + + it("rolls back archive state when enqueue fails and wakes on a later successful archive", async () => { + await start(); await quiet(); + await db.execute(sql`create function archive_delivery_fail() returns trigger language plpgsql as $$ + begin raise exception 'archive enqueue fixture failure'; end $$`); + await db.execute(sql`create trigger archive_delivery_fail before insert on decision_archive_notification_outbox + for each row execute function archive_delivery_fail()`); + await expect(archive()).rejects.toThrow(); + await db.execute(sql`drop trigger archive_delivery_fail on decision_archive_notification_outbox`); + await quiet(); expect(notify).not.toHaveBeenCalled(); + expect(await db.select().from(decisionArchiveNotificationOutbox)).toEqual([]); + expect((await db.select().from(decisionRetention))[0].archivedAt).toBeNull(); + await archive(); await settle(); await delivered(); await quiet(); + }); + + it("keeps an enqueue committed during delivery and drains it on the next pass", async () => { + await start(); await quiet(); + let entered!: () => void, release!: () => void; + const ready = new Promise(resolve => { entered = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + notify.mockImplementationOnce(async () => { entered(); await gate; }); + await archive(); + await vi.advanceTimersByTimeAsync(30_000); await ready; + try { + fixture = await seedReviewItems(1); await archive(); + expect(notify).toHaveBeenCalledTimes(1); + } finally { release(); } + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); await delivered(); }, { timeout: 5000 }); + await quiet(); expect(notify).toHaveBeenCalledTimes(2); + }); + + it("does not let an older failed batch starve rows beyond the batch limit", async () => { + await archive(); const [row] = await db.select().from(decisionArchiveNotificationOutbox); + await db.insert(decisionArchiveNotificationOutbox).values(Array.from({ length: 501 }, (_, index) => ({ + ...row, id: randomUUID(), sourceId: `additional-${index}` }))); + notify.mockRejectedValueOnce(new Error("first batch unavailable")); + await start(); expect(notify.mock.calls[0][0]).toMatchObject({ items: expect.any(Array) }); + expect((notify.mock.calls[0][0] as { items: unknown[] }).items).toHaveLength(500); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); await delivered(); }, { timeout: 5000 }); + await quiet(); + expect((notify.mock.calls[1][0] as { items: unknown[] }).items).toHaveLength(500); + expect((notify.mock.calls[2][0] as { items: unknown[] }).items).toHaveLength(2); + }); + + it("does not settle a replacement claim from an older delivery result", async () => { + await archive(); + let entered!: () => void, release!: () => void; + const ready = new Promise(resolve => { entered = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + notify.mockImplementationOnce(async () => { entered(); await gate; }); + const first = service.deliverNotifications(); await ready; + const replacementAt = new Date(Date.now() + 6 * 60_000); + try { await db.update(decisionArchiveNotificationOutbox).set({ lastAttemptAt: replacementAt, attemptCount: 2 }); } + finally { release(); } + expect(await first).toEqual({ notifiedAgents: 0, delivered: 0 }); + expect((await db.select().from(decisionArchiveNotificationOutbox))[0]).toMatchObject({ status: "delivering", lastAttemptAt: replacementAt }); + await start(); expect(coordinator!.nextWakeAt()).toBe(replacementAt.getTime() + 5 * 60_000); + await settle(12 * 60_000); await delivered(); await quiet(); + }); + }); + it("auto-archives, excludes by default, returns archived rows, and revives", async () => { const { companyId, issueIds } = await seedReviewItems(1); const now = new Date("2026-08-02T00:00:00.000Z"); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index cb7da55e61a..2f454b226e2 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -6,8 +6,8 @@ import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; import * as idleLocalWork from "../services/idle-local-work.js"; import { idleBackupWakeMarker, markIdleBackupWakeRequired } from "../services/idle-database-backup.js"; import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; -import { reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js"; -import { runDatabaseBackup } from "@paperclipai/db"; +import { decisionRetentionService, reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js"; +import { companies, runDatabaseBackup, type Db } from "@paperclipai/db"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { chmodSync, existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, symlinkSync, writeFileSync } from "node:fs"; import { gzipSync } from "node:zlib"; @@ -352,6 +352,7 @@ vi.mock("../services/index.js", () => ({ decisionRetentionService: vi.fn(() => ({ autoArchive: vi.fn(async () => 0), deliverNotifications: vi.fn(async () => ({ notifiedAgents: 0, delivered: 0 })), + nextArchiveNotificationAt: vi.fn(async () => null), })), feedbackService: feedbackServiceFactoryMock, bootstrapExecutionPolicyFromEnv: vi.fn(async () => null), @@ -521,6 +522,40 @@ describe("startServer feedback export wiring", () => { process.env.BETTER_AUTH_SECRET = "test-secret"; }); + it.each([false, true])("registers archive delivery separately from retention sweeps (scheduler %s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled, heartbeatSchedulerIntervalMs: 30000 })); + let tick: (() => void) | undefined; + const timer = vi.spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void, interval: number) => { + if (interval === 30000) tick = callback; + return 1 as unknown as ReturnType; + }) as typeof setInterval); + try { + await startServer(); await settleStartupHousekeeping(); + const registration = deliveryWorkRegisterMock.mock.calls.find(([queue]) => queue === DELIVERY_QUEUES.decisionArchiveDelivery); + if (!enabled) { + expect(registration).toBeUndefined(); + expect(decisionRetentionService).not.toHaveBeenCalled(); + expect(tick).toBeDefined(); tick!(); await settleStartupHousekeeping(); + expect(decisionRetentionService).not.toHaveBeenCalled(); + return; + } + expect(registration).toBeDefined(); + const retention = vi.mocked(decisionRetentionService).mock.results[0]!.value; + expect(retention.deliverNotifications).toHaveBeenCalledOnce(); + expect(retention.nextArchiveNotificationAt).toHaveBeenCalledOnce(); + const db = createDbMock.mock.results[0]!.value as Db; + const companyReads = () => vi.mocked(db.select).mock.results.filter(({ value }) => + vi.mocked(value.from).mock.calls.some(([table]) => table === companies)).length; + const before = companyReads(); + expect(before).toBeGreaterThan(0); expect(tick).toBeDefined(); + tick!(); await settleStartupHousekeeping(); + expect(companyReads()).toBeGreaterThan(before); + expect(retention.deliverNotifications).toHaveBeenCalledOnce(); + await registration![1].run(new AbortController().signal); + expect(retention.deliverNotifications).toHaveBeenCalledTimes(2); + } finally { timer.mockRestore(); } + }); + it("registers durable case handoffs independently of the heartbeat scheduler", async () => { await startServer(); expect(caseHandoffDeliveryServiceFactoryMock).toHaveBeenCalledWith(createDbMock.mock.results[0]!.value, { heartbeat: heartbeatServiceMock }); diff --git a/server/src/index.ts b/server/src/index.ts index bef68b441e7..37661c37c0e 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,5 @@ +import { registerDecisionArchiveDeliveryWork } from "./services/decision-archive-delivery-work.js"; + import { registerToolReviewProjectionWork } from "./services/tool-review-projection-work.js"; import { registerToolReviewInterruptionWork } from "./services/tool-review-interruption-work.js"; import { registerToolReviewRecoveryWork } from "./services/tool-review-recovery-work.js"; @@ -1809,8 +1811,7 @@ async function startServerWithDatabaseTeardown( }); archived += await retentionExecutor.autoArchive({ companyId: company.id, items: page.items }); } - const notifications = await retentionExecutor.deliverNotifications(); - return { archived, ...notifications }; + return { archived }; }; // Startup housekeeping: provider and connection sweeps that every timer tick // repeats and that neither serving requests nor run recovery depends on. @@ -1843,6 +1844,8 @@ async function startServerWithDatabaseTeardown( // Retry any orphan sandbox teardown left by a failed acquire before a // server restart, so a leaked sandbox does not stay allocated. await step("environment_lease_cleanup", startEnvironmentCleanup); + await step("archive_notification_worker", () => registerDecisionArchiveDeliveryWork(deliveryWork, retentionExecutor, + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); await step("retention_sweep", runRetentionSweep); }; diff --git a/server/src/services/decision-archive-delivery-work.ts b/server/src/services/decision-archive-delivery-work.ts new file mode 100644 index 00000000000..7b75cd921be --- /dev/null +++ b/server/src/services/decision-archive-delivery-work.ts @@ -0,0 +1,17 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { registerDeadlineWork } from "./deadline-work.js"; +import { DECISION_ARCHIVE_RETRY_MS } from "./decision-archive-delivery.js"; +import type { decisionRetentionService } from "./decision-retention.js"; + +export function registerDecisionArchiveDeliveryWork( + coordinator: ReturnType, + service: Pick, "deliverNotifications" | "nextArchiveNotificationAt">, + canRun: () => boolean, +) { + return registerDeadlineWork(coordinator, DELIVERY_QUEUES.decisionArchiveDelivery, { + retryMs: DECISION_ARCHIVE_RETRY_MS, canRun, + run: () => service.deliverNotifications(), + next: service.nextArchiveNotificationAt, + }); +} diff --git a/server/src/services/decision-archive-delivery.ts b/server/src/services/decision-archive-delivery.ts new file mode 100644 index 00000000000..df7bd7ee1f2 --- /dev/null +++ b/server/src/services/decision-archive-delivery.ts @@ -0,0 +1,115 @@ +import { and, asc, eq, inArray, isNull, lte, or, sql } from "drizzle-orm"; +import { decisionArchiveNotificationOutbox, type Db } from "@paperclipai/db"; +import type { AttentionSourceKind } from "@paperclipai/shared"; + +const CLAIM_MS = 5 * 60_000; +export const DECISION_ARCHIVE_RETRY_MS = 30_000; + +export type ArchiveNotificationBatch = { + companyId: string; + agentId: string; + items: Array<{ + sourceKind: AttentionSourceKind; + sourceId: string; + archiveVersion: number; + issueId: string; + }>; +}; + +export function decisionArchiveDeliveryService( + db: Db, + options: { notifyOriginAgent?: (batch: ArchiveNotificationBatch) => Promise }, +) { + async function deliverNotifications(limit = 500) { + if (!options.notifyOriginAgent) return { notifiedAgents: 0, delivered: 0 }; + const claimAt = new Date(); + const staleClaimAt = new Date(claimAt.getTime() - CLAIM_MS); + await db.update(decisionArchiveNotificationOutbox).set({ status: "pending", updatedAt: claimAt }) + .where(and( + eq(decisionArchiveNotificationOutbox.status, "delivering"), + or(isNull(decisionArchiveNotificationOutbox.lastAttemptAt), lte(decisionArchiveNotificationOutbox.lastAttemptAt, staleClaimAt)), + )); + const rows = await db.transaction(async (tx) => { + const pendingIds = await tx.select({ id: decisionArchiveNotificationOutbox.id }) + .from(decisionArchiveNotificationOutbox) + .where(eq(decisionArchiveNotificationOutbox.status, "pending")) + .orderBy(sql`${decisionArchiveNotificationOutbox.lastAttemptAt} asc nulls first`, + asc(decisionArchiveNotificationOutbox.createdAt), asc(decisionArchiveNotificationOutbox.id)) + .limit(limit) + .then((values) => values.map((value) => value.id)); + if (pendingIds.length === 0) return []; + return tx.update(decisionArchiveNotificationOutbox).set({ + status: "delivering", + lastAttemptAt: claimAt, + attemptCount: sql`${decisionArchiveNotificationOutbox.attemptCount} + 1`, + updatedAt: claimAt, + }).where(and( + inArray(decisionArchiveNotificationOutbox.id, pendingIds), + eq(decisionArchiveNotificationOutbox.status, "pending"), + )).returning(); + }); + const groups = new Map(); + for (const row of rows) { + const key = `${row.companyId}:${row.originAgentId}`; + const grouped = groups.get(key) ?? []; + grouped.push(row); + groups.set(key, grouped); + } + let notifiedAgents = 0; + let delivered = 0; + for (const grouped of groups.values()) { + const first = grouped[0]!; + try { + await options.notifyOriginAgent({ + companyId: first.companyId, + agentId: first.originAgentId, + items: grouped.map((row) => ({ + sourceKind: row.sourceKind as AttentionSourceKind, + sourceId: row.sourceId, + archiveVersion: row.archiveVersion, + issueId: row.originIssueId, + })), + }); + const deliveredAt = new Date(); + const settled = await db.update(decisionArchiveNotificationOutbox).set({ + status: "delivered", + deliveredAt, + lastAttemptAt: deliveredAt, + updatedAt: deliveredAt, + }).where(and( + inArray(decisionArchiveNotificationOutbox.id, grouped.map((row) => row.id)), + eq(decisionArchiveNotificationOutbox.status, "delivering"), + eq(decisionArchiveNotificationOutbox.lastAttemptAt, claimAt), + )).returning({ id: decisionArchiveNotificationOutbox.id }); + if (settled.length > 0) notifiedAgents += 1; + delivered += settled.length; + } catch { + const failedAt = new Date(); + await db.update(decisionArchiveNotificationOutbox).set({ + status: "pending", + lastAttemptAt: failedAt, + updatedAt: failedAt, + }).where(and( + inArray(decisionArchiveNotificationOutbox.id, grouped.map((row) => row.id)), + eq(decisionArchiveNotificationOutbox.status, "delivering"), + eq(decisionArchiveNotificationOutbox.lastAttemptAt, claimAt), + )); + } + } + return { notifiedAgents, delivered }; + } + + async function next() { + if (!options.notifyOriginAgent) return null; + const [pending] = await db.select({ id: decisionArchiveNotificationOutbox.id }) + .from(decisionArchiveNotificationOutbox).where(eq(decisionArchiveNotificationOutbox.status, "pending")).limit(1); + const [claimed] = await db.select({ attemptedAt: decisionArchiveNotificationOutbox.lastAttemptAt }) + .from(decisionArchiveNotificationOutbox).where(eq(decisionArchiveNotificationOutbox.status, "delivering")) + .orderBy(sql`${decisionArchiveNotificationOutbox.lastAttemptAt} asc nulls first`).limit(1); + const pendingAt = pending ? Date.now() + DECISION_ARCHIVE_RETRY_MS : null; + const claimAt = claimed ? (claimed.attemptedAt?.getTime() ?? Date.now() - CLAIM_MS) + CLAIM_MS : null; + return pendingAt === null ? claimAt : claimAt === null ? pendingAt : Math.min(pendingAt, claimAt); + } + + return { deliver: deliverNotifications, next }; +} diff --git a/server/src/services/decision-retention.ts b/server/src/services/decision-retention.ts index 18616e44d71..b65e6e3bb77 100644 --- a/server/src/services/decision-retention.ts +++ b/server/src/services/decision-retention.ts @@ -1,5 +1,7 @@ +import { decisionArchiveDeliveryService, type ArchiveNotificationBatch } from "./decision-archive-delivery.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { createHash } from "node:crypto"; -import { and, asc, eq, inArray, isNotNull, isNull, lte, ne, or, sql } from "drizzle-orm"; +import { and, eq, inArray, isNotNull, isNull, lte, ne, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { agents, @@ -31,16 +33,7 @@ export const DEFAULT_DECISION_ARCHIVE_DAYS = 90; const DAY_MS = 86_400_000; export type DecisionRetentionState = typeof decisionRetention.$inferSelect; -export type ArchiveNotificationBatch = { - companyId: string; - agentId: string; - items: Array<{ - sourceKind: AttentionSourceKind; - sourceId: string; - archiveVersion: number; - issueId: string; - }>; -}; +export type { ArchiveNotificationBatch } from "./decision-archive-delivery.js"; function sourceKey(sourceKind: string, sourceId: string) { return `${sourceKind}:${sourceId}`; @@ -371,6 +364,7 @@ export function decisionRetentionService( archived += 1; const origin = await resolveOrigin(txDb, input.companyId, item.sourceKind, item.subject.id); if (origin) { + await notifyDeliveryWork(txDb, DELIVERY_QUEUES.decisionArchiveDelivery); await txDb.insert(decisionArchiveNotificationOutbox).values({ companyId: input.companyId, sourceKind: item.sourceKind, @@ -394,80 +388,7 @@ export function decisionRetentionService( return archived; } - async function deliverNotifications(limit = 500) { - if (!options.notifyOriginAgent) return { notifiedAgents: 0, delivered: 0 }; - const claimAt = new Date(); - const staleClaimAt = new Date(claimAt.getTime() - 5 * 60_000); - await db.update(decisionArchiveNotificationOutbox).set({ status: "pending", updatedAt: claimAt }) - .where(and( - eq(decisionArchiveNotificationOutbox.status, "delivering"), - or(isNull(decisionArchiveNotificationOutbox.lastAttemptAt), lte(decisionArchiveNotificationOutbox.lastAttemptAt, staleClaimAt)), - )); - const rows = await db.transaction(async (tx) => { - const pendingIds = await tx.select({ id: decisionArchiveNotificationOutbox.id }) - .from(decisionArchiveNotificationOutbox) - .where(eq(decisionArchiveNotificationOutbox.status, "pending")) - .orderBy(asc(decisionArchiveNotificationOutbox.createdAt), asc(decisionArchiveNotificationOutbox.id)) - .limit(limit) - .then((values) => values.map((value) => value.id)); - if (pendingIds.length === 0) return []; - return tx.update(decisionArchiveNotificationOutbox).set({ - status: "delivering", - lastAttemptAt: claimAt, - attemptCount: sql`${decisionArchiveNotificationOutbox.attemptCount} + 1`, - updatedAt: claimAt, - }).where(and( - inArray(decisionArchiveNotificationOutbox.id, pendingIds), - eq(decisionArchiveNotificationOutbox.status, "pending"), - )).returning(); - }); - const groups = new Map(); - for (const row of rows) { - const key = `${row.companyId}:${row.originAgentId}`; - const grouped = groups.get(key) ?? []; - grouped.push(row); - groups.set(key, grouped); - } - let notifiedAgents = 0; - let delivered = 0; - for (const grouped of groups.values()) { - const first = grouped[0]!; - try { - await options.notifyOriginAgent({ - companyId: first.companyId, - agentId: first.originAgentId, - items: grouped.map((row) => ({ - sourceKind: row.sourceKind as AttentionSourceKind, - sourceId: row.sourceId, - archiveVersion: row.archiveVersion, - issueId: row.originIssueId, - })), - }); - const deliveredAt = new Date(); - await db.update(decisionArchiveNotificationOutbox).set({ - status: "delivered", - deliveredAt, - lastAttemptAt: deliveredAt, - updatedAt: deliveredAt, - }).where(and( - inArray(decisionArchiveNotificationOutbox.id, grouped.map((row) => row.id)), - eq(decisionArchiveNotificationOutbox.status, "delivering"), - )); - notifiedAgents += 1; - delivered += grouped.length; - } catch { - await db.update(decisionArchiveNotificationOutbox).set({ - status: "pending", - lastAttemptAt: new Date(), - updatedAt: new Date(), - }).where(and( - inArray(decisionArchiveNotificationOutbox.id, grouped.map((row) => row.id)), - eq(decisionArchiveNotificationOutbox.status, "delivering"), - )); - } - } - return { notifiedAgents, delivered }; - } + const archiveDelivery = decisionArchiveDeliveryService(db, options); return { syncItems, @@ -477,6 +398,7 @@ export function decisionRetentionService( revive: (input: Omit[0], "archived">) => changeArchived({ ...input, archived: false }), archiveReviewedManifest, autoArchive, - deliverNotifications, + deliverNotifications: archiveDelivery.deliver, + nextArchiveNotificationAt: archiveDelivery.next, }; } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 8af5f57babe..230be70dcf8 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -9,6 +9,8 @@ export const DELIVERY_QUEUES = { executionStatus: "execution-status", executionDisposition: "execution-disposition", environmentCleanup: "environment-cleanup", + decisionArchiveDelivery: "decision-archive-delivery", + toolReviewProjection: "tool-review-projection", toolReviewInterruption: "tool-review-interruption", toolReviewRecovery: "tool-review-recovery",