diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 959658e603..a580717f0b 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -716,7 +716,7 @@ not revoke grants used by other bots. See [Slack setup](connections/slack-setup. ## Transaction-aware delivery work signals -Five existing delivery queues use process-local work signals to avoid empty +Durable delivery queues, including plugin resource lifecycle events, use process-local work signals to avoid empty polling. Register work **before** writing a queue row, on that row's transaction: ```ts diff --git a/doc/plugins/PLUGIN_AUTHORING_GUIDE.md b/doc/plugins/PLUGIN_AUTHORING_GUIDE.md index cae51f0cdc..f4b4c06a8d 100644 --- a/doc/plugins/PLUGIN_AUTHORING_GUIDE.md +++ b/doc/plugins/PLUGIN_AUTHORING_GUIDE.md @@ -114,48 +114,71 @@ agents. The durable event inbox below is separate: acknowledging an event does not complete required lifecycle work. See [Agent lifecycle](../AGENT-LIFECYCLE.md) for phase ordering, recovery, and existing-installation limits. -## Durable resource lifecycle inbox +## Durable resource lifecycle delivery -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. +Declare `resourceLifecycle: ["project"]` (or `["agent", "project"]`) in the +manifest, with `events.subscribe`, and implement `onResourceLifecycle(event)`: -An event has `id`, `companyId`, `resourceType`, `resourceId`, `action`, and +```ts +async onResourceLifecycle(event) { + const config = await ctx.config.get(event.companyId); + if (!config.provider) return { eventId: event.id, status: "deferred" }; + const finished = await reconcileCurrentResource(event); + return { eventId: event.id, status: finished ? "acknowledged" : "retry" }; +} +``` + +Core delivers events to ready plugins with a configuration row for that company, +unless the plugin is disabled there. Each invocation has the event's company +scope. The dispatcher wakes after the mutation's outer transaction settles, +on host startup, plugin activation/worker start, and company configuration or +enablement changes. Successful worker `onConfigChanged` completion wakes delivery +again, so a callback that deferred while configuration was still applying is +revisited. Failed scans retain pending work through idle drain. Empty and deferred +queues have no periodic scans. `retry`, +exceptions, invalid results, and unavailable workers leave the event pending and +schedule another attempt after 30 seconds. Other acknowledgments or mutations +do not bypass that event's delay; a host restart may replay it immediately. +`deferred` retains the event without +a retry deadline, useful while waiting for operator configuration. A later wake +revisits deferred events. Acknowledgment immediately exposes subsequent work. + +Events contain `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. +project actions are `create`, `update`, `archive`, and `delete`. Pending hires +produce creation after approval. Project updates include repository/workspace +mutations and restoring an archived project. Deletion is recorded in the same +transaction as removal and survives the project row. Earlier deletions cannot +be reconstructed. 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. +backfilled ID is newer. Remaining events follow ID order. Failed or deferred +resources do not block independent resources. Acknowledgments are stored per +plugin, survive worker/host restarts, and are not a global sequence cursor. +The initial journal migration seeds existing hired agents and projects, +including archived projects; this is a desired-state baseline, not history. +Agent lifecycle participation (`onAgentLifecycle`) remains a separate contract: +acknowledging a resource event does not advance an agent lifecycle phase. -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()`. +Delivery is at least once. Re-read current authorized state and make provider +operations idempotent; a worker can finish an external effect and exit before +Core commits its acknowledgment. Content-free events contain no repository +credentials or configuration snapshots. Deleted projects may return `null`. +Use stable resource identifiers and fence delayed provider requests when removing +resources. Neither the callback nor an acknowledgment proves mounts are unused. -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. +Delivery uses the same process-local transaction signals as Core's other durable +queues. Startup replays committed work. Independent database writers and separate +active server processes need an explicit ownership/wake protocol; see +[database work signals](../DATABASE.md#transaction-aware-delivery-work-signals). + +The existing `ctx.events.listLifecycle(companyId, limit?, afterId?)` and +`acknowledgeLifecycle(companyId, eventId)` remain available for manual consumers. +Do not mix automatic and manual acknowledgment for the same resources. Reads +return one pending event per resource (default 50, maximum 100). `afterId` is a +page cursor, never a durable high-water mark: restart pagination to revisit +failures and transactions that commit later with lower IDs. Offline SDK tests +can seed `lifecycleEvents` with `createTestHarness().seed()`. ## External object reference providers diff --git a/packages/db/src/migrations/0340_open_warlock.sql b/packages/db/src/migrations/0340_open_warlock.sql new file mode 100644 index 0000000000..81041a3a4d --- /dev/null +++ b/packages/db/src/migrations/0340_open_warlock.sql @@ -0,0 +1,2 @@ +ALTER TABLE "resource_lifecycle_events" DROP CONSTRAINT "resource_lifecycle_events_action_check";--> statement-breakpoint +ALTER TABLE "resource_lifecycle_events" ADD CONSTRAINT "resource_lifecycle_events_action_check" CHECK ("resource_lifecycle_events"."action" = 'create' OR ("resource_lifecycle_events"."resource_type" = 'project' AND "resource_lifecycle_events"."action" IN ('update', 'archive', 'delete')) OR ("resource_lifecycle_events"."resource_type" = 'agent' AND "resource_lifecycle_events"."action" IN ('pause', 'resume', 'terminate'))); \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0335_snapshot.json b/packages/db/src/migrations/meta/0340_snapshot.json similarity index 98% rename from packages/db/src/migrations/meta/0335_snapshot.json rename to packages/db/src/migrations/meta/0340_snapshot.json index b348334bbd..188aeb107d 100644 --- a/packages/db/src/migrations/meta/0335_snapshot.json +++ b/packages/db/src/migrations/meta/0340_snapshot.json @@ -1,6 +1,6 @@ { - "id": "18c156f2-8344-4282-98b7-f702f0035c81", - "prevId": "39d82f2c-8fd5-4606-84df-f51880dbd6a7", + "id": "b32fa655-79c7-4793-a021-22258b6c369c", + "prevId": "045fe4f7-1636-4e1b-b304-11b6128fefcf", "version": "7", "dialect": "postgresql", "tables": { @@ -9858,6 +9858,12 @@ "notNull": true, "default": false }, + "execution_defaults": { + "name": "execution_defaults", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, "metadata": { "name": "metadata", "type": "jsonb", @@ -10098,6 +10104,12 @@ "notNull": true, "default": "''" }, + "execution_defaults": { + "name": "execution_defaults", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, "allow_direct_messages": { "name": "allow_direct_messages", "type": "boolean", @@ -16312,6 +16324,18 @@ "primaryKey": false, "notNull": false }, + "owner_agent_id": { + "name": "owner_agent_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "owner_plugin_id": { + "name": "owner_plugin_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, "user_secret_definition_id": { "name": "user_secret_definition_id", "type": "uuid", @@ -16540,6 +16564,40 @@ "method": "btree", "with": {} }, + "company_secrets_instance_owner_key_uq": { + "name": "company_secrets_instance_owner_key_uq", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner_agent_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner_plugin_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"company_secrets\".\"scope\" = 'instance' and \"company_secrets\".\"deleted_at\" is null", + "concurrently": false, + "method": "btree", + "with": {} + }, "company_secrets_provider_config_idx": { "name": "company_secrets_provider_config_idx", "columns": [ @@ -16642,6 +16700,19 @@ "onDelete": "no action", "onUpdate": "no action" }, + "company_secrets_owner_plugin_id_plugins_id_fk": { + "name": "company_secrets_owner_plugin_id_plugins_id_fk", + "tableFrom": "company_secrets", + "tableTo": "plugins", + "columnsFrom": [ + "owner_plugin_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, "company_secrets_user_secret_definition_id_user_secret_definitions_id_fk": { "name": "company_secrets_user_secret_definition_id_user_secret_definitions_id_fk", "tableFrom": "company_secrets", @@ -16680,6 +16751,21 @@ ], "onDelete": "set null", "onUpdate": "no action" + }, + "company_secrets_company_id_owner_agent_id_agents_company_id_id_fk": { + "name": "company_secrets_company_id_owner_agent_id_agents_company_id_id_fk", + "tableFrom": "company_secrets", + "tableTo": "agents", + "columnsFrom": [ + "company_id", + "owner_agent_id" + ], + "columnsTo": [ + "company_id", + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" } }, "compositePrimaryKeys": {}, @@ -16688,7 +16774,7 @@ "checkConstraints": { "company_secrets_scope_shape_check": { "name": "company_secrets_scope_shape_check", - "value": "(\n \"company_secrets\".\"scope\" = 'company'\n and \"company_secrets\".\"owner_user_id\" is null\n and \"company_secrets\".\"user_secret_definition_id\" is null\n ) or (\n \"company_secrets\".\"scope\" = 'user'\n and \"company_secrets\".\"owner_user_id\" is not null\n and \"company_secrets\".\"user_secret_definition_id\" is not null\n )" + "value": "(\n \"company_secrets\".\"scope\" = 'company'\n and \"company_secrets\".\"owner_user_id\" is null\n and \"company_secrets\".\"user_secret_definition_id\" is null\n and \"company_secrets\".\"owner_agent_id\" is null and \"company_secrets\".\"owner_plugin_id\" is null\n ) or (\n \"company_secrets\".\"scope\" = 'user'\n and \"company_secrets\".\"owner_user_id\" is not null\n and \"company_secrets\".\"user_secret_definition_id\" is not null\n and \"company_secrets\".\"owner_agent_id\" is null and \"company_secrets\".\"owner_plugin_id\" is null\n ) or (\n \"company_secrets\".\"scope\" = 'instance'\n and \"company_secrets\".\"owner_user_id\" is null and \"company_secrets\".\"user_secret_definition_id\" is null\n and \"company_secrets\".\"owner_agent_id\" is not null and \"company_secrets\".\"owner_plugin_id\" is not null\n )" } }, "isRLSEnabled": false @@ -26581,6 +26667,476 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.execution_workspace_repositories": { + "name": "execution_workspace_repositories", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "execution_workspace_id": { + "name": "execution_workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "requested_by_issue_id": { + "name": "requested_by_issue_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "repository_identity": { + "name": "repository_identity", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "catalog_repository_id": { + "name": "catalog_repository_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "repo_url": { + "name": "repo_url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "relative_path": { + "name": "relative_path", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "requested_ref": { + "name": "requested_ref", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'HEAD'" + }, + "pinned_commit": { + "name": "pinned_commit", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "branch_name": { + "name": "branch_name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "request_key": { + "name": "request_key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "request_keys": { + "name": "request_keys", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'[]'::jsonb" + }, + "state": { + "name": "state", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'pending'" + }, + "failure_code": { + "name": "failure_code", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "execution_workspace_repositories_identity_idx": { + "name": "execution_workspace_repositories_identity_idx", + "columns": [ + { + "expression": "execution_workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "repository_identity", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "execution_workspace_repositories_path_idx": { + "name": "execution_workspace_repositories_path_idx", + "columns": [ + { + "expression": "execution_workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "relative_path", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "execution_workspace_repositories_request_idx": { + "name": "execution_workspace_repositories_request_idx", + "columns": [ + { + "expression": "execution_workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "request_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "execution_workspace_repositories_company_workspace_idx": { + "name": "execution_workspace_repositories_company_workspace_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "execution_workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "execution_workspace_repositories_company_id_companies_id_fk": { + "name": "execution_workspace_repositories_company_id_companies_id_fk", + "tableFrom": "execution_workspace_repositories", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "execution_workspace_repositories_execution_workspace_id_execution_workspaces_id_fk": { + "name": "execution_workspace_repositories_execution_workspace_id_execution_workspaces_id_fk", + "tableFrom": "execution_workspace_repositories", + "tableTo": "execution_workspaces", + "columnsFrom": [ + "execution_workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "execution_workspace_repositories_requested_by_issue_id_issues_id_fk": { + "name": "execution_workspace_repositories_requested_by_issue_id_issues_id_fk", + "tableFrom": "execution_workspace_repositories", + "tableTo": "issues", + "columnsFrom": [ + "requested_by_issue_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.execution_workspace_repository_access": { + "name": "execution_workspace_repository_access", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "root_key": { + "name": "root_key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "canonical_cwd": { + "name": "canonical_cwd", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "scope_id": { + "name": "scope_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "issue_id": { + "name": "issue_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "agent_id": { + "name": "agent_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "responsible_user_id": { + "name": "responsible_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "execution_workspace_repository_access_scope_idx": { + "name": "execution_workspace_repository_access_scope_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "root_key", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "kind", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "scope_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "execution_workspace_repository_access_company_id_companies_id_fk": { + "name": "execution_workspace_repository_access_company_id_companies_id_fk", + "tableFrom": "execution_workspace_repository_access", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.execution_workspace_repository_roots": { + "name": "execution_workspace_repository_roots", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "canonical_cwd": { + "name": "canonical_cwd", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "root_key": { + "name": "root_key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "mode": { + "name": "mode", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "execution_workspace_repository_roots_company_root_idx": { + "name": "execution_workspace_repository_roots_company_root_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "root_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "execution_workspace_repository_roots_company_id_companies_id_fk": { + "name": "execution_workspace_repository_roots_company_id_companies_id_fk", + "tableFrom": "execution_workspace_repository_roots", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.execution_workspace_runtime_leases": { "name": "execution_workspace_runtime_leases", "schema": "", @@ -26836,7 +27392,7 @@ "name": "project_id", "type": "uuid", "primaryKey": false, - "notNull": true + "notNull": false }, "project_workspace_id": { "name": "project_workspace_id", @@ -27114,7 +27670,7 @@ "columnsTo": [ "id" ], - "onDelete": "cascade", + "onDelete": "restrict", "onUpdate": "no action" }, "execution_workspaces_project_workspace_id_project_workspaces_id_fk": { @@ -37902,6 +38458,25 @@ "primaryKey": false, "notNull": false }, + "workspace_binding_revision": { + "name": "workspace_binding_revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "workspace_selection": { + "name": "workspace_selection", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "workspace_pending_selection": { + "name": "workspace_pending_selection", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, "execution_workspace_preference": { "name": "execution_workspace_preference", "type": "text", @@ -45625,7 +46200,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\" IN ('update', 'archive')) 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', 'delete')) OR (\"resource_lifecycle_events\".\"resource_type\" = 'agent' AND \"resource_lifecycle_events\".\"action\" IN ('pause', 'resume', 'terminate'))" } }, "isRLSEnabled": false diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 4e82cc8592..248bad8e5b 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2367,6 +2367,13 @@ "when": 1791903526575, "tag": "0339_luxuriant_iron_monger", "breakpoints": true + }, + { + "idx": 340, + "version": "7", + "when": 1791911743153, + "tag": "0340_open_warlock", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/resource_lifecycle_events.ts b/packages/db/src/schema/resource_lifecycle_events.ts index cc0747373a..d107f7f1f7 100644 --- a/packages/db/src/schema/resource_lifecycle_events.ts +++ b/packages/db/src/schema/resource_lifecycle_events.ts @@ -8,12 +8,12 @@ export const resourceLifecycleEvents = pgTable("resource_lifecycle_events", { companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }), resourceType: text("resource_type").$type<"agent" | "project">().notNull(), resourceId: uuid("resource_id").notNull(), - action: text("action").$type<"create" | "update" | "archive" | "pause" | "resume" | "terminate">().notNull(), + action: text("action").$type<"create" | "update" | "archive" | "delete" | "pause" | "resume" | "terminate">().notNull(), createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), }, (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'))`), + actionCheck: check("resource_lifecycle_events_action_check", sql`${table.action} = 'create' OR (${table.resourceType} = 'project' AND ${table.action} IN ('update', 'archive', 'delete')) OR (${table.resourceType} = 'agent' AND ${table.action} IN ('pause', 'resume', 'terminate'))`), })); diff --git a/packages/plugins/sdk/src/define-plugin.ts b/packages/plugins/sdk/src/define-plugin.ts index f6f2021470..f206f8c839 100644 --- a/packages/plugins/sdk/src/define-plugin.ts +++ b/packages/plugins/sdk/src/define-plugin.ts @@ -51,7 +51,7 @@ import type { AiConnectionRouterRequest, AiConnectionRouterResult } from "@paper * ``` */ -import type { PluginContext } from "./types.js"; +import type { PluginContext, ResourceLifecycleEvent, ResourceLifecycleResult } from "./types.js"; import type { PluginEnvironmentAcquireLeaseParams, PluginEnvironmentDestroyLeaseParams, @@ -360,6 +360,14 @@ export interface PluginDefinition { */ onAgentLifecycle?(params: AgentLifecycleRequest): Promise; + /** + * Reconcile a committed resource event. Requires manifest resourceLifecycle and events.subscribe. + * The host acknowledges successful results and replays interrupted/unfinished work. + * Calls are at least once, in order per resource. Re-read current state and use + * idempotent external operations; deletion events remain available after the row is gone. + */ + onResourceLifecycle?(event: ResourceLifecycleEvent): Promise; + /** Propose a member from host-authorized candidates. Requires ai.connections.route. */ onRouteAiConnection?(params: AiConnectionRouterRequest): Promise; diff --git a/packages/plugins/sdk/src/index.ts b/packages/plugins/sdk/src/index.ts index 98fa405f85..d6ab601b7a 100644 --- a/packages/plugins/sdk/src/index.ts +++ b/packages/plugins/sdk/src/index.ts @@ -239,6 +239,7 @@ export type { PluginLocalFoldersClient, PluginEventsClient, ResourceLifecycleEvent, + ResourceLifecycleResult, PluginJobsClient, PluginLaunchersClient, PluginHttpClient, diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index a851294d7a..cc5d11d154 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -56,6 +56,7 @@ export type { PluginLauncherRenderContextSnapshot } from "@paperclipai/shared"; import type { PluginEvent, ResourceLifecycleEvent, + ResourceLifecycleResult, PluginIssueCheckoutOwnership, PluginIssueOrchestrationSummary, PluginIssueRelationSummary, @@ -1351,6 +1352,7 @@ export interface HostToWorkerMethods { handleApiRequest: [params: PluginApiRequestInput, result: PluginApiResponse]; /** @see PLUGIN_SPEC.md §13.8 */ agentLifecycle: [params: AgentLifecycleRequest, result: AgentLifecycleResult]; + resourceLifecycle: [params: ResourceLifecycleEvent, result: ResourceLifecycleResult]; getData: [params: GetDataParams, result: unknown]; /** @see PLUGIN_SPEC.md §13.9 */ performAction: [params: PerformActionParams, result: unknown]; @@ -1488,6 +1490,7 @@ export const HOST_TO_WORKER_OPTIONAL_METHODS: readonly HostToWorkerMethodName[] "handleWebhook", "handleApiRequest", "agentLifecycle", + "resourceLifecycle", "getData", "performAction", "executeTool", diff --git a/packages/plugins/sdk/src/types.ts b/packages/plugins/sdk/src/types.ts index b1e2980315..d553744bbb 100644 --- a/packages/plugins/sdk/src/types.ts +++ b/packages/plugins/sdk/src/types.ts @@ -547,7 +547,8 @@ export interface PluginLocalFoldersClient { 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. + * Prefer manifest resourceLifecycle + onResourceLifecycle for automatic delivery. + * Manual consumers page using afterId and restart pagination to revisit unacknowledged events. * Events repeat until acknowledged; use a company-scoped provider idempotency key. */ listLifecycle(companyId: string, limit?: number, afterId?: string): Promise; @@ -597,9 +598,18 @@ export type ResourceLifecycleEvent = { createdAt: string; } & ( | { resourceType: "agent"; action: "create" | "pause" | "resume" | "terminate" } - | { resourceType: "project"; action: "create" | "update" | "archive" } + | { resourceType: "project"; action: "create" | "update" | "archive" | "delete" } ); +/** Return acknowledged only after durable effects finish. Retry unfinished work; + * defer when configuration is missing (replayed on configuration changes or startup). + * Both retry and deferred leave the event unacknowledged. + */ +export interface ResourceLifecycleResult { + eventId: string; + status: "acknowledged" | "retry" | "deferred"; +} + /** * `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 5b1492106b..9f08052250 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -39,6 +39,7 @@ import { createPluginIdleDrain } from "./idle-drain.js"; * @see PLUGIN_SPEC.md §14 — SDK Surface */ +import type { ResourceLifecycleEvent } from "./types.js"; import fs from "node:fs"; import { AsyncLocalStorage } from "node:async_hooks"; import path from "node:path"; @@ -1660,6 +1661,9 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost return handleExecuteTool(params as ExecuteToolParams); case "detectExternalObjects": return handleDetectExternalObjects(params as DetectExternalObjectsParams); + case "resourceLifecycle": + if (!plugin.definition.onResourceLifecycle) throw methodNotImplemented("resourceLifecycle"); + return plugin.definition.onResourceLifecycle(params as ResourceLifecycleEvent); case "agentLifecycle": if (!plugin.definition.onAgentLifecycle) throw methodNotImplemented("agentLifecycle"); return plugin.definition.onAgentLifecycle(params as AgentLifecycleRequest); @@ -1783,6 +1787,7 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost if (plugin.definition.onIdleDrain) supportedMethods.push("prepareIdleSleep", "releaseIdleSleep"); if (plugin.definition.onApiRequest) supportedMethods.push("handleApiRequest"); if (plugin.definition.onDetectExternalObjects) supportedMethods.push("detectExternalObjects"); + if (plugin.definition.onResourceLifecycle) supportedMethods.push("resourceLifecycle"); if (plugin.definition.onAgentLifecycle) supportedMethods.push("agentLifecycle"); if (plugin.definition.onRouteAiConnection) supportedMethods.push("routeAiConnection"); if (plugin.definition.onResolveExternalObject) supportedMethods.push("resolveExternalObject"); diff --git a/packages/plugins/sdk/tests/worker-rpc-host.test.ts b/packages/plugins/sdk/tests/worker-rpc-host.test.ts index 2895d57ece..0c759222c3 100644 --- a/packages/plugins/sdk/tests/worker-rpc-host.test.ts +++ b/packages/plugins/sdk/tests/worker-rpc-host.test.ts @@ -73,6 +73,38 @@ describe("isWorkerEntrypoint", () => { }); }); +describe("worker resource lifecycle delivery", () => { + it("advertises and dispatches the native lifecycle callback", async () => { + const stdin = new PassThrough(), stdout = new PassThrough(); + const lines = createInterface({ input: stdout }); + const plugin = definePlugin({ + async setup() {}, + async onResourceLifecycle(event) { return { eventId: event.id, status: "retry" }; }, + }); + const worker = startWorkerRpcHost({ plugin, stdin, stdout }); + async function call(method: string, params: unknown) { + const response = new Promise(resolve => { + const handler = (line: string) => { + const message = parseMessage(line); + if (isJsonRpcResponse(message)) { lines.off("line", handler); resolve(message); } + }; + lines.on("line", handler); + }); + stdin.write(serializeMessage(createRequest(method, params, method))); + return response; + } + try { + const initialized = await call("initialize", { + manifest: { id: "test.lifecycle", apiVersion: 1, version: "1.0.0", displayName: "Lifecycle", description: "Test", author: "Tests", categories: ["automation"], capabilities: ["events.subscribe"], resourceLifecycle: ["project"], entrypoints: { worker: "worker.js" } }, + config: {}, databaseNamespace: null, + }); + expect(initialized).toMatchObject({ result: { supportedMethods: expect.arrayContaining(["resourceLifecycle"]) } }); + expect(await call("resourceLifecycle", { id: "17", companyId: "company", resourceType: "project", resourceId: "project", action: "delete", createdAt: "2020-01-01T00:00:00Z" })) + .toMatchObject({ result: { eventId: "17", status: "retry" } }); + } finally { worker.stop(); lines.close(); stdin.destroy(); stdout.destroy(); } + }); +}); + describe("worker performAction context", () => { it("does not derive context companyId from caller params without host actor context", async () => { const hostToWorker = new PassThrough(); diff --git a/packages/shared/src/types/plugin.ts b/packages/shared/src/types/plugin.ts index 95fa9dc8a5..7a44d3aa7f 100644 --- a/packages/shared/src/types/plugin.ts +++ b/packages/shared/src/types/plugin.ts @@ -689,6 +689,8 @@ export interface PaperclipPluginManifestV1 { capabilities: PluginCapability[]; /** Required participant in agent preparation, stop, resume, and cleanup. */ agentLifecycle?: true; + /** Durable, acknowledged resource events delivered to onResourceLifecycle. Requires events.subscribe. */ + resourceLifecycle?: Array<"agent" | "project">; /** Opt into the native pooled-connection catalog, setup, and management UI. */ aiConnectionRouter?: { name: string; description: string }; /** Entrypoint paths relative to the package root. */ diff --git a/packages/shared/src/validators/plugin.test.ts b/packages/shared/src/validators/plugin.test.ts index 683272d130..e4a56eb919 100644 --- a/packages/shared/src/validators/plugin.test.ts +++ b/packages/shared/src/validators/plugin.test.ts @@ -39,6 +39,15 @@ describe("plugin manifest validators", () => { expect(pluginManifestV1Schema.parse({ ...manifest, capabilities: ["agents.lifecycle.manage"] }).agentLifecycle).toBe(true); }); + it("requires subscription authority and known resource types for durable delivery", () => { + const manifest = { id: "example.lifecycle", apiVersion: 1, version: "1.0.0", displayName: "Lifecycle", description: "Lifecycle tests", author: "Tests", categories: ["automation"], entrypoints: { worker: "worker.js" }, resourceLifecycle: ["project"] }; + expect(pluginManifestV1Schema.safeParse({ ...manifest, capabilities: ["projects.read"] }).success).toBe(false); + expect(pluginManifestV1Schema.parse({ ...manifest, capabilities: ["events.subscribe"] }).resourceLifecycle).toEqual(["project"]); + for (const resourceLifecycle of [[], ["issue"]]) { + expect(pluginManifestV1Schema.safeParse({ ...manifest, resourceLifecycle, capabilities: ["events.subscribe"] }).success).toBe(false); + } + }); + it("requires routing authority for native pooled connector declarations without a custom UI bundle", () => { const manifest = { id: "example.pool", apiVersion: 1, version: "0.1.0", displayName: "Pool", description: "Pool", author: "Tests", categories: ["connector"], entrypoints: { worker: "worker.js" }, aiConnectionRouter: { name: "AI connection pool", description: "Use saved connections" } }; expect(pluginManifestV1Schema.safeParse({ ...manifest, capabilities: ["ui.page.register"] }).success).toBe(false); diff --git a/packages/shared/src/validators/plugin.ts b/packages/shared/src/validators/plugin.ts index 8f67466da5..24faa172fb 100644 --- a/packages/shared/src/validators/plugin.ts +++ b/packages/shared/src/validators/plugin.ts @@ -783,6 +783,7 @@ export const pluginManifestV1Schema = z.object({ ).optional(), capabilities: z.array(z.enum(PLUGIN_CAPABILITIES)).min(1), agentLifecycle: z.literal(true).optional(), + resourceLifecycle: z.array(z.enum(["agent", "project"])).min(1).max(2).optional(), aiConnectionRouter: z.object({ name: z.string().trim().min(1).max(100), description: z.string().trim().min(1).max(500) }).strict().optional(), entrypoints: z.object({ worker: z.string().min(1), @@ -811,6 +812,10 @@ export const pluginManifestV1Schema = z.object({ ctx.addIssue({ code: z.ZodIssueCode.custom, message: "Agent lifecycle participation requires agents.lifecycle.manage", path: ["capabilities"] }); } + if (manifest.resourceLifecycle && !manifest.capabilities.includes("events.subscribe")) { + ctx.addIssue({ code: z.ZodIssueCode.custom, message: "Resource lifecycle delivery requires events.subscribe", path: ["capabilities"] }); + } + // ── Entrypoint ↔ UI slot consistency ────────────────────────────────── // Plugins that declare UI slots must also declare a UI entrypoint so the // host knows where to load the bundle from (PLUGIN_SPEC.md §10.1). diff --git a/server/src/__tests__/fixtures/plugin-worker-delayed.cjs b/server/src/__tests__/fixtures/plugin-worker-delayed.cjs index 8aa7e4efa8..48a68cfcc5 100644 --- a/server/src/__tests__/fixtures/plugin-worker-delayed.cjs +++ b/server/src/__tests__/fixtures/plugin-worker-delayed.cjs @@ -20,7 +20,7 @@ rl.on("line", (line) => { id: message.id, result: { ok: true, - supportedMethods: ["environmentExecute"], + supportedMethods: ["environmentExecute", "configChanged"], }, }); return; @@ -44,6 +44,16 @@ rl.on("line", (line) => { return; } + if (method === "configChanged") { + setTimeout(() => send({ + jsonrpc: "2.0", id: message.id, + ...(message.params.config.fail + ? { error: { code: -32000, message: "configuration rejected" } } + : { result: {} }), + }), Number(message.params.config.delayMs ?? 0)); + return; + } + if (method === "shutdown") { send({ jsonrpc: "2.0", diff --git a/server/src/__tests__/plugin-lifecycle-delivery.test.ts b/server/src/__tests__/plugin-lifecycle-delivery.test.ts new file mode 100644 index 0000000000..f745dcdc0e --- /dev/null +++ b/server/src/__tests__/plugin-lifecycle-delivery.test.ts @@ -0,0 +1,217 @@ +import { randomUUID } from "node:crypto"; +import { EventEmitter } from "node:events"; +import { eq } from "drizzle-orm"; +import { companies, createDb, pluginConfig, plugins, resourceLifecycleEvents, type Db } from "@paperclipai/db"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { logger } from "../middleware/logger.js"; +import { registerPluginLifecycleDelivery } from "../services/plugin-lifecycle-delivery.js"; +import { pluginLifecycleInbox } from "../services/plugin-lifecycle-inbox.js"; +import { pluginRegistryService } from "../services/plugin-registry.js"; +import { recordProjectLifecycleEvent, recordResourceCreationEvent } from "../services/resource-lifecycle-events.js"; +import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js"; +import type { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import type { PluginLifecycleManager } from "../services/plugin-lifecycle.js"; +import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const describePostgres = support.supported ? describe : describe.skip; +if (!support.supported) console.warn(`Skipping lifecycle delivery database tests: ${support.reason}`); + +describePostgres("durable plugin lifecycle delivery", () => { + let database: Awaited>; + let db: Db; + let companyId: string; + let pluginId: string; + beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("paperclip-plugin-delivery-"); db = createDb(database.connectionString); }, 90_000); + afterAll(async () => { await database?.cleanup(); }); + afterEach(() => { vi.restoreAllMocks(); }); + beforeEach(async () => { + await db.delete(plugins); + companyId = randomUUID(); pluginId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Delivery fixture", issuePrefix: `L${companyId.slice(0, 6)}` }); + await db.insert(plugins).values({ id: pluginId, pluginKey: pluginId, packageName: "delivery-fixture", version: "1.0.0", status: "ready", + manifestJson: { resourceLifecycle: ["project"], capabilities: ["events.subscribe"] } as never }); + await db.insert(pluginConfig).values({ pluginId, companyId, configJson: {} }); + }); + async function event(resourceId = randomUUID(), action: "create" | "update" | "delete" = "create") { + await db.transaction(async tx => { + if (action === "create") await recordResourceCreationEvent(tx as unknown as Db, companyId, "project", resourceId); + else await recordProjectLifecycleEvent(tx as unknown as Db, companyId, resourceId, action); + }); + return resourceId; + } + function delivery() { + let task!: Parameters["register"]>[1]; + const wake = vi.fn(); + const coordinator = { register: vi.fn((_queue, input) => { task = input; return { wake, ready: Promise.resolve() }; }) }; + const call = vi.fn(async (_pluginId, _method, input) => ({ eventId: input.id, status: "acknowledged" })); + const running = vi.fn(() => true); + const lifecycle = new EventEmitter(); + const configApplied = new Set<() => void>(); + const onConfigApplied = (listener: () => void) => { configApplied.add(listener); return () => { configApplied.delete(listener); }; }; + const worker = registerPluginLifecycleDelivery(coordinator as never, db, { call, isRunning: running, onConfigApplied } as unknown as PluginWorkerManager, lifecycle as unknown as PluginLifecycleManager); + return { call, running, wake, lifecycle, worker, applied: () => { for (const listener of configApplied) listener(); }, run: () => task.run(new AbortController().signal), pending: () => task.hasPending(), nextRunAt: () => task.nextRunAt?.(), settled: () => task.onSettled?.() }; + } + it("drains creation before transitions, recovers acknowledgments across dispatcher replacement, and ignores agent events", async () => { + const id = await event(); await event(id, "update"); await event(id, "delete"); + await db.transaction(tx => recordResourceCreationEvent(tx as unknown as Db, companyId, "agent", randomUUID())); + const first = delivery(); await first.run(); first.worker.dispose(); + expect(first.call.mock.calls.map(c => c[2].action)).toEqual(["create"]); + const restarted = delivery(); await restarted.run(); await restarted.run(); await restarted.run(); + expect(restarted.call.mock.calls.map(c => c[2].action)).toEqual(["update", "delete"]); + expect(await restarted.pending()).toBe(false); + expect((await pluginLifecycleInbox(db, pluginId).list(companyId)).map(e => e.resourceType)).toEqual(["agent"]); + restarted.worker.dispose(); + }); + it("replays failures and invalid acknowledgments without blocking independent resources", async () => { + let now = Date.now(); vi.spyOn(Date, "now").mockImplementation(() => now); + const failed = await event(); await event(failed, "delete"); await event(); + const d = delivery(); + d.call.mockImplementationOnce(async () => { throw new Error("worker exited after provider effect"); }); + await d.run(); + expect(d.call).toHaveBeenCalledTimes(2); expect(await d.pending()).toBe(true); + d.call.mockImplementationOnce(async () => ({ eventId: "wrong", status: "acknowledged" })); + now += 30_000; await d.run(); expect(await d.pending()).toBe(true); + now += 30_000; await d.run(); await d.run(); + expect(d.call.mock.calls.map(c => c[2].action)).toEqual(["create", "create", "create", "create", "delete"]); + expect(await pluginLifecycleInbox(db, pluginId).list(companyId)).toEqual([]); + d.worker.dispose(); + }); + it("healthy acknowledgments and new events do not bypass a failed event's retry deadline", async () => { + let now = Date.now(); vi.spyOn(Date, "now").mockImplementation(() => now); + const failed = await event(), healthy = await event(); + await event(healthy, "update"); await event(healthy, "delete"); + const d = delivery(); + d.call.mockImplementation(async (_p, _m, e) => ({ eventId: e.id, status: e.resourceId === failed ? "retry" : "acknowledged" })); + await d.run(); + const deadline = now + 30_000; + expect(d.nextRunAt()).toBe(deadline); + now += 1000; + await d.run(); await d.run(); await event(); d.settled(); await d.run(); + expect(d.call.mock.calls.filter(c => c[2].resourceId === healthy).map(c => c[2].action)).toEqual(["create", "update", "delete"]); + expect(d.call.mock.calls.filter(c => c[2].resourceId === failed)).toHaveLength(1); + expect(d.nextRunAt()).toBe(deadline); + now = deadline; + expect(d.nextRunAt()).toBeNull(); // Paused dispatch waits for the normal retry, not a zero-delay loop. + await d.run(); + expect(d.call.mock.calls.filter(c => c[2].resourceId === failed)).toHaveLength(2); + expect(d.nextRunAt()).toBe(now + 30_000); + d.worker.dispose(); + }); + it("defers without a retry deadline, retries unfinished work, and wakes on plugin startup", async () => { + await event(); const d = delivery(); + d.call.mockImplementationOnce(async (_p, _m, e) => ({ eventId: e.id, status: "deferred" })); + await d.run(); expect(await d.pending()).toBe(false); expect(d.wake).not.toHaveBeenCalled(); + d.lifecycle.emit("plugin.worker_started"); expect(d.wake).toHaveBeenCalledTimes(1); + d.call.mockImplementationOnce(async (_p, _m, e) => ({ eventId: e.id, status: "retry" })); + await d.run(); expect(await d.pending()).toBe(true); + d.running.mockReturnValue(false); await d.run(); expect(await d.pending()).toBe(true); + expect(d.call).toHaveBeenCalledTimes(2); d.worker.dispose(); + d.lifecycle.emit("plugin.worker_started"); expect(d.wake).toHaveBeenCalledTimes(1); + }); + it("requires ready, configured, company-enabled plugins and rechecks disabling within a batch", async () => { + await event(); await event(); const d = delivery(); const registry = pluginRegistryService(db); + await registry.upsertCompanySettings(pluginId, companyId, { enabled: false, settingsJson: {} }); + await d.run(); expect(d.call).not.toHaveBeenCalled(); + await registry.upsertCompanySettings(pluginId, companyId, { enabled: true, settingsJson: {} }); + d.call.mockImplementationOnce(async (_p, _m, e) => { + await registry.upsertCompanySettings(pluginId, companyId, { enabled: false, settingsJson: {} }); + return { eventId: e.id, status: "acknowledged" }; + }); + await d.run(); expect(d.call).toHaveBeenCalledTimes(1); + await registry.upsertCompanySettings(pluginId, companyId, { enabled: true, settingsJson: {} }); + await db.update(plugins).set({ status: "disabled" }).where(eq(plugins.id, pluginId)); + await d.run(); expect(d.call).toHaveBeenCalledTimes(1); + await db.update(plugins).set({ status: "ready" }).where(eq(plugins.id, pluginId)); + await db.delete(pluginConfig).where(eq(pluginConfig.pluginId, pluginId)); + await d.run(); expect(d.call).toHaveBeenCalledTimes(1); d.worker.dispose(); + }); + it("notifies only after the outer transaction settles and wakes deferred work for config changes", async () => { + const wake = vi.fn(); const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle, wake); + try { + await db.transaction(async tx => { + await eventInTransaction(tx as unknown as Db); + expect(wake).not.toHaveBeenCalled(); + }); + expect(wake).toHaveBeenCalledTimes(1); + const registry = pluginRegistryService(db); + await registry.upsertConfig(pluginId, companyId, { configJson: { enabled: true } }); + await registry.patchConfig(pluginId, companyId, { configJson: { enabled: false } }); + await registry.upsertCompanySettings(pluginId, companyId, { enabled: true, settingsJson: {} }); + expect(wake).toHaveBeenCalledTimes(4); + } finally { unsubscribe(); } + async function eventInTransaction(tx: Db) { + await tx.transaction(inner => recordResourceCreationEvent(inner as unknown as Db, companyId, "project", randomUUID())); + } + }); + it("retains startup and settled work while idle drain prevents dispatch", async () => { + const d = delivery(); + // The coordinator can call hasPending without run when admission is paused. + expect(await d.pending()).toBe(true); + await d.run(); expect(await d.pending()).toBe(false); + await event(); d.settled(); + expect(await d.pending()).toBe(true); + expect(d.call).not.toHaveBeenCalled(); + await d.run(); + expect(d.call).toHaveBeenCalledTimes(1); + expect(await d.pending()).toBe(true); // An acknowledgment must drain its successor. + await d.run(); expect(await d.pending()).toBe(false); + d.lifecycle.emit("plugin.enabled"); expect(await d.pending()).toBe(true); + await d.run(); expect(await d.pending()).toBe(false); + d.worker.wake(); expect(await d.pending()).toBe(true); + d.worker.dispose(); + }); + it("wakes deferred events after live configuration finishes applying", async () => { + await event(); const d = delivery(); + let configured = false; + d.call.mockImplementation(async (_p, _m, e) => ({ eventId: e.id, status: configured ? "acknowledged" : "deferred" })); + await d.run(); expect(await d.pending()).toBe(false); + // Commit notification races ahead of the asynchronous worker callback. + d.settled(); await d.run(); expect(await d.pending()).toBe(false); + configured = true; d.applied(); + expect(await d.pending()).toBe(true); expect(d.wake).toHaveBeenCalledTimes(1); + await d.run(); expect(await pluginLifecycleInbox(db, pluginId).list(companyId)).toEqual([]); + d.worker.dispose(); d.wake.mockClear(); d.applied(); expect(d.wake).not.toHaveBeenCalled(); + }); + it("retains a failed scan across a retry skipped during idle drain", async () => { + await event(); const d = delivery(); + const select = vi.spyOn(db, "select").mockImplementationOnce(() => { throw new Error("database unavailable"); }); + await expect(d.run()).rejects.toThrow("database unavailable"); + // The coordinator's drain path checks hasPending without calling run. + expect(await d.pending()).toBe(true); expect(d.call).not.toHaveBeenCalled(); + select.mockRestore(); await d.run(); await d.run(); + expect(await d.pending()).toBe(false); + expect(await pluginLifecycleInbox(db, pluginId).list(companyId)).toEqual([]); + d.worker.dispose(); + }); + it("saves config and settings without borrowing another pool connection", async () => { + const single = createDb(database.connectionString, { maxConnections: 1 }); + const registry = pluginRegistryService(single); + await registry.upsertConfig(pluginId, companyId, { configJson: { first: true } }); + await registry.patchConfig(pluginId, companyId, { configJson: { second: true } }); + await registry.upsertCompanySettings(pluginId, companyId, { enabled: true, settingsJson: {} }); + expect((await registry.getConfig(pluginId, companyId))?.configJson).toEqual({ first: true, second: true }); + }, 5000); + it("records failed delivery scope and stage without provider secrets", async () => { + await event(); const d = delivery(); + const warn = vi.spyOn(logger, "warn").mockImplementation(() => {}); + try { + d.call.mockImplementationOnce(async () => { throw Object.assign(new Error("provider-secret-fixture"), { code: -32001 }); }); + await d.run(); + expect(warn).toHaveBeenCalledWith(expect.objectContaining({ pluginId, companyId, eventId: expect.any(String), stage: "callback", rpcCode: -32001 }), expect.any(String)); + expect(JSON.stringify(warn.mock.calls)).not.toContain("provider-secret-fixture"); + expect(await d.pending()).toBe(true); + } finally { warn.mockRestore(); d.worker.dispose(); } + }); + it("rollbacks never expose queue rows even when settlement wakes the dispatcher", async () => { + const d = delivery(); + await expect(db.transaction(async tx => { + await recordResourceCreationEvent(tx as unknown as Db, companyId, "project", randomUUID()); + throw new Error("rollback"); + })).rejects.toThrow("rollback"); + await d.run(); expect(d.call).not.toHaveBeenCalled(); expect(await d.pending()).toBe(false); + expect(await db.select().from(resourceLifecycleEvents).where(eq(resourceLifecycleEvents.companyId, companyId))).toEqual([]); + d.worker.dispose(); + }); +}); diff --git a/server/src/__tests__/plugin-worker-manager.test.ts b/server/src/__tests__/plugin-worker-manager.test.ts index 58b0c9a668..c97c14a835 100644 --- a/server/src/__tests__/plugin-worker-manager.test.ts +++ b/server/src/__tests__/plugin-worker-manager.test.ts @@ -31,6 +31,7 @@ import { appendStderrExcerpt, createDuplexRouteSlotController, createPluginWorkerHandle, + createPluginWorkerManager, formatWorkerFailureMessage, resolveRpcCallTimeoutMs, } from "../services/plugin-worker-manager.js"; @@ -56,6 +57,28 @@ const TEST_MANIFEST: PaperclipPluginManifestV1 = { entrypoints: { worker: "dist/worker.js" }, }; +describe("worker configuration completion", () => { + it("notifies only after a successful config callback and releases subscriptions", async () => { + const manager = createPluginWorkerManager(); + const applied = vi.fn(), unsubscribe = manager.onConfigApplied(applied); + try { + await manager.startWorker("test.plugin", { + entrypointPath: DELAYED_WORKER_ENTRYPOINT, manifest: TEST_MANIFEST, config: {}, + instanceInfo: { instanceId: "instance-1", hostVersion: "1.0.0" }, apiVersion: 1, hostHandlers: {}, + }); + expect(applied).not.toHaveBeenCalled(); + const updating = manager.call("test.plugin", "configChanged", { companyId: "company-1", config: { delayMs: 20 } }); + expect(applied).not.toHaveBeenCalled(); + await updating; expect(applied).toHaveBeenCalledTimes(1); + await expect(manager.call("test.plugin", "configChanged", { companyId: "company-1", config: { fail: true } })).rejects.toThrow("configuration rejected"); + expect(applied).toHaveBeenCalledTimes(1); + unsubscribe(); + await manager.call("test.plugin", "configChanged", { companyId: "company-1", config: {} }); + expect(applied).toHaveBeenCalledTimes(1); + } finally { unsubscribe(); await manager.stopAll(); } + }); +}); + describe("resolveRpcCallTimeoutMs", () => { const MAX_RPC_TIMEOUT_MS = 15 * 60 * 1_000; const MAX_NODE_TIMER_TIMEOUT_MS = 2_147_483_647; diff --git a/server/src/__tests__/resource-lifecycle-events.test.ts b/server/src/__tests__/resource-lifecycle-events.test.ts index dd54856dfd..f7b263d40f 100644 --- a/server/src/__tests__/resource-lifecycle-events.test.ts +++ b/server/src/__tests__/resource-lifecycle-events.test.ts @@ -53,7 +53,7 @@ describePostgres("Resource lifecycle events", () => { it("records direct and approval-free hires once, including concurrent duplicate submissions", async () => { const agent = await createAgent(); - await Promise.all(Array.from({ length: 5 }, () => recordResourceCreationEvent(db, companyId, "agent", agent.id))); + await Promise.all(Array.from({ length: 5 }, () => db.transaction(tx => recordResourceCreationEvent(tx as unknown as Db, companyId, "agent", agent.id)))); expect(await events()).toEqual([expect.objectContaining({ resourceType: "agent", resourceId: agent.id, action: "create" })]); }); @@ -235,11 +235,11 @@ describePostgres("Resource lifecycle events", () => { vi.stubEnv("PAPERCLIP_MANAGED_CONFIG", undefined); vi.stubEnv("PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN", "fixture-token"); const id = randomUUID(); - await recordResourceCreationEvent(db, companyId, "agent", id); - await recordResourceCreationEvent(db, companyId, "project", id); + await db.transaction(tx => recordResourceCreationEvent(tx as unknown as Db, companyId, "agent", id)); + await db.transaction(tx => recordResourceCreationEvent(tx as unknown as Db, companyId, "project", id)); const otherCompanyId = randomUUID(); await db.insert(companies).values({ id: otherCompanyId, name: "Other company", issuePrefix: "OTHER" }); - await recordResourceCreationEvent(db, otherCompanyId, "agent", id); + await db.transaction(tx => recordResourceCreationEvent(tx as unknown as Db, otherCompanyId, "agent", id)); expect(await events()).toHaveLength(2); expect(await db.select().from(resourceLifecycleEvents).where(eq(resourceLifecycleEvents.companyId, otherCompanyId))).toHaveLength(1); await db.delete(companies).where(eq(companies.id, otherCompanyId)); @@ -267,6 +267,23 @@ describePostgres("Resource lifecycle events", () => { expect(beforeArchive.sort((a, b) => a.id - b.id).map(row => row.action)).toEqual(["create", "update", "update", "update", "update", "update"]); }); + it("records deletion atomically and retains the event after removing the project", async () => { + const service = projectService(db); + const project = await service.create(companyId, { name: "Delete fixture" }); + await db.execute(sql`ALTER TABLE resource_lifecycle_events ADD CONSTRAINT fixture_reject_delete CHECK (action <> 'delete') NOT VALID`); + try { + await expect(service.remove(project.id)).rejects.toThrow(); + expect(await service.getById(project.id)).not.toBeNull(); + expect((await events()).map(row => row.action)).toEqual(["create"]); + } finally { + await db.execute(sql`ALTER TABLE resource_lifecycle_events DROP CONSTRAINT fixture_reject_delete`); + } + await service.remove(project.id); + await service.remove(project.id); + expect(await service.getById(project.id)).toBeNull(); + expect((await events()).map(row => row.action)).toEqual(["create", "delete"]); + }); + it("records archive once per transition and update on restore, retaining repositories", async () => { const service = projectService(db); const project = await service.createWithRepositories(companyId, { name: "Archive fixture" }, [ diff --git a/server/src/app.ts b/server/src/app.ts index c45961b203..270c368602 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -191,6 +191,7 @@ import { import { subscribeAllCompanyLiveEvents } from "./services/live-events.js"; import { heartbeatService } from "./services/heartbeat.js"; import { pluginLifecycleManager } from "./services/plugin-lifecycle.js"; +import { registerPluginLifecycleDelivery } from "./services/plugin-lifecycle-delivery.js"; import { createPluginJobCoordinator } from "./services/plugin-job-coordinator.js"; import { buildHostServices, @@ -1251,6 +1252,7 @@ export async function createApp( onError: (err, queue) => logger.error({ err, queue }, "Delivery reconciliation failed"), }); app.locals.deliveryWork = deliveryWork; + const resourceLifecycleDelivery = registerPluginLifecycleDelivery(deliveryWork, db, workerManager, lifecycle); if (opts.feedbackExportService) { deliveryWork.register(DELIVERY_QUEUES.feedback, { retryMs: FEEDBACK_EXPORT_FLUSH_INTERVAL_MS, @@ -1409,7 +1411,7 @@ export async function createApp( logger.error({ err }, "Failed to load ready plugins on startup"); }); app.locals.bundledPluginsStartup = trackIdleWork(bundledPluginsStartup); - void bundledPluginsStartup.then(() => { lifecyclePluginsReady = true; return agentLifecycle.sweep(); }).catch(() => logger.warn("Agent lifecycle recovery failed; retrying.")); + void bundledPluginsStartup.then(() => { lifecyclePluginsReady = true; resourceLifecycleDelivery.wake(); return agentLifecycle.sweep(); }).catch(() => logger.warn("Agent lifecycle recovery failed; retrying.")); // The shutdown hook runs at most once. It caches the in-flight promise, so a // second caller (for example the `exit` handler) awaits the same completion // instead of starting a second teardown. @@ -1424,6 +1426,7 @@ export async function createApp( await publicMcpEvents?.stop(); await dotMcpEvents?.stop(); jobCoordinator.stop(); + resourceLifecycleDelivery.dispose(); await deliveryWork.stop(); unsubscribeChatPublicationSignals(); chatReconciliation.stop(); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 2bc6840e69..315b548b40 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -23,6 +23,7 @@ export const DELIVERY_QUEUES = { chatCompletion: "chat-completions", connection: "connection-continuations", question: "question-responses", + pluginLifecycle: "plugin-resource-lifecycle", toolAction: "tool-action-receipts", } as const; export type DeliveryQueue = typeof DELIVERY_QUEUES[keyof typeof DELIVERY_QUEUES]; diff --git a/server/src/services/plugin-lifecycle-delivery.ts b/server/src/services/plugin-lifecycle-delivery.ts new file mode 100644 index 0000000000..24c5dbc26d --- /dev/null +++ b/server/src/services/plugin-lifecycle-delivery.ts @@ -0,0 +1,122 @@ +import { and, asc, eq, isNull, or } from "drizzle-orm"; +import { pluginCompanySettings, pluginConfig, plugins, type Db } from "@paperclipai/db"; +import { logger } from "../middleware/logger.js"; +import { pluginLifecycleInbox } from "./plugin-lifecycle-inbox.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import type { PluginWorkerManager } from "./plugin-worker-manager.js"; +import type { PluginLifecycleManager } from "./plugin-lifecycle.js"; + +type Coordinator = ReturnType; + +/** The journal is the queue. Notifications only wake its single local dispatcher. */ +export function registerPluginLifecycleDelivery( + coordinator: Coordinator, + db: Db, + manager: Pick, + lifecycle: Pick, +) { + // Startup and every wake remain pending until an admitted run inspects the inbox. + // The coordinator may reconcile signals while idle drain prevents dispatch. + let pending = true; + const retryAt = new Map(); + const retryMs = 30_000; + const delivery = coordinator.register(DELIVERY_QUEUES.pluginLifecycle, { + retryMs, + hasPending: async () => pending, + onSettled: () => { pending = true; }, + nextRunAt() { + if (!pending) return null; + let next: number | null = null; + for (const at of retryAt.values()) { + // An expired deadline during idle drain must not schedule a busy loop. + if (at > Date.now() && (next === null || at < next)) next = at; + } + return next; + }, + async run(signal) { + pending = false; + try { + for (const [key, at] of retryAt) if (at <= Date.now()) retryAt.delete(key); + const configured = await db.select({ plugin: plugins, companyId: pluginConfig.companyId }) + .from(plugins).innerJoin(pluginConfig, eq(pluginConfig.pluginId, plugins.id)) + .where(eq(plugins.status, "ready")) + .orderBy(asc(plugins.installOrder), asc(pluginConfig.companyId)); + for (const { plugin, companyId } of configured) { + const types = plugin.manifestJson.resourceLifecycle; + if (!types?.length || !plugin.manifestJson.capabilities.includes("events.subscribe")) continue; + const [disabled] = await db.select({ id: pluginCompanySettings.id }).from(pluginCompanySettings) + .where(and(eq(pluginCompanySettings.pluginId, plugin.id), eq(pluginCompanySettings.companyId, companyId), eq(pluginCompanySettings.enabled, false))); + if (disabled) continue; + const inbox = pluginLifecycleInbox(db, plugin.id); + let afterId: string | undefined; + // Visit independent resources past a failed or deferred head. + for (;;) { + if (signal.aborted) return; + const events = await inbox.list(companyId, 100, afterId, types); + for (const event of events) { + if (signal.aborted) return; + const retryKey = `${plugin.id}:${event.id}`; + if (retryAt.has(retryKey)) { pending = true; continue; } + if (!manager.isRunning(plugin.id)) { pending = true; break; } + let stage = "eligibility"; + try { + // Recheck enablement for each invocation; a disable during a batch must stop delivery. + const [enabled] = await db.select({ manifest: plugins.manifestJson }).from(plugins) + .leftJoin(pluginCompanySettings, and(eq(pluginCompanySettings.pluginId, plugins.id), eq(pluginCompanySettings.companyId, companyId))) + .where(and(eq(plugins.id, plugin.id), eq(plugins.status, "ready"), + or(isNull(pluginCompanySettings.enabled), eq(pluginCompanySettings.enabled, true)))); + if (!enabled?.manifest.resourceLifecycle?.includes(event.resourceType) || !enabled.manifest.capabilities.includes("events.subscribe")) break; + stage = "callback"; + const result = await manager.call(plugin.id, "resourceLifecycle", event, 30_000); + stage = "result"; + if (result?.eventId !== event.id || !["acknowledged", "retry", "deferred"].includes(result.status)) { + throw new Error("Invalid resource lifecycle result"); + } + if (signal.aborted) return; + if (result.status === "acknowledged") { + stage = "acknowledgment"; + await inbox.acknowledge(companyId, event.id); + // An acknowledgment exposes the next event for this resource. Drain immediately. + wake(); + } else if (result.status === "retry") { + retryAt.set(retryKey, Date.now() + retryMs); + pending = true; + } + } catch (error) { + retryAt.set(retryKey, Date.now() + retryMs); + // Provider error messages/data may contain secrets; retain only routing + // identifiers, the failed stage, and the protocol's numeric error code. + const code = (error as { code?: unknown } | null)?.code; + logger.warn({ pluginId: plugin.id, companyId, eventId: event.id, stage, + ...(typeof code === "number" ? { rpcCode: code } : {}) }, "Resource lifecycle delivery failed; retry scheduled"); + // Failed RPCs and uncertain acknowledgments leave durable work for replay. + pending = true; + } + } + if (events.length < 100) break; + afterId = events.at(-1)!.id; + } + } + } catch (error) { + pending = true; // A failed scan must survive a retry skipped during idle drain. + throw error; + } + }, + }); + function wake() { pending = true; delivery.wake(); } + const unsubscribeConfig = manager.onConfigApplied(wake); + lifecycle.on("plugin.loaded", wake); + lifecycle.on("plugin.enabled", wake); + lifecycle.on("plugin.worker_started", wake); + return { + ...delivery, + wake, + dispose() { + unsubscribeConfig(); + lifecycle.off("plugin.loaded", wake); + lifecycle.off("plugin.enabled", wake); + lifecycle.off("plugin.worker_started", wake); + }, + }; +} diff --git a/server/src/services/plugin-lifecycle-inbox.ts b/server/src/services/plugin-lifecycle-inbox.ts index 7187d83b06..e7aa0a4bf7 100644 --- a/server/src/services/plugin-lifecycle-inbox.ts +++ b/server/src/services/plugin-lifecycle-inbox.ts @@ -1,10 +1,10 @@ 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 { and, asc, eq, gt, inArray, sql } from "drizzle-orm"; import { badRequest, conflict, forbidden, notFound } from "../errors.js"; -/** Durable pull delivery. Provider work stays outside tenant transactions. */ +/** Durable delivery inbox. 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 @@ -31,12 +31,12 @@ export function pluginLifecycleInbox(db: Db, pluginId: string) { } return { - async list(companyId: string, limit = 50, afterId?: string): Promise { + async list(companyId: string, limit = 50, afterId?: string, resourceTypes?: ResourceLifecycleEvent["resourceType"][]): 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)))) + .where(and(eq(resourceLifecycleEvents.companyId, companyId), unacknowledged, firstForResource, resourceTypes ? inArray(resourceLifecycleEvents.resourceType, resourceTypes) : undefined, 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[]; }, diff --git a/server/src/services/plugin-registry.ts b/server/src/services/plugin-registry.ts index f86a550b94..2ff529cc2d 100644 --- a/server/src/services/plugin-registry.ts +++ b/server/src/services/plugin-registry.ts @@ -28,6 +28,7 @@ import type { PluginWebhookDeliveryStatus, } from "@paperclipai/shared"; import { conflict, notFound } from "../errors.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; // --------------------------------------------------------------------------- // Helpers @@ -65,8 +66,8 @@ export function pluginRegistryService(db: Db) { // Internal helpers // ----------------------------------------------------------------------- - async function getById(id: string) { - return db + async function getById(id: string, database: Db = db) { + return database .select() .from(plugins) .where(eq(plugins.id, id)) @@ -323,37 +324,41 @@ export function pluginRegistryService(db: Db) { * otherwise a new row is inserted. */ upsertConfig: async (pluginId: string, companyId: string, input: UpsertPluginConfig) => { - const plugin = await getById(pluginId); - if (!plugin) throw notFound("Plugin not found"); + return db.transaction(async tx => { + const db = tx as unknown as Db; + await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle); + const plugin = await getById(pluginId, db); + if (!plugin) throw notFound("Plugin not found"); - const existing = await db - .select() - .from(pluginConfig) - .where(and(eq(pluginConfig.pluginId, pluginId), eq(pluginConfig.companyId, companyId))) - .then((rows) => rows[0] ?? null); - - if (existing) { - return db - .update(pluginConfig) - .set({ - configJson: input.configJson, - lastError: null, - updatedAt: new Date(), - }) + const existing = await db + .select() + .from(pluginConfig) .where(and(eq(pluginConfig.pluginId, pluginId), eq(pluginConfig.companyId, companyId))) + .then((rows) => rows[0] ?? null); + + if (existing) { + return db + .update(pluginConfig) + .set({ + configJson: input.configJson, + lastError: null, + updatedAt: new Date(), + }) + .where(and(eq(pluginConfig.pluginId, pluginId), eq(pluginConfig.companyId, companyId))) + .returning() + .then((rows) => rows[0]); + } + + return db + .insert(pluginConfig) + .values({ + pluginId, + companyId, + configJson: input.configJson, + }) .returning() .then((rows) => rows[0]); - } - - return db - .insert(pluginConfig) - .values({ - pluginId, - companyId, - configJson: input.configJson, - }) - .returning() - .then((rows) => rows[0]); + }); }, /** @@ -361,38 +366,42 @@ export function pluginRegistryService(db: Db) { * If no config row exists yet one is created with the supplied values. */ patchConfig: async (pluginId: string, companyId: string, input: PatchPluginConfig) => { - const plugin = await getById(pluginId); - if (!plugin) throw notFound("Plugin not found"); + return db.transaction(async tx => { + const db = tx as unknown as Db; + await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle); + const plugin = await getById(pluginId, db); + if (!plugin) throw notFound("Plugin not found"); - const existing = await db - .select() - .from(pluginConfig) - .where(and(eq(pluginConfig.pluginId, pluginId), eq(pluginConfig.companyId, companyId))) - .then((rows) => rows[0] ?? null); - - if (existing) { - const merged = { ...existing.configJson, ...input.configJson }; - return db - .update(pluginConfig) - .set({ - configJson: merged, - lastError: null, - updatedAt: new Date(), - }) + const existing = await db + .select() + .from(pluginConfig) .where(and(eq(pluginConfig.pluginId, pluginId), eq(pluginConfig.companyId, companyId))) + .then((rows) => rows[0] ?? null); + + if (existing) { + const merged = { ...existing.configJson, ...input.configJson }; + return db + .update(pluginConfig) + .set({ + configJson: merged, + lastError: null, + updatedAt: new Date(), + }) + .where(and(eq(pluginConfig.pluginId, pluginId), eq(pluginConfig.companyId, companyId))) + .returning() + .then((rows) => rows[0]); + } + + return db + .insert(pluginConfig) + .values({ + pluginId, + companyId, + configJson: input.configJson, + }) .returning() .then((rows) => rows[0]); - } - - return db - .insert(pluginConfig) - .values({ - pluginId, - companyId, - configJson: input.configJson, - }) - .returning() - .then((rows) => rows[0]); + }); }, /** @@ -439,43 +448,47 @@ export function pluginRegistryService(db: Db) { companyId: string, input: { enabled?: boolean; settingsJson: Record; lastError?: string | null }, ): Promise => { - const plugin = await getById(pluginId); - if (!plugin) throw notFound("Plugin not found"); + return db.transaction(async tx => { + const db = tx as unknown as Db; + await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle); + const plugin = await getById(pluginId, db); + if (!plugin) throw notFound("Plugin not found"); - const existing = await db - .select() - .from(pluginCompanySettings) - .where(and( - eq(pluginCompanySettings.pluginId, pluginId), - eq(pluginCompanySettings.companyId, companyId), - )) - .then((rows) => rows[0] ?? null); + const existing = await db + .select() + .from(pluginCompanySettings) + .where(and( + eq(pluginCompanySettings.pluginId, pluginId), + eq(pluginCompanySettings.companyId, companyId), + )) + .then((rows) => rows[0] ?? null); + + if (existing) { + return db + .update(pluginCompanySettings) + .set({ + enabled: input.enabled ?? existing.enabled, + settingsJson: input.settingsJson, + lastError: input.lastError ?? null, + updatedAt: new Date(), + }) + .where(eq(pluginCompanySettings.id, existing.id)) + .returning() + .then((rows) => rows[0]) as Promise; + } - if (existing) { return db - .update(pluginCompanySettings) - .set({ - enabled: input.enabled ?? existing.enabled, + .insert(pluginCompanySettings) + .values({ + pluginId, + companyId, + enabled: input.enabled ?? true, settingsJson: input.settingsJson, lastError: input.lastError ?? null, - updatedAt: new Date(), }) - .where(eq(pluginCompanySettings.id, existing.id)) .returning() .then((rows) => rows[0]) as Promise; - } - - return db - .insert(pluginCompanySettings) - .values({ - pluginId, - companyId, - enabled: input.enabled ?? true, - settingsJson: input.settingsJson, - lastError: input.lastError ?? null, - }) - .returning() - .then((rows) => rows[0]) as Promise; + }); }, // ----- Entities ------------------------------------------------------- diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index 2d3f961d53..5569f80c65 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -796,6 +796,8 @@ export interface WorkerDiagnostics { * for starting/stopping all workers and routing RPC calls. */ export interface PluginWorkerManager { + /** Subscribe after a worker successfully applies live company configuration. */ + onConfigApplied(listener: () => void): () => void; inspectIdleSleep?(hold: { ownerId: string; expiresAt: number }): Promise<{ backgroundWork: "none" | "present" | "unknown"; pluginIds: string[]; }>; @@ -3726,6 +3728,7 @@ export function createPluginWorkerManager( ): PluginWorkerManager { const log = logger.child({ service: "plugin-worker-manager" }); const workers = new Map(); + const configApplied = new Set<() => void>(); /** Per-plugin startup locks to prevent concurrent spawn races. */ const startupLocks = new Map>(); // The one shared, process-scoped aggregate route-slot controller. The manager @@ -3736,6 +3739,10 @@ export function createPluginWorkerManager( ); return { + onConfigApplied(listener) { + configApplied.add(listener); + return () => { configApplied.delete(listener); }; + }, async inspectIdleSleep(hold) { if (startupLocks.size) return { backgroundWork: "unknown", pluginIds: [] }; const snapshot = [...workers.entries()]; @@ -3872,7 +3879,14 @@ export function createPluginWorkerManager( new Error(`No worker registered for plugin "${pluginId}"`), ); } - return handle.call(method, params, timeoutMs, executeLogSink); + const result = handle.call(method, params, timeoutMs, executeLogSink); + if (method === "configChanged") void result.then(() => { + for (const listener of configApplied) { + try { listener(); } + catch { log.warn({ pluginId }, "Plugin configuration listener failed"); } + } + }, () => {}); + return result; }, openLoginPtySession(pluginId: string, input: LoginPtyOpenInput) { diff --git a/server/src/services/projects.ts b/server/src/services/projects.ts index cdd88c948c..67c320f530 100644 --- a/server/src/services/projects.ts +++ b/server/src/services/projects.ts @@ -1073,6 +1073,7 @@ export function projectService(db: Db, options: { captureWorkspaceUpdates?: bool if (retained) throw conflict("This project still supplies access policy for retained task workspaces. Archive the project or explicitly clean up its workspaces before deleting it."); const workspaces = await tx.select().from(projectWorkspaces).where(eq(projectWorkspaces.projectId, id)).for("update"); for (const workspace of workspaces) await preserveConfiguredProjectWorkspaceRootProvenance(tx, workspace); + await recordProjectLifecycleEvent(tx as unknown as Db, project.companyId, id, "delete"); const [row] = await tx.delete(projects).where(eq(projects.id, id)).returning(); return row ? { ...row, urlKey: deriveProjectUrlKey(row.name, row.id) } : null; }), diff --git a/server/src/services/resource-lifecycle-events.ts b/server/src/services/resource-lifecycle-events.ts index 73bd9444fb..7d4c29692f 100644 --- a/server/src/services/resource-lifecycle-events.ts +++ b/server/src/services/resource-lifecycle-events.ts @@ -1,5 +1,6 @@ import { resourceLifecycleEvents, type Db } from "@paperclipai/db"; import { sql } from "drizzle-orm"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; /** Call inside the transaction that creates the resource or approves the hire. */ export async function recordResourceCreationEvent( @@ -8,6 +9,7 @@ export async function recordResourceCreationEvent( resourceType: "agent" | "project", resourceId: string, ): Promise { + await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle); await db.insert(resourceLifecycleEvents).values({ companyId, resourceType, resourceId, action: "create" }).onConflictDoNothing({ target: [resourceLifecycleEvents.companyId, resourceLifecycleEvents.resourceType, resourceLifecycleEvents.resourceId], where: sql`${resourceLifecycleEvents.action} = 'create'`, @@ -21,10 +23,12 @@ export async function recordAgentStatusEvent(db: Db, companyId: string, agentId: : after === "paused" ? "pause" : before === "paused" && ["active", "idle", "running", "error"].includes(after) ? "resume" : null; if (!action) return; + await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle); await db.insert(resourceLifecycleEvents).values({ companyId, resourceType: "agent", resourceId: agentId, action }); } /** Call while holding the project row lock, in the mutation transaction. */ -export async function recordProjectLifecycleEvent(db: Db, companyId: string, projectId: string, action: "update" | "archive" = "update"): Promise { +export async function recordProjectLifecycleEvent(db: Db, companyId: string, projectId: string, action: "update" | "archive" | "delete" = "update"): Promise { + await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle); await db.insert(resourceLifecycleEvents).values({ companyId, resourceType: "project", resourceId: projectId, action }); }