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 <noreply@paperclip.ing>
This commit is contained in:
Nicky LeachandPaperclip authored and GitHub committed 2026-10-06 12:27:09 -07:00
1 parent 854af7df19
commit 1cebdd4c4f
23 files changed
+930 -20

No files matched your search

+23 -1
View File
@@ -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");
},
);
+34 -5
View File
@@ -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
+43
View File
@@ -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
+28
View File
@@ -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 () => {
+35 -6
View File
@@ -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<SequenceDefinition[]>`
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<SequenceDefinition[]>`
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();
}
@@ -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';
@@ -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": "",
+7
View File
@@ -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
}
]
}
+1
View File
@@ -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";
@@ -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),
}));
@@ -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'))`),
}));
@@ -139,6 +139,8 @@ export interface HostServices {
/** Provides `events.emit` and `events.subscribe`. */
events: {
listLifecycle(params: WorkerToHostMethods["events.listLifecycle"][0]): Promise<WorkerToHostMethods["events.listLifecycle"][1]>;
acknowledgeLifecycle(params: WorkerToHostMethods["events.acknowledgeLifecycle"][0]): Promise<void>;
emit(params: WorkerToHostMethods["events.emit"][0]): Promise<void>;
subscribe(params: WorkerToHostMethods["events.subscribe"][0]): Promise<void>;
};
@@ -405,6 +407,8 @@ const METHOD_CAPABILITY_MAP: Record<WorkerToHostMethodName, PluginCapability | n
// Events
"events.emit": "events.emit",
"events.subscribe": "events.subscribe",
"events.listLifecycle": "events.subscribe",
"events.acknowledgeLifecycle": "events.subscribe",
// HTTP
"http.fetch": "http.outbound",
@@ -762,6 +766,8 @@ export function createHostClientHandlers(
}),
// Events
"events.listLifecycle": gated("events.listLifecycle", async (params) => 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);
}),
+1
View File
@@ -238,6 +238,7 @@ export type {
PluginLocalFolderListing,
PluginLocalFoldersClient,
PluginEventsClient,
ResourceLifecycleEvent,
PluginJobsClient,
PluginLaunchersClient,
PluginHttpClient,
+3
View File
@@ -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,
+33
View File
@@ -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<string, PluginEntityRecord>();
const entityExternalIndex = new Map<string, string>();
const companies = new Map<string, Company>();
const lifecycleEvents = new Map<string, ResourceLifecycleEvent>();
const lifecycleAcknowledgments = new Set<string>();
const projects = new Map<string, Project>();
const routines = new Map<string, Routine>();
const routineRuns = new Map<string, RoutineRun>();
@@ -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<string>();
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<void>), maybeFn?: (event: PluginEvent) => Promise<void>): () => 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 ?? []) {
+18
View File
@@ -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<ResourceLifecycleEvent[]>;
/** Acknowledge only after the provider operation succeeds. */
acknowledgeLifecycle(companyId: string, eventId: string): Promise<void>;
/**
* 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<void>;
}
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.
*
@@ -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<void>),
@@ -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" })]);
});
});
@@ -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<string, (value: JsonRpcResponse) => 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<JsonRpcResponse>(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(); }
});
});
@@ -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 {
@@ -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();
}
});
});
@@ -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<typeof setTimeout> }>();
@@ -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);
@@ -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<ResourceLifecycleEvent[]> {
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<void> {
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();
});
},
};
}