mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 20:21:50 +01:00
fix(server): wake archive notification delivery from committed changes (#17475)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Auto-archive writes an existing notification outbox in its archive transaction. > - The maintenance sweep polls that outbox even when it is empty. > - The committing producer can wake delivery. Saved rows and claim timestamps can restore unfinished work. > - This PR separates delivery from retention scanning and uses the existing coordinator for retries and deadlines. > - Settled delivery becomes quiet while archive and notification behavior remain intact. ## Linked Issues or Issue Description **What happened?** The retention maintenance sweep calls archive notification delivery on every tick. An empty outbox and retained delivered history still cause database work. **What did you expect?** Committed auto-archive should wake delivery. Pending work and existing claims should retain retry or expiry deadlines. Settled history should cause no recurring delivery SQL. **Steps to reproduce** Start a configured instance with an empty or fully delivered archive outbox. Leave it idle. The old maintenance tick still checks the outbox. This is slice 08 of the sleep migration. It remains stacked on projection repair #17468. Retention eligibility and scanning, decision TTL expiry and decision continuation recovery have separate owners and PRs. ## What Changed - Notify archive delivery before the existing outbox insert in the auto-archive transaction. Preserve outer transaction commit, rollback and uncertainty handling. - Register a delivery worker with the existing coordinator. Remove notification delivery from retention sweeps. - Keep claim recovery, selection, company/agent grouping and timestamp-fenced settlement in the delivery service. Keep the existing service API and payload. - Count a notified group only when it settles rows. Use one timestamp for failure reset. - Add generated concurrent migration `0387_boring_professor_monster.sql`. Its partial index matches pending order: `last_attempt_at NULLS FIRST, created_at, id`. Add no table. - Test real startup registration and prove a retention tick does not send notifications. Preserve the disabled-scheduler no-notification contract. ## Verification Current candidate: `457c70fc6acbd681fcccd90d08b6391efb202bce`, normally merged with pinned master `8161f2a813` after projection #17468 landed. All 39 independent files from that master integration are preserved exactly. Archive owner behavior and migration 0387 are unchanged. - **58 archive/startup/scoped-deadline cases pass** after master reconciliation. Server typecheck and build pass. The preceding projection-parent integration passed all **239** gateway/projection/archive/startup/deadline cases. - All 385 parent journal entries remain an exact prefix, with own migration 0387 appended; the five newest generated snapshots remain. Generated index apply, replay, schema drift and real pending-order query measurements passed in the prior owner verification. - Failure accounting: the broader local command also attempted six upstream heartbeat-session-preparation and 55 native control-plane cases. The embedded PostgreSQL runtime was unavailable: the six heartbeat cases skipped, and native setup failed before its 55 cases could execute. This is an incomplete additional local check, not a passing full run. The owning 58 cases passed with the existing external fixture where applicable. CI must exercise its supported embedded environment. - The first independent-master comparison used the shared mutable remote ref, which another worktree advanced during verification. Pinning the actual merged master SHA confirms all 39 independent files match exactly. No incoming source change was dropped. The behavior audit in `doc/idle-decision-archive-delivery.md` records earlier fixture failures, the stale-claim regression and the index probe. Local tests do not prove deployed suspension or fleet savings. Fresh current-head CI and both reviews remain required. ## Risks - Delivery remains at least once. A wake accepted before an acknowledgement is lost can be retried. - Each pass keeps the existing 500-row limit. Failed or remaining pending work retains the existing thirty-second retry. Saved delivery claims retain their five-minute deadline. - Auto-archive is the existing outbox producer. Manual archive/revive and reviewed-manifest actions do not gain notifications. - Signals use the existing single-active-writer database owner. Independent SQL or another active server needs a wake contract. - Concurrent index creation uses the existing top-level executor and invalid-index retry path. Existing SQL and migration receipts are unchanged. - Cloud sleep guards and external-wake coverage do not change. Local worker quiet is not proof of database or host suspension. ## Model Used OpenAI Codex, GPT-6 family, with reasoning and code execution. The exact serving model identifier and context window were not exposed. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge The final projection efficiency parent is integrated normally. All 239 affected gateway/projection, archive, startup and scoped-deadline cases pass, with server typecheck/build. The archive owner behavior and index remain unchanged. Fresh reviews and CI are required on this parent integration. --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
3f8149066c
commit
5785090ff1
14 files changed
+649
-99
No files matched your search
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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",
|
||||
]);
|
||||
|
||||
@@ -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';
|
||||
+111
-2
@@ -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": [
|
||||
+7
@@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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')`,
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
|
||||
let stopWorker: (() => Promise<void>) | 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<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let fixture: Awaited<ReturnType<typeof seedReviewItems>>;
|
||||
let service: ReturnType<typeof decisionRetentionService>;
|
||||
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<void>(resolve => { entered = resolve; });
|
||||
const gate = new Promise<void>(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<void>(resolve => { entered = resolve; });
|
||||
const gate = new Promise<void>(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<void>(resolve => { entered = resolve; });
|
||||
const gate = new Promise<void>(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");
|
||||
|
||||
@@ -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<typeof setInterval>;
|
||||
}) 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 });
|
||||
|
||||
+5
-2
@@ -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);
|
||||
};
|
||||
|
||||
|
||||
@@ -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<typeof createDeliveryWorkCoordinator>,
|
||||
service: Pick<ReturnType<typeof decisionRetentionService>, "deliverNotifications" | "nextArchiveNotificationAt">,
|
||||
canRun: () => boolean,
|
||||
) {
|
||||
return registerDeadlineWork(coordinator, DELIVERY_QUEUES.decisionArchiveDelivery, {
|
||||
retryMs: DECISION_ARCHIVE_RETRY_MS, canRun,
|
||||
run: () => service.deliverNotifications(),
|
||||
next: service.nextArchiveNotificationAt,
|
||||
});
|
||||
}
|
||||
@@ -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<unknown> },
|
||||
) {
|
||||
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<string, typeof rows>();
|
||||
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 };
|
||||
}
|
||||
@@ -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<string, typeof rows>();
|
||||
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<Parameters<typeof changeArchived>[0], "archived">) => changeArchived({ ...input, archived: false }),
|
||||
archiveReviewedManifest,
|
||||
autoArchive,
|
||||
deliverNotifications,
|
||||
deliverNotifications: archiveDelivery.deliver,
|
||||
nextArchiveNotificationAt: archiveDelivery.next,
|
||||
};
|
||||
}
|
||||
@@ -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",
|
||||
|
||||
Reference in new issue
Block a user