mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-05 12:29:37 +01:00
feat(execution-admission): add project concurrency policy and admission storage (#18746)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Projects group tasks. Agents run tasks as heartbeat runs. > - Today, workspace mode and shared-workspace flags decide whether tasks run at the same time. A project cannot set its own limit on concurrent task executions. > - A correct project limit must count every project that a task uses, including @-mentioned projects. It must also keep the count when a project is archived or deleted, and when a run is admitted more than one time. > - This change cannot go into one pull request with claims and settlement. Thus this pull request adds only the policy semantics and the durable accounting storage that later claims will use. > - This pull request adds a new `execution-admission` module, three tables and the deletion guards. No route writes a cap and no claim path enforces it. > - The benefit is a tested storage contract (constraints, idempotency, rollback, indexed counts, lifecycle preservation) before any behavior changes. ## Linked Issues or Issue Description Refs #2831, Refs #14333 (related agent-level and adapter-level limits; not duplicates). **Subsystem affected** Server execution admission (new module `server/src/modules/execution-admission/`) and the `packages/db` schema. **Problem or motivation** A project cannot limit how many of its task executions are active. The current shared-workspace flags mix resource safety with capacity. A future project limit needs storage that records each admission attempt and every project that the attempt charges. Run IDs cannot be the identity, because a run can be released before dispatch and admitted again. **Proposed solution** Add a separate policy table that keeps "never configured" apart from "explicit no limit". Add admission epochs with one charge row per distinct project. Settlement deletes the charge set in one transaction. Later pull requests integrate claims, settlement, waits and the public control. **Alternatives considered** Columns on `projects` would leak an unenforced cap through every project reader. A project column or a `uuid[]` on `heartbeat_runs` cannot enforce per-project uniqueness or company scope. `budget_reservations` has no project set. A counter would be a second source of truth. ## What Changed - Added `project_task_execution_policies`. No row means never configured. A null maximum means an explicit "no limit". A check accepts only positive maximums. A composite foreign key keeps the project and company together. The row is deleted with its project. - Added `execution_admissions`. Each row is one admission attempt of a run: `(run_id, attempt)` is unique, and a partial unique index allows one unsettled attempt per run. It keeps the context revision, the nullable primary project (provenance, no foreign key) and the settlement identity. A check sets `settlement_id` and `settled_at` together. - Added `execution_admission_project_charges`. It has one row per distinct admitted project, unlimited projects included. Composite foreign keys keep the admission and the project in one company. The project foreign key has no cascade, so a charge cannot disappear with its project. `(company_id, project_id)` is indexed for active counts. - Added migration `0426_absent_guardian.sql` (renumbered from `0423`, then `0425`, as master added `0423`–`0425`). The generated SQL was reordered so the unique indexes exist before the composite foreign keys that need them. Snapshots `0419` and `0420` were pruned to keep five snapshots. - Added the `execution-admission` module (domain → application → adapters): - Domain: policy patch parsing and application (omitted keeps, null clears, partial updates keep other fields, malformed values are rejected), canonical project-set preparation (primary ∪ references, deduplicated, ascending, bounded with an explicit `too_many_projects` result), and the settlement replay decision. - Application: internal `readTaskExecutionPolicy` and `updateTaskExecutionPolicy` commands. No route, tool, CLI or import calls them. - Adapter: the project-set lock (`FOR NO KEY UPDATE`, ascending, one statement), the policy store, and the storage transitions `recordAdmission`, `settleAdmission` and `countActiveProjectCharges`. These transitions are not exported from the module entry point. Claims (later) must add the capacity check and the lifecycle evidence. - Project deletion returns `409 project_execution_admission_pending` while the project has a charge. Agent deletion returns `409 agent_execution_admission_pending` while its runs have an unsettled admission, and removes settled history. Company deletion calls a new module participant before it deletes runs and projects. - Documented the tables, the lifecycle rules and the proposed global lock order in `doc/DATABASE.md`. Added module ownership to `server/src/modules/README.md`. ## Verification - `cd server && npx vitest run src/modules/execution-admission` → 2 files, 26 tests pass (the Greptile finding on an empty primary ID added an assertion to an existing test). The adapter tests use embedded PostgreSQL and cover: - policy semantics, the maximum check and the company foreign key; - charges for every distinct project, a primaryless set and an empty set (no rows); - replay of one admission ID, conflict on a changed record, refusal of a second live attempt and the partial unique index; - settlement of the whole set once, replay, a different settlement, then attempt 2 for the same run; - company isolation of projects, runs and admission IDs, the duplicate-charge key, and rollback of an aborted transaction; - active counts per project, and an `EXPLAIN` that uses `execution_admission_project_charges_project_idx`; - archive keeps charges, `projectService.remove` returns 409, a direct project delete fails on the foreign key, and delete succeeds after settlement; - `deleteAgentDependencies` returns 409 while unsettled, and `deleteCompany` removes only that company's admissions. - `cd packages/db && npx vitest run src/execution-admission-migration.test.ts` → pass. It removes the new tables and receipt, seeds a legacy project and run, applies pending migrations, and compares the existing rows. - `pnpm --filter @paperclipai/db check:migrations` → pass. Five snapshots remain (`0421`, `0422`, `0424`, `0425`, `0426`; `0423` is SQL-only on master). - `node scripts/check-module-boundaries.mjs` → pass. - `packages/db` `tsc --noEmit` → pass. A scoped `tsc` of the module entry points and domain tests → pass. - Existing tests for the changed deletion services → 6 files, 161 tests pass (`resource-lifecycle-events`, `prompt-injection-prevention`, `agent-lifecycle/lifecycle`, `case-deletion`, `project-workspace-root-provenance`, `companies-service`). - Not run locally: the full `server` typecheck. `tsc` and `tsgo` were stopped by the out-of-memory killer on the 7 GB machine (exit 137, also single-threaded). The full `pnpm -r typecheck`, `pnpm test:run` and `pnpm build` were not run locally; CI ran them on head `520b69d553`: all gates pass, including Typecheck, Build and the 12 server test shards. After merging master (`1c9ca5829f`) into head `c088729361`, the migration check, migration upgrade test, module tests, module-boundary check and `packages/db` typecheck were rerun locally and pass; CI is rerunning the full suite. ## Risks - Migration risk is low. It creates three empty tables and indexes. It does not change existing tables or rows. - Behavior changes only when admission rows exist. Nothing writes them yet, so project, agent and company deletion behave as before. - Later claim work must take the project-set lock in the documented order. A claim that takes project locks after issue locks could deadlock with project deletion. - Settlement deletes charge rows, so a settled admission keeps its primary project and context revision but not its full project set. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use (shell, file edits, embedded PostgreSQL tests). Default context window and reasoning mode. ## 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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: nickyleach <331803+nickyleach@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
20 files changed
+1526
-2
No files matched your search
@@ -56,3 +56,9 @@ Ordinary configuration updates use the root lifecycle command.
|
||||
|
||||
The [agent lifecycle module](../../../doc/AGENT-LIFECYCLE.md) owns agent creation,
|
||||
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).
|
||||
@@ -0,0 +1,11 @@
|
||||
import { executionAdmissions } from "@paperclipai/db";
|
||||
import { eq } from "drizzle-orm";
|
||||
import type { CompanyDeletionParticipant } from "../../../lib/company-deletion.js";
|
||||
|
||||
/** Removes every admission, settled or not, before the company's runs and projects are deleted. */
|
||||
export const executionAdmissionCompanyDeletion: CompanyDeletionParticipant = {
|
||||
async deleteCompanyData(tx, companyId) {
|
||||
if (!("nestedIndex" in tx)) throw new Error("Company deletion requires a database transaction");
|
||||
await tx.delete(executionAdmissions).where(eq(executionAdmissions.companyId, companyId));
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,239 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import {
|
||||
agents,
|
||||
companies,
|
||||
createDb,
|
||||
executionAdmissionProjectCharges as charges,
|
||||
executionAdmissions as admissions,
|
||||
heartbeatRuns,
|
||||
projectTaskExecutionPolicies as policies,
|
||||
projects,
|
||||
} from "@paperclipai/db";
|
||||
import { asc, eq, sql } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../../__tests__/helpers/embedded-postgres.js";
|
||||
import { withAccountingTransaction } from "../../../services/accounting-transaction.js";
|
||||
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 { createExecutionAdmission } from "../index.js";
|
||||
import { countActiveProjectCharges, recordAdmission, settleAdmission } from "./postgres.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
type TestDb = ReturnType<typeof createDb>;
|
||||
|
||||
function record(primaryProjectId: string | null, referencedProjectIds: string[], contextRevision = "issue:1") {
|
||||
const result = prepareAdmissionRecord({ primaryProjectId, referencedProjectIds, contextRevision });
|
||||
if (result.kind !== "ok") throw new Error(result.kind);
|
||||
return result.record;
|
||||
}
|
||||
|
||||
describe.skipIf(!support.supported)("execution admission storage", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: TestDb;
|
||||
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("execution-admission-");
|
||||
db = createDb(database.connectionString);
|
||||
}, 60_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await db?.$client.end();
|
||||
await database?.cleanup();
|
||||
});
|
||||
|
||||
async function fixture() {
|
||||
const [company] = await db.insert(companies).values({ name: "Admission fixture", issuePrefix: `A${randomUUID().slice(0, 6)}` }).returning();
|
||||
const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Admission agent", adapterType: "process" }).returning();
|
||||
const projectRows = await db.insert(projects).values(["Design", "Docs", "Unlimited"].map(name => ({ companyId: company.id, name })))
|
||||
.returning();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id }).returning();
|
||||
return { companyId: company.id, agentId: agent.id, projectIds: projectRows.map(row => row.id).sort(), runId: run.id };
|
||||
}
|
||||
|
||||
const inTransaction = <T>(work: (tx: TestDb) => Promise<T>) => db.transaction(tx => work(tx as unknown as TestDb));
|
||||
const admit = (input: { companyId: string; runId: string; admissionId?: string; record: AdmissionRecord }) =>
|
||||
inTransaction(tx => recordAdmission(tx, { ...input, admissionId: input.admissionId ?? randomUUID() }));
|
||||
const settle = (companyId: string, admissionId: string, settlementId = randomUUID()) =>
|
||||
inTransaction(tx => settleAdmission(tx, { companyId, admissionId, settlementId }));
|
||||
const chargedProjects = async (admissionId: string) => (await db.select({ projectId: charges.projectId }).from(charges)
|
||||
.where(eq(charges.admissionId, admissionId)).orderBy(asc(charges.projectId))).map(row => row.projectId);
|
||||
|
||||
it("stores absent, explicit unlimited and positive maximums with partial updates", async () => {
|
||||
const { companyId, projectIds: [projectId] } = await fixture();
|
||||
const policy = createExecutionAdmission(db);
|
||||
const scope = { companyId, projectId };
|
||||
expect(await policy.readTaskExecutionPolicy(scope)).toEqual({ kind: "found", policy: { state: "absent", maxConcurrentTasks: null } });
|
||||
expect(await policy.updateTaskExecutionPolicy(scope, {})).toMatchObject({ kind: "unchanged", policy: { state: "absent" } });
|
||||
expect(await db.select().from(policies)).not.toContainEqual(expect.objectContaining({ projectId }));
|
||||
|
||||
expect(await policy.updateTaskExecutionPolicy(scope, { maxConcurrentTasks: 2 }))
|
||||
.toEqual({ kind: "updated", policy: { state: "configured", maxConcurrentTasks: 2 } });
|
||||
expect(await policy.updateTaskExecutionPolicy(scope, undefined)).toMatchObject({ kind: "unchanged", policy: { maxConcurrentTasks: 2 } });
|
||||
expect(await policy.updateTaskExecutionPolicy(scope, { maxConcurrentTasks: 0 })).toMatchObject({ kind: "invalid" });
|
||||
expect(await policy.updateTaskExecutionPolicy(scope, null))
|
||||
.toEqual({ kind: "updated", policy: { state: "configured", maxConcurrentTasks: null } });
|
||||
expect(await policy.readTaskExecutionPolicy(scope)).toEqual({ kind: "found", policy: { state: "configured", maxConcurrentTasks: null } });
|
||||
|
||||
const foreign = { companyId: (await fixture()).companyId, projectId };
|
||||
expect(await policy.readTaskExecutionPolicy(foreign)).toEqual({ kind: "project_not_found" });
|
||||
expect(await policy.updateTaskExecutionPolicy(foreign, { maxConcurrentTasks: 1 })).toEqual({ kind: "project_not_found" });
|
||||
expect(await policy.readTaskExecutionPolicy(scope)).toMatchObject({ policy: { maxConcurrentTasks: null } });
|
||||
});
|
||||
|
||||
it("enforces policy constraints and company scope in the database", async () => {
|
||||
const { companyId, projectIds: [projectId] } = await fixture();
|
||||
const other = await fixture();
|
||||
await expect(db.insert(policies).values({ companyId, projectId, maxConcurrentTasks: 0 }))
|
||||
.rejects.toMatchObject({ cause: { constraint_name: "project_task_execution_policies_maximum_check" } });
|
||||
await expect(db.insert(policies).values({ companyId: other.companyId, projectId, maxConcurrentTasks: 1 }))
|
||||
.rejects.toMatchObject({ cause: { constraint_name: "project_task_execution_policies_company_project_fk" } });
|
||||
});
|
||||
|
||||
it("charges every distinct project, including unlimited and primaryless sets", async () => {
|
||||
const { companyId, runId, projectIds: [a, b, c] } = await fixture();
|
||||
await createExecutionAdmission(db).updateTaskExecutionPolicy({ companyId, projectId: c }, null);
|
||||
const admissionId = randomUUID();
|
||||
const admitted = record(a, [c, b, a]);
|
||||
expect(await admit({ companyId, runId, admissionId, record: admitted })).toEqual({ kind: "recorded", attempt: 1 });
|
||||
expect(await chargedProjects(admissionId)).toEqual([a, b, c]);
|
||||
const [epoch] = await db.select().from(admissions).where(eq(admissions.id, admissionId));
|
||||
expect(epoch).toMatchObject({ companyId, runId, attempt: 1, primaryProjectId: a, contextRevision: "issue:1", settlementId: null });
|
||||
|
||||
const other = await fixture();
|
||||
const primaryless = randomUUID();
|
||||
expect(await admit({ ...other, admissionId: primaryless, record: record(null, [other.projectIds[1]]) }))
|
||||
.toMatchObject({ kind: "recorded" });
|
||||
expect(await chargedProjects(primaryless)).toEqual([other.projectIds[1]]);
|
||||
|
||||
const empty = await fixture();
|
||||
const emptyId = randomUUID();
|
||||
expect(await admit({ ...empty, admissionId: emptyId, record: record(null, []) })).toMatchObject({ kind: "recorded" });
|
||||
expect(await chargedProjects(emptyId)).toEqual([]);
|
||||
expect([...(await countActiveProjectCharges(db, empty.companyId, empty.projectIds)).values()]).toEqual([0, 0, 0]);
|
||||
});
|
||||
|
||||
it("replays one admission identity and refuses a second live epoch for the run", async () => {
|
||||
const { companyId, runId, projectIds: [a, b] } = await fixture();
|
||||
const admissionId = randomUUID();
|
||||
const admitted = record(a, [b]);
|
||||
await admit({ companyId, runId, admissionId, record: admitted });
|
||||
expect(await admit({ companyId, runId, admissionId, record: admitted })).toEqual({ kind: "replayed", attempt: 1 });
|
||||
expect(await admit({ companyId, runId, admissionId, record: record(a, []) })).toEqual({ kind: "admission_conflict" });
|
||||
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" }))
|
||||
.rejects.toMatchObject({ cause: { constraint_name: "execution_admissions_live_run_uq" } });
|
||||
});
|
||||
|
||||
it("settles the whole set once and starts a new attempt for the same run", async () => {
|
||||
const { companyId, runId, projectIds: [a, b] } = await fixture();
|
||||
const first = randomUUID();
|
||||
await admit({ companyId, runId, admissionId: first, record: record(a, [b]) });
|
||||
const settlementId = randomUUID();
|
||||
expect(await settle(companyId, first, settlementId)).toEqual({ kind: "settled", releasedProjectIds: [a, b] });
|
||||
expect(await settle(companyId, first, settlementId)).toEqual({ kind: "already_settled" });
|
||||
expect(await settle(companyId, first)).toEqual({ kind: "settled_by_other" });
|
||||
expect(await settle(randomUUID(), first)).toEqual({ kind: "not_found" });
|
||||
expect(await chargedProjects(first)).toEqual([]);
|
||||
expect(await admit({ companyId, runId, admissionId: first, record: record(a, [b]) })).toEqual({ kind: "admission_settled" });
|
||||
|
||||
const second = randomUUID();
|
||||
expect(await admit({ companyId, runId, admissionId: second, record: record(a, [], "issue:2") })).toEqual({ kind: "recorded", attempt: 2 });
|
||||
expect(await db.select({ attempt: admissions.attempt, settled: sql<boolean>`${admissions.settledAt} is not null` })
|
||||
.from(admissions).where(eq(admissions.runId, runId)).orderBy(asc(admissions.attempt)))
|
||||
.toEqual([{ attempt: 1, settled: true }, { attempt: 2, settled: false }]);
|
||||
await expect(db.update(admissions).set({ settlementId: randomUUID() }).where(eq(admissions.id, second)))
|
||||
.rejects.toMatchObject({ cause: { constraint_name: "execution_admissions_settlement_check" } });
|
||||
});
|
||||
|
||||
it("keeps company isolation and rolls back the whole record", async () => {
|
||||
const { companyId, runId, projectIds: [a] } = await fixture();
|
||||
const other = await fixture();
|
||||
expect(await admit({ companyId, runId, record: record(a, [other.projectIds[0]]) }))
|
||||
.toEqual({ kind: "projects_unavailable", projectIds: [other.projectIds[0]] });
|
||||
expect(await admit({ companyId, runId: other.runId, record: record(a, []) })).toEqual({ kind: "run_not_found" });
|
||||
const admissionId = randomUUID();
|
||||
await admit({ companyId, runId, admissionId, record: record(a, []) });
|
||||
expect(await admit({ companyId: other.companyId, runId: other.runId, admissionId, record: record(null, []) }))
|
||||
.toEqual({ kind: "admission_conflict" });
|
||||
await expect(db.insert(charges).values({ admissionId, companyId, projectId: other.projectIds[0] }))
|
||||
.rejects.toMatchObject({ cause: { constraint_name: "execution_admission_project_charges_company_project_fk" } });
|
||||
await expect(db.insert(charges).values({ admissionId, companyId, projectId: a }))
|
||||
.rejects.toMatchObject({ cause: { constraint_name: "execution_admission_project_charges_admission_id_project_id_pk" } });
|
||||
|
||||
const rolledBack = randomUUID();
|
||||
await expect(inTransaction(async tx => {
|
||||
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 chargedProjects(rolledBack)).toEqual([]);
|
||||
await expect(recordAdmission(db, { companyId, runId, admissionId: randomUUID(), record: record(a, []) }))
|
||||
.rejects.toThrow("requires a database transaction");
|
||||
});
|
||||
|
||||
it("counts active charges per project through the project index", async () => {
|
||||
const first = await fixture();
|
||||
const [a, b, c] = first.projectIds;
|
||||
const [secondRun] = await db.insert(heartbeatRuns).values({ companyId: first.companyId, agentId: first.agentId }).returning();
|
||||
const held = randomUUID();
|
||||
await admit({ ...first, admissionId: held, record: record(a, [b]) });
|
||||
await admit({ companyId: first.companyId, runId: secondRun.id, record: record(null, [a]) });
|
||||
expect(Object.fromEntries(await countActiveProjectCharges(db, first.companyId, [a, b, c]))).toEqual({ [a]: 2, [b]: 1, [c]: 0 });
|
||||
await settle(first.companyId, held);
|
||||
expect(Object.fromEntries(await countActiveProjectCharges(db, first.companyId, [a, b, c]))).toEqual({ [a]: 1, [b]: 0, [c]: 0 });
|
||||
expect(Object.fromEntries(await countActiveProjectCharges(db, randomUUID(), [a]))).toEqual({ [a]: 0 });
|
||||
|
||||
const plan = await inTransaction(async tx => {
|
||||
await tx.execute(sql`set local enable_seqscan = off`);
|
||||
return tx.execute(sql`explain select project_id, count(*) from execution_admission_project_charges
|
||||
where company_id = ${first.companyId} and project_id in (${a}, ${b}) group by project_id`);
|
||||
});
|
||||
expect(JSON.stringify(plan)).toContain("execution_admission_project_charges_project_idx");
|
||||
});
|
||||
|
||||
it("preserves unsettled charges through project archive and delete", async () => {
|
||||
const { companyId, runId, projectIds: [a, b] } = await fixture();
|
||||
const admissionId = randomUUID();
|
||||
await admit({ companyId, runId, admissionId, record: record(a, [b]) });
|
||||
const projectsService = projectService(db);
|
||||
await db.update(projects).set({ archivedAt: new Date() }).where(eq(projects.id, b));
|
||||
expect(await chargedProjects(admissionId)).toEqual([a, b]);
|
||||
await expect(projectsService.remove(a)).rejects.toMatchObject({ status: 409, details: { code: "project_execution_admission_pending" } });
|
||||
await expect(db.delete(projects).where(eq(projects.id, b)))
|
||||
.rejects.toMatchObject({ cause: { constraint_name: "execution_admission_project_charges_company_project_fk" } });
|
||||
expect(await chargedProjects(admissionId)).toEqual([a, b]);
|
||||
|
||||
await createExecutionAdmission(db).updateTaskExecutionPolicy({ companyId, projectId: a }, { maxConcurrentTasks: 1 });
|
||||
await settle(companyId, admissionId);
|
||||
expect(await projectsService.remove(a)).toMatchObject({ id: a });
|
||||
expect(await db.select().from(policies).where(eq(policies.projectId, a))).toEqual([]);
|
||||
expect(await db.select({ primaryProjectId: admissions.primaryProjectId }).from(admissions).where(eq(admissions.id, admissionId)))
|
||||
.toEqual([{ primaryProjectId: a }]);
|
||||
});
|
||||
|
||||
it("guards agent deletion while unsettled and removes admissions with the company", async () => {
|
||||
const { companyId, agentId, runId, projectIds: [a] } = await fixture();
|
||||
const admissionId = randomUUID();
|
||||
await admit({ companyId, runId, admissionId, record: record(a, []) });
|
||||
await expect(withAccountingTransaction(db, companyId, tx => deleteAgentDependencies(tx, companyId, agentId)))
|
||||
.rejects.toMatchObject({ status: 409, details: { code: "agent_execution_admission_pending" } });
|
||||
expect(await chargedProjects(admissionId)).toEqual([a]);
|
||||
await settle(companyId, admissionId);
|
||||
await withAccountingTransaction(db, companyId, tx => deleteAgentDependencies(tx, companyId, agentId));
|
||||
expect(await db.select().from(admissions).where(eq(admissions.runId, runId))).toEqual([]);
|
||||
|
||||
const deleted = await fixture();
|
||||
const retained = await fixture();
|
||||
const retainedAdmission = randomUUID();
|
||||
await admit({ ...retained, admissionId: retainedAdmission, record: record(retained.projectIds[0], []) });
|
||||
await admit({ ...deleted, record: record(deleted.projectIds[0], deleted.projectIds) });
|
||||
await db.update(agents).set({ lifecycleState: "terminated", status: "terminated" }).where(eq(agents.id, deleted.agentId));
|
||||
expect(await deleteCompany(db, deleted.companyId)).toMatchObject({ id: deleted.companyId });
|
||||
expect(await db.select().from(admissions).where(eq(admissions.companyId, deleted.companyId))).toEqual([]);
|
||||
expect(await chargedProjects(retainedAdmission)).toEqual([retained.projectIds[0]]);
|
||||
}, 60_000);
|
||||
});
|
||||
@@ -0,0 +1,165 @@
|
||||
import {
|
||||
executionAdmissionProjectCharges as charges,
|
||||
executionAdmissions as admissions,
|
||||
heartbeatRuns,
|
||||
projectTaskExecutionPolicies as policies,
|
||||
projects,
|
||||
type Db,
|
||||
} 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 { ABSENT_TASK_EXECUTION_POLICY, type TaskExecutionPolicy } from "../domain/policy.js";
|
||||
|
||||
function assertTransaction(tx: Db) {
|
||||
if (!("nestedIndex" in tx)) throw new Error("Execution admission storage requires a database transaction");
|
||||
}
|
||||
|
||||
/**
|
||||
* Global lock order step 3: locks the whole project set in ascending ID order
|
||||
* in one statement. Returns the IDs that are not projects of the company.
|
||||
*/
|
||||
export async function lockAdmissionProjects(tx: Db, companyId: string, projectIds: readonly string[]) {
|
||||
assertTransaction(tx);
|
||||
if (projectIds.length === 0) return [];
|
||||
const rows = await tx.select({ id: projects.id }).from(projects)
|
||||
.where(and(eq(projects.companyId, companyId), inArray(projects.id, [...projectIds])))
|
||||
.orderBy(asc(projects.id)).for("no key update");
|
||||
const found = new Set(rows.map(row => row.id));
|
||||
return projectIds.filter(id => !found.has(id));
|
||||
}
|
||||
|
||||
function toPolicy(row: { maxConcurrentTasks: number | null } | undefined): TaskExecutionPolicy {
|
||||
return row ? { state: "configured", maxConcurrentTasks: row.maxConcurrentTasks } : ABSENT_TASK_EXECUTION_POLICY;
|
||||
}
|
||||
|
||||
export function createPostgresTaskExecutionPolicyStore(db: Db): TaskExecutionPolicyStore {
|
||||
return {
|
||||
async read({ companyId, projectId }) {
|
||||
const [row] = await db.select({ configured: policies.projectId, maxConcurrentTasks: policies.maxConcurrentTasks })
|
||||
.from(projects).leftJoin(policies, eq(policies.projectId, projects.id))
|
||||
.where(and(eq(projects.companyId, companyId), eq(projects.id, projectId)));
|
||||
if (!row) return null;
|
||||
return row.configured ? toPolicy(row) : ABSENT_TASK_EXECUTION_POLICY;
|
||||
},
|
||||
|
||||
update: ({ companyId, projectId }, change) => db.transaction(async transaction => {
|
||||
const tx = transaction as unknown as Db;
|
||||
if ((await lockAdmissionProjects(tx, companyId, [projectId])).length > 0) return null;
|
||||
const [row] = await tx.select({ maxConcurrentTasks: policies.maxConcurrentTasks }).from(policies)
|
||||
.where(eq(policies.projectId, projectId));
|
||||
const current = toPolicy(row);
|
||||
const next = change(current);
|
||||
if (next === current) return current;
|
||||
if (next.state === "absent") throw new Error("A stored task execution policy cannot become absent");
|
||||
await tx.insert(policies).values({ projectId, companyId, maxConcurrentTasks: next.maxConcurrentTasks })
|
||||
.onConflictDoUpdate({ target: policies.projectId, set: { maxConcurrentTasks: next.maxConcurrentTasks, updatedAt: sql`now()` } });
|
||||
return next;
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
export type RecordAdmissionInput = { companyId: string; runId: string; admissionId: string; record: AdmissionRecord };
|
||||
|
||||
export type RecordAdmissionOutcome =
|
||||
| { kind: "recorded" | "replayed"; attempt: number }
|
||||
| { kind: "projects_unavailable"; projectIds: string[] }
|
||||
| { kind: "run_not_found" }
|
||||
| { kind: "run_already_admitted"; admissionId: string }
|
||||
| { kind: "admission_settled" }
|
||||
| { kind: "admission_conflict" };
|
||||
|
||||
/**
|
||||
* Storage transition only: persists one admission attempt and its charges in
|
||||
* the caller's transaction. It neither checks capacity nor claims work; the
|
||||
* composed admission command must do both under the same locks. Takes lock
|
||||
* steps 3 (projects) and 6 (run); composed callers take them earlier.
|
||||
*/
|
||||
export async function recordAdmission(tx: Db, input: RecordAdmissionInput): Promise<RecordAdmissionOutcome> {
|
||||
assertTransaction(tx);
|
||||
const { companyId, runId, admissionId, record } = input;
|
||||
const missing = await lockAdmissionProjects(tx, companyId, record.projectIds);
|
||||
if (missing.length > 0) return { kind: "projects_unavailable", projectIds: missing };
|
||||
const [run] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId))).for("no key update");
|
||||
if (!run) return { kind: "run_not_found" };
|
||||
|
||||
const [existing] = await tx.select().from(admissions).where(eq(admissions.id, admissionId));
|
||||
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) };
|
||||
return sameAdmissionRecord(stored, record) ? { kind: "replayed", attempt: existing.attempt } : { kind: "admission_conflict" };
|
||||
}
|
||||
const [live] = await tx.select({ id: admissions.id }).from(admissions)
|
||||
.where(and(eq(admissions.runId, runId), isNull(admissions.settledAt)));
|
||||
if (live) return { kind: "run_already_admitted", admissionId: live.id };
|
||||
|
||||
const [{ attempt }] = await tx.select({ attempt: sql<number>`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 });
|
||||
if (record.projectIds.length > 0) {
|
||||
await tx.insert(charges).values(record.projectIds.map(projectId => ({ admissionId, companyId, projectId })));
|
||||
}
|
||||
return { kind: "recorded", attempt };
|
||||
}
|
||||
|
||||
export type SettleAdmissionOutcome =
|
||||
| { kind: "settled"; releasedProjectIds: string[] }
|
||||
| { 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.
|
||||
*/
|
||||
export async function settleAdmission(tx: Db, input: { companyId: string; admissionId: string; settlementId: string }):
|
||||
Promise<SettleAdmissionOutcome> {
|
||||
assertTransaction(tx);
|
||||
const { companyId, admissionId, settlementId } = input;
|
||||
const [row] = await tx.select({ settlementId: admissions.settlementId }).from(admissions)
|
||||
.where(and(eq(admissions.companyId, companyId), eq(admissions.id, admissionId))).for("no key update");
|
||||
if (!row) return { kind: "not_found" };
|
||||
const decision = decideSettlement(row.settlementId, settlementId);
|
||||
if (decision !== "settle") return { kind: decision };
|
||||
const released = await tx.delete(charges)
|
||||
.where(and(eq(charges.companyId, companyId), eq(charges.admissionId, admissionId)))
|
||||
.returning({ projectId: charges.projectId });
|
||||
await tx.update(admissions).set({ settlementId, settledAt: sql`now()` }).where(eq(admissions.id, admissionId));
|
||||
return { kind: "settled", releasedProjectIds: released.map(charge => charge.projectId).sort() };
|
||||
}
|
||||
|
||||
/** 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]));
|
||||
if (projectIds.length === 0) return counts;
|
||||
const rows = await db.select({ projectId: charges.projectId, active: sql<number>`count(*)::int` }).from(charges)
|
||||
.where(and(eq(charges.companyId, companyId), inArray(charges.projectId, [...projectIds])))
|
||||
.groupBy(charges.projectId);
|
||||
for (const row of rows) counts.set(row.projectId, row.active);
|
||||
return counts;
|
||||
}
|
||||
|
||||
/** Project deletion holds the project row lock, so no charge can be added concurrently. */
|
||||
export async function hasUnsettledProjectCharges(db: Db, scope: { companyId: string; projectId: string }) {
|
||||
const [row] = await db.select({ admissionId: charges.admissionId }).from(charges)
|
||||
.where(and(eq(charges.companyId, scope.companyId), eq(charges.projectId, scope.projectId))).limit(1);
|
||||
return Boolean(row);
|
||||
}
|
||||
|
||||
/** Removes settled history of an agent's runs before they are deleted; refuses while any admission is unsettled. */
|
||||
export async function deleteAgentAdmissionHistory(tx: Db, scope: { companyId: string; agentId: string }) {
|
||||
assertTransaction(tx);
|
||||
const agentRuns = tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.companyId, scope.companyId), eq(heartbeatRuns.agentId, scope.agentId)));
|
||||
const [live] = await tx.select({ id: admissions.id }).from(admissions)
|
||||
.where(and(eq(admissions.companyId, scope.companyId), inArray(admissions.runId, agentRuns), isNull(admissions.settledAt)))
|
||||
.limit(1);
|
||||
if (live) return { kind: "unsettled" as const };
|
||||
await tx.delete(admissions).where(and(eq(admissions.companyId, scope.companyId), inArray(admissions.runId, agentRuns)));
|
||||
return { kind: "deleted" as const };
|
||||
}
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
import {
|
||||
applyTaskExecutionPolicyPatch,
|
||||
parseTaskExecutionPolicyPatch,
|
||||
type TaskExecutionPolicy,
|
||||
} from "../domain/policy.js";
|
||||
|
||||
export type ProjectScope = { companyId: string; projectId: string };
|
||||
|
||||
export interface TaskExecutionPolicyStore {
|
||||
/** Null when the project is not in the company. */
|
||||
read(scope: ProjectScope): Promise<TaskExecutionPolicy | null>;
|
||||
/**
|
||||
* Locks the project in admission lock order and stores `change(current)`.
|
||||
* Returning `current` itself writes nothing. Null when the project is not in the company.
|
||||
*/
|
||||
update(scope: ProjectScope, change: (current: TaskExecutionPolicy) => TaskExecutionPolicy): Promise<TaskExecutionPolicy | null>;
|
||||
}
|
||||
|
||||
export type PolicyReadResult = { kind: "found"; policy: TaskExecutionPolicy } | { kind: "project_not_found" };
|
||||
|
||||
export type PolicyUpdateResult =
|
||||
| { kind: "updated" | "unchanged"; policy: TaskExecutionPolicy }
|
||||
| { kind: "invalid"; field: "policy" | "maxConcurrentTasks"; reason: string }
|
||||
| { kind: "project_not_found" };
|
||||
|
||||
/**
|
||||
* Internal policy commands. No route, tool or import calls `update` until
|
||||
* admission enforces the maximum; a cap written earlier would mislead users.
|
||||
*/
|
||||
export function createTaskExecutionPolicyCommands(store: TaskExecutionPolicyStore) {
|
||||
return {
|
||||
async readTaskExecutionPolicy(scope: ProjectScope): Promise<PolicyReadResult> {
|
||||
const policy = await store.read(scope);
|
||||
return policy ? { kind: "found", policy } : { kind: "project_not_found" };
|
||||
},
|
||||
|
||||
async updateTaskExecutionPolicy(scope: ProjectScope, input: unknown): Promise<PolicyUpdateResult> {
|
||||
const parsed = parseTaskExecutionPolicyPatch(input);
|
||||
if (parsed.kind === "invalid") return parsed;
|
||||
let changed = false;
|
||||
const policy = await store.update(scope, current => {
|
||||
const next = applyTaskExecutionPolicyPatch(current, parsed.patch);
|
||||
changed = next !== current;
|
||||
return next;
|
||||
});
|
||||
if (!policy) return { kind: "project_not_found" };
|
||||
return { kind: changed ? "updated" : "unchanged", policy };
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
export { executionAdmissionCompanyDeletion } from "./adapters/company-deletion.js";
|
||||
@@ -0,0 +1,53 @@
|
||||
/** Referenced projects are rejected beyond this bound instead of admitting a subset. */
|
||||
export const MAX_ADMISSION_PROJECTS = 100;
|
||||
const MAX_CONTEXT_REVISION_LENGTH = 256;
|
||||
const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/;
|
||||
|
||||
export type AdmissionRecordInput = {
|
||||
primaryProjectId: string | null;
|
||||
referencedProjectIds: readonly string[];
|
||||
contextRevision: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* `projectIds` is the complete distinct set in canonical lock order. Lowercase
|
||||
* UUID text order matches PostgreSQL uuid order. An empty set is valid and
|
||||
* charges no project.
|
||||
*/
|
||||
export type AdmissionRecord = {
|
||||
primaryProjectId: string | null;
|
||||
projectIds: string[];
|
||||
contextRevision: string;
|
||||
};
|
||||
|
||||
export type AdmissionRecordResult =
|
||||
| { kind: "ok"; record: AdmissionRecord }
|
||||
| { kind: "invalid_project_id" }
|
||||
| { kind: "too_many_projects"; count: number; limit: number }
|
||||
| { kind: "invalid_context_revision" };
|
||||
|
||||
export function prepareAdmissionRecord(input: AdmissionRecordInput): AdmissionRecordResult {
|
||||
const contextRevision = input.contextRevision.trim();
|
||||
if (!contextRevision || contextRevision.length > MAX_CONTEXT_REVISION_LENGTH) return { kind: "invalid_context_revision" };
|
||||
const primaryProjectId = input.primaryProjectId?.toLowerCase() ?? null;
|
||||
const candidates = [...input.referencedProjectIds, ...(primaryProjectId === null ? [] : [primaryProjectId])].map(id => id.toLowerCase());
|
||||
if (candidates.some(id => !UUID.test(id))) return { kind: "invalid_project_id" };
|
||||
const projectIds = [...new Set(candidates)].sort();
|
||||
if (projectIds.length > MAX_ADMISSION_PROJECTS) {
|
||||
return { kind: "too_many_projects", count: projectIds.length, limit: MAX_ADMISSION_PROJECTS };
|
||||
}
|
||||
return { kind: "ok", record: { primaryProjectId, projectIds, contextRevision } };
|
||||
}
|
||||
|
||||
export function sameAdmissionRecord(left: AdmissionRecord, right: AdmissionRecord) {
|
||||
return left.primaryProjectId === right.primaryProjectId && left.contextRevision === right.contextRevision
|
||||
&& left.projectIds.length === right.projectIds.length && left.projectIds.every((id, index) => id === right.projectIds[index]);
|
||||
}
|
||||
|
||||
export type SettlementDecision = "settle" | "already_settled" | "settled_by_other";
|
||||
|
||||
/** A settlement identity can be replayed; a second, different settlement cannot take over. */
|
||||
export function decideSettlement(existingSettlementId: string | null, settlementId: string): SettlementDecision {
|
||||
if (existingSettlementId === null) return "settle";
|
||||
return existingSettlementId === settlementId ? "already_settled" : "settled_by_other";
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { decideSettlement, MAX_ADMISSION_PROJECTS, prepareAdmissionRecord } from "./admission.js";
|
||||
import { ABSENT_TASK_EXECUTION_POLICY, applyTaskExecutionPolicyPatch, parseTaskExecutionPolicyPatch, type TaskExecutionPolicy } from "./policy.js";
|
||||
|
||||
const configured = (maxConcurrentTasks: number | null): TaskExecutionPolicy => ({ state: "configured", maxConcurrentTasks });
|
||||
const apply = (current: TaskExecutionPolicy, input: unknown) => {
|
||||
const parsed = parseTaskExecutionPolicyPatch(input);
|
||||
if (parsed.kind !== "ok") throw new Error(parsed.reason);
|
||||
return applyTaskExecutionPolicyPatch(current, parsed.patch);
|
||||
};
|
||||
|
||||
describe("task execution policy", () => {
|
||||
it("keeps never-configured distinct from explicit unlimited", () => {
|
||||
expect(apply(ABSENT_TASK_EXECUTION_POLICY, null)).toEqual(configured(null));
|
||||
expect(apply(ABSENT_TASK_EXECUTION_POLICY, { maxConcurrentTasks: null })).toEqual(configured(null));
|
||||
expect(apply(ABSENT_TASK_EXECUTION_POLICY, { maxConcurrentTasks: 3 })).toEqual(configured(3));
|
||||
});
|
||||
|
||||
it("preserves the current value for omitted policies and fields", () => {
|
||||
for (const current of [ABSENT_TASK_EXECUTION_POLICY, configured(null), configured(2)]) {
|
||||
expect(apply(current, undefined)).toBe(current);
|
||||
expect(apply(current, {})).toBe(current);
|
||||
expect(apply(current, { maxConcurrentTasks: undefined })).toBe(current);
|
||||
}
|
||||
const two = configured(2);
|
||||
expect(apply(two, { maxConcurrentTasks: 2 })).toBe(two);
|
||||
const unlimited = configured(null);
|
||||
expect(apply(unlimited, null)).toBe(unlimited);
|
||||
});
|
||||
|
||||
it("clears or replaces a configured maximum", () => {
|
||||
expect(apply(configured(2), null)).toEqual(configured(null));
|
||||
expect(apply(configured(2), { maxConcurrentTasks: null })).toEqual(configured(null));
|
||||
expect(apply(configured(null), { maxConcurrentTasks: 1 })).toEqual(configured(1));
|
||||
});
|
||||
|
||||
it.each([0, -1, 1.5, Number.NaN, Number.POSITIVE_INFINITY, 2_147_483_648, "2", true, {}])(
|
||||
"rejects maximum %s", value => {
|
||||
expect(parseTaskExecutionPolicyPatch({ maxConcurrentTasks: value })).toMatchObject({ kind: "invalid", field: "maxConcurrentTasks" });
|
||||
},
|
||||
);
|
||||
|
||||
it("rejects malformed policies and unknown fields", () => {
|
||||
for (const input of [1, "unlimited", [], { maxConcurrentTasks: 1, perTaskOverride: 3 }]) {
|
||||
expect(parseTaskExecutionPolicyPatch(input)).toMatchObject({ kind: "invalid", field: "policy" });
|
||||
}
|
||||
expect(parseTaskExecutionPolicyPatch({ maxConcurrentTasks: 2_147_483_647 })).toEqual({ kind: "ok", patch: { maxConcurrentTasks: 2_147_483_647 } });
|
||||
});
|
||||
});
|
||||
|
||||
describe("admission record", () => {
|
||||
const a = "0b9d8e52-6f2e-4bb0-9a4a-1f0a7f2e0a01";
|
||||
const b = "7c1f3d9a-2b4e-4d8f-8a6b-3e5c7d9f1b02";
|
||||
const c = "f3a2b1c0-9d8e-4f7a-8b6c-5d4e3f2a1b03";
|
||||
|
||||
it("unions the primary with referenced projects in canonical order", () => {
|
||||
expect(prepareAdmissionRecord({ primaryProjectId: c, referencedProjectIds: [b, a, b.toUpperCase(), c], contextRevision: " r1 " }))
|
||||
.toEqual({ kind: "ok", record: { primaryProjectId: c, projectIds: [a, b, c], contextRevision: "r1" } });
|
||||
});
|
||||
|
||||
it("keeps a primaryless set and an empty set without a project bucket", () => {
|
||||
expect(prepareAdmissionRecord({ primaryProjectId: null, referencedProjectIds: [b], contextRevision: "r" }))
|
||||
.toEqual({ kind: "ok", record: { primaryProjectId: null, projectIds: [b], contextRevision: "r" } });
|
||||
expect(prepareAdmissionRecord({ primaryProjectId: null, referencedProjectIds: [], contextRevision: "r" }))
|
||||
.toEqual({ kind: "ok", record: { primaryProjectId: null, projectIds: [], contextRevision: "r" } });
|
||||
});
|
||||
|
||||
it("rejects invalid references, oversized sets and missing context identity instead of narrowing", () => {
|
||||
expect(prepareAdmissionRecord({ primaryProjectId: null, referencedProjectIds: [a, "Design"], contextRevision: "r" }))
|
||||
.toEqual({ kind: "invalid_project_id" });
|
||||
expect(prepareAdmissionRecord({ primaryProjectId: "", referencedProjectIds: [a], contextRevision: "r" }))
|
||||
.toEqual({ kind: "invalid_project_id" });
|
||||
const many = Array.from({ length: MAX_ADMISSION_PROJECTS + 1 }, (_, index) =>
|
||||
`00000000-0000-4000-8000-${index.toString(16).padStart(12, "0")}`);
|
||||
expect(prepareAdmissionRecord({ primaryProjectId: null, referencedProjectIds: many, contextRevision: "r" }))
|
||||
.toEqual({ kind: "too_many_projects", count: MAX_ADMISSION_PROJECTS + 1, limit: MAX_ADMISSION_PROJECTS });
|
||||
expect(prepareAdmissionRecord({ primaryProjectId: a, referencedProjectIds: [], contextRevision: " " }))
|
||||
.toEqual({ kind: "invalid_context_revision" });
|
||||
expect(prepareAdmissionRecord({ primaryProjectId: a, referencedProjectIds: [], contextRevision: "x".repeat(257) }))
|
||||
.toEqual({ kind: "invalid_context_revision" });
|
||||
});
|
||||
|
||||
it("settles once and replays only the same settlement", () => {
|
||||
expect(decideSettlement(null, a)).toBe("settle");
|
||||
expect(decideSettlement(a, a)).toBe("already_settled");
|
||||
expect(decideSettlement(a, b)).toBe("settled_by_other");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,47 @@
|
||||
/**
|
||||
* A project's task-execution policy. `absent` was never configured, so a later
|
||||
* legacy migration may still choose a value. `configured` with a null maximum
|
||||
* is an explicit "no limit" that such a migration must preserve.
|
||||
*/
|
||||
export type TaskExecutionPolicy =
|
||||
| { state: "absent"; maxConcurrentTasks: null }
|
||||
| { state: "configured"; maxConcurrentTasks: number | null };
|
||||
|
||||
/** Null clears the cap explicitly. An omitted field preserves its value. */
|
||||
export type TaskExecutionPolicyPatch = { maxConcurrentTasks?: number | null } | null;
|
||||
|
||||
export type PolicyPatchResult =
|
||||
| { kind: "ok"; patch: TaskExecutionPolicyPatch }
|
||||
| { kind: "invalid"; field: "policy" | "maxConcurrentTasks"; reason: string };
|
||||
|
||||
export const ABSENT_TASK_EXECUTION_POLICY: TaskExecutionPolicy = { state: "absent", maxConcurrentTasks: null };
|
||||
|
||||
// The column is a PostgreSQL integer.
|
||||
const MAX_STORED_MAXIMUM = 2_147_483_647;
|
||||
|
||||
/** Undefined (an omitted policy) preserves everything; null clears the cap. */
|
||||
export function parseTaskExecutionPolicyPatch(input: unknown): PolicyPatchResult {
|
||||
if (input === undefined) return { kind: "ok", patch: {} };
|
||||
if (input === null) return { kind: "ok", patch: null };
|
||||
if (typeof input !== "object" || Array.isArray(input)) {
|
||||
return { kind: "invalid", field: "policy", reason: "Policy must be an object or null" };
|
||||
}
|
||||
const unknownKey = Object.keys(input).find(key => key !== "maxConcurrentTasks");
|
||||
if (unknownKey) return { kind: "invalid", field: "policy", reason: `Unknown policy field ${unknownKey}` };
|
||||
if (!("maxConcurrentTasks" in input)) return { kind: "ok", patch: {} };
|
||||
const value = (input as { maxConcurrentTasks: unknown }).maxConcurrentTasks;
|
||||
if (value === undefined) return { kind: "ok", patch: {} };
|
||||
if (value === null) return { kind: "ok", patch: { maxConcurrentTasks: null } };
|
||||
if (typeof value !== "number" || !Number.isInteger(value) || value < 1 || value > MAX_STORED_MAXIMUM) {
|
||||
return { kind: "invalid", field: "maxConcurrentTasks", reason: "Maximum concurrent tasks must be a positive whole number" };
|
||||
}
|
||||
return { kind: "ok", patch: { maxConcurrentTasks: value } };
|
||||
}
|
||||
|
||||
/** Applies a validated patch. Returns the current policy when nothing changes. */
|
||||
export function applyTaskExecutionPolicyPatch(current: TaskExecutionPolicy, patch: TaskExecutionPolicyPatch): TaskExecutionPolicy {
|
||||
if (patch !== null && patch.maxConcurrentTasks === undefined) return current;
|
||||
const maxConcurrentTasks = patch === null ? null : patch.maxConcurrentTasks ?? null;
|
||||
if (current.state === "configured" && current.maxConcurrentTasks === maxConcurrentTasks) return current;
|
||||
return { state: "configured", maxConcurrentTasks };
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { createPostgresTaskExecutionPolicyStore } from "./adapters/postgres.js";
|
||||
import { createTaskExecutionPolicyCommands } from "./application/policy.js";
|
||||
|
||||
export type { TaskExecutionPolicy, TaskExecutionPolicyPatch } from "./domain/policy.js";
|
||||
export type { PolicyReadResult, PolicyUpdateResult, ProjectScope } from "./application/policy.js";
|
||||
// Lifecycle guards run inside the caller's deletion transaction.
|
||||
export { deleteAgentAdmissionHistory, hasUnsettledProjectCharges } from "./adapters/postgres.js";
|
||||
|
||||
/**
|
||||
* Execution admission owns project task-execution policy and the durable
|
||||
* admission epochs that charge project capacity. Admission and settlement
|
||||
* commands are not composed yet; nothing enforces the maximum.
|
||||
*/
|
||||
export function createExecutionAdmission(db: Db) {
|
||||
return createTaskExecutionPolicyCommands(createPostgresTaskExecutionPolicyStore(db));
|
||||
}
|
||||
Reference in new issue
Block a user