Files
PaperClipAI/server/src/services/approvals.ts
T
Devin Foley dc12197cce fix: prevent duplicate built-in agents and self-heal reconciliation (#10223)
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Every company is auto-provisioned a set of built-in agents (e.g. the
Summarizer), and a startup reconciler keeps that set correct across
every company on boot.
> - Provisioning marks these agents with
`metadata.paperclipBuiltInAgent.key`, but nothing in the database
enforced one active agent per `(company, key)` —
`provision()`/`ensure()` did a check-then-insert with no guard.
> - Two concurrent server processes (e.g. a `tsx watch` double-boot)
could both read "no summarizer exists" and both insert, leaving a
company with duplicate built-in agents plus paired orphan pending
`hire_agent` approvals.
> - That data blemish then became a recurring outage: `findSingleAgent`
throws on >1 marked row, and because the throw escaped
`reconcileBuiltInAgentsOnStartup`'s sequential loop, **every company
after the affected one was silently skipped** on each boot — no
auto-provisioning, no default grants — until manual DB surgery.
> - This pull request closes the race at the database level and makes
reconciliation self-healing and fault-isolated.
> - The benefit is that concurrent provisioning can no longer create
duplicates, and even pre-existing duplicates are resolved automatically
instead of bricking startup reconciliation for unrelated companies.

## Linked Issues or Issue Description

- [x] I searched the GitHub PR list (open and recently closed) for
similar PRs and confirmed this is not a duplicate.

No public GitHub issue exists; describing the bug in-PR (bug-report
shape):

**What happened**

A dev instance booted with two concurrent server processes. Both ran
built-in agent provisioning for the same company at the same time, and
the check-then-insert in `provision()`/`ensure()`
(`server/src/services/built-in-agents.ts`) let both writers see "no
summarizer exists" and each create one — the company ended up with two
identical Summarizer agents (identical `paperclipBuiltInAgent` markers)
plus two paired pending `hire_agent` approvals.

From then on, **every** server boot logged:

```
ERROR: startup reconciliation of built-in agents failed
       Multiple built-in agents found for summarizer (built_in_agent_duplicate_instance)
```

because `findSingleAgent` throws on >1 marked row rather than resolving
the duplicate. Worse, `reconcileBuiltInAgentsOnStartup` loops companies
sequentially and the throw escaped the loop, so every company *after*
the affected one was silently skipped on every boot.

**Expected behavior**

1. Concurrent provisioning must not create duplicate built-in agents
(there was no DB uniqueness constraint on the marker key per company).
2. Reconciliation should be resilient: if duplicates exist anyway,
self-heal (keep the oldest row, terminate the newer dupe, cancel its
orphan pending `hire_agent` approval), and never let one bad company
abort reconciliation for the rest.

**Steps to reproduce**

- Race two `provision(companyId, "summarizer")` calls for a company with
board approval for new agents enabled (or simulate a double-boot); both
insert.
- Restart the server → startup reconciliation error fires, companies
later in the loop are never reconciled.

## What Changed

**Part 1 — stop creating duplicates**

- Migration `0192_built_in_agent_unique_marker` adds a **partial unique
index** on `(company_id, metadata->'paperclipBuiltInAgent'->>'key')`
where the marker exists and `status != 'terminated'`. It first resolves
any pre-existing duplicates (keep oldest by `created_at`, terminate
newer dupes, cancel their orphan pending `hire_agent` approvals, revoke
their API keys) so the index can be created on already-affected
instances.
- `provision()`/`ensure()` now catch the losing race's `23505` unique
violation (walking the driver's wrapped cause chain) and re-resolve to
the winning row instead of surfacing the error.

**Part 2 — resilient reconciliation**

- `findSingleAgent` self-heals: keeps the oldest marked row, terminates
the newer duplicates, and cancels each one's orphan pending `hire_agent`
approval (idempotent) instead of throwing.
- `reconcileBuiltInAgentsOnStartup` isolates per-company failures in
both loops so one bad company can't abort reconciliation for the rest;
it surfaces a `companyFailures` count in the startup log.
- Adds `approvalService.cancel()` for system-initiated cancellation of
an orphan approval.

## Verification

- `pnpm --filter @paperclipai/db run check:migrations` → numbering +
safety checks pass.
- `packages/db` migration test (real embedded Postgres) — seeds
pre-index duplicate state, runs the migration, asserts dupes resolved +
index enforced: **1 passed**.
- `server` `built-in-agents.test.ts` — self-heal, concurrent races
(plain and board-gated), and startup
self-heal-without-aborting-later-companies: **34 passed**.

```
pnpm --filter @paperclipai/db exec vitest run src/built-in-agent-unique-marker-migration.test.ts
pnpm --filter @paperclipai/server exec vitest run src/__tests__/built-in-agents.test.ts
```

## Risks

- **Migration safety**: the migration mutates data (terminates duplicate
rows, cancels their orphan pending approvals, revokes their API keys)
before creating the index. It keeps the oldest row per `(company, key)`
and only touches non-terminated marked rows; the destructive step is
covered by the migration test and the safety-check baseline. On a clean
instance it is a no-op cleanup followed by `CREATE UNIQUE INDEX IF NOT
EXISTS`.
- Otherwise low risk: the unique index is partial (excludes terminated
rows, so re-provisioning after a termination stays possible), and the
conflict handling degrades gracefully to re-resolving the existing
winner.

## Model Used

Claude Opus 4.8 (`claude-opus-4-8`), 1M context window, extended
thinking, with tool use.
2026-07-28 11:12:58 -07:00

315 lines
11 KiB
TypeScript

import { and, asc, eq, inArray, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import { approvalComments, approvals } from "@paperclipai/db";
import { notFound, unprocessable } from "../errors.js";
import { redactCurrentUserText } from "../log-redaction.js";
import { agentService } from "./agents.js";
import { budgetService } from "./budgets.js";
import { notifyHireApproved } from "./hire-hook.js";
import { instanceSettingsService } from "./instance-settings.js";
export function approvalService(db: Db) {
const agentsSvc = agentService(db);
const budgets = budgetService(db);
const instanceSettings = instanceSettingsService(db);
const canResolveStatuses = new Set(["pending", "revision_requested"]);
const resolvableStatuses = Array.from(canResolveStatuses);
type ApprovalRecord = typeof approvals.$inferSelect;
type ResolutionResult = { approval: ApprovalRecord; applied: boolean };
function redactApprovalComment<T extends { body: string }>(comment: T, censorUsernameInLogs: boolean): T {
return {
...comment,
body: redactCurrentUserText(comment.body, { enabled: censorUsernameInLogs }),
};
}
async function reconcileApprovedBuiltInAgent(companyId: string, payload: Record<string, unknown>) {
const sourceBuiltInAgentKey = typeof payload.sourceBuiltInAgentKey === "string" ? payload.sourceBuiltInAgentKey : null;
if (!sourceBuiltInAgentKey) return;
const { builtInAgentService } = await import("./built-in-agents.js");
await builtInAgentService(db).ensure(companyId, sourceBuiltInAgentKey);
}
async function getExistingApproval(id: string) {
const existing = await db
.select()
.from(approvals)
.where(eq(approvals.id, id))
.then((rows) => rows[0] ?? null);
if (!existing) throw notFound("Approval not found");
return existing;
}
async function resolveApproval(
id: string,
targetStatus: "approved" | "rejected",
decidedByUserId: string,
decisionNote: string | null | undefined,
): Promise<ResolutionResult> {
const existing = await getExistingApproval(id);
if (!canResolveStatuses.has(existing.status)) {
if (existing.status === targetStatus) {
return { approval: existing, applied: false };
}
throw unprocessable(
`Only pending or revision requested approvals can be ${targetStatus === "approved" ? "approved" : "rejected"}`,
);
}
const now = new Date();
const updated = await db
.update(approvals)
.set({
status: targetStatus,
decidedByUserId,
decisionNote: decisionNote ?? null,
decidedAt: now,
updatedAt: now,
})
.where(and(eq(approvals.id, id), inArray(approvals.status, resolvableStatuses)))
.returning()
.then((rows) => rows[0] ?? null);
if (updated) {
return { approval: updated, applied: true };
}
const latest = await getExistingApproval(id);
if (latest.status === targetStatus) {
return { approval: latest, applied: false };
}
throw unprocessable(
`Only pending or revision requested approvals can be ${targetStatus === "approved" ? "approved" : "rejected"}`,
);
}
return {
list: (companyId: string, status?: string) => {
const conditions = [eq(approvals.companyId, companyId)];
if (status) conditions.push(eq(approvals.status, status));
return db.select().from(approvals).where(and(...conditions));
},
getById: (id: string) =>
db
.select()
.from(approvals)
.where(eq(approvals.id, id))
.then((rows) => rows[0] ?? null),
findOpenHireApprovalForAgent: async (companyId: string, agentId: string) => {
const rows = await db
.select()
.from(approvals)
.where(
and(
eq(approvals.companyId, companyId),
eq(approvals.type, "hire_agent"),
inArray(approvals.status, resolvableStatuses),
sql`${approvals.payload} ->> 'agentId' = ${agentId}`,
),
);
return rows[0] ?? null;
},
create: (companyId: string, data: Omit<typeof approvals.$inferInsert, "companyId">) =>
db
.insert(approvals)
.values({ ...data, companyId })
.returning()
.then((rows) => rows[0]),
// Cancel an open (pending/revision_requested) approval without a board
// decision — e.g. when its paired agent is terminated during duplicate
// cleanup. Idempotent: a no-op on already-resolved approvals.
cancel: async (id: string, reason?: string | null) => {
const now = new Date();
const updated = await db
.update(approvals)
.set({
status: "cancelled",
decisionNote: reason ?? null,
decidedAt: now,
updatedAt: now,
})
.where(and(eq(approvals.id, id), inArray(approvals.status, resolvableStatuses)))
.returning()
.then((rows) => rows[0] ?? null);
return updated;
},
approve: async (id: string, decidedByUserId: string, decisionNote?: string | null) => {
const { approval: updated, applied } = await resolveApproval(
id,
"approved",
decidedByUserId,
decisionNote,
);
let hireApprovedAgentId: string | null = null;
const now = new Date();
if (applied && updated.type === "hire_agent") {
const payload = updated.payload as Record<string, unknown>;
const payloadAgentId = typeof payload.agentId === "string" ? payload.agentId : null;
if (payloadAgentId) {
await agentsSvc.activatePendingApproval(payloadAgentId, payload);
await reconcileApprovedBuiltInAgent(updated.companyId, payload);
hireApprovedAgentId = payloadAgentId;
} else {
const created = await agentsSvc.create(updated.companyId, {
name: String(payload.name ?? "New Agent"),
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,
});
hireApprovedAgentId = created?.id ?? null;
}
if (hireApprovedAgentId) {
const budgetMonthlyCents =
typeof payload.budgetMonthlyCents === "number" ? payload.budgetMonthlyCents : 0;
if (budgetMonthlyCents > 0) {
await budgets.upsertPolicy(
updated.companyId,
{
scopeType: "agent",
scopeId: hireApprovedAgentId,
amount: budgetMonthlyCents,
windowKind: "calendar_month_utc",
},
decidedByUserId,
);
}
void notifyHireApproved(db, {
companyId: updated.companyId,
agentId: hireApprovedAgentId,
source: "approval",
sourceId: id,
approvedAt: now,
}).catch(() => {});
}
}
return { approval: updated, applied };
},
reject: async (id: string, decidedByUserId: string, decisionNote?: string | null) => {
const { approval: updated, applied } = await resolveApproval(
id,
"rejected",
decidedByUserId,
decisionNote,
);
if (applied && updated.type === "hire_agent") {
const payload = updated.payload as Record<string, unknown>;
const payloadAgentId = typeof payload.agentId === "string" ? payload.agentId : null;
if (payloadAgentId) {
await agentsSvc.terminate(payloadAgentId);
}
}
return { approval: updated, applied };
},
requestRevision: async (id: string, decidedByUserId: string, decisionNote?: string | null) => {
const existing = await getExistingApproval(id);
if (existing.status !== "pending") {
throw unprocessable("Only pending approvals can request revision");
}
const now = new Date();
return db
.update(approvals)
.set({
status: "revision_requested",
decidedByUserId,
decisionNote: decisionNote ?? null,
decidedAt: now,
updatedAt: now,
})
.where(eq(approvals.id, id))
.returning()
.then((rows) => rows[0]);
},
resubmit: async (id: string, payload?: Record<string, unknown>) => {
const existing = await getExistingApproval(id);
if (existing.status !== "revision_requested") {
throw unprocessable("Only revision requested approvals can be resubmitted");
}
const now = new Date();
return db
.update(approvals)
.set({
status: "pending",
payload: payload ?? existing.payload,
decisionNote: null,
decidedByUserId: null,
decidedAt: null,
updatedAt: now,
})
.where(eq(approvals.id, id))
.returning()
.then((rows) => rows[0]);
},
listComments: async (approvalId: string) => {
const existing = await getExistingApproval(approvalId);
const { censorUsernameInLogs } = await instanceSettings.getGeneral();
return db
.select()
.from(approvalComments)
.where(
and(
eq(approvalComments.approvalId, approvalId),
eq(approvalComments.companyId, existing.companyId),
),
)
.orderBy(asc(approvalComments.createdAt))
.then((comments) => comments.map((comment) => redactApprovalComment(comment, censorUsernameInLogs)));
},
addComment: async (
approvalId: string,
body: string,
actor: { agentId?: string; userId?: string },
) => {
const existing = await getExistingApproval(approvalId);
const currentUserRedactionOptions = {
enabled: (await instanceSettings.getGeneral()).censorUsernameInLogs,
};
const redactedBody = redactCurrentUserText(body, currentUserRedactionOptions);
return db
.insert(approvalComments)
.values({
companyId: existing.companyId,
approvalId,
authorAgentId: actor.agentId ?? null,
authorUserId: actor.userId ?? null,
body: redactedBody,
})
.returning()
.then((rows) => redactApprovalComment(rows[0], currentUserRedactionOptions.enabled));
},
};
}