From 1cebdd4c4f9d8a4cb9e9eb1e71642e23605d04f9 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Tue, 6 Oct 2026 12:27:09 -0700 Subject: [PATCH] Add plugin lifecycle delivery and an initial resource baseline (#15343) ## Thinking Path > - Paperclip manages agents and projects for work. > - Their lifecycle changes already commit to a durable journal. > - Plugins need to read these records and track completed work. > - Existing resources need a one-time baseline before delivery is enabled. > - Worker crashes must leave unfinished records available for retry. > - Each plugin needs its own progress and company access checks. > - This PR adds a pull inbox through the existing plugin SDK and job system. ## Linked Issues or Issue Description **Problem or motivation** Plugins cannot consume the durable resource lifecycle journal. In-process notifications can disappear during a restart and cannot record successful completion. Resource plugins need retries, company boundaries, and ordered transitions. **Proposed solution** Add `ctx.events.listLifecycle(companyId, limit?, afterId?)` and `ctx.events.acknowledgeLifecycle(companyId, eventId)` under `events.subscribe`. Seed a one-time baseline from current resource state. Deliver creation first, then the remaining transitions in ID order. Store acknowledgments for each plugin. Use existing plugin jobs to poll configured companies. **Alternatives considered** A global sequence cursor can skip lower IDs that commit later. Fire-and-forget subscriptions cannot record completion. A new dispatcher is unnecessary because plugin jobs support polling. Consumers serialize polling and use provider idempotency keys. **Roadmap alignment** This extends the existing plugin system and builds on #15280 and #15306. Searches found no duplicate resource inbox work. Related #13306 exposes decision events on the in-process bus. This PR includes the initial journal baseline. Provider provisioning remains separate work. ## What Changed - Add company-scoped lifecycle reads and acknowledgments to the SDK and worker RPC host. - Gate both methods by capability, invocation or proactive company scope, plugin readiness, and company enablement. - Seed hired agents and all projects once. Preserve paused/terminated agent state and archived project state. Keep pending hires behind approval. - Preserve existing history and deliver backfilled creation before partial transition histories. - Deliver project archive events from #15371 through the SDK. Seed archive intents for archived projects and update intents for active projects with a partial archive history. - Store acknowledgments per plugin and reject acknowledgments that skip earlier resource events. - Page past failed resources while retaining their pending records. Reset the page cursor each sweep to include late commits. - Add acknowledgment storage, an index for resource ordering, tests, and authoring guidance. - Preserve native identity definitions and sequence progress in JavaScript backups. Repair journal ID generators lost by older backups before seeding the baseline, without changing existing IDs. ## Verification - `pnpm -r typecheck` passed after the final origin/master rebase. - `pnpm build` passed before the final metadata rebase. The final CI build also passed. - All 46 focused lifecycle, SDK harness/RPC, migration snapshot, and legacy restore checks passed again with migration 0309. - Checks cover retries, per-plugin progress, resource order, capability/company boundaries, baseline idempotency, approval gates, archive delivery, restored projects with partial histories, and atomic migration rollback. - Full CI passed on final commit `0e674b87fc`. One unrelated Cursor fixture hit a 10-second timeout; it passed locally in under one second, and the failed server shard passed on retry. - Greptile scored the final commit 5/5 with no unresolved review threads. - The branch is rebased onto origin/master and is conflict-free. - Earlier full local runs were stopped as scope changed. CI ran the complete repository suite. - `git diff --check` and a local secret/PII scan passed. ## Risks - Apply migration `0309_loving_the_hood.sql` before starting the new server. It creates acknowledgment storage and an index, then seeds the baseline in the same transaction. Agent/project writes wait for the migration to commit. Keep these writes quiesced through the migration and activation of the new capture-capable server; do not resume an older runtime that lacks capture after the baseline. - The baseline records current desired state, not historical transitions. It includes archived projects and terminated-agent cleanup intents. It runs once with the delivery migration; no later or runtime journal backfill is planned. - Delivery is at least once. Concurrent reads can repeat an event. Consumers must serialize polling, use stable company/event idempotency keys, and acknowledge successful operations only. - Reset `afterId` at the start of every polling sweep. It is a page cursor, not a persisted high-water mark. - Deleting a plugin removes its acknowledgment records. A new installation may replay existing journal events. - Records contain identity and action. Consumers must load current authorized data before acting. Cleanup and retention policy belong to the provider plugin. - Provider calls, VM/volume provisioning, and journal retention are outside this PR. ## Model Used OpenAI Codex, based on GPT-6, with reasoning, repository inspection, code execution, and tool use. The exact deployment model ID and context window are not exposed in this session. ## 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 --------- Co-authored-by: Paperclip --- cli/src/__tests__/worktree.test.ts | 24 +- doc/DATABASE.md | 39 ++- doc/plugins/PLUGIN_AUTHORING_GUIDE.md | 43 +++ packages/db/src/backup-lib.test.ts | 28 ++ packages/db/src/backup-lib.ts | 41 ++- .../src/migrations/0309_loving_the_hood.sql | 73 +++++ ...{0304_snapshot.json => 0309_snapshot.json} | 251 +++++++++++++++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/index.ts | 1 + .../plugin_lifecycle_acknowledgments.ts | 12 + .../src/schema/resource_lifecycle_events.ts | 1 + .../plugins/sdk/src/host-client-factory.ts | 6 + packages/plugins/sdk/src/index.ts | 1 + packages/plugins/sdk/src/protocol.ts | 3 + packages/plugins/sdk/src/testing.ts | 33 +++ packages/plugins/sdk/src/types.ts | 18 ++ packages/plugins/sdk/src/worker-rpc-host.ts | 6 + .../sdk/tests/testing-lifecycle.test.ts | 52 ++++ .../plugins/sdk/tests/worker-rpc-host.test.ts | 42 +++ .../__tests__/plugin-worker-manager.test.ts | 25 +- .../resource-lifecycle-events.test.ts | 178 ++++++++++++- server/src/services/plugin-host-services.ts | 8 + server/src/services/plugin-lifecycle-inbox.ts | 58 ++++ 23 files changed, 930 insertions(+), 20 deletions(-) create mode 100644 packages/db/src/migrations/0309_loving_the_hood.sql rename packages/db/src/migrations/meta/{0304_snapshot.json => 0309_snapshot.json} (99%) create mode 100644 packages/db/src/schema/plugin_lifecycle_acknowledgments.ts create mode 100644 packages/plugins/sdk/tests/testing-lifecycle.test.ts create mode 100644 server/src/services/plugin-lifecycle-inbox.ts diff --git a/cli/src/__tests__/worktree.test.ts b/cli/src/__tests__/worktree.test.ts index a2164aa7e4e..5a6ef211675 100644 --- a/cli/src/__tests__/worktree.test.ts +++ b/cli/src/__tests__/worktree.test.ts @@ -1674,7 +1674,19 @@ describe("worktree helpers", () => { } const sourceDbClient = createDb(sourceDb.connectionString); await migrate(drizzle(sourceDbClient.$client), { migrationsFolder: priorMigrations }); - await seedValidWorktreeSource(sourceDb.connectionString); + const seed = await seedValidWorktreeSource(sourceDb.connectionString); + // An older filtered JavaScript backup retained event IDs but lost the + // identity generator. The pending migration must repair that schema. + const legacyAgentId = randomUUID(); + await sourceDbClient.$client` + INSERT INTO agents (id, company_id, name, status) + VALUES (${legacyAgentId}, ${seed.companyId}, 'Legacy paused agent', 'paused') + `; + await sourceDbClient.$client` + INSERT INTO resource_lifecycle_events (company_id, resource_type, resource_id, action) + VALUES (${seed.companyId}, 'agent', ${legacyAgentId}, 'pause') + `; + await sourceDbClient.$client.unsafe('ALTER TABLE resource_lifecycle_events ALTER COLUMN id DROP IDENTITY'); await sourceDbClient.$client.unsafe(` WITH pair AS ( SELECT @@ -1787,6 +1799,16 @@ describe("worktree helpers", () => { ); const seededUsers = await targetDb.select().from(authUsers); expect(seededUsers.some((row) => row.email === "existing@paperclip.ing")).toBe(true); + const restoredEvents = await targetDb.$client` + SELECT id, action FROM resource_lifecycle_events WHERE resource_id = ${legacyAgentId} ORDER BY id + `; + expect(restoredEvents.map(row => row.action)).toEqual(["pause", "create"]); + expect(Number(restoredEvents[1].id)).toBeGreaterThan(Number(restoredEvents[0].id)); + const [identity] = await targetDb.$client` + SELECT is_identity FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = 'resource_lifecycle_events' AND column_name = 'id' + `; + expect(identity.is_identity).toBe("YES"); }, ); diff --git a/doc/DATABASE.md b/doc/DATABASE.md index c05b9d6d3aa..e2cb9a9d33b 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -515,7 +515,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. Future 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, @@ -523,11 +525,38 @@ action, and timestamps, not repository snapshots, credentials, provider config, or resource health. Company deletion cascades to its events. Resource deletion retains events, so consumers must revalidate existence and eligibility and load current authorized repository data. A termination hook does not authorize -removing persistent VM or project data. +provider cleanup without the plugin's own authorization and retention policy. -The migration creates an empty table. It does not scan or backfill existing -installs. Plugin delivery, retention, retries, and provider integration are -separate work. This change makes no provider calls and adds no plugin read API. +Migration `0309_loving_the_hood.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. +Archived projects receive missing archive intents. A partial journal ending at +archive receives update when the current project is active. +Existing records remain intact, and rerunning the baseline does not duplicate it. +The migration also repairs the journal ID generator in older JavaScript restores +that lost identity metadata, starting above existing IDs. New JavaScript backups +preserve identity generation, sequence options, and sequence progress. +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 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 runtime backfill scan. Consumers must serialize their +processing and make provider operations idempotent before acknowledging success. +Retention and provider integration remain separate work. + +Lifecycle polls can page past failed resources using the last returned event id as +`afterId`. Reset `afterId` at the start of every polling sweep: it is a page +cursor, never a persisted high-water mark. This retries failures and includes +transactions that commit later with lower ids. ## Legacy controller ownership diff --git a/doc/plugins/PLUGIN_AUTHORING_GUIDE.md b/doc/plugins/PLUGIN_AUTHORING_GUIDE.md index 2d18119ea60..a3728720e1c 100644 --- a/doc/plugins/PLUGIN_AUTHORING_GUIDE.md +++ b/doc/plugins/PLUGIN_AUTHORING_GUIDE.md @@ -25,6 +25,49 @@ It is intentionally narrower than [PLUGIN_SPEC.md](./PLUGIN_SPEC.md). The spec i building custom versions. - `ctx.assets` is not supported in the current runtime. +## Durable resource lifecycle inbox + +Plugins with `events.subscribe` can read durable, company-scoped resource hooks +through `ctx.events.listLifecycle(companyId, limit?, afterId?)` and acknowledge successful +work with `ctx.events.acknowledgeLifecycle(companyId, eventId)`. Use an existing +plugin job to poll each configured company; the host checks invocation scope and whether +the plugin is ready and enabled for that company. These methods are separate +from the fire-and-forget `ctx.events.on()` bus. +Proactive jobs and timers may read only companies authorized by the plugin's +company configuration. Calls inside a host-issued invocation must match its company. + +An event has `id`, `companyId`, `resourceType`, `resourceId`, `action`, and +`createdAt`. Agent actions are `create`, `pause`, `resume`, and `terminate`; +project actions are `create`, `update`, and `archive`. Pending hires produce creation +after approval. Project updates include repository/workspace mutations and restoring +an archived project. Provider cleanup and retention policy belong to the plugin. + +Reads return at most one pending event per resource (default 50, maximum 100). +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. 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. +The baseline includes archive for existing archived projects and update for restored projects whose +journal still ends at archive. Archiving does not authorize provider cleanup. +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 +keys, then acknowledge only after successful completion. Load current authorized +agent/project/workspace data before acting; the journal contains no configuration +snapshots, repository credentials, or deletion authority. For offline plugin tests, +seed `lifecycleEvents` with `createTestHarness().seed()`. + +Lifecycle polls can page past failed resources using the last returned event id as +`afterId`. Reset `afterId` at the start of every polling sweep: it is a page +cursor, never a persisted high-water mark. This retries failures and includes +transactions that commit later with lower ids. + ## External object reference providers Plugins can contribute provider-neutral object reference detection and status diff --git a/packages/db/src/backup-lib.test.ts b/packages/db/src/backup-lib.test.ts index 13c5edc9d8a..b6851534991 100644 --- a/packages/db/src/backup-lib.test.ts +++ b/packages/db/src/backup-lib.test.ts @@ -75,6 +75,34 @@ describe("createBufferedTextFileWriter", () => { }); describeEmbeddedPostgres("runDatabaseBackup", () => { + it("preserves identity generation, sequence options and progress in JavaScript backups", async () => { + const source = await createTempDatabase(); + const target = await createSiblingDatabase(source, "identity_restore_target"); + const sourceSql = postgres(source, { max: 1, onnotice: () => {} }); + const targetSql = postgres(target, { max: 1, onnotice: () => {} }); + try { + await sourceSql.unsafe(` + CREATE SCHEMA identity_fixture; + CREATE TABLE identity_fixture.records ( + id bigint GENERATED ALWAYS AS IDENTITY (START WITH 7 INCREMENT BY 2) PRIMARY KEY, + default_id integer GENERATED BY DEFAULT AS IDENTITY, + value text NOT NULL + ); + INSERT INTO identity_fixture.records (value) VALUES ('one'), ('two'); + `); + const backup = await runDatabaseBackup({ connectionString: source, backupDir: createTempDir("identity-backup-"), backupEngine: "javascript", retention: { dailyDays: 7, weeklyWeeks: 4, monthlyMonths: 1 } }); + await runDatabaseRestore({ connectionString: target, backupFile: backup.backupFile }); + expect(await targetSql`SELECT id::text, default_id, value FROM identity_fixture.records ORDER BY id`) + .toEqual([{ id: "7", default_id: 1, value: "one" }, { id: "9", default_id: 2, value: "two" }]); + expect(await targetSql`INSERT INTO identity_fixture.records (value) VALUES ('three') RETURNING id::text, default_id`) + .toEqual([{ id: "11", default_id: 3 }]); + await expect(targetSql`INSERT INTO identity_fixture.records (id, value) VALUES (99, 'explicit')`).rejects.toThrow("non-DEFAULT"); + } finally { + await sourceSql.end(); + await targetSql.end(); + } + }, 30_000); + it( "keeps the newest backup for each retained calendar month", async () => { diff --git a/packages/db/src/backup-lib.ts b/packages/db/src/backup-lib.ts index f066f22e235..a95ac3f1be8 100644 --- a/packages/db/src/backup-lib.ts +++ b/packages/db/src/backup-lib.ts @@ -43,6 +43,7 @@ export type RunDatabaseRestoreOptions = { }; type SequenceDefinition = { + is_identity: boolean; sequence_schema: string; sequence_name: string; data_type: string; @@ -621,6 +622,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise const allSequences = await sql` SELECT + false AS is_identity, s.sequence_schema, s.sequence_name, s.data_type, @@ -642,7 +644,26 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise WHERE ${sql.unsafe(nonSystemSchemaPredicate("s.sequence_schema"))} ORDER BY s.sequence_schema, s.sequence_name `; - const sequences = allSequences.filter( + const identitySequences = await sql` + SELECT true AS is_identity, n.nspname AS sequence_schema, seq.relname AS sequence_name, + format_type(options.seqtypid, NULL) AS data_type, + options.seqstart::text AS start_value, options.seqmin::text AS minimum_value, + options.seqmax::text AS maximum_value, options.seqincrement::text AS increment, + CASE WHEN options.seqcycle THEN 'YES' ELSE 'NO' END AS cycle_option, + owner_namespace.nspname AS owner_schema, owner.relname AS owner_table, + attribute.attname AS owner_column + FROM pg_sequence options + JOIN pg_class seq ON seq.oid = options.seqrelid + JOIN pg_namespace n ON n.oid = seq.relnamespace + JOIN pg_depend dependency ON dependency.objid = seq.oid AND dependency.deptype = 'i' + AND dependency.classid = 'pg_class'::regclass AND dependency.refclassid = 'pg_class'::regclass + JOIN pg_class owner ON owner.oid = dependency.refobjid + JOIN pg_namespace owner_namespace ON owner_namespace.oid = owner.relnamespace + JOIN pg_attribute attribute ON attribute.attrelid = owner.oid AND attribute.attnum = dependency.refobjsubid + WHERE ${sql.unsafe(nonSystemSchemaPredicate("n.nspname"))} + ORDER BY n.nspname, seq.relname + `; + const sequences = [...allSequences, ...identitySequences].filter( (seq) => !seq.owner_table || includedTableNames.has(tableKey(seq.owner_schema ?? "public", seq.owner_table)), ); @@ -684,7 +705,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise if (sequences.length > 0) { emit("-- Sequences"); - for (const seq of sequences) { + for (const seq of sequences.filter(seq => !seq.is_identity)) { const qualifiedSequenceName = quoteQualifiedName(seq.sequence_schema, seq.sequence_name); emitStatement(`DROP SEQUENCE IF EXISTS ${qualifiedSequenceName} CASCADE;`); emitStatement( @@ -704,11 +725,13 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise udt_name: string; is_nullable: string; column_default: string | null; + is_identity: string; + identity_generation: "ALWAYS" | "BY DEFAULT" | null; character_maximum_length: number | null; numeric_precision: number | null; numeric_scale: number | null; }[]>` - SELECT column_name, data_type, udt_schema, udt_name, is_nullable, column_default, + SELECT column_name, data_type, udt_schema, udt_name, is_nullable, column_default, is_identity, identity_generation, character_maximum_length, numeric_precision, numeric_scale FROM information_schema.columns WHERE table_schema = ${schema_name} AND table_name = ${tablename} @@ -742,7 +765,12 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise } let def = ` "${col.column_name}" ${typeStr}`; - if (col.column_default != null) def += ` DEFAULT ${col.column_default}`; + if (col.is_identity === "YES") { + const sequence = sequences.find(seq => seq.is_identity && seq.owner_schema === schema_name + && seq.owner_table === tablename && seq.owner_column === col.column_name); + if (!sequence || !col.identity_generation) throw new Error(`Missing identity sequence for ${schema_name}.${tablename}.${col.column_name}`); + def += ` GENERATED ${col.identity_generation} AS IDENTITY (SEQUENCE NAME ${quoteQualifiedName(sequence.sequence_schema, sequence.sequence_name)} START WITH ${sequence.start_value} INCREMENT BY ${sequence.increment} MINVALUE ${sequence.minimum_value} MAXVALUE ${sequence.maximum_value} ${sequence.cycle_option === "YES" ? "CYCLE" : "NO CYCLE"})`; + } else if (col.column_default != null) def += ` DEFAULT ${col.column_default}`; if (col.is_nullable === "NO") def += " NOT NULL"; colDefs.push(def); } @@ -770,7 +798,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise emit(""); } - const ownedSequences = sequences.filter((seq) => seq.owner_table && seq.owner_column); + const ownedSequences = sequences.filter((seq) => !seq.is_identity && seq.owner_table && seq.owner_column); if (ownedSequences.length > 0) { emit("-- Sequence ownership"); for (const seq of ownedSequences) { @@ -919,6 +947,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise for (const { schema_name, tablename } of tables) { const currentTableKey = tableKey(schema_name, tablename); const qualifiedTableName = quoteQualifiedName(schema_name, tablename); + const overrideIdentity = identitySequences.some(seq => seq.owner_schema === schema_name && seq.owner_table === tablename); const count = await sql.unsafe<{ n: number }[]>(`SELECT count(*)::int AS n FROM ${qualifiedTableName}`); if (excludedTableNames.has(currentTableKey) || (count[0]?.n ?? 0) === 0) continue; @@ -963,7 +992,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise const values = row.map((rawValue, index) => formatSqlValue(rawValue, cols[index]?.column_name, nullifiedColumns, cols[index]?.data_type), ); - emitStatement(`INSERT INTO ${qualifiedTableName} (${colNames}) VALUES (${values.join(", ")});`); + emitStatement(`INSERT INTO ${qualifiedTableName} (${colNames})${overrideIdentity ? " OVERRIDING SYSTEM VALUE" : ""} VALUES (${values.join(", ")});`); } await writer.drain(); } diff --git a/packages/db/src/migrations/0309_loving_the_hood.sql b/packages/db/src/migrations/0309_loving_the_hood.sql new file mode 100644 index 00000000000..d03f30a1ac2 --- /dev/null +++ b/packages/db/src/migrations/0309_loving_the_hood.sql @@ -0,0 +1,73 @@ +-- 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, + "acknowledged_at" timestamp with time zone DEFAULT now() NOT NULL, + CONSTRAINT "plugin_lifecycle_acknowledgments_plugin_id_event_id_pk" PRIMARY KEY("plugin_id","event_id") +); +--> statement-breakpoint +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");--> statement-breakpoint +-- Older JavaScript backups lost identity metadata. Repair before seeding records. +DO $$ +DECLARE next_id bigint; +BEGIN + IF EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = 'resource_lifecycle_events' + AND column_name = 'id' AND is_identity = 'NO' AND column_default IS NULL + ) THEN + SELECT COALESCE(MAX("id"), 0) + 1 INTO next_id FROM "resource_lifecycle_events"; + EXECUTE format('ALTER TABLE "resource_lifecycle_events" ALTER COLUMN "id" ADD GENERATED ALWAYS AS IDENTITY (START WITH %s)', next_id); + END IF; +END $$;--> 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'; +--> statement-breakpoint +-- Preserve project archive state, including resources created before capture. +INSERT INTO "resource_lifecycle_events" ("company_id", "resource_type", "resource_id", "action") +SELECT p."company_id", 'project', p."id", 'archive' +FROM "projects" p +WHERE p."archived_at" IS NOT NULL + AND (SELECT e."action" FROM "resource_lifecycle_events" e + WHERE e."company_id" = p."company_id" AND e."resource_type" = 'project' + AND e."resource_id" = p."id" AND e."action" <> 'create' + ORDER BY e."id" DESC LIMIT 1) IS DISTINCT FROM 'archive';--> statement-breakpoint +-- A partial journal may end at archive even though the project was restored. +INSERT INTO "resource_lifecycle_events" ("company_id", "resource_type", "resource_id", "action") +SELECT p."company_id", 'project', p."id", 'update' +FROM "projects" p +WHERE p."archived_at" IS NULL + AND (SELECT e."action" FROM "resource_lifecycle_events" e + WHERE e."company_id" = p."company_id" AND e."resource_type" = 'project' + AND e."resource_id" = p."id" AND e."action" <> 'create' + ORDER BY e."id" DESC LIMIT 1) = 'archive'; diff --git a/packages/db/src/migrations/meta/0304_snapshot.json b/packages/db/src/migrations/meta/0309_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0304_snapshot.json rename to packages/db/src/migrations/meta/0309_snapshot.json index 394e357e8f9..73605603da1 100644 --- a/packages/db/src/migrations/meta/0304_snapshot.json +++ b/packages/db/src/migrations/meta/0309_snapshot.json @@ -1,6 +1,6 @@ { - "id": "0a7e411a-1041-42e2-92d7-424a90683f5f", - "prevId": "300dc8d4-29f3-4eb5-8498-3ecc69e4ea0a", + "id": "593431cc-d6a8-4daf-a3e4-9e3c892d9419", + "prevId": "00d98f2c-c054-4ca3-91eb-08211bdf418d", "version": "7", "dialect": "postgresql", "tables": { @@ -829,6 +829,83 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.agent_identity_keys": { + "name": "agent_identity_keys", + "schema": "", + "columns": { + "agent_id": { + "name": "agent_id", + "type": "uuid", + "primaryKey": true, + "notNull": true + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "algorithm": { + "name": "algorithm", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "key_id": { + "name": "key_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "public_key_pem": { + "name": "public_key_pem", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "private_key_material": { + "name": "private_key_material", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "agent_identity_keys_owner_fk": { + "name": "agent_identity_keys_owner_fk", + "tableFrom": "agent_identity_keys", + "tableTo": "agents", + "columnsFrom": [ + "company_id", + "agent_id" + ], + "columnsTo": [ + "company_id", + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": { + "agent_identity_keys_algorithm_check": { + "name": "agent_identity_keys_algorithm_check", + "value": "\"agent_identity_keys\".\"algorithm\" = 'Ed25519'" + } + }, + "isRLSEnabled": false + }, "public.agent_instruction_heads": { "name": "agent_instruction_heads", "schema": "", @@ -2894,7 +2971,7 @@ "checkConstraints": { "ai_connection_defaults_provider_check": { "name": "ai_connection_defaults_provider_check", - "value": "\"ai_connection_defaults\".\"provider\" in ('anthropic','openai','openrouter','xai')" + "value": "\"ai_connection_defaults\".\"provider\" in ('anthropic','openai','openrouter','xai','google')" }, "ai_connection_defaults_method_check": { "name": "ai_connection_defaults_method_check", @@ -3367,7 +3444,7 @@ "checkConstraints": { "ai_provider_defaults_provider_check": { "name": "ai_provider_defaults_provider_check", - "value": "\"ai_provider_defaults\".\"provider\" in ('anthropic','openai','openrouter','xai')" + "value": "\"ai_provider_defaults\".\"provider\" in ('anthropic','openai','openrouter','xai','google')" } }, "isRLSEnabled": false @@ -37125,6 +37202,89 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.plugin_lifecycle_acknowledgments": { + "name": "plugin_lifecycle_acknowledgments", + "schema": "", + "columns": { + "plugin_id": { + "name": "plugin_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "event_id": { + "name": "event_id", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "acknowledged_at": { + "name": "acknowledged_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "plugin_lifecycle_acknowledgments_event_idx": { + "name": "plugin_lifecycle_acknowledgments_event_idx", + "columns": [ + { + "expression": "event_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "plugin_lifecycle_acknowledgments_plugin_id_plugins_id_fk": { + "name": "plugin_lifecycle_acknowledgments_plugin_id_plugins_id_fk", + "tableFrom": "plugin_lifecycle_acknowledgments", + "tableTo": "plugins", + "columnsFrom": [ + "plugin_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "plugin_lifecycle_acknowledgments_event_id_resource_lifecycle_events_id_fk": { + "name": "plugin_lifecycle_acknowledgments_event_id_resource_lifecycle_events_id_fk", + "tableFrom": "plugin_lifecycle_acknowledgments", + "tableTo": "resource_lifecycle_events", + "columnsFrom": [ + "event_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "plugin_lifecycle_acknowledgments_plugin_id_event_id_pk": { + "name": "plugin_lifecycle_acknowledgments_plugin_id_event_id_pk", + "columns": [ + "plugin_id", + "event_id" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.plugin_logs": { "name": "plugin_logs", "schema": "", @@ -39383,6 +39543,39 @@ "concurrently": false, "method": "btree", "with": {} + }, + "resource_lifecycle_events_resource_order_idx": { + "name": "resource_lifecycle_events_resource_order_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "resource_type", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "resource_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -39410,7 +39603,7 @@ }, "resource_lifecycle_events_action_check": { "name": "resource_lifecycle_events_action_check", - "value": "\"resource_lifecycle_events\".\"action\" = 'create' OR (\"resource_lifecycle_events\".\"resource_type\" = 'project' AND \"resource_lifecycle_events\".\"action\" = 'update') OR (\"resource_lifecycle_events\".\"resource_type\" = 'agent' AND \"resource_lifecycle_events\".\"action\" IN ('pause', 'resume', 'terminate'))" + "value": "\"resource_lifecycle_events\".\"action\" = 'create' OR (\"resource_lifecycle_events\".\"resource_type\" = 'project' AND \"resource_lifecycle_events\".\"action\" IN ('update', 'archive')) OR (\"resource_lifecycle_events\".\"resource_type\" = 'agent' AND \"resource_lifecycle_events\".\"action\" IN ('pause', 'resume', 'terminate'))" } }, "isRLSEnabled": false @@ -52024,6 +52217,54 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.mcp_oauth_metadata_admissions": { + "name": "mcp_oauth_metadata_admissions", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "source_hash": { + "name": "source_hash", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "mcp_oauth_metadata_admissions_expiry_idx": { + "name": "mcp_oauth_metadata_admissions_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.mcp_oauth_requests": { "name": "mcp_oauth_requests", "schema": "", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 5daf7757de5..91b0d3cffc2 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2150,6 +2150,13 @@ "when": 1791312251106, "tag": "0308_whole_steel_serpent", "breakpoints": true + }, + { + "idx": 309, + "version": "7", + "when": 1791313098905, + "tag": "0309_loving_the_hood", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/index.ts b/packages/db/src/schema/index.ts index 0ef185057b9..53f42b9e8c5 100644 --- a/packages/db/src/schema/index.ts +++ b/packages/db/src/schema/index.ts @@ -201,6 +201,7 @@ export { pluginLogs } from "./plugin_logs.js"; export { runIdentityContexts } from "./run_identity_contexts.js"; export { connectionIntentDeliveries } from "./connection_intent_deliveries.js"; export { resourceLifecycleEvents } from "./resource_lifecycle_events.js"; +export { pluginLifecycleAcknowledgments } from "./plugin_lifecycle_acknowledgments.js"; export { toolActionDeliveries } from "./tool_action_deliveries.js"; export { chatTeamsFileTransfers } from "./chat_teams_file_transfers.js"; diff --git a/packages/db/src/schema/plugin_lifecycle_acknowledgments.ts b/packages/db/src/schema/plugin_lifecycle_acknowledgments.ts new file mode 100644 index 00000000000..beffa3d3d73 --- /dev/null +++ b/packages/db/src/schema/plugin_lifecycle_acknowledgments.ts @@ -0,0 +1,12 @@ +import { bigint, index, pgTable, primaryKey, timestamp, uuid } from "drizzle-orm/pg-core"; +import { plugins } from "./plugins.js"; +import { resourceLifecycleEvents } from "./resource_lifecycle_events.js"; + +export const pluginLifecycleAcknowledgments = pgTable("plugin_lifecycle_acknowledgments", { + pluginId: uuid("plugin_id").notNull().references(() => plugins.id, { onDelete: "cascade" }), + eventId: bigint("event_id", { mode: "number" }).notNull().references(() => resourceLifecycleEvents.id, { onDelete: "cascade" }), + acknowledgedAt: timestamp("acknowledged_at", { withTimezone: true }).notNull().defaultNow(), +}, table => ({ + pk: primaryKey({ columns: [table.pluginId, table.eventId] }), + eventIdx: index("plugin_lifecycle_acknowledgments_event_idx").on(table.eventId), +})); diff --git a/packages/db/src/schema/resource_lifecycle_events.ts b/packages/db/src/schema/resource_lifecycle_events.ts index 6eb56240d09..cc0747373a3 100644 --- a/packages/db/src/schema/resource_lifecycle_events.ts +++ b/packages/db/src/schema/resource_lifecycle_events.ts @@ -13,6 +13,7 @@ export const resourceLifecycleEvents = pgTable("resource_lifecycle_events", { }, (table) => ({ resourceIdx: uniqueIndex("resource_lifecycle_events_creation_idx").on(table.companyId, table.resourceType, table.resourceId).where(sql`${table.action} = 'create'`), companyIdx: index("resource_lifecycle_events_company_idx").on(table.companyId, table.id), + resourceOrderIdx: index("resource_lifecycle_events_resource_order_idx").on(table.companyId, table.resourceType, table.resourceId, table.id), resourceTypeCheck: check("resource_lifecycle_events_resource_type_check", sql`${table.resourceType} IN ('agent', 'project')`), actionCheck: check("resource_lifecycle_events_action_check", sql`${table.action} = 'create' OR (${table.resourceType} = 'project' AND ${table.action} IN ('update', 'archive')) OR (${table.resourceType} = 'agent' AND ${table.action} IN ('pause', 'resume', 'terminate'))`), })); diff --git a/packages/plugins/sdk/src/host-client-factory.ts b/packages/plugins/sdk/src/host-client-factory.ts index f0c4db2de77..ce90ceeaf71 100644 --- a/packages/plugins/sdk/src/host-client-factory.ts +++ b/packages/plugins/sdk/src/host-client-factory.ts @@ -139,6 +139,8 @@ export interface HostServices { /** Provides `events.emit` and `events.subscribe`. */ events: { + listLifecycle(params: WorkerToHostMethods["events.listLifecycle"][0]): Promise; + acknowledgeLifecycle(params: WorkerToHostMethods["events.acknowledgeLifecycle"][0]): Promise; emit(params: WorkerToHostMethods["events.emit"][0]): Promise; subscribe(params: WorkerToHostMethods["events.subscribe"][0]): Promise; }; @@ -405,6 +407,8 @@ const METHOD_CAPABILITY_MAP: Record services.events.listLifecycle(params)), + "events.acknowledgeLifecycle": gated("events.acknowledgeLifecycle", async (params) => services.events.acknowledgeLifecycle(params)), "events.emit": gated("events.emit", async (params) => { return services.events.emit(params); }), diff --git a/packages/plugins/sdk/src/index.ts b/packages/plugins/sdk/src/index.ts index 57f86281178..b834be1a600 100644 --- a/packages/plugins/sdk/src/index.ts +++ b/packages/plugins/sdk/src/index.ts @@ -238,6 +238,7 @@ export type { PluginLocalFolderListing, PluginLocalFoldersClient, PluginEventsClient, + ResourceLifecycleEvent, PluginJobsClient, PluginLaunchersClient, PluginHttpClient, diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index 9c08a45c42b..76f0cd42284 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -54,6 +54,7 @@ export type { PluginLauncherRenderContextSnapshot } from "@paperclipai/shared"; import type { PluginEvent, + ResourceLifecycleEvent, PluginIssueCheckoutOwnership, PluginIssueOrchestrationSummary, PluginIssueRelationSummary, @@ -1642,6 +1643,8 @@ export interface WorkerToHostMethods { ]; // Events + "events.listLifecycle": [params: { companyId: string; limit?: number; afterId?: string }, result: ResourceLifecycleEvent[]]; + "events.acknowledgeLifecycle": [params: { companyId: string; eventId: string }, result: void]; "events.emit": [ params: { name: string; companyId: string; payload: unknown }, result: void, diff --git a/packages/plugins/sdk/src/testing.ts b/packages/plugins/sdk/src/testing.ts index 35d23c68072..8f18c9d59ed 100644 --- a/packages/plugins/sdk/src/testing.ts +++ b/packages/plugins/sdk/src/testing.ts @@ -30,6 +30,7 @@ import type { PluginJobContext, PluginLauncherRegistration, PluginEvent, + ResourceLifecycleEvent, ScopeKey, ToolResult, ToolRunContext, @@ -108,6 +109,7 @@ export interface TestHarness { /** Seed host entities for `ctx.companies/projects/issues/agents/goals/access/authorization` reads. */ seed(input: { companies?: Company[]; + lifecycleEvents?: ResourceLifecycleEvent[]; projects?: Project[]; issues?: Issue[]; issueComments?: IssueComment[]; @@ -495,6 +497,8 @@ export function createTestHarness(options: TestHarnessOptions): TestHarness { const entities = new Map(); const entityExternalIndex = new Map(); const companies = new Map(); + const lifecycleEvents = new Map(); + const lifecycleAcknowledgments = new Set(); const projects = new Map(); const routines = new Map(); const routineRuns = new Map(); @@ -862,6 +866,34 @@ export function createTestHarness(options: TestHarnessOptions): TestHarness { }, }, events: { + async listLifecycle(companyId, limit = 50, afterId) { + requireCapability(manifest, capabilitySet, "events.subscribe"); + requireCompanyId(companyId); + 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) => (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)) + .sort((a, b) => Number(a.id) - Number(b.id)).slice(0, limit); + }, + async acknowledgeLifecycle(companyId, eventId) { + requireCapability(manifest, capabilitySet, "events.subscribe"); + requireCompanyId(companyId); + const event = lifecycleEvents.get(eventId); + 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 + && 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); + }, on(name: PluginEventType | `plugin.${string}`, filterOrFn: EventFilter | ((event: PluginEvent) => Promise), maybeFn?: (event: PluginEvent) => Promise): () => void { requireCapability(manifest, capabilitySet, "events.subscribe"); let registration: EventRegistration; @@ -2539,6 +2571,7 @@ export function createTestHarness(options: TestHarnessOptions): TestHarness { const harness: TestHarness = { ctx, seed(input) { + for (const row of input.lifecycleEvents ?? []) lifecycleEvents.set(row.id, row); for (const row of input.companies ?? []) companies.set(row.id, row); for (const row of input.projects ?? []) projects.set(row.id, row); for (const row of input.issues ?? []) { diff --git a/packages/plugins/sdk/src/types.ts b/packages/plugins/sdk/src/types.ts index 55d0d0f850e..3378d248c4d 100644 --- a/packages/plugins/sdk/src/types.ts +++ b/packages/plugins/sdk/src/types.ts @@ -546,6 +546,14 @@ export interface PluginLocalFoldersClient { * @see PLUGIN_SPEC.md §16 — Event System */ export interface PluginEventsClient { + /** Read durable resource hooks. Requires events.subscribe and a company scope. + * 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. + */ + listLifecycle(companyId: string, limit?: number, afterId?: string): Promise; + /** Acknowledge only after the provider operation succeeds. */ + acknowledgeLifecycle(companyId: string, eventId: string): Promise; /** * Subscribe to a core Paperclip domain event or a plugin-namespaced event. * @@ -583,6 +591,16 @@ export interface PluginEventsClient { emit(name: string, companyId: string, payload: unknown): Promise; } +export type ResourceLifecycleEvent = { + id: string; + companyId: string; + resourceId: string; + createdAt: string; +} & ( + | { resourceType: "agent"; action: "create" | "pause" | "resume" | "terminate" } + | { resourceType: "project"; action: "create" | "update" | "archive" } +); + /** * `ctx.jobs` — register handlers for scheduled jobs declared in the manifest. * diff --git a/packages/plugins/sdk/src/worker-rpc-host.ts b/packages/plugins/sdk/src/worker-rpc-host.ts index 0f0d4eb4cfd..6684f8fdd18 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -525,6 +525,12 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost }, events: { + async listLifecycle(companyId: string, limit?: number, afterId?: string) { + return callHost("events.listLifecycle", { companyId, limit, afterId }); + }, + async acknowledgeLifecycle(companyId: string, eventId: string) { + await callHost("events.acknowledgeLifecycle", { companyId, eventId }); + }, on( name: string, filterOrFn: EventFilter | ((event: PluginEvent) => Promise), diff --git a/packages/plugins/sdk/tests/testing-lifecycle.test.ts b/packages/plugins/sdk/tests/testing-lifecycle.test.ts new file mode 100644 index 00000000000..dc8963f6bfa --- /dev/null +++ b/packages/plugins/sdk/tests/testing-lifecycle.test.ts @@ -0,0 +1,52 @@ +import { describe, expect, it } from "vitest"; +import { createTestHarness } from "../src/testing.js"; +import type { PaperclipPluginManifestV1, ResourceLifecycleEvent } from "../src/types.js"; + +const manifest: PaperclipPluginManifestV1 = { + id: "fixture.lifecycle", apiVersion: 1, version: "1.0.0", displayName: "Lifecycle", description: "Fixture", + author: "Tests", categories: ["automation"], capabilities: ["events.subscribe"], entrypoints: {}, +}; + +describe("Lifecycle inbox test harness", () => { + it("delivers project creation, archive, and restore in order", async () => { + const harness = createTestHarness({ manifest }); + const event: ResourceLifecycleEvent = { id: "1", companyId: "a", resourceType: "project", resourceId: "project", action: "create", createdAt: new Date(0).toISOString() }; + harness.seed({ lifecycleEvents: [event, { ...event, id: "2", action: "archive" }, { ...event, id: "3", action: "update" }] }); + for (const action of ["create", "archive", "update"]) { + const [next] = await harness.ctx.events.listLifecycle("a"); + expect(next.action).toBe(action); + await harness.ctx.events.acknowledgeLifecycle("a", next.id); + } + expect(await harness.ctx.events.listLifecycle("a")).toEqual([]); + }); + it("lets a plugin exercise retries, acknowledgment order, and company scope", async () => { + const harness = createTestHarness({ manifest }); + const event: ResourceLifecycleEvent = { id: "1", companyId: "a", resourceType: "agent", resourceId: "agent", action: "create", createdAt: new Date(0).toISOString() }; + harness.seed({ lifecycleEvents: [event, { ...event, id: "2", action: "pause" }, { ...event, id: "3", companyId: "b" }] }); + expect(await harness.ctx.events.listLifecycle("a")).toEqual([event]); + expect(await harness.ctx.events.listLifecycle("a", 100, "1")).toEqual([]); + expect(await harness.ctx.events.listLifecycle("a")).toEqual([event]); + await expect(harness.ctx.events.acknowledgeLifecycle("a", "2")).rejects.toThrow("earlier lifecycle events"); + await expect(harness.ctx.events.acknowledgeLifecycle("a", "3")).rejects.toThrow("not found"); + await harness.ctx.events.acknowledgeLifecycle("a", "1"); + await harness.ctx.events.acknowledgeLifecycle("a", "1"); + expect(await harness.ctx.events.listLifecycle("a")).toEqual([expect.objectContaining({ action: "pause" })]); + expect(await harness.ctx.events.listLifecycle("b")).toEqual([expect.objectContaining({ id: "3" })]); + const denied = createTestHarness({ manifest, capabilities: [] }); + 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/packages/plugins/sdk/tests/worker-rpc-host.test.ts b/packages/plugins/sdk/tests/worker-rpc-host.test.ts index cce6a3ae1a7..2895d57ece4 100644 --- a/packages/plugins/sdk/tests/worker-rpc-host.test.ts +++ b/packages/plugins/sdk/tests/worker-rpc-host.test.ts @@ -1137,3 +1137,45 @@ describe("AI connection router RPC", () => { } finally { worker.stop(); reader.close(); input.destroy(); output.destroy(); } }); }); + +describe("Durable lifecycle inbox RPC", () => { + it("sends company-scoped reads and acknowledgments through the worker SDK", async () => { + const input = new PassThrough(), output = new PassThrough(); + const reader = createInterface({ input: output }); + const calls: Array<{ method: string; params: unknown }> = []; + const event = { id: "7", companyId: "company-a", resourceType: "agent", resourceId: "agent-a", action: "pause", createdAt: new Date(0).toISOString() }; + const plugin = definePlugin({ async setup(ctx) { + ctx.data.register("drain", async ({ companyId }) => { + const events = await ctx.events.listLifecycle(companyId!, 10, "6"); + await ctx.events.acknowledgeLifecycle(companyId!, events[0].id); + return events; + }); + } }); + const worker = startWorkerRpcHost({ plugin, stdin: input, stdout: output }); + let sequence = 0; + const pending = new Map void>(); + reader.on("line", line => { + const message = parseMessage(line); + if (isJsonRpcResponse(message)) { + pending.get(String(message.id))?.(message); + pending.delete(String(message.id)); + } else if (isJsonRpcRequest(message)) { + calls.push({ method: message.method, params: message.params }); + input.write(serializeMessage(createSuccessResponse(message.id, message.method === "events.listLifecycle" ? [event] : null))); + } + }); + const call = (method: string, params: unknown) => new Promise(resolve => { + const id = String(++sequence); + pending.set(id, resolve); + input.write(serializeMessage(createRequest(method, params, id))); + }); + try { + await call("initialize", { manifest: { id: "fixture.lifecycle", apiVersion: 1, version: "1.0.0", displayName: "Lifecycle", description: "Fixture", author: "Tests", categories: ["automation"], capabilities: ["events.subscribe"], entrypoints: {} }, config: {}, databaseNamespace: null }); + expect((await call("getData", { key: "drain", companyId: "company-a", params: {} }) as { result: unknown }).result).toEqual([event]); + expect(calls).toEqual([ + { method: "events.listLifecycle", params: { companyId: "company-a", limit: 10, afterId: "6" } }, + { method: "events.acknowledgeLifecycle", params: { companyId: "company-a", eventId: "7" } }, + ]); + } finally { worker.stop(); reader.close(); input.destroy(); output.destroy(); } + }); +}); diff --git a/server/src/__tests__/plugin-worker-manager.test.ts b/server/src/__tests__/plugin-worker-manager.test.ts index 2940b0c56b4..58b0c9a6682 100644 --- a/server/src/__tests__/plugin-worker-manager.test.ts +++ b/server/src/__tests__/plugin-worker-manager.test.ts @@ -605,12 +605,14 @@ describe("plugin proactive company scope (LOOA-629)", () => { }) { const companiesGet = overrides?.companiesGet ?? vi.fn(async () => ({ id: "company-1", name: "Co" })); const stateGet = overrides?.stateGet ?? vi.fn(async () => ({ value: "ok" })); + const listLifecycle = vi.fn(async () => [{ id: "1", companyId: "company-1" }]); const hostHandlers = createHostClientHandlers({ pluginId: "test.plugin", - capabilities: ["companies.read", "plugin.state.read"], + capabilities: ["companies.read", "plugin.state.read", "events.subscribe"], services: { companies: { get: companiesGet }, state: { get: stateGet }, + events: { listLifecycle }, } as unknown as HostServices, }); const handle = createPluginWorkerHandle("test.plugin", { @@ -621,7 +623,7 @@ describe("plugin proactive company scope (LOOA-629)", () => { apiVersion: 1, hostHandlers, }); - return { handle, companiesGet, stateGet }; + return { handle, companiesGet, stateGet, listLifecycle }; } it("denies a proactive company-scoped call when no company is authorized", async () => { @@ -640,6 +642,25 @@ describe("plugin proactive company scope (LOOA-629)", () => { } }); + it("admits lifecycle polling only for configured proactive companies", async () => { + const { handle, listLifecycle } = makeHandle(); + const poll = (companyId: string) => handle.call("getData", { + params: { mode: "omit", hostMethod: "events.listLifecycle", requestedCompanyId: companyId }, + } as unknown as HostToWorkerMethods["getData"][0]); + try { + await handle.start(); + await expect(poll("company-1")).rejects.toMatchObject({ code: PLUGIN_RPC_ERROR_CODES.INVOCATION_SCOPE_DENIED }); + handle.setProactiveCompanyScopes(["company-1"]); + expect(await poll("company-1")).toEqual([{ id: "1", companyId: "company-1" }]); + await expect(poll("company-2")).rejects.toMatchObject({ code: PLUGIN_RPC_ERROR_CODES.INVOCATION_SCOPE_DENIED }); + handle.setProactiveCompanyScopes([]); + await expect(poll("company-1")).rejects.toMatchObject({ code: PLUGIN_RPC_ERROR_CODES.INVOCATION_SCOPE_DENIED }); + expect(listLifecycle).toHaveBeenCalledTimes(1); + } finally { + await handle.stop().catch(() => undefined); + } + }); + it("admits a proactive company-scoped call for an authorized company", async () => { const { handle, companiesGet } = makeHandle(); try { diff --git a/server/src/__tests__/resource-lifecycle-events.test.ts b/server/src/__tests__/resource-lifecycle-events.test.ts index 54de2a4b349..95aae83d32b 100644 --- a/server/src/__tests__/resource-lifecycle-events.test.ts +++ b/server/src/__tests__/resource-lifecycle-events.test.ts @@ -1,6 +1,8 @@ import { randomUUID } from "node:crypto"; +import { readFileSync } from "node:fs"; import { eq, sql } from "drizzle-orm"; -import { agentApiKeys, agents, budgetPolicies, companies, costEvents, createDb, projects, resourceLifecycleEvents, type Db } from "@paperclipai/db"; +import { agentApiKeys, agents, budgetPolicies, companies, costEvents, createDb, pluginCompanySettings, pluginLifecycleAcknowledgments, plugins, projects, resourceLifecycleEvents, type Db } from "@paperclipai/db"; +import { createHostClientHandlers } from "@paperclipai/plugin-sdk"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { agentService } from "../services/agents.js"; @@ -8,6 +10,9 @@ import { approvalService } from "../services/approvals.js"; import { budgetService } from "../services/budgets.js"; import { projectService } from "../services/projects.js"; import { recordResourceCreationEvent } from "../services/resource-lifecycle-events.js"; +import { pluginLifecycleInbox } from "../services/plugin-lifecycle-inbox.js"; +import { buildHostServices } from "../services/plugin-host-services.js"; +import { createPluginEventBus } from "../services/plugin-event-bus.js"; const support = await getEmbeddedPostgresTestSupport(); const describePostgres = support.supported ? describe : describe.skip; @@ -34,6 +39,10 @@ describePostgres("Resource lifecycle events", () => { const events = () => db.select().from(resourceLifecycleEvents).where(eq(resourceLifecycleEvents.companyId, companyId)); const createAgent = (status: "idle" | "pending_approval" | "terminated" = "idle", database: Db = db) => agentService(database).create(companyId, { name: "Lifecycle agent", adapterType: "process", adapterConfig: {}, status }); + const createPlugin = async () => { + const [plugin] = await db.insert(plugins).values({ pluginKey: randomUUID(), packageName: "lifecycle-fixture", version: "1.0.0", status: "ready", manifestJson: {} as never }).returning(); + return plugin; + }; it("records direct and approval-free hires once, including concurrent duplicate submissions", async () => { const agent = await createAgent(); @@ -291,4 +300,171 @@ describePostgres("Resource lifecycle events", () => { } }); + const backfillBaseline = () => db.transaction(async tx => { + const migration = readFileSync(new URL("../../../packages/db/src/migrations/0309_loving_the_hood.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 [restored] = await db.insert(projects).values({ companyId, name: "Restored baseline" }).returning(); + await db.insert(resourceLifecycleEvents).values({ companyId, resourceType: "project", resourceId: restored.id, action: "archive" }); + 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", "archive"]); + expect(actions(restored.id)).toEqual(["archive", "create", "update"]); + 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" }); + const projectCreate = (await inbox.list(companyId)).find(e => e.resourceId === archived.id)!; + await inbox.acknowledge(companyId, projectCreate.id); + const archive = (await inbox.list(companyId)).find(e => e.resourceId === archived.id)!; + expect(archive.action).toBe("archive"); + await inbox.acknowledge(companyId, archive.id); + expect((await inbox.list(companyId)).find(e => e.resourceId === archived.id)).toBeUndefined(); + 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`ALTER TABLE resource_lifecycle_events ALTER COLUMN id DROP IDENTITY`); + 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/0309_loving_the_hood.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.toMatchObject({ cause: { constraint_name: "fixture_reject_baseline" } }); + expect(await events()).toEqual([]); + expect((await db.execute(sql`SELECT is_identity FROM information_schema.columns WHERE table_schema = 'public' AND table_name = 'resource_lifecycle_events' AND column_name = 'id'`))[0]).toMatchObject({ is_identity: "NO" }); + 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(); + expect((await db.execute(sql`SELECT is_identity FROM information_schema.columns WHERE table_schema = 'public' AND table_name = 'resource_lifecycle_events' AND column_name = 'id'`))[0]).toMatchObject({ is_identity: "YES" }); + } + }); + + it("retries until acknowledgment, preserves resource order, and isolates progress per plugin", async () => { + const first = await createPlugin(); + const second = await createPlugin(); + const agent = await createAgent(); + await agentService(db).pause(agent.id); + await agentService(db).resume(agent.id); + const journal = (await events()).sort((a, b) => a.id - b.id); + const inbox = pluginLifecycleInbox(db, first.id); + const [event] = await inbox.list(companyId); + expect(event).toMatchObject({ id: String(journal[0].id), resourceId: agent.id, action: "create" }); + expect(await pluginLifecycleInbox(db, first.id).list(companyId)).toEqual([event]); + await expect(inbox.acknowledge(companyId, String(journal[1].id))).rejects.toMatchObject({ status: 409 }); + await Promise.all([inbox.acknowledge(companyId, event.id), inbox.acknowledge(companyId, event.id)]); + expect(await inbox.list(companyId)).toEqual([expect.objectContaining({ action: "pause" })]); + expect(await pluginLifecycleInbox(db, second.id).list(companyId)).toEqual([event]); + await inbox.acknowledge(companyId, String(journal[1].id)); + await inbox.acknowledge(companyId, String(journal[2].id)); + expect(await inbox.list(companyId)).toEqual([]); + await db.delete(plugins).where(eq(plugins.id, first.id)); + expect(await db.select().from(pluginLifecycleAcknowledgments).where(eq(pluginLifecycleAcknowledgments.pluginId, first.id))).toEqual([]); + }); + + it("pages past 100 failing resources without skipping their later events", async () => { + const plugin = await createPlugin(); + const inbox = pluginLifecycleInbox(db, plugin.id); + await db.insert(resourceLifecycleEvents).values(Array.from({ length: 101 }, () => ({ + companyId, resourceType: "agent" as const, resourceId: randomUUID(), action: "create" as const, + }))); + const first = await inbox.list(companyId, 100); + await db.insert(resourceLifecycleEvents).values({ companyId, resourceType: "agent", resourceId: first[0].resourceId, action: "pause" }); + const next = await inbox.list(companyId, 100, first.at(-1)!.id); + expect(next).toHaveLength(1); + expect(next[0].resourceId).not.toBe(first[0].resourceId); + await inbox.acknowledge(companyId, next[0].id); + expect(await inbox.list(companyId, 100, first.at(-1)!.id)).toEqual([]); + expect(await inbox.list(companyId, 100)).toEqual(first); + await expect(inbox.list(companyId, 100, "invalid")).rejects.toMatchObject({ status: 400 }); + }); + + it("never skips a lower event id that commits after a higher id is acknowledged", async () => { + const plugin = await createPlugin(); + const inbox = pluginLifecycleInbox(db, plugin.id); + await db.transaction(async tx => { + const low = await agentService(tx as unknown as Db).create(companyId, { name: "Delayed commit" }); + const high = await createAgent(); + const [visible] = await inbox.list(companyId); + expect(visible.resourceId).toBe(high.id); + await inbox.acknowledge(companyId, visible.id); + expect(await inbox.list(companyId)).toEqual([]); + expect(low.id).not.toBe(high.id); + }); + expect(await inbox.list(companyId)).toEqual([expect.objectContaining({ action: "create" })]); + }); + + it("enforces capability, invocation company, plugin availability, and company boundaries through host RPC", async () => { + const plugin = await createPlugin(); + await createAgent(); + const services = buildHostServices(db, plugin.id, plugin.pluginKey, createPluginEventBus()); + const handlers = createHostClientHandlers({ pluginId: plugin.id, capabilities: ["events.subscribe"], services }); + const scope = { invocationScope: { companyId } }; + try { + await expect(handlers["events.listLifecycle"]({ companyId })).rejects.toThrow("company context"); + const other = randomUUID(); + await expect(handlers["events.listLifecycle"]({ companyId: other }, scope)).rejects.toThrow("requested company"); + const denied = createHostClientHandlers({ pluginId: plugin.id, capabilities: [], services }); + await expect(denied["events.listLifecycle"]({ companyId }, scope)).rejects.toThrow("events.subscribe"); + await expect(denied["events.acknowledgeLifecycle"]({ companyId, eventId: "1" }, scope)).rejects.toThrow("events.subscribe"); + const [event] = await handlers["events.listLifecycle"]({ companyId }, scope); + await expect(handlers["events.acknowledgeLifecycle"]({ companyId: other, eventId: event.id }, scope)).rejects.toThrow("requested company"); + await expect(pluginLifecycleInbox(db, plugin.id).acknowledge(other, event.id)).rejects.toMatchObject({ status: 404 }); + await db.insert(pluginCompanySettings).values({ pluginId: plugin.id, companyId, enabled: false }); + await expect(handlers["events.listLifecycle"]({ companyId }, scope)).rejects.toMatchObject({ status: 403 }); + await expect(handlers["events.acknowledgeLifecycle"]({ companyId, eventId: event.id }, scope)).rejects.toMatchObject({ status: 403 }); + await db.update(pluginCompanySettings).set({ enabled: true }).where(eq(pluginCompanySettings.pluginId, plugin.id)); + await expect(handlers["events.listLifecycle"]({ companyId, limit: 0 }, scope)).rejects.toMatchObject({ status: 400 }); + await expect(handlers["events.acknowledgeLifecycle"]({ companyId, eventId: "1e2" }, scope)).rejects.toMatchObject({ status: 400 }); + await handlers["events.acknowledgeLifecycle"]({ companyId, eventId: event.id }, scope); + expect(await handlers["events.listLifecycle"]({ companyId }, scope)).toEqual([]); + await db.update(plugins).set({ status: "disabled" }).where(eq(plugins.id, plugin.id)); + await expect(handlers["events.listLifecycle"]({ companyId }, scope)).rejects.toMatchObject({ status: 403 }); + } finally { + services.dispose(); + } + }); }); diff --git a/server/src/services/plugin-host-services.ts b/server/src/services/plugin-host-services.ts index 185524fc97e..2fac28c3b91 100644 --- a/server/src/services/plugin-host-services.ts +++ b/server/src/services/plugin-host-services.ts @@ -47,6 +47,7 @@ import { createHash, randomBytes, randomUUID } from "node:crypto"; import path from "node:path"; import { pluginRegistryService } from "./plugin-registry.js"; import { pluginStateStore } from "./plugin-state-store.js"; +import { pluginLifecycleInbox } from "./plugin-lifecycle-inbox.js"; import { pluginDatabaseService } from "./plugin-database.js"; import { pluginManagedAgentService } from "./plugin-managed-agents.js"; import { pluginManagedRoutineService } from "./plugin-managed-routines.js"; @@ -757,6 +758,7 @@ export function buildHostServices( const approvalSvc = approvalService(db); const interactions = issueThreadInteractionService(db); const scopedBus = eventBus.forPlugin(pluginKey); + const lifecycleInbox = pluginLifecycleInbox(db, pluginId); // Track active session event subscriptions for cleanup const activeSubscriptions = new Set<{ unsubscribe: () => void; timer: ReturnType }>(); @@ -1557,6 +1559,12 @@ export function buildHostServices( }, events: { + async listLifecycle(params) { + return lifecycleInbox.list(ensureCompanyId(params.companyId), params.limit, params.afterId); + }, + async acknowledgeLifecycle(params) { + await lifecycleInbox.acknowledge(ensureCompanyId(params.companyId), params.eventId); + }, async emit(params) { if (params.companyId) { await ensurePluginAvailableForCompany(params.companyId); diff --git a/server/src/services/plugin-lifecycle-inbox.ts b/server/src/services/plugin-lifecycle-inbox.ts new file mode 100644 index 00000000000..7187d83b069 --- /dev/null +++ b/server/src/services/plugin-lifecycle-inbox.ts @@ -0,0 +1,58 @@ +import { pluginCompanySettings, pluginLifecycleAcknowledgments, plugins, resourceLifecycleEvents, type Db } from "@paperclipai/db"; +import type { ResourceLifecycleEvent } from "@paperclipai/plugin-sdk"; +import { z } from "zod"; +import { and, asc, eq, gt, sql } from "drizzle-orm"; +import { badRequest, conflict, forbidden, notFound } from "../errors.js"; + +/** Durable pull delivery. Provider work stays outside tenant transactions. */ +export function pluginLifecycleInbox(db: Db, pluginId: string) { + const unacknowledged = sql`NOT EXISTS ( + SELECT 1 FROM plugin_lifecycle_acknowledgments ack + WHERE ack.plugin_id = ${pluginId} AND ack.event_id = ${resourceLifecycleEvents.id} + )`; + const firstForResource = sql`NOT EXISTS ( + SELECT 1 FROM resource_lifecycle_events older + WHERE older.company_id = ${resourceLifecycleEvents.companyId} + AND older.resource_type = ${resourceLifecycleEvents.resourceType} + AND older.resource_id = ${resourceLifecycleEvents.resourceId} + -- 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) + )`; + + async function assertAvailable(database: Db, companyId: string) { + if (!z.string().uuid().safeParse(companyId).success) throw badRequest("companyId must be a UUID"); + const [plugin] = await database.select({ status: plugins.status }).from(plugins).where(eq(plugins.id, pluginId)); + const [settings] = await database.select({ enabled: pluginCompanySettings.enabled }).from(pluginCompanySettings) + .where(and(eq(pluginCompanySettings.pluginId, pluginId), eq(pluginCompanySettings.companyId, companyId))); + if (plugin?.status !== "ready" || settings?.enabled === false) throw forbidden("Plugin is not enabled for this company"); + } + + return { + async list(companyId: string, limit = 50, afterId?: string): Promise { + if (!Number.isInteger(limit) || limit < 1 || limit > 100) throw badRequest("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 badRequest("Invalid lifecycle page id"); + await assertAvailable(db, companyId); + const rows = await db.select().from(resourceLifecycleEvents) + .where(and(eq(resourceLifecycleEvents.companyId, companyId), unacknowledged, firstForResource, afterId === undefined ? undefined : gt(resourceLifecycleEvents.id, Number(afterId)))) + .orderBy(asc(resourceLifecycleEvents.id)).limit(limit); + return rows.map(row => ({ ...row, id: String(row.id), createdAt: row.createdAt.toISOString() })) as ResourceLifecycleEvent[]; + }, + async acknowledge(companyId: string, eventId: string): Promise { + if (typeof eventId !== "string" || !/^[1-9]\d*$/.test(eventId) || !Number.isSafeInteger(Number(eventId))) throw badRequest("Invalid lifecycle event id"); + await db.transaction(async tx => { + const txDb = tx as unknown as Db; + await assertAvailable(txDb, companyId); + const [event] = await tx.select().from(resourceLifecycleEvents) + .where(and(eq(resourceLifecycleEvents.companyId, companyId), eq(resourceLifecycleEvents.id, Number(eventId)))).for("update"); + if (!event) throw notFound("Lifecycle event not found"); + const [eligible] = await tx.select({ id: resourceLifecycleEvents.id }).from(resourceLifecycleEvents) + .where(and(eq(resourceLifecycleEvents.id, event.id), firstForResource)); + if (!eligible) throw conflict("Acknowledge earlier lifecycle events for this resource first"); + await tx.insert(pluginLifecycleAcknowledgments).values({ pluginId, eventId: event.id }).onConflictDoNothing(); + }); + }, + }; +}