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) )`;