diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 3f1c105a130..b3f7070103c 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -762,17 +762,27 @@ path enforces a project maximum. and a partial unique index allows one unsettled attempt per run. The row keeps the context revision, the nullable primary project as provenance (no foreign key) and the settlement identity, which is set together with `settled_at`. + `admitted_project_ids` is the complete admitted project set, sorted, kept + after settlement and immutable (a trigger rejects any change). It has no + project foreign key, so settled history never blocks project deletion and + keeps the deleted project's ID. `{}` is a known empty set. Null means + unknown and is allowed only on settled rows: migration `0428` filled it from + the charges of unsettled rows, but settled `0426` rows had already lost + their set. A retained provider session resumes with the previous set ∪ the + current references; an unknown previous set requires a fresh context, never + an empty or primary-only guess. - `execution_admission_project_charges` stores one row per distinct admitted project, including unlimited projects. An admission without projects has no rows. Composite `(company_id, …)` foreign keys keep admissions and projects in - one company. Rows exist only while the admission is unsettled: settlement - deletes the whole set in one transaction, and `(company_id, project_id)` is - indexed for active counts, so lookups scale with active work, not history. + one company. Rows are the only active capacity and exist only while the + admission is unsettled: settlement deletes the whole set in one transaction, + and `(company_id, project_id)` is indexed for active counts, so lookups + scale with active work, not history. Archiving a project changes no charge. Deleting a project with a charge returns `409 project_execution_admission_pending`; the foreign key is the backstop. Deleting an agent refuses unsettled admissions on its runs and removes settled -ones. Company deletion removes the company's admissions before its runs and +ones, with their sets, together with the runs. Company deletion removes the company's admissions before its runs and projects. The proposed global lock order for the admission transaction is: company row diff --git a/packages/db/src/execution-admission-migration.test.ts b/packages/db/src/execution-admission-migration.test.ts index 6fefa53f9bd..82de5325181 100644 --- a/packages/db/src/execution-admission-migration.test.ts +++ b/packages/db/src/execution-admission-migration.test.ts @@ -5,9 +5,11 @@ import { describe, expect, it } from "vitest"; import { applyPendingMigrations, inspectMigrations } from "./client.js"; import { EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./test-embedded-postgres.js"; +const hashMigration = async (name: string) => createHash("sha256") + .update(await readFile(new URL(`./migrations/${name}`, import.meta.url), "utf8")).digest("hex"); const migrationName = "0426_absent_guardian.sql"; -const migrationHash = createHash("sha256") - .update(await readFile(new URL(`./migrations/${migrationName}`, import.meta.url), "utf8")).digest("hex"); +const migrationHash = await hashMigration(migrationName); +const retainedSetMigration = "0428_retained_admitted_project_sets.sql"; const support = await getEmbeddedPostgresTestSupport(); const describePostgres = support.supported ? describe : describe.skip; @@ -17,7 +19,8 @@ describePostgres("execution admission migration", () => { const sql = postgres(database.connectionString, { max: 1, onnotice: () => {} }); try { await sql`DROP TABLE execution_admission_project_charges, execution_admissions, project_task_execution_policies`; - await sql`DELETE FROM drizzle.__drizzle_migrations WHERE hash = ${migrationHash}`; + await sql`DROP FUNCTION paperclip_execution_admission_project_set_immutable`; + await sql`DELETE FROM drizzle.__drizzle_migrations WHERE hash IN (${migrationHash}, ${await hashMigration(retainedSetMigration)})`; const companyId = randomUUID(), agentId = randomUUID(), projectId = randomUUID(); await sql`INSERT INTO companies (id, name, issue_prefix) VALUES (${companyId}, 'Upgrade', 'UPG')`; await sql`INSERT INTO agents (id, company_id, name, adapter_type) VALUES (${agentId}, ${companyId}, 'Agent', 'process')`; @@ -31,7 +34,7 @@ describePostgres("execution admission migration", () => { }); const before = await contents(); expect(await inspectMigrations(database.connectionString)) - .toMatchObject({ status: "needsMigrations", pendingMigrations: [migrationName] }); + .toMatchObject({ status: "needsMigrations", pendingMigrations: [migrationName, retainedSetMigration] }); await applyPendingMigrations(database.connectionString); @@ -50,4 +53,49 @@ describePostgres("execution admission migration", () => { await database.cleanup(); } }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + + it("keeps the admitted sets of shipped 0426 rows that still have charges and marks settled ones unknown", async () => { + const database = await startEmbeddedPostgresTestDatabase("paperclip-execution-admission-sets-"); + const sql = postgres(database.connectionString, { max: 1, onnotice: () => {} }); + try { + await sql`DROP TRIGGER paperclip_execution_admission_project_set_immutable ON execution_admissions`; + await sql`DROP FUNCTION paperclip_execution_admission_project_set_immutable`; + await sql`ALTER TABLE execution_admissions DROP CONSTRAINT execution_admissions_project_set_check, DROP COLUMN admitted_project_ids`; + await sql`DELETE FROM drizzle.__drizzle_migrations WHERE hash = ${await hashMigration(retainedSetMigration)}`; + const companyId = randomUUID(), agentId = randomUUID(); + const [a, b] = [randomUUID(), randomUUID()].sort(); + await sql`INSERT INTO companies (id, name, issue_prefix) VALUES (${companyId}, 'Upgrade', 'UPS')`; + await sql`INSERT INTO agents (id, company_id, name, adapter_type) VALUES (${agentId}, ${companyId}, 'Agent', 'process')`; + await sql`INSERT INTO projects (id, company_id, name) VALUES (${b}, ${companyId}, 'B'), (${a}, ${companyId}, 'A')`; + const [live, empty, settled] = [randomUUID(), randomUUID(), randomUUID()]; + for (const id of [live, empty, settled]) { + const [run] = await sql`INSERT INTO heartbeat_runs (company_id, agent_id) VALUES (${companyId}, ${agentId}) RETURNING id`; + await sql`INSERT INTO execution_admissions (id, company_id, run_id, attempt, primary_project_id, context_revision, + settlement_id, settled_at) VALUES (${id}, ${companyId}, ${run.id}, 1, ${a}, 'issue:1', + ${id === settled ? randomUUID() : null}, ${id === settled ? sql`now()` : null})`; + } + await sql`INSERT INTO execution_admission_project_charges (admission_id, company_id, project_id) + VALUES (${live}, ${companyId}, ${b}), (${live}, ${companyId}, ${a})`; + + const sets = () => sql`SELECT id, admitted_project_ids FROM execution_admissions ORDER BY id`; + await applyPendingMigrations(database.connectionString); + + const upgraded = await sets(); + expect(upgraded).toEqual(expect.arrayContaining([ + { id: live, admitted_project_ids: [a, b] }, + { id: empty, admitted_project_ids: [] }, + { id: settled, admitted_project_ids: null }, + ])); + await sql`DELETE FROM drizzle.__drizzle_migrations WHERE hash = ${await hashMigration(retainedSetMigration)}`; + await applyPendingMigrations(database.connectionString); + expect(await sets()).toEqual(upgraded); + expect(await sql`SELECT count(*)::int AS count FROM execution_admission_project_charges`).toEqual([{ count: 2 }]); + await expect(sql`UPDATE execution_admissions SET admitted_project_ids = '{}' WHERE id = ${settled}`) + .rejects.toMatchObject({ message: "execution admission project set is immutable" }); + expect(await inspectMigrations(database.connectionString)).toMatchObject({ status: "upToDate" }); + } finally { + await sql.end(); + await database.cleanup(); + } + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); }); diff --git a/packages/db/src/migrations/0428_retained_admitted_project_sets.sql b/packages/db/src/migrations/0428_retained_admitted_project_sets.sql new file mode 100644 index 00000000000..30c4d953746 --- /dev/null +++ b/packages/db/src/migrations/0428_retained_admitted_project_sets.sql @@ -0,0 +1,22 @@ +ALTER TABLE "execution_admissions" ADD COLUMN IF NOT EXISTS "admitted_project_ids" uuid[];--> statement-breakpoint +-- Unsettled charges are still the complete admitted set. Settled rows already lost theirs and stay unknown (null). +UPDATE "execution_admissions" AS "admission" SET "admitted_project_ids" = coalesce(( + SELECT array_agg("charge"."project_id" ORDER BY "charge"."project_id") FROM "execution_admission_project_charges" AS "charge" + WHERE "charge"."company_id" = "admission"."company_id" AND "charge"."admission_id" = "admission"."id" +), '{}'::uuid[]) WHERE "admission"."settled_at" IS NULL AND "admission"."admitted_project_ids" IS NULL;--> statement-breakpoint +ALTER TABLE "execution_admissions" DROP CONSTRAINT IF EXISTS "execution_admissions_project_set_check";--> statement-breakpoint +ALTER TABLE "execution_admissions" ADD CONSTRAINT "execution_admissions_project_set_check" CHECK (case when "execution_admissions"."admitted_project_ids" is null + then "execution_admissions"."settled_at" is not null else array_position("execution_admissions"."admitted_project_ids", null) is null and cardinality("execution_admissions"."admitted_project_ids") <= 100 end);--> statement-breakpoint +CREATE OR REPLACE FUNCTION paperclip_execution_admission_project_set_immutable() RETURNS trigger +LANGUAGE plpgsql SET search_path = public, pg_temp AS $$ +BEGIN + IF NEW.admitted_project_ids IS DISTINCT FROM OLD.admitted_project_ids THEN + RAISE EXCEPTION 'execution admission project set is immutable' USING ERRCODE = 'check_violation'; + END IF; + RETURN NEW; +END; +$$; +--> statement-breakpoint +DROP TRIGGER IF EXISTS paperclip_execution_admission_project_set_immutable ON "execution_admissions";--> statement-breakpoint +CREATE TRIGGER paperclip_execution_admission_project_set_immutable BEFORE UPDATE OF admitted_project_ids ON "execution_admissions" +FOR EACH ROW EXECUTE FUNCTION paperclip_execution_admission_project_set_immutable(); diff --git a/packages/db/src/migrations/meta/0422_snapshot.json b/packages/db/src/migrations/meta/0428_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0422_snapshot.json rename to packages/db/src/migrations/meta/0428_snapshot.json index 66b83d1df9a..1dad6da7c8a 100644 --- a/packages/db/src/migrations/meta/0422_snapshot.json +++ b/packages/db/src/migrations/meta/0428_snapshot.json @@ -1,6 +1,6 @@ { - "id": "fe0b877c-d63e-46d9-8798-c7729fd13eb7", - "prevId": "deb3f1b9-6eb7-447f-adf2-dee69f943c59", + "id": "fb4383f6-cbe6-4aff-9fd2-6eeb69a6927a", + "prevId": "dd860bff-55de-4849-8787-58b9de729258", "version": "7", "dialect": "postgresql", "tables": { @@ -28825,6 +28825,337 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.execution_admission_project_charges": { + "name": "execution_admission_project_charges", + "schema": "", + "columns": { + "admission_id": { + "name": "admission_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "project_id": { + "name": "project_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "execution_admission_project_charges_project_idx": { + "name": "execution_admission_project_charges_project_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "project_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "execution_admission_project_charges_company_admission_fk": { + "name": "execution_admission_project_charges_company_admission_fk", + "tableFrom": "execution_admission_project_charges", + "tableTo": "execution_admissions", + "columnsFrom": [ + "company_id", + "admission_id" + ], + "columnsTo": [ + "company_id", + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "execution_admission_project_charges_company_project_fk": { + "name": "execution_admission_project_charges_company_project_fk", + "tableFrom": "execution_admission_project_charges", + "tableTo": "projects", + "columnsFrom": [ + "company_id", + "project_id" + ], + "columnsTo": [ + "company_id", + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "execution_admission_project_charges_admission_id_project_id_pk": { + "name": "execution_admission_project_charges_admission_id_project_id_pk", + "columns": [ + "admission_id", + "project_id" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.execution_admissions": { + "name": "execution_admissions", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "run_id": { + "name": "run_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "attempt": { + "name": "attempt", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "primary_project_id": { + "name": "primary_project_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "context_revision": { + "name": "context_revision", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "admitted_project_ids": { + "name": "admitted_project_ids", + "type": "uuid[]", + "primaryKey": false, + "notNull": false + }, + "settlement_id": { + "name": "settlement_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "settled_at": { + "name": "settled_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "execution_admissions_company_id_uq": { + "name": "execution_admissions_company_id_uq", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "execution_admissions_run_attempt_uq": { + "name": "execution_admissions_run_attempt_uq", + "columns": [ + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "attempt", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "execution_admissions_live_run_uq": { + "name": "execution_admissions_live_run_uq", + "columns": [ + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"execution_admissions\".\"settled_at\" is null", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "execution_admissions_company_id_companies_id_fk": { + "name": "execution_admissions_company_id_companies_id_fk", + "tableFrom": "execution_admissions", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "execution_admissions_run_id_heartbeat_runs_id_fk": { + "name": "execution_admissions_run_id_heartbeat_runs_id_fk", + "tableFrom": "execution_admissions", + "tableTo": "heartbeat_runs", + "columnsFrom": [ + "run_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": { + "execution_admissions_attempt_check": { + "name": "execution_admissions_attempt_check", + "value": "\"execution_admissions\".\"attempt\" > 0" + }, + "execution_admissions_settlement_check": { + "name": "execution_admissions_settlement_check", + "value": "(\"execution_admissions\".\"settlement_id\" is null) = (\"execution_admissions\".\"settled_at\" is null)" + }, + "execution_admissions_project_set_check": { + "name": "execution_admissions_project_set_check", + "value": "case when \"execution_admissions\".\"admitted_project_ids\" is null\n then \"execution_admissions\".\"settled_at\" is not null else array_position(\"execution_admissions\".\"admitted_project_ids\", null) is null and cardinality(\"execution_admissions\".\"admitted_project_ids\") <= 100 end" + } + }, + "isRLSEnabled": false + }, + "public.project_task_execution_policies": { + "name": "project_task_execution_policies", + "schema": "", + "columns": { + "project_id": { + "name": "project_id", + "type": "uuid", + "primaryKey": true, + "notNull": true + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "max_concurrent_tasks": { + "name": "max_concurrent_tasks", + "type": "integer", + "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": {}, + "foreignKeys": { + "project_task_execution_policies_company_project_fk": { + "name": "project_task_execution_policies_company_project_fk", + "tableFrom": "project_task_execution_policies", + "tableTo": "projects", + "columnsFrom": [ + "company_id", + "project_id" + ], + "columnsTo": [ + "company_id", + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": { + "project_task_execution_policies_maximum_check": { + "name": "project_task_execution_policies_maximum_check", + "value": "\"project_task_execution_policies\".\"max_concurrent_tasks\" is null or \"project_task_execution_policies\".\"max_concurrent_tasks\" > 0" + } + }, + "isRLSEnabled": false + }, "public.execution_workspace_repositories": { "name": "execution_workspace_repositories", "schema": "", @@ -29999,6 +30330,22 @@ "concurrently": false, "method": "btree", "with": {} + }, + "execution_workspace_snapshots_imported_children_idx": { + "name": "execution_workspace_snapshots_imported_children_idx", + "columns": [ + { + "expression": "(\"metadata\"->'importedWorkspaceChildren') jsonb_path_ops", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"execution_workspace_snapshots\".\"metadata\" ? 'importedWorkspaceChildren'", + "concurrently": false, + "method": "gin", + "with": {} } }, "foreignKeys": {}, @@ -50480,6 +50827,12 @@ "primaryKey": false, "notNull": false }, + "created_by_run_id": { + "name": "created_by_run_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, "dispatch_fingerprint": { "name": "dispatch_fingerprint", "type": "text", @@ -62442,6 +62795,12 @@ "primaryKey": false, "notNull": false }, + "cleanup_verified_at": { + "name": "cleanup_verified_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, "created_at": { "name": "created_at", "type": "timestamp with time zone", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 75ac82c14f3..a2f1da3ce13 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2983,6 +2983,13 @@ "when": 1793736361959, "tag": "0427_ordinary_xavin", "breakpoints": true + }, + { + "idx": 428, + "version": "7", + "when": 1793743828770, + "tag": "0428_retained_admitted_project_sets", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/execution_admissions.ts b/packages/db/src/schema/execution_admissions.ts index 2a9e9fb641f..d97900bbd79 100644 --- a/packages/db/src/schema/execution_admissions.ts +++ b/packages/db/src/schema/execution_admissions.ts @@ -24,7 +24,10 @@ export const projectTaskExecutionPolicies = pgTable("project_task_execution_poli /** * One admission attempt of a run. A run can be admitted again after a settled * attempt, so the run ID is not the admission identity. Primary project is - * provenance only; the charges below are the admitted project set. + * provenance only. `admittedProjectIds` is the complete admitted set, sorted and + * immutable (trigger), kept after settlement without project foreign keys. Null + * means unknown: settled before the set was recorded. Charges below are the + * active capacity only. */ export const executionAdmissions = pgTable("execution_admissions", { id: uuid("id").primaryKey(), @@ -33,6 +36,7 @@ export const executionAdmissions = pgTable("execution_admissions", { attempt: integer("attempt").notNull(), primaryProjectId: uuid("primary_project_id"), contextRevision: text("context_revision").notNull(), + admittedProjectIds: uuid("admitted_project_ids").array(), settlementId: uuid("settlement_id"), settledAt: timestamp("settled_at", { withTimezone: true }), createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), @@ -42,6 +46,8 @@ export const executionAdmissions = pgTable("execution_admissions", { liveRun: uniqueIndex("execution_admissions_live_run_uq").on(t.runId).where(sql`${t.settledAt} is null`), attemptCheck: check("execution_admissions_attempt_check", sql`${t.attempt} > 0`), settlementCheck: check("execution_admissions_settlement_check", sql`(${t.settlementId} is null) = (${t.settledAt} is null)`), + projectSetCheck: check("execution_admissions_project_set_check", sql`case when ${t.admittedProjectIds} is null + then ${t.settledAt} is not null else array_position(${t.admittedProjectIds}, null) is null and cardinality(${t.admittedProjectIds}) <= 100 end`), })); /** One active charge per distinct admitted project. Settlement deletes the whole set; a charged project cannot be deleted. */ diff --git a/server/src/modules/README.md b/server/src/modules/README.md index 1da9c9b7c67..9b4447083df 100644 --- a/server/src/modules/README.md +++ b/server/src/modules/README.md @@ -60,5 +60,7 @@ pause, resume, and termination. Use its commands for these changes. The execution admission module owns project task-execution policy and the durable admission records that charge project capacity. Today it exposes only internal policy commands and the deletion guards that project, agent, and -company deletion call. Execution claims still belong to `services/heartbeat/queue.ts` -and native review dispatch. See [execution admission accounting](../../../doc/DATABASE.md#execution-admission-accounting). +company deletion call. Settlement releases active charges but keeps each +admission's complete project set for later retained-session resolution. +Execution claims still belong to `services/heartbeat/queue.ts` and native +review dispatch. See [execution admission accounting](../../../doc/DATABASE.md#execution-admission-accounting). diff --git a/server/src/modules/execution-admission/adapters/postgres.test.ts b/server/src/modules/execution-admission/adapters/postgres.test.ts index ecdfef3cf07..bf509cb0f05 100644 --- a/server/src/modules/execution-admission/adapters/postgres.test.ts +++ b/server/src/modules/execution-admission/adapters/postgres.test.ts @@ -16,9 +16,9 @@ import { withAccountingTransaction } from "../../../services/accounting-transact import { deleteAgentDependencies } from "../../../services/agent-deletion.js"; import { deleteCompany } from "../../../services/company-deletion.js"; import { projectService } from "../../../services/projects.js"; -import { prepareAdmissionRecord, type AdmissionRecord } from "../domain/admission.js"; +import { prepareAdmissionRecord, retainedSessionProjects, type AdmissionRecord } from "../domain/admission.js"; import { createExecutionAdmission } from "../index.js"; -import { countActiveProjectCharges, recordAdmission, settleAdmission } from "./postgres.js"; +import { countActiveProjectCharges, readAdmittedProjects, recordAdmission, settleAdmission } from "./postgres.js"; const support = await getEmbeddedPostgresTestSupport(); type TestDb = ReturnType; @@ -124,7 +124,7 @@ describe.skipIf(!support.supported)("execution admission storage", () => { expect(await admit({ companyId, runId, admissionId, record: record(a, [b], "issue:2") })).toEqual({ kind: "admission_conflict" }); expect(await admit({ companyId, runId, record: admitted })).toEqual({ kind: "run_already_admitted", admissionId }); expect(await db.select({ id: admissions.id }).from(admissions).where(eq(admissions.runId, runId))).toHaveLength(1); - await expect(db.insert(admissions).values({ id: randomUUID(), companyId, runId, attempt: 2, contextRevision: "x" })) + await expect(db.insert(admissions).values({ id: randomUUID(), companyId, runId, attempt: 2, contextRevision: "x", admittedProjectIds: [] })) .rejects.toMatchObject({ cause: { constraint_name: "execution_admissions_live_run_uq" } }); }); @@ -149,6 +149,51 @@ describe.skipIf(!support.supported)("execution admission storage", () => { .rejects.toMatchObject({ cause: { constraint_name: "execution_admissions_settlement_check" } }); }); + it("keeps each admitted set readable after settlement releases its capacity", async () => { + const { companyId, agentId, runId, projectIds: [a, b, c] } = await fixture(); + const first = randomUUID(); + await admit({ companyId, runId, admissionId: first, record: record(a, [b]) }); + const settlementId = randomUUID(); + await settle(companyId, first, settlementId); + const settledRead = { kind: "found", runId, attempt: 1, primaryProjectId: a, contextRevision: "issue:1", settled: true, + projects: { kind: "known", projectIds: [a, b] } }; + expect(await chargedProjects(first)).toEqual([]); + expect(Object.fromEntries(await countActiveProjectCharges(db, companyId, [a, b]))).toEqual({ [a]: 0, [b]: 0 }); + expect(await readAdmittedProjects(db, { companyId, admissionId: first })).toEqual(settledRead); + expect(await settle(companyId, first, settlementId)).toEqual({ kind: "already_settled" }); + expect(await readAdmittedProjects(db, { companyId, admissionId: first })).toEqual(settledRead); + + // B was removed from the task text; the retained session still carries it. + const second = randomUUID(); + await admit({ companyId, runId, admissionId: second, record: record(a, [c], "issue:2") }); + expect(await readAdmittedProjects(db, { companyId, admissionId: second })) + .toMatchObject({ attempt: 2, settled: false, projects: { kind: "known", projectIds: [a, c] } }); + expect(retainedSessionProjects(settledRead.projects as { kind: "known"; projectIds: string[] }, [a])) + .toEqual({ kind: "retain", projectIds: [a, b] }); + expect(retainedSessionProjects({ kind: "unknown" }, [a])).toEqual({ kind: "fresh_context_required" }); + + const [otherRun, emptyRun] = await db.insert(heartbeatRuns).values([{ companyId, agentId }, { companyId, agentId }]).returning(); + const primaryless = randomUUID(), empty = randomUUID(); + await admit({ companyId, runId: otherRun.id, admissionId: primaryless, record: record(null, [b]) }); + await admit({ companyId, runId: emptyRun.id, admissionId: empty, record: record(null, []) }); + await settle(companyId, primaryless); + await settle(companyId, empty); + expect(await readAdmittedProjects(db, { companyId, admissionId: primaryless })) + .toMatchObject({ primaryProjectId: null, projects: { kind: "known", projectIds: [b] } }); + expect(await readAdmittedProjects(db, { companyId, admissionId: empty })).toMatchObject({ projects: { kind: "known", projectIds: [] } }); + expect(await readAdmittedProjects(db, { companyId: (await fixture()).companyId, admissionId: first })).toEqual({ kind: "not_found" }); + + await expect(db.update(admissions).set({ admittedProjectIds: [a] }).where(eq(admissions.id, first))) + .rejects.toMatchObject({ cause: { message: "execution admission project set is immutable" } }); + await expect(db.insert(admissions).values({ id: randomUUID(), companyId, runId: emptyRun.id, attempt: 9, contextRevision: "x" })) + .rejects.toMatchObject({ cause: { constraint_name: "execution_admissions_project_set_check" } }); + + // Settled history neither blocks deleting B nor loses its identity. + expect(await projectService(db).remove(b)).toMatchObject({ id: b }); + expect(await readAdmittedProjects(db, { companyId, admissionId: first })).toEqual(settledRead); + await expect(projectService(db).remove(c)).rejects.toMatchObject({ details: { code: "project_execution_admission_pending" } }); + }); + it("keeps company isolation and rolls back the whole record", async () => { const { companyId, runId, projectIds: [a] } = await fixture(); const other = await fixture(); @@ -169,7 +214,7 @@ describe.skipIf(!support.supported)("execution admission storage", () => { await recordAdmission(tx, { companyId: other.companyId, runId: other.runId, admissionId: rolledBack, record: record(null, other.projectIds) }); throw new Error("claim lost"); })).rejects.toThrow("claim lost"); - expect(await db.select().from(admissions).where(eq(admissions.id, rolledBack))).toEqual([]); + expect(await readAdmittedProjects(db, { companyId: other.companyId, admissionId: rolledBack })).toEqual({ kind: "not_found" }); expect(await chargedProjects(rolledBack)).toEqual([]); await expect(recordAdmission(db, { companyId, runId, admissionId: randomUUID(), record: record(a, []) })) .rejects.toThrow("requires a database transaction"); diff --git a/server/src/modules/execution-admission/adapters/postgres.ts b/server/src/modules/execution-admission/adapters/postgres.ts index 937eef72e14..972af1c81fa 100644 --- a/server/src/modules/execution-admission/adapters/postgres.ts +++ b/server/src/modules/execution-admission/adapters/postgres.ts @@ -8,7 +8,7 @@ import { } from "@paperclipai/db"; import { and, asc, eq, inArray, isNull, sql } from "drizzle-orm"; import type { TaskExecutionPolicyStore } from "../application/policy.js"; -import { decideSettlement, sameAdmissionRecord, type AdmissionRecord } from "../domain/admission.js"; +import { decideSettlement, sameAdmissionRecord, type AdmissionRecord, type AdmittedProjectSet } from "../domain/admission.js"; import { ABSENT_TASK_EXECUTION_POLICY, type TaskExecutionPolicy } from "../domain/policy.js"; function assertTransaction(tx: Db) { @@ -88,10 +88,8 @@ export async function recordAdmission(tx: Db, input: RecordAdmissionInput): Prom if (existing) { if (existing.companyId !== companyId || existing.runId !== runId) return { kind: "admission_conflict" }; if (existing.settledAt) return { kind: "admission_settled" }; - const charged = await tx.select({ projectId: charges.projectId }).from(charges) - .where(and(eq(charges.companyId, companyId), eq(charges.admissionId, admissionId))).orderBy(asc(charges.projectId)); const stored = { primaryProjectId: existing.primaryProjectId, contextRevision: existing.contextRevision, - projectIds: charged.map(row => row.projectId) }; + projectIds: existing.admittedProjectIds ?? [] }; return sameAdmissionRecord(stored, record) ? { kind: "replayed", attempt: existing.attempt } : { kind: "admission_conflict" }; } const [live] = await tx.select({ id: admissions.id }).from(admissions) @@ -101,7 +99,7 @@ export async function recordAdmission(tx: Db, input: RecordAdmissionInput): Prom const [{ attempt }] = await tx.select({ attempt: sql`coalesce(max(${admissions.attempt}), 0)::int + 1` }) .from(admissions).where(eq(admissions.runId, runId)); await tx.insert(admissions).values({ id: admissionId, companyId, runId, attempt, - primaryProjectId: record.primaryProjectId, contextRevision: record.contextRevision }); + primaryProjectId: record.primaryProjectId, contextRevision: record.contextRevision, admittedProjectIds: record.projectIds }); if (record.projectIds.length > 0) { await tx.insert(charges).values(record.projectIds.map(projectId => ({ admissionId, companyId, projectId }))); } @@ -113,8 +111,9 @@ export type SettleAdmissionOutcome = | { kind: "already_settled" | "settled_by_other" | "not_found" }; /** - * Storage transition only: releases the admission's whole charge set once. - * Callers must first verify authoritative lifecycle facts. Takes no project lock. + * Storage transition only: releases the admission's whole charge set once and + * keeps its admitted project set. Callers must first verify authoritative + * lifecycle facts. Takes no project lock. */ export async function settleAdmission(tx: Db, input: { companyId: string; admissionId: string; settlementId: string }): Promise { @@ -132,6 +131,21 @@ Promise { return { kind: "settled", releasedProjectIds: released.map(charge => charge.projectId).sort() }; } +export type AdmittedProjectsRead = + | { kind: "found"; runId: string; attempt: number; primaryProjectId: string | null; contextRevision: string; + settled: boolean; projects: AdmittedProjectSet } + | { kind: "not_found" }; + +/** The admission's complete admitted set, settled or not, for retained-session resolution. */ +export async function readAdmittedProjects(db: Db, scope: { companyId: string; admissionId: string }): Promise { + const [row] = await db.select().from(admissions) + .where(and(eq(admissions.companyId, scope.companyId), eq(admissions.id, scope.admissionId))); + if (!row) return { kind: "not_found" }; + return { kind: "found", runId: row.runId, attempt: row.attempt, primaryProjectId: row.primaryProjectId, + contextRevision: row.contextRevision, settled: row.settledAt !== null, + projects: row.admittedProjectIds ? { kind: "known", projectIds: row.admittedProjectIds } : { kind: "unknown" } }; +} + /** Active charge counts per project. Every stored charge is unsettled. */ export async function countActiveProjectCharges(db: Db, companyId: string, projectIds: readonly string[]) { const counts = new Map(projectIds.map(id => [id, 0])); diff --git a/server/src/modules/execution-admission/domain/admission.ts b/server/src/modules/execution-admission/domain/admission.ts index 4177e9c4c06..bf091e3c2ed 100644 --- a/server/src/modules/execution-admission/domain/admission.ts +++ b/server/src/modules/execution-admission/domain/admission.ts @@ -51,3 +51,18 @@ export function decideSettlement(existingSettlementId: string | null, settlement if (existingSettlementId === null) return "settle"; return existingSettlementId === settlementId ? "already_settled" : "settled_by_other"; } + +/** A recorded complete set, or unknown: settled before admitted sets were kept. Unknown is never empty. */ +export type AdmittedProjectSet = { kind: "known"; projectIds: string[] } | { kind: "unknown" }; + +export type RetainedSessionProjects = { kind: "retain"; projectIds: string[] } | { kind: "fresh_context_required" }; + +/** + * A resumed provider session still carries the previous admission's references, + * so its next set is current ∪ previous. An unknown previous set cannot be + * bounded: the consumer must start a fresh context instead of resuming. + */ +export function retainedSessionProjects(previous: AdmittedProjectSet, currentProjectIds: readonly string[]): RetainedSessionProjects { + if (previous.kind === "unknown") return { kind: "fresh_context_required" }; + return { kind: "retain", projectIds: [...new Set([...previous.projectIds, ...currentProjectIds])].sort() }; +}