From 453d54f1efbc82e2c537bcc664b69d18bdc08c6d Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Tue, 6 Oct 2026 08:04:37 -0700 Subject: [PATCH] Seed the resource baseline before lifecycle delivery Backfill current hired agents and projects in the delivery migration. Preserve existing records and approval gates, seed paused/terminated state, and deliver creation before earlier partial transition histories. Acquire resource locks before journal locks and commit the baseline with its delivery schema. No later journal backfill is needed. Verified targeted database, SDK, worker, tracing and migration checks, workspace typechecks, and the production build. Full local tests are running. Co-Authored-By: Paperclip --- doc/DATABASE.md | 17 +++-- doc/plugins/PLUGIN_AUTHORING_GUIDE.md | 11 ++- .../db/src/migrations/0301_whole_venom.sql | 32 ++++++++- packages/plugins/sdk/src/testing.ts | 8 ++- packages/plugins/sdk/src/types.ts | 2 +- .../sdk/tests/testing-lifecycle.test.ts | 13 ++++ .../resource-lifecycle-events.test.ts | 72 +++++++++++++++++++ server/src/services/plugin-lifecycle-inbox.ts | 4 +- 8 files changed, 146 insertions(+), 13 deletions(-) diff --git a/doc/DATABASE.md b/doc/DATABASE.md index c7458d91450..c3f363bb545 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -511,7 +511,9 @@ Termination commits API-key revocation in that same transaction. Hire approval and rejection commit with agent activation or termination, so a failed event write leaves the decision pending and retryable. -The numeric event ID orders transitions for a resource. Plugin delivery +Creation is delivered first for each resource, including a backfilled creation +whose ID is newer than earlier captured transitions. The remaining events follow +numeric ID order. Plugin delivery must enforce company scope, preserve resource order, and track acknowledgments per plugin. A global high-water mark can skip transactions that have not yet committed; it is not a safe delivery cursor. The journal stores only identity, @@ -521,14 +523,21 @@ retains events, so consumers must revalidate existence and eligibility and load current authorized repository data. A termination hook does not authorize provider cleanup without the plugin's own authorization and retention policy. -The migrations create empty journal and acknowledgment tables. They do not scan or backfill existing -installs. Plugins use `ctx.events.listLifecycle(companyId, limit?, afterId?)` and +Migration `0301_whole_venom.sql` seeds a one-time current-state baseline before +plugin delivery is available. It records creation for existing hired agents and +all projects, including archived projects. Pending hires stay behind approval. +Paused and terminated agents receive missing final status intents. A partial +journal ending at pause receives resume when the current agent is running. +Existing records remain intact, and rerunning the baseline does not duplicate it. +Resource writes wait for the migration transaction to commit. These records +represent current desired state, not reconstructed historical transitions. +There is no later or runtime journal backfill. Plugins use `ctx.events.listLifecycle(companyId, limit?, afterId?)` and `ctx.events.acknowledgeLifecycle(companyId, eventId)` with `events.subscribe`. The host requires a matching company invocation (or configured-company proactive access) and a ready plugin enabled for that company. `plugin_lifecycle_acknowledgments` stores progress independently for each plugin and event; plugin and event deletion cascade acknowledgments. -Reads return only the earliest unacknowledged event for each resource, up to 100 +Reads return creation first, then the earliest unacknowledged transition for each resource, up to 100 resources. Acknowledging a later event is rejected. Reads never consume work, so crashes, retries, and restarts cannot lose a hook; concurrent reads can repeat an event. There is no global cursor or backfill scan. Consumers must serialize their diff --git a/doc/plugins/PLUGIN_AUTHORING_GUIDE.md b/doc/plugins/PLUGIN_AUTHORING_GUIDE.md index 1d401190cc1..6624d556761 100644 --- a/doc/plugins/PLUGIN_AUTHORING_GUIDE.md +++ b/doc/plugins/PLUGIN_AUTHORING_GUIDE.md @@ -43,11 +43,16 @@ approval. Project updates include repository/workspace mutations, but archive-on changes produce no hook. Provider cleanup and retention policy belong to the plugin. Reads return at most one pending event per resource (default 50, maximum 100). -After acknowledging that event, a later read exposes its successor. A failed +Creation is delivered before other events for that resource, even when its +backfilled ID is newer. Remaining events follow ID order. After acknowledging +an event, a later read exposes its successor. A failed resource remains pending without blocking other resources; process the rest of the batch independently. Progress is stored per plugin, survives worker restarts, -and is not a global sequence cursor. New consumers can see previously recorded -events, but no events are synthesized for resources that predate capture. +and is not a global sequence cursor. The delivery migration seeds existing hired +agents and all projects once, including archived projects. Pending hires require +approval; paused and terminated agents retain their current status intents. +This baseline represents current desired state, not reconstructed history. +After that migration, capture stays forward-only; no later journal backfill runs. Delivery is at least once: concurrent reads or a crash after a provider operation can repeat an event. Serialize polling and use stable company/event idempotency diff --git a/packages/db/src/migrations/0301_whole_venom.sql b/packages/db/src/migrations/0301_whole_venom.sql index 25b6d2ef0d0..68273746f92 100644 --- a/packages/db/src/migrations/0301_whole_venom.sql +++ b/packages/db/src/migrations/0301_whole_venom.sql @@ -1,3 +1,5 @@ +-- Lock resource writers before the journal foreign key and index take locks. +LOCK TABLE "agents", "projects" IN SHARE ROW EXCLUSIVE MODE;--> statement-breakpoint CREATE TABLE "plugin_lifecycle_acknowledgments" ( "plugin_id" uuid NOT NULL, "event_id" bigint NOT NULL, @@ -8,4 +10,32 @@ CREATE TABLE "plugin_lifecycle_acknowledgments" ( ALTER TABLE "plugin_lifecycle_acknowledgments" ADD CONSTRAINT "plugin_lifecycle_acknowledgments_plugin_id_plugins_id_fk" FOREIGN KEY ("plugin_id") REFERENCES "public"."plugins"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint ALTER TABLE "plugin_lifecycle_acknowledgments" ADD CONSTRAINT "plugin_lifecycle_acknowledgments_event_id_resource_lifecycle_events_id_fk" FOREIGN KEY ("event_id") REFERENCES "public"."resource_lifecycle_events"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint CREATE INDEX "plugin_lifecycle_acknowledgments_event_idx" ON "plugin_lifecycle_acknowledgments" USING btree ("event_id");--> statement-breakpoint -CREATE INDEX "resource_lifecycle_events_resource_order_idx" ON "resource_lifecycle_events" USING btree ("company_id","resource_type","resource_id","id"); \ No newline at end of file +CREATE INDEX "resource_lifecycle_events_resource_order_idx" ON "resource_lifecycle_events" USING btree ("company_id","resource_type","resource_id","id");--> statement-breakpoint +-- Seed a one-time current-state baseline before plugins can consume the journal. +INSERT INTO "resource_lifecycle_events" ("company_id", "resource_type", "resource_id", "action") +SELECT "company_id", 'agent', "id", 'create' +FROM "agents" +WHERE "status" NOT IN ('pending_approval', 'terminated') +UNION ALL +SELECT "company_id", 'project', "id", 'create' +FROM "projects" +ON CONFLICT ("company_id", "resource_type", "resource_id") WHERE "action" = 'create' DO NOTHING;--> statement-breakpoint +-- Restore current stop/delete intent without repeating an existing final transition. +INSERT INTO "resource_lifecycle_events" ("company_id", "resource_type", "resource_id", "action") +SELECT a."company_id", 'agent', a."id", CASE a."status" WHEN 'paused' THEN 'pause' ELSE 'terminate' END +FROM "agents" a +WHERE a."status" IN ('paused', 'terminated') + AND (SELECT e."action" FROM "resource_lifecycle_events" e + WHERE e."company_id" = a."company_id" AND e."resource_type" = 'agent' + AND e."resource_id" = a."id" AND e."action" <> 'create' + ORDER BY e."id" DESC LIMIT 1) + IS DISTINCT FROM CASE a."status" WHEN 'paused' THEN 'pause' ELSE 'terminate' END;--> statement-breakpoint +-- A partial journal may end at pause even though the current agent is running. +INSERT INTO "resource_lifecycle_events" ("company_id", "resource_type", "resource_id", "action") +SELECT a."company_id", 'agent', a."id", 'resume' +FROM "agents" a +WHERE a."status" NOT IN ('pending_approval', 'paused', 'terminated') + AND (SELECT e."action" FROM "resource_lifecycle_events" e + WHERE e."company_id" = a."company_id" AND e."resource_type" = 'agent' + AND e."resource_id" = a."id" AND e."action" <> 'create' + ORDER BY e."id" DESC LIMIT 1) = 'pause'; diff --git a/packages/plugins/sdk/src/testing.ts b/packages/plugins/sdk/src/testing.ts index 3c6658cd793..8f18c9d59ed 100644 --- a/packages/plugins/sdk/src/testing.ts +++ b/packages/plugins/sdk/src/testing.ts @@ -872,14 +872,15 @@ export function createTestHarness(options: TestHarnessOptions): TestHarness { if (!Number.isInteger(limit) || limit < 1 || limit > 100) throw new Error("limit must be an integer from 1 to 100"); if (afterId !== undefined && (typeof afterId !== "string" || !/^[1-9]\d*$/.test(afterId) || !Number.isSafeInteger(Number(afterId)))) throw new Error("Invalid lifecycle page id"); const pending = [...lifecycleEvents.values()].filter(event => event.companyId === companyId && !lifecycleAcknowledgments.has(event.id)) - .sort((a, b) => Number(a.id) - Number(b.id)); + .sort((a, b) => (a.action === "create" ? 0 : 1) - (b.action === "create" ? 0 : 1) || Number(a.id) - Number(b.id)); const resources = new Set(); return pending.filter(event => { const key = `${event.resourceType}:${event.resourceId}`; if (resources.has(key)) return false; resources.add(key); return true; - }).filter(event => afterId === undefined || Number(event.id) > Number(afterId)).slice(0, limit); + }).filter(event => afterId === undefined || Number(event.id) > Number(afterId)) + .sort((a, b) => Number(a.id) - Number(b.id)).slice(0, limit); }, async acknowledgeLifecycle(companyId, eventId) { requireCapability(manifest, capabilitySet, "events.subscribe"); @@ -888,7 +889,8 @@ export function createTestHarness(options: TestHarnessOptions): TestHarness { if (!event || event.companyId !== companyId) throw new Error("Lifecycle event not found"); const earlier = [...lifecycleEvents.values()].some(other => other.companyId === companyId && other.resourceType === event.resourceType && other.resourceId === event.resourceId - && Number(other.id) < Number(event.id) && !lifecycleAcknowledgments.has(other.id)); + && event.action !== "create" && (other.action === "create" || Number(other.id) < Number(event.id)) + && !lifecycleAcknowledgments.has(other.id)); if (earlier) throw new Error("Acknowledge earlier lifecycle events for this resource first"); lifecycleAcknowledgments.add(eventId); }, diff --git a/packages/plugins/sdk/src/types.ts b/packages/plugins/sdk/src/types.ts index 0c80ce45e16..021ab3b2dfd 100644 --- a/packages/plugins/sdk/src/types.ts +++ b/packages/plugins/sdk/src/types.ts @@ -547,7 +547,7 @@ export interface PluginLocalFoldersClient { */ export interface PluginEventsClient { /** Read durable resource hooks. Requires events.subscribe and a company scope. - * Only the earliest unacknowledged event for each resource is returned. + * Creation is delivered first, then the remaining events in id order. * Page using afterId; reset it each polling sweep to retry failures and late commits. * Events repeat until acknowledged; use a company-scoped provider idempotency key. */ diff --git a/packages/plugins/sdk/tests/testing-lifecycle.test.ts b/packages/plugins/sdk/tests/testing-lifecycle.test.ts index e58f1a671c2..88a33aeb58b 100644 --- a/packages/plugins/sdk/tests/testing-lifecycle.test.ts +++ b/packages/plugins/sdk/tests/testing-lifecycle.test.ts @@ -25,4 +25,17 @@ describe("Lifecycle inbox test harness", () => { await expect(denied.ctx.events.listLifecycle("a")).rejects.toThrow("events.subscribe"); await expect(denied.ctx.events.acknowledgeLifecycle("a", "1")).rejects.toThrow("events.subscribe"); }); + it("delivers a backfilled creation before earlier transitions", async () => { + const harness = createTestHarness({ manifest }); + const event: ResourceLifecycleEvent = { id: "1", companyId: "a", resourceType: "agent", resourceId: "agent", action: "pause", createdAt: new Date(0).toISOString() }; + harness.seed({ lifecycleEvents: [event, { ...event, id: "2", action: "resume" }, { ...event, id: "3", action: "create" }] }); + expect(await harness.ctx.events.listLifecycle("a")).toEqual([expect.objectContaining({ id: "3", action: "create" })]); + await expect(harness.ctx.events.acknowledgeLifecycle("a", "1")).rejects.toThrow("earlier lifecycle events"); + await harness.ctx.events.acknowledgeLifecycle("a", "3"); + expect(await harness.ctx.events.listLifecycle("a", 100, "3")).toEqual([]); + expect(await harness.ctx.events.listLifecycle("a")).toEqual([event]); + await harness.ctx.events.acknowledgeLifecycle("a", "1"); + expect(await harness.ctx.events.listLifecycle("a")).toEqual([expect.objectContaining({ id: "2" })]); + }); + }); diff --git a/server/src/__tests__/resource-lifecycle-events.test.ts b/server/src/__tests__/resource-lifecycle-events.test.ts index ab35f703025..488a1cd5732 100644 --- a/server/src/__tests__/resource-lifecycle-events.test.ts +++ b/server/src/__tests__/resource-lifecycle-events.test.ts @@ -1,4 +1,5 @@ import { randomUUID } from "node:crypto"; +import { readFileSync } from "node:fs"; import { eq, sql } from "drizzle-orm"; import { agentApiKeys, agents, budgetPolicies, companies, costEvents, createDb, pluginCompanySettings, pluginLifecycleAcknowledgments, plugins, projects, resourceLifecycleEvents, type Db } from "@paperclipai/db"; import { createHostClientHandlers } from "@paperclipai/plugin-sdk"; @@ -269,6 +270,77 @@ describePostgres("Resource lifecycle events", () => { } }); + const backfillBaseline = () => db.transaction(async tx => { + const migration = readFileSync(new URL("../../../packages/db/src/migrations/0301_whole_venom.sql", import.meta.url), "utf8"); + const lock = migration.split("--> statement-breakpoint")[0]; + const baseline = migration.slice(migration.indexOf("-- Seed a one-time current-state baseline")); + for (const statement of [lock, ...baseline.split("--> statement-breakpoint")]) { + if (statement.trim()) await tx.execute(sql.raw(statement)); + } + }); + + it("backfills a current-state baseline once without bypassing approval or reordering partial histories", async () => { + const [live, paused, terminated, pending, partial, resumed] = await db.insert(agents).values([ + { companyId, name: "Legacy live", status: "idle" }, + { companyId, name: "Legacy paused", status: "paused" }, + { companyId, name: "Legacy terminated", status: "terminated" }, + { companyId, name: "Legacy pending", status: "pending_approval" }, + { companyId, name: "Partial pause", status: "paused" }, + { companyId, name: "Partial resume", status: "idle" }, + ]).returning(); + const [archived] = await db.insert(projects).values({ companyId, name: "Archived baseline", archivedAt: new Date(0) }).returning(); + const existing = await createAgent(); + const [earlyPause] = await db.insert(resourceLifecycleEvents).values([ + { companyId, resourceType: "agent", resourceId: partial.id, action: "pause" }, + { companyId, resourceType: "agent", resourceId: resumed.id, action: "pause" }, + ]).returning(); + const deletedId = randomUUID(); + const [deleted] = await db.insert(resourceLifecycleEvents).values({ companyId, resourceType: "agent", resourceId: deletedId, action: "terminate" }).returning(); + await backfillBaseline(); + const baseline = (await events()).sort((a, b) => a.id - b.id); + const actions = (id: string) => baseline.filter(e => e.resourceId === id).map(e => e.action); + expect(actions(live.id)).toEqual(["create"]); + expect(actions(paused.id)).toEqual(["create", "pause"]); + expect(actions(terminated.id)).toEqual(["terminate"]); + expect(actions(pending.id)).toEqual([]); + expect(actions(partial.id)).toEqual(["pause", "create"]); + expect(actions(resumed.id)).toEqual(["pause", "create", "resume"]); + expect(actions(archived.id)).toEqual(["create"]); + expect(actions(existing.id)).toEqual(["create"]); + expect(baseline.find(e => e.id === deleted.id)).toEqual(deleted); + const plugin = await createPlugin(); + const inbox = pluginLifecycleInbox(db, plugin.id); + const creation = (await inbox.list(companyId)).find(e => e.resourceId === partial.id)!; + expect(creation.action).toBe("create"); + expect(Number(creation.id)).toBeGreaterThan(earlyPause.id); + await expect(inbox.acknowledge(companyId, String(earlyPause.id))).rejects.toMatchObject({ status: 409 }); + await inbox.acknowledge(companyId, creation.id); + expect((await inbox.list(companyId)).find(e => e.resourceId === partial.id)).toMatchObject({ id: String(earlyPause.id), action: "pause" }); + await backfillBaseline(); + expect((await events()).sort((a, b) => a.id - b.id)).toEqual(baseline); + }); + + it("keeps delivery unavailable when the baseline migration fails", async () => { + await db.insert(agents).values({ companyId, name: "Legacy paused", status: "paused" }); + await db.execute(sql`DROP TABLE plugin_lifecycle_acknowledgments`); + await db.execute(sql`DROP INDEX resource_lifecycle_events_resource_order_idx`); + const applyDeliveryMigration = () => db.transaction(async tx => { + const migration = readFileSync(new URL("../../../packages/db/src/migrations/0301_whole_venom.sql", import.meta.url), "utf8"); + for (const statement of migration.split("--> statement-breakpoint")) { + if (statement.trim()) await tx.execute(sql.raw(statement)); + } + }); + await db.execute(sql`ALTER TABLE resource_lifecycle_events ADD CONSTRAINT fixture_reject_baseline CHECK (action <> 'pause') NOT VALID`); + try { + await expect(applyDeliveryMigration()).rejects.toThrow(); + expect(await events()).toEqual([]); + expect((await db.execute(sql`SELECT to_regclass('public.plugin_lifecycle_acknowledgments') AS relation`))[0]).toMatchObject({ relation: null }); + } finally { + await db.execute(sql`ALTER TABLE resource_lifecycle_events DROP CONSTRAINT fixture_reject_baseline`); + await applyDeliveryMigration(); + } + }); + it("retries until acknowledgment, preserves resource order, and isolates progress per plugin", async () => { const first = await createPlugin(); const second = await createPlugin(); diff --git a/server/src/services/plugin-lifecycle-inbox.ts b/server/src/services/plugin-lifecycle-inbox.ts index 931405e675d..7187d83b069 100644 --- a/server/src/services/plugin-lifecycle-inbox.ts +++ b/server/src/services/plugin-lifecycle-inbox.ts @@ -15,7 +15,9 @@ export function pluginLifecycleInbox(db: Db, pluginId: string) { WHERE older.company_id = ${resourceLifecycleEvents.companyId} AND older.resource_type = ${resourceLifecycleEvents.resourceType} AND older.resource_id = ${resourceLifecycleEvents.resourceId} - AND older.id < ${resourceLifecycleEvents.id} + -- Backfilled creation must precede transitions captured before this baseline. + AND ${resourceLifecycleEvents.action} <> 'create' + AND (older.action = 'create' OR older.id < ${resourceLifecycleEvents.id}) AND NOT EXISTS (SELECT 1 FROM plugin_lifecycle_acknowledgments ack WHERE ack.plugin_id = ${pluginId} AND ack.event_id = older.id) )`;