mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 02:32:07 +01:00
merge: integrate agent lifecycle ownership from master
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
commit
ad5d8826ad
159 files changed
+7173
-2805
No files matched your search
@@ -0,0 +1,96 @@
|
||||
import { agentAppearanceSchema } from "@paperclipai/shared";
|
||||
import { approvals, type Db } from "@paperclipai/db";
|
||||
import { and, desc, eq, inArray, sql } from "drizzle-orm";
|
||||
import { unprocessable } from "../../../errors.js";
|
||||
import { approvalResolutionRecords, resolvableApprovalStatuses } from "../../../lib/approval-records.js";
|
||||
import { agentRecords } from "./records.js";
|
||||
import { createLifecycleStore } from "./postgres.js";
|
||||
import type { LifecycleEffects } from "./effects.js";
|
||||
|
||||
export type HireDecisionTarget = string | { approvalId: string };
|
||||
type HireDecisionResult = {
|
||||
approval: typeof approvals.$inferSelect | null;
|
||||
agent: Awaited<ReturnType<ReturnType<typeof agentRecords>["getById"]>>;
|
||||
applied: boolean;
|
||||
activated: boolean;
|
||||
hireApprovedAgentId: string | null;
|
||||
};
|
||||
|
||||
export function hireApprovalService(db: Db, effects: LifecycleEffects) {
|
||||
const { getExistingApproval, resolveApproval } = approvalResolutionRecords(db);
|
||||
async function decide(target: HireDecisionTarget, status: "approved" | "rejected", userId: string, note?: string | null): Promise<HireDecisionResult | null> {
|
||||
const initial = typeof target === "string"
|
||||
? await agentRecords(db, effects).getById(target)
|
||||
: await getExistingApproval(target.approvalId);
|
||||
if (!initial) return null;
|
||||
if (typeof target !== "string" && (!("type" in initial) || initial.type !== "hire_agent")) throw unprocessable("Expected a hire approval");
|
||||
const companyId = initial.companyId;
|
||||
// Hire decisions acquire the company lock before changing an approval or agent.
|
||||
return effects.transaction(db, companyId, async (tx, publications) => {
|
||||
const records = agentRecords(tx, effects);
|
||||
let approvalId = typeof target === "string" ? null : target.approvalId;
|
||||
if (typeof target === "string") {
|
||||
const [open] = await tx.select({ id: approvals.id }).from(approvals).where(and(
|
||||
eq(approvals.companyId, companyId), eq(approvals.type, "hire_agent"),
|
||||
inArray(approvals.status, resolvableApprovalStatuses),
|
||||
sql`${approvals.payload}->>'agentId' = ${target}`,
|
||||
)).orderBy(desc(approvals.createdAt)).limit(1);
|
||||
approvalId = open?.id ?? null;
|
||||
}
|
||||
let approval: typeof approvals.$inferSelect | null = null;
|
||||
let applied = true;
|
||||
if (approvalId) {
|
||||
const result = await resolveApproval(approvalId, status, userId, note, tx);
|
||||
approval = result.approval;
|
||||
applied = result.applied;
|
||||
}
|
||||
const payload = approval?.payload ?? {};
|
||||
let agentId = typeof target === "string" ? target : typeof payload.agentId === "string" ? payload.agentId : null;
|
||||
if (applied && status === "approved") {
|
||||
if (agentId) {
|
||||
const activated = await records.activatePendingApproval(agentId, approval ? payload : null, approval ? approval.requestedByUserId : userId);
|
||||
if (!approval) applied = activated?.activated ?? false;
|
||||
} else {
|
||||
const created = await records.create(companyId, {
|
||||
name: String(payload.name ?? "New Agent"),
|
||||
appearance: payload.appearance == null ? undefined : agentAppearanceSchema.parse(payload.appearance),
|
||||
role: String(payload.role ?? "general"),
|
||||
title: typeof payload.title === "string" ? payload.title : null,
|
||||
reportsTo: typeof payload.reportsTo === "string" ? payload.reportsTo : null,
|
||||
capabilities: typeof payload.capabilities === "string" ? payload.capabilities : null,
|
||||
adapterType: String(payload.adapterType ?? "process"),
|
||||
adapterConfig:
|
||||
typeof payload.adapterConfig === "object" && payload.adapterConfig !== null
|
||||
? (payload.adapterConfig as Record<string, unknown>)
|
||||
: {},
|
||||
budgetMonthlyCents:
|
||||
typeof payload.budgetMonthlyCents === "number" ? payload.budgetMonthlyCents : 0,
|
||||
metadata:
|
||||
typeof payload.metadata === "object" && payload.metadata !== null
|
||||
? (payload.metadata as Record<string, unknown>)
|
||||
: null,
|
||||
status: "idle",
|
||||
spentMonthlyCents: 0,
|
||||
permissions: undefined,
|
||||
lastHeartbeatAt: null,
|
||||
}, { runnerResolved: true, createdByUserId: approval!.requestedByAgentId ? null : approval!.requestedByUserId, responsibleUserId: approval!.requestedByUserId });
|
||||
agentId = created.id;
|
||||
}
|
||||
if (approval && agentId && typeof payload.budgetMonthlyCents === "number" && payload.budgetMonthlyCents > 0) {
|
||||
await effects.setAgentBudget(tx, publications, companyId, agentId, payload.budgetMonthlyCents, userId);
|
||||
}
|
||||
} else if (applied && agentId) {
|
||||
if (approval) await records.rejectPendingHire(agentId);
|
||||
else {
|
||||
const before = await records.getById(agentId);
|
||||
await createLifecycleStore(tx, effects).change(agentId, "reject");
|
||||
applied = before?.lifecycleState !== "rejected";
|
||||
}
|
||||
}
|
||||
return { approval, agent: agentId ? await records.getById(agentId) : null, applied,
|
||||
activated: status === "approved" && applied,
|
||||
hireApprovedAgentId: status === "approved" && applied ? agentId : null };
|
||||
});
|
||||
}
|
||||
return { decide };
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
import { agents } from "@paperclipai/db";
|
||||
import { and, eq, inArray } from "drizzle-orm";
|
||||
import type { CompanyDeletionParticipant } from "../../../lib/company-deletion.js";
|
||||
import { AgentLifecycleConflict, assertAgentPurgeAllowed } from "../domain/policy.js";
|
||||
|
||||
export const agentLifecycleCompanyDeletion: CompanyDeletionParticipant = {
|
||||
async deleteCompanyData(tx, companyId) {
|
||||
if (!("nestedIndex" in tx)) {
|
||||
throw new AgentLifecycleConflict("Company deletion requires a database transaction");
|
||||
}
|
||||
const rows = await tx.select({ lifecycleState: agents.lifecycleState }).from(agents)
|
||||
.where(eq(agents.companyId, companyId)).for("update");
|
||||
for (const agent of rows) assertAgentPurgeAllowed(agent);
|
||||
await tx.delete(agents).where(and(
|
||||
eq(agents.companyId, companyId),
|
||||
inArray(agents.lifecycleState, ["terminated", "rejected"]),
|
||||
));
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,32 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { agents, type Db } from "@paperclipai/db";
|
||||
import { agentRecordQueries, AGENT_CONFIGURATION_FIELDS, jsonEqual, type AgentConfigurationPatch, type UpdateAgentOptions } from "../../../lib/agent-records.js";
|
||||
import type { ActivityPublication } from "../../../types/activity-publication.js";
|
||||
import { AgentLifecycleConflict } from "../domain/policy.js";
|
||||
import type { LifecycleEffects } from "./effects.js";
|
||||
|
||||
export async function updateAgentConfiguration(tx: Db, effects: LifecycleEffects, id: string, data: AgentConfigurationPatch, options?: UpdateAgentOptions, publications: ActivityPublication[] = []) {
|
||||
if (!("nestedIndex" in tx)) throw new AgentLifecycleConflict("Configuration integration requires a database transaction");
|
||||
assertConfigurationFields(data);
|
||||
const [current] = await tx.select().from(agents).where(eq(agents.id, id)).for("update");
|
||||
if (!current) return null;
|
||||
const patch = await effects.prepareConfiguration(tx, current, data, options);
|
||||
assertConfigurationFields(patch);
|
||||
const changed = ["adapterType", "adapterConfig", "runtimeConfig", "defaultEnvironmentId"].some(key =>
|
||||
Object.prototype.hasOwnProperty.call(patch, key) && !jsonEqual(patch[key as keyof typeof patch], current[key as keyof typeof current]));
|
||||
const invalidate = changed && ["preparing", "verifying", "resuming"].includes(current.lifecycleState);
|
||||
const [updated] = await tx.update(agents).set({ ...patch, updatedAt: new Date(), ...(invalidate ? {
|
||||
lifecycleVersion: current.lifecycleVersion + 1, lifecycleError: null,
|
||||
lifecycleOperation: { ...current.lifecycleOperation!, id: randomUUID(), hostComplete: false,
|
||||
completedPluginIds: [], leaseOwner: undefined, leaseUntil: undefined, retryAt: undefined },
|
||||
} : {}) }).where(eq(agents.id, id)).returning();
|
||||
await effects.completeConfiguration(tx, current, updated, patch, options, publications);
|
||||
return agentRecordQueries(tx).getById(id);
|
||||
}
|
||||
|
||||
function assertConfigurationFields(data: AgentConfigurationPatch) {
|
||||
if (Object.keys(data).some(field => !AGENT_CONFIGURATION_FIELDS.some(allowed => allowed === field))) {
|
||||
throw new AgentLifecycleConflict("Unsupported configuration field; use the command that owns this field, such as a lifecycle command");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
import { agents, type Db } from "@paperclipai/db";
|
||||
import { and, eq, inArray } from "drizzle-orm";
|
||||
import { conflict } from "../../../errors.js";
|
||||
import { agentRecordQueries } from "../../../lib/agent-records.js";
|
||||
import { readBuiltInAgentMarker } from "../../../lib/built-in-agent-metadata.js";
|
||||
import { assertAgentPurgeAllowed } from "../domain/policy.js";
|
||||
import type { LifecycleEffects } from "./effects.js";
|
||||
|
||||
export async function purgeAgent(db: Db, effects: LifecycleEffects, id: string) {
|
||||
const { getById, normalizeAgentRow } = agentRecordQueries(db);
|
||||
const existing = await getById(id);
|
||||
if (!existing) return null;
|
||||
return effects.transaction(db, existing.companyId, async tx => {
|
||||
const [agent] = await tx.select().from(agents).where(eq(agents.id, id)).for("update");
|
||||
if (!agent) return null;
|
||||
const marker = readBuiltInAgentMarker(agent.metadata);
|
||||
if (marker) throw conflict("Built-in agents cannot be deleted; pause them instead", {
|
||||
code: "built_in_agent_undeletable", key: marker.key, featureKeys: marker.featureKeys,
|
||||
});
|
||||
assertAgentPurgeAllowed(agent);
|
||||
await effects.deleteDependencies(tx, agent.companyId, id);
|
||||
const [deleted] = await tx.delete(agents).where(and(
|
||||
eq(agents.id, id), inArray(agents.lifecycleState, ["terminated", "rejected"]),
|
||||
)).returning();
|
||||
return deleted ? normalizeAgentRow(deleted) : null;
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
import type { LifecycleFailure } from "../application/ports.js";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import type { budgetPolicies } from "@paperclipai/db";
|
||||
import type { CreateAgentData, CreateAgentOptions, AgentConfigurationPatch, AgentConfigurationRecord, AgentHireRecord, UpdateAgentOptions } from "../../../lib/agent-records.js";
|
||||
import type { ActivityPublication } from "../../../types/activity-publication.js";
|
||||
|
||||
/** Services supply integrations; lifecycle owns the transaction and state writes. */
|
||||
export interface LifecycleEffects {
|
||||
reportFailure(context: LifecycleFailure, error: unknown): void;
|
||||
deleteDependencies(db: Db, companyId: string, agentId: string): Promise<void>;
|
||||
prepareConfiguration(db: Db, current: AgentConfigurationRecord, data: AgentConfigurationPatch, options?: UpdateAgentOptions): Promise<AgentConfigurationPatch>;
|
||||
completeConfiguration(db: Db, before: AgentConfigurationRecord, after: AgentConfigurationRecord, patch: AgentConfigurationPatch, options: UpdateAgentOptions | undefined, publications: ActivityPublication[]): Promise<void>;
|
||||
transaction<T>(db: Db, companyId: string, work: (tx: Db, publications: ActivityPublication[]) => Promise<T>): Promise<T>;
|
||||
policyBlocks(db: Db, policy: typeof budgetPolicies.$inferSelect): Promise<boolean>;
|
||||
setAgentBudget(db: Db, publications: ActivityPublication[], companyId: string, agentId: string, amount: number, userId: string | null, isActive?: boolean): Promise<unknown>;
|
||||
enforceBudget(db: Db, companyId: string, agentId: string): Promise<void>;
|
||||
recordCreation(db: Db, companyId: string, agentId: string): Promise<void>;
|
||||
recordStatus(db: Db, companyId: string, agentId: string, before: string, after: string): Promise<void>;
|
||||
clearPrimary(db: Db, companyId: string, agentId: string): Promise<unknown>;
|
||||
prepareHire(db: Db, companyId: string, data: CreateAgentData, options?: CreateAgentOptions): Promise<CreateAgentData>;
|
||||
initializeHire(db: Db, agent: AgentHireRecord, options?: CreateAgentOptions): Promise<void>;
|
||||
prepareHireApproval(db: Db, agent: AgentHireRecord, payload?: Record<string, unknown> | null): Promise<AgentConfigurationPatch>;
|
||||
completeHireApproval(db: Db, agent: AgentHireRecord, previous: AgentHireRecord): Promise<void>;
|
||||
}
|
||||
@@ -0,0 +1,151 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, asc, eq, inArray, sql } from "drizzle-orm";
|
||||
import { agents, agentApiKeys, companies, budgetPolicies, type Db } from "@paperclipai/db";
|
||||
import type { AgentLifecycleOperation } from "@paperclipai/shared";
|
||||
import { AgentLifecycleConflict, compatibilityStatus, transition } from "../domain/policy.js";
|
||||
import type { LifecycleEffects } from "./effects.js";
|
||||
import type { LifecycleStore } from "../application/ports.js";
|
||||
|
||||
export function assertRootDatabase(db: Db) {
|
||||
if ("nestedIndex" in db) throw new AgentLifecycleConflict("Lifecycle commands require the root database connection");
|
||||
}
|
||||
|
||||
export function createLifecycleStore(db: Db, effects: LifecycleEffects): LifecycleStore {
|
||||
const get = (id: string) => db.select().from(agents).where(eq(agents.id, id)).then(rows => rows[0] ?? null);
|
||||
return {
|
||||
get,
|
||||
async change(id, command, options = {}) {
|
||||
const existing = await get(id);
|
||||
if (!existing) return null;
|
||||
return effects.transaction(db, existing.companyId, async tx => {
|
||||
const [agent] = await tx.select().from(agents).where(eq(agents.id, id)).for("update", command === "reconcile" ? { skipLocked: true } : undefined);
|
||||
if (!agent) return null;
|
||||
if (options.version !== undefined && options.version !== agent.lifecycleVersion) return null;
|
||||
if (command === "complete" && (!options.owner || !agent.lifecycleOperation?.leaseUntil || agent.lifecycleOperation?.leaseOwner !== options.owner ||
|
||||
Date.parse(agent.lifecycleOperation?.leaseUntil ?? "") <= Date.now() ||
|
||||
!agent.lifecycleOperation.hostComplete || agent.lifecycleRequiredPluginIds === null ||
|
||||
agent.lifecycleRequiredPluginIds.some(id => !agent.lifecycleOperation!.completedPluginIds.includes(id)))) return null;
|
||||
if (command === "retry") {
|
||||
transition(agent.lifecycleState, command);
|
||||
if (agent.lifecycleOperation?.leaseUntil && Date.parse(agent.lifecycleOperation.leaseUntil) > Date.now()) return agent;
|
||||
const [updated] = await tx.update(agents).set({ lifecycleError: null,
|
||||
lifecycleOperation: sql`${agents.lifecycleOperation} - 'retryAt' - 'leaseOwner' - 'leaseUntil'`, updatedAt: new Date() }).where(eq(agents.id, id)).returning();
|
||||
return updated;
|
||||
}
|
||||
if (command === "resume" && !["ready", "paused", "pausing", "resuming"].includes(agent.lifecycleState)) transition(agent.lifecycleState, command);
|
||||
let holds = agent.lifecycleHolds;
|
||||
const reason = options.reason ?? "manual";
|
||||
if (command === "reconcile") {
|
||||
if (["pending_approval", "terminating", "cleaning_up", "terminated", "rejected"].includes(agent.lifecycleState)) return agent;
|
||||
const [company] = await tx.select().from(companies).where(eq(companies.id, agent.companyId));
|
||||
const policies = await tx.select().from(budgetPolicies).where(and(eq(budgetPolicies.companyId, agent.companyId),
|
||||
eq(budgetPolicies.scopeType, "agent"), eq(budgetPolicies.scopeId, id)));
|
||||
let budgetBlocked = false;
|
||||
for (const policy of policies) if (await effects.policyBlocks(tx, policy)) budgetBlocked = true;
|
||||
holds = holds.filter(hold => hold !== "budget" && hold !== "company_archived" && hold !== "company_paused");
|
||||
if (budgetBlocked) holds.push("budget");
|
||||
if (company.status === "archived") holds.push("company_archived");
|
||||
if (company.status === "paused") holds.push("company_paused");
|
||||
}
|
||||
if (command === "pause") holds = [...new Set([...holds, reason])];
|
||||
if (command === "resume") holds = holds.filter(hold => reason === "user" ? ["budget", "company_archived", "company_paused"].includes(hold) : hold !== reason);
|
||||
let next = command === "reconcile"
|
||||
? holds.length && !["pausing", "paused"].includes(agent.lifecycleState) ? "pausing" as const
|
||||
: !holds.length && agent.lifecycleState === "paused" ? "resuming" as const : agent.lifecycleState
|
||||
: command === "resume" && (holds.length || agent.lifecycleState === "pausing") ? agent.lifecycleState
|
||||
: transition(agent.lifecycleState, command, agent.lifecycleOperation?.resumeState);
|
||||
if (command === "complete" && next === "ready" && holds.length) next = "pausing";
|
||||
if (command === "complete" && next === "paused" && !holds.length) next = "resuming";
|
||||
if (next === agent.lifecycleState) {
|
||||
if (JSON.stringify(holds) !== JSON.stringify(agent.lifecycleHolds) || agent.pauseReason !== (holds[0] ?? null)) {
|
||||
await tx.update(agents).set({ lifecycleHolds: holds, pauseReason: holds[0] ?? null }).where(eq(agents.id, id));
|
||||
}
|
||||
return agent;
|
||||
}
|
||||
const operation: AgentLifecycleOperation = {
|
||||
responsibleUserId: agent.lifecycleOperation?.responsibleUserId,
|
||||
id: randomUUID(), hostComplete: false, completedPluginIds: [], attempts: 0,
|
||||
resumeState: next === "pausing"
|
||||
? agent.lifecycleState === "preparing" || agent.lifecycleState === "verifying" ? agent.lifecycleState : agent.lifecycleOperation?.resumeState ?? "ready"
|
||||
: agent.lifecycleOperation?.resumeState,
|
||||
};
|
||||
if (next === "ready") operation.resumeState = "ready";
|
||||
const status = compatibilityStatus(next, agent.status);
|
||||
const [updated] = await tx.update(agents).set({
|
||||
lifecycleHolds: holds, lifecycleState: next, lifecycleVersion: agent.lifecycleVersion + 1, lifecycleError: null,
|
||||
lifecycleOperation: operation, status,
|
||||
pauseReason: holds[0] ?? null,
|
||||
pausedAt: next === "pausing" ? new Date() : next === "ready" ? null : agent.pausedAt,
|
||||
updatedAt: new Date(),
|
||||
}).where(eq(agents.id, id)).returning();
|
||||
if (next === "terminating" || next === "rejected") {
|
||||
await tx.update(agentApiKeys).set({ revokedAt: new Date() }).where(eq(agentApiKeys.agentId, id));
|
||||
await effects.clearPrimary(tx, agent.companyId, id);
|
||||
}
|
||||
if (command === "approve") await effects.recordCreation(tx, agent.companyId, id);
|
||||
if (next === "pausing" || next === "resuming" || next === "terminating" || next === "rejected") {
|
||||
await effects.recordStatus(tx, agent.companyId, id,
|
||||
next === "resuming" ? "paused" : "idle",
|
||||
next === "resuming" ? "idle" : next === "pausing" ? "paused" : "terminated");
|
||||
}
|
||||
return updated;
|
||||
});
|
||||
},
|
||||
async claim(id, owner, now) {
|
||||
const [agent] = await db.update(agents).set({
|
||||
lifecycleOperation: sql`coalesce(${agents.lifecycleOperation}, '{"id":"", "hostComplete":false, "completedPluginIds":[], "attempts":0}'::jsonb)
|
||||
|| jsonb_build_object('leaseOwner', ${owner}::text, 'leaseUntil', ${(new Date(now.getTime() + 120_000)).toISOString()}::text,
|
||||
'attempts', coalesce((${agents.lifecycleOperation}->>'attempts')::integer, 0) + 1)`,
|
||||
}).where(and(eq(agents.id, id), inArray(agents.lifecycleState, ["preparing", "verifying", "pausing", "resuming", "terminating", "cleaning_up"]),
|
||||
sql`coalesce((${agents.lifecycleOperation}->>'leaseUntil')::timestamptz, '-infinity') <= ${now.toISOString()}::timestamptz`,
|
||||
sql`coalesce((${agents.lifecycleOperation}->>'retryAt')::timestamptz, '-infinity') <= ${now.toISOString()}::timestamptz`,
|
||||
)).returning();
|
||||
return agent ?? null;
|
||||
},
|
||||
async renew(agent, owner, now) {
|
||||
const rows = await db.update(agents).set({ lifecycleOperation: sql`jsonb_set(${agents.lifecycleOperation}, '{leaseUntil}', ${JSON.stringify(new Date(now.getTime() + 120_000).toISOString())}::jsonb)` })
|
||||
.where(and(eq(agents.id, agent.id), eq(agents.lifecycleVersion, agent.lifecycleVersion), sql`${agents.lifecycleOperation}->>'leaseOwner' = ${owner}`)).returning({ id: agents.id });
|
||||
return rows.length > 0;
|
||||
},
|
||||
async setRequiredPlugins(agent, owner, pluginIds) {
|
||||
const rows = await db.update(agents).set({ lifecycleRequiredPluginIds: pluginIds })
|
||||
.where(and(eq(agents.id, agent.id), eq(agents.lifecycleVersion, agent.lifecycleVersion),
|
||||
sql`${agents.lifecycleRequiredPluginIds} is null`,
|
||||
sql`${agents.lifecycleOperation}->>'leaseOwner' = ${owner}`)).returning({ id: agents.id });
|
||||
return rows.length > 0;
|
||||
},
|
||||
async completeHost(agent, owner) {
|
||||
const rows = await db.update(agents).set({
|
||||
lifecycleOperation: sql`jsonb_set(${agents.lifecycleOperation}, '{hostComplete}', 'true'::jsonb)`,
|
||||
}).where(and(eq(agents.id, agent.id), eq(agents.lifecycleVersion, agent.lifecycleVersion),
|
||||
sql`${agents.lifecycleOperation}->>'leaseOwner' = ${owner}`)).returning({ id: agents.id });
|
||||
return rows.length > 0;
|
||||
},
|
||||
async completePlugin(agent, owner, pluginId) {
|
||||
const rows = await db.update(agents).set({
|
||||
lifecycleOperation: sql`jsonb_set(${agents.lifecycleOperation}, '{completedPluginIds}',
|
||||
coalesce(${agents.lifecycleOperation}->'completedPluginIds', '[]'::jsonb) || ${JSON.stringify([pluginId])}::jsonb)`,
|
||||
}).where(and(eq(agents.id, agent.id), eq(agents.lifecycleVersion, agent.lifecycleVersion),
|
||||
sql`${agents.lifecycleRequiredPluginIds} @> ${JSON.stringify([pluginId])}::jsonb`,
|
||||
sql`${agents.lifecycleOperation}->>'leaseOwner' = ${owner}`)).returning({ id: agents.id });
|
||||
return rows.length > 0;
|
||||
},
|
||||
async defer(agent, owner, error, now) {
|
||||
const rows = await db.update(agents).set({
|
||||
lifecycleError: error,
|
||||
lifecycleOperation: sql`(${agents.lifecycleOperation} - 'leaseOwner' - 'leaseUntil')
|
||||
|| jsonb_build_object('retryAt', ${new Date(now.getTime() + (error ? 60_000 : 2_000)).toISOString()}::text)`,
|
||||
updatedAt: now,
|
||||
}).where(and(eq(agents.id, agent.id), eq(agents.lifecycleVersion, agent.lifecycleVersion),
|
||||
sql`${agents.lifecycleOperation}->>'leaseOwner' = ${owner}`)).returning({ id: agents.id });
|
||||
return rows.length > 0;
|
||||
},
|
||||
async pending(limit, now) {
|
||||
return db.select({ id: agents.id }).from(agents).where(and(
|
||||
inArray(agents.lifecycleState, ["preparing", "verifying", "pausing", "resuming", "terminating", "cleaning_up"]),
|
||||
sql`coalesce((${agents.lifecycleOperation}->>'leaseUntil')::timestamptz, '-infinity') <= ${now.toISOString()}::timestamptz`,
|
||||
sql`coalesce((${agents.lifecycleOperation}->>'retryAt')::timestamptz, '-infinity') <= ${now.toISOString()}::timestamptz`,
|
||||
)).orderBy(asc(agents.updatedAt), asc(agents.id)).limit(limit).then(rows => rows.map(row => row.id));
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
import type { LifecycleEffects } from "./effects.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { agents, agentApiKeys, type Db } from "@paperclipai/db";
|
||||
import { conflict, notFound } from "../../../errors.js";
|
||||
import { agentRecordQueries, type CreateAgentOptions, type CreateAgentData } from "../../../lib/agent-records.js";
|
||||
|
||||
export function agentRecords(db: Db, effects: LifecycleEffects) {
|
||||
const { getById } = agentRecordQueries(db);
|
||||
|
||||
return {
|
||||
getById,
|
||||
create: async (companyId: string, data: CreateAgentData, options?: CreateAgentOptions) => {
|
||||
if (Object.keys(data).some(key => key.startsWith("lifecycle"))) throw conflict("Lifecycle fields belong to the lifecycle module");
|
||||
return effects.transaction(db, companyId, async (txDb) => {
|
||||
const tx = txDb;
|
||||
const prepared = await effects.prepareHire(txDb, companyId, data, options);
|
||||
const created = await tx
|
||||
.insert(agents)
|
||||
.values({
|
||||
...prepared,
|
||||
lifecycleState: data.status === "pending_approval" ? "pending_approval" : data.status === "terminated" ? "terminated" : data.status === "paused" ? "paused" : "preparing",
|
||||
lifecycleVersion: 1,
|
||||
lifecycleHolds: data.status === "paused" ? [data.pauseReason ?? "manual"] : [],
|
||||
pauseReason: data.status === "paused" ? data.pauseReason ?? "manual" : null,
|
||||
pausedAt: data.status === "paused" ? new Date() : null,
|
||||
lifecycleOperation: { id: randomUUID(), hostComplete: false, completedPluginIds: [], attempts: 0, resumeState: "preparing", responsibleUserId: options?.responsibleUserId ?? options?.createdByUserId },
|
||||
status: data.status === "pending_approval" ? "pending_approval" : data.status === "terminated" ? "terminated" : "paused",
|
||||
companyId,
|
||||
})
|
||||
.returning()
|
||||
.then((rows) => rows[0]);
|
||||
await effects.initializeHire(txDb, created, options);
|
||||
if (created.status !== "pending_approval" && created.status !== "terminated") {
|
||||
await effects.recordCreation(txDb, companyId, created.id);
|
||||
}
|
||||
const normalizedCreated = await agentRecordQueries(txDb).getById(created.id);
|
||||
if (!normalizedCreated) {
|
||||
throw notFound("Agent not found");
|
||||
}
|
||||
return normalizedCreated;
|
||||
});
|
||||
},
|
||||
|
||||
clearError: async (id: string) => {
|
||||
const existing = await getById(id);
|
||||
if (!existing) return null;
|
||||
if (existing.status === "terminated") throw conflict("Cannot clear error on terminated agent");
|
||||
if (existing.status === "pending_approval") {
|
||||
throw conflict("Pending approval agents cannot have errors cleared");
|
||||
}
|
||||
if (existing.status !== "error") {
|
||||
throw conflict("Only agents in error status can have their error cleared");
|
||||
}
|
||||
|
||||
const updated = await db
|
||||
.update(agents)
|
||||
.set({
|
||||
status: "idle",
|
||||
pauseReason: null,
|
||||
pausedAt: null,
|
||||
errorReason: null,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(and(eq(agents.id, id), eq(agents.status, "error"), eq(agents.lifecycleState, "ready")))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
if (!updated) {
|
||||
throw conflict("Only agents in error status can have their error cleared");
|
||||
}
|
||||
return getById(updated.id);
|
||||
},
|
||||
|
||||
rejectPendingHire: async (id: string) => {
|
||||
const rows = await db.update(agents).set({ status: "terminated", lifecycleState: "rejected",
|
||||
lifecycleVersion: sql`${agents.lifecycleVersion} + 1`, lifecycleOperation: null,
|
||||
updatedAt: new Date() }).where(and(eq(agents.id, id), eq(agents.lifecycleState, "pending_approval"))).returning();
|
||||
for (const row of rows) {
|
||||
await effects.clearPrimary(db, row.companyId, id);
|
||||
await db.update(agentApiKeys).set({ revokedAt: new Date() }).where(eq(agentApiKeys.agentId, id));
|
||||
await effects.recordStatus(db, row.companyId, id, "pending_approval", "terminated");
|
||||
}
|
||||
return rows;
|
||||
},
|
||||
|
||||
activatePendingApproval: async (id: string, approvedPayload?: Record<string, unknown> | null, requestedByUserId?: string | null) => {
|
||||
const activatedAgent = await db.transaction(async (tx) => {
|
||||
const txDb = tx as unknown as Db;
|
||||
const existing = await agentRecordQueries(txDb).getById(id);
|
||||
if (!existing || existing.status !== "pending_approval") return null;
|
||||
const patch = await effects.prepareHireApproval(txDb, existing, approvedPayload);
|
||||
const updated = await tx
|
||||
.update(agents)
|
||||
.set({ ...patch, status: "paused", lifecycleState: "preparing", lifecycleVersion: existing.lifecycleVersion + 1, lifecycleError: null, lifecycleOperation: sql`jsonb_build_object('id', ${randomUUID()}::text, 'hostComplete', false, 'completedPluginIds', '[]'::jsonb, 'attempts', 0, 'responsibleUserId', coalesce(nullif(${agents.lifecycleOperation}->'responsibleUserId', 'null'::jsonb), to_jsonb(${requestedByUserId ?? null}::text)))`, updatedAt: new Date() })
|
||||
.where(and(eq(agents.id, id), eq(agents.status, "pending_approval")))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!updated) return null;
|
||||
await effects.completeHireApproval(txDb, updated, existing);
|
||||
await effects.recordCreation(txDb, existing.companyId, updated.id);
|
||||
const agent = await agentRecordQueries(txDb).getById(updated.id);
|
||||
if (!agent) {
|
||||
throw notFound("Agent not found");
|
||||
}
|
||||
return agent;
|
||||
});
|
||||
|
||||
if (activatedAgent) {
|
||||
return { agent: activatedAgent, activated: true };
|
||||
}
|
||||
|
||||
const existing = await getById(id);
|
||||
return existing ? { agent: existing, activated: false } : null;
|
||||
},
|
||||
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user