RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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.
This commit is contained in:
Devin Foley 2026-07-28 11:12:58 -07:00 committed by GitHub
parent c8c2ae82a3
commit dc12197cce
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 507 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

119
packages/db/src/built-in-agent-unique-marker-migration.test.ts Normal file
View File

@ -0,0 +1,119 @@
import { createHash, randomUUID } from "node:crypto";
import fs from "node:fs";
import { afterEach, describe, expect, it } from "vitest";
import postgres from "postgres";
import { applyPendingMigrations } from "./client.js";
import {
getEmbeddedPostgresTestSupport,
startEmbeddedPostgresTestDatabase,
} from "./test-embedded-postgres.js";
const MIGRATION_FILE = "0195_built_in_agent_unique_marker.sql";
const UNIQUE_INDEX = "agents_company_built_in_agent_key_unique_idx";
const cleanups: Array<() => Promise<void>> = [];
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
async function migrationHash() {
const content = await fs.promises.readFile(new URL(`./migrations/${MIGRATION_FILE}`, import.meta.url), "utf8");
return createHash("sha256").update(content).digest("hex");
}
function markerObj(key: string) {
// Return a plain object (not a JSON string) so `sql.json` stores it as a jsonb
// object; a JSON string param + `::jsonb` would double-encode into a scalar.
return { paperclipBuiltInAgent: { key, featureKeys: [key] } };
}
describeEmbeddedPostgres("built-in agent unique marker migration", () => {
afterEach(async () => {
await Promise.all(cleanups.splice(0).map((cleanup) => cleanup()));
});
it("resolves pre-existing duplicates and enforces uniqueness going forward", async () => {
const database = await startEmbeddedPostgresTestDatabase("paperclip-built-in-unique-marker-");
cleanups.push(database.cleanup);
const sql = postgres(database.connectionString, { max: 1 });
cleanups.push(async () => sql.end());
// Rewind the migration so we can seed the pre-index (duplicate) state.
await sql`DELETE FROM "drizzle"."__drizzle_migrations" WHERE "hash" = ${await migrationHash()}`;
await sql`DROP INDEX IF EXISTS ${sql(UNIQUE_INDEX)}`;
const affectedCompanyId = randomUUID();
const cleanCompanyId = randomUUID();
const olderId = randomUUID();
const newerId = randomUUID();
const terminatedDupeId = randomUUID();
const cleanSummarizerId = randomUUID();
const approvalId = randomUUID();
const apiKeyId = randomUUID();
await sql`
INSERT INTO "companies" ("id", "name", "issue_prefix")
VALUES
(${affectedCompanyId}, 'Affected', 'AFF'),
(${cleanCompanyId}, 'Clean', 'CLN')
`;
await sql`
INSERT INTO "agents" ("id", "company_id", "name", "status", "created_at", "metadata")
VALUES
(${olderId}, ${affectedCompanyId}, 'Summarizer One', 'idle', '2026-07-18T00:00:00.000Z', ${sql.json(markerObj("summarizer"))}),
(${newerId}, ${affectedCompanyId}, 'Summarizer Two', 'pending_approval', '2026-07-18T00:00:00.025Z', ${sql.json(markerObj("summarizer"))}),
(${terminatedDupeId}, ${affectedCompanyId}, 'Summarizer Old', 'terminated', '2026-07-17T00:00:00.000Z', ${sql.json(markerObj("summarizer"))}),
(${cleanSummarizerId}, ${cleanCompanyId}, 'Summarizer', 'idle', '2026-07-18T00:00:00.000Z', ${sql.json(markerObj("summarizer"))})
`;
await sql`
INSERT INTO "approvals" ("id", "company_id", "type", "status", "payload")
VALUES (${approvalId}, ${affectedCompanyId}, 'hire_agent', 'pending', ${sql.json({ agentId: newerId, sourceBuiltInAgentKey: "summarizer" })})
`;
await sql`
INSERT INTO "agent_api_keys" ("id", "agent_id", "company_id", "name", "key_hash")
VALUES (${apiKeyId}, ${newerId}, ${affectedCompanyId}, 'dupe-key', 'hash')
`;
// Re-run the migration: it should clean up the duplicates, then recreate the index.
await applyPendingMigrations(database.connectionString);
const agentRows = await sql<{ id: string; status: string }[]>`
SELECT "id", "status" FROM "agents"
WHERE "company_id" IN (${affectedCompanyId}, ${cleanCompanyId})
`;
const statusById = new Map(agentRows.map((row) => [row.id, row.status]));
// Oldest active row kept; newer duplicate terminated; the clean company untouched.
expect(statusById.get(olderId)).toBe("idle");
expect(statusById.get(newerId)).toBe("terminated");
expect(statusById.get(cleanSummarizerId)).toBe("idle");
const activeMarked = await sql<{ count: string }[]>`
SELECT count(*)::text AS count FROM "agents"
WHERE "company_id" = ${affectedCompanyId}
AND "status" <> 'terminated'
AND ("metadata" -> 'paperclipBuiltInAgent' ->> 'key') = 'summarizer'
`;
expect(activeMarked[0]!.count).toBe("1");
// The newer duplicate's orphan approval was cancelled and its API key revoked.
const [approval] = await sql<{ status: string }[]>`
SELECT "status" FROM "approvals" WHERE "id" = ${approvalId}
`;
expect(approval!.status).toBe("cancelled");
const [apiKey] = await sql<{ revoked_at: Date | null }[]>`
SELECT "revoked_at" FROM "agent_api_keys" WHERE "id" = ${apiKeyId}
`;
expect(apiKey!.revoked_at).not.toBeNull();
// The partial unique index exists again and now rejects a second active row.
const indexes = await sql<{ indexname: string }[]>`
SELECT "indexname" FROM "pg_indexes"
WHERE "tablename" = 'agents' AND "indexname" = ${UNIQUE_INDEX}
`;
expect(indexes).toHaveLength(1);
await expect(
sql`
INSERT INTO "agents" ("id", "company_id", "name", "status", "metadata")
VALUES (${randomUUID()}, ${affectedCompanyId}, 'Summarizer Dupe', 'idle', ${sql.json(markerObj("summarizer"))})
`,
).rejects.toMatchObject({ code: "23505", constraint_name: UNIQUE_INDEX });
}, 30_000);
});

53
packages/db/src/migrations/0195_built_in_agent_unique_marker.sql Normal file
View File

@ -0,0 +1,53 @@
-- Enforce one active built-in agent per (company, marker key).
--
-- Provisioning used to check-then-insert with no database guard, so two
-- concurrent server processes could each see "no summarizer exists" and both
-- insert, leaving a company with duplicate built-in agents plus paired orphan
-- pending hire_agent approvals.
--
-- Step 1: resolve any pre-existing duplicates so the unique index below can be
-- created. Keep the oldest marked row per (company, key), terminate the newer
-- duplicates, cancel their orphan pending hire_agent approvals, and revoke any
-- API keys those terminated rows held.
WITH ranked AS (
SELECT
id,
row_number() OVER (
PARTITION BY company_id, (metadata -> 'paperclipBuiltInAgent' ->> 'key')
ORDER BY created_at ASC, id ASC
) AS rn
FROM agents
WHERE (metadata -> 'paperclipBuiltInAgent' ->> 'key') IS NOT NULL
AND status <> 'terminated'
),
duplicates AS (
SELECT id FROM ranked WHERE rn > 1
),
terminated AS (
UPDATE agents
SET status = 'terminated', updated_at = now()
WHERE id IN (SELECT id FROM duplicates)
RETURNING id
),
cancelled_approvals AS (
UPDATE approvals
SET status = 'cancelled', updated_at = now()
WHERE type = 'hire_agent'
AND status IN ('pending', 'revision_requested')
AND (payload ->> 'agentId') IN (SELECT id::text FROM terminated)
RETURNING id
)
UPDATE agent_api_keys
SET revoked_at = now()
WHERE revoked_at IS NULL
AND agent_id IN (SELECT id FROM terminated);
--> statement-breakpoint
-- Step 2: enforce uniqueness going forward. A partial unique index lets the
-- provisioning race lose cleanly (the losing INSERT raises 23505, which
-- provision()/ensure() catch and re-resolve to the winner's row). Terminated
-- rows are excluded so re-provisioning after a termination stays possible.
CREATE UNIQUE INDEX IF NOT EXISTS "agents_company_built_in_agent_key_unique_idx"
ON "agents" ("company_id", ((metadata -> 'paperclipBuiltInAgent' ->> 'key')))
WHERE (metadata -> 'paperclipBuiltInAgent' ->> 'key') IS NOT NULL
AND status <> 'terminated';

7
packages/db/src/migrations/meta/_journal.json
View File

@ -1352,6 +1352,13 @@
"when": 1784920485226,
"tag": "0194_company_skill_releases",
"breakpoints": true
},
{
"idx": 195,
"version": "7",
"when": 1785170000001,
"tag": "0195_built_in_agent_unique_marker",
"breakpoints": true
}
]
}

169
server/src/__tests__/built-in-agents.test.ts
View File

@ -3,7 +3,7 @@ import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
import { and, eq } from "drizzle-orm";
import { and, eq, sql } from "drizzle-orm";
import {
activityLog,
agentConfigRevisions,
@ -27,7 +27,6 @@ import {
getEmbeddedPostgresTestSupport,
startEmbeddedPostgresTestDatabase,
} from "./helpers/embedded-postgres.js";
import { HttpError } from "../errors.ts";
import { agentInstructionsService } from "../services/agent-instructions.ts";
import { agentService } from "../services/agents.ts";
import { approvalService } from "../services/approvals.ts";
@ -45,6 +44,16 @@ import { issueThreadInteractionService } from "../services/issue-thread-interact
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
const BUILT_IN_MARKER_UNIQUE_INDEX = "agents_company_built_in_agent_key_unique_idx";
// Mirrors migration 0192. Used to drop/restore the partial unique index when a
// test needs to simulate legacy duplicates that predate the constraint.
const BUILT_IN_MARKER_UNIQUE_INDEX_DDL = `
CREATE UNIQUE INDEX IF NOT EXISTS "${BUILT_IN_MARKER_UNIQUE_INDEX}"
ON "agents" ("company_id", ((metadata -> 'paperclipBuiltInAgent' ->> 'key')))
WHERE (metadata -> 'paperclipBuiltInAgent' ->> 'key') IS NOT NULL
AND status <> 'terminated'
`;
function issuePrefix(id: string) {
return `T${id.replace(/-/g, "").slice(0, 6).toUpperCase()}`;
}
@ -129,6 +138,9 @@ describeEmbeddedPostgres("built-in agents", () => {
await db.delete(agents);
await db.delete(budgetPolicies);
await db.delete(companies);
// Some tests drop the built-in marker unique index to simulate legacy
// (pre-migration 0192) duplicates; restore it now that all rows are gone.
await db.execute(sql.raw(BUILT_IN_MARKER_UNIQUE_INDEX_DDL));
});
afterAll(async () => {
@ -901,11 +913,16 @@ describeEmbeddedPostgres("built-in agents", () => {
expect(readBuiltInAgentMarker(row?.metadata)).toEqual({ key: "briefs", featureKeys: ["briefs"] });
});
it("reports duplicate active instances for a company/key", async () => {
const companyId = await seedCompany();
// Reproduce a company that already carries duplicate marked rows from before
// migration 0192 by dropping the unique index, inserting two active rows, and
// pairing a pending hire_agent approval with the newer one.
async function seedLegacyDuplicateBriefs(companyId: string) {
await db.execute(sql.raw(`DROP INDEX IF EXISTS "${BUILT_IN_MARKER_UNIQUE_INDEX}"`));
const olderId = randomUUID();
const newerId = randomUUID();
await db.insert(agents).values([
{
id: randomUUID(),
id: olderId,
companyId,
name: "Briefs One",
role: "general",
@ -914,29 +931,155 @@ describeEmbeddedPostgres("built-in agents", () => {
adapterConfig: { model: "gpt-5.4" },
runtimeConfig: {},
permissions: {},
createdAt: new Date("2026-07-18T00:00:00.000Z"),
metadata: withBuiltInAgentMarker({}, { key: "briefs", featureKeys: ["briefs"] }),
},
{
id: randomUUID(),
id: newerId,
companyId,
name: "Briefs Two",
role: "general",
status: "pending_approval",
adapterType: "codex_local",
adapterConfig: { model: "gpt-5.4" },
runtimeConfig: {},
permissions: {},
createdAt: new Date("2026-07-18T00:00:00.025Z"),
metadata: withBuiltInAgentMarker({}, { key: "briefs", featureKeys: ["briefs"] }),
},
]);
const approval = await approvalService(db).create(companyId, {
type: "hire_agent",
requestedByAgentId: null,
requestedByUserId: null,
status: "pending",
payload: { agentId: newerId, sourceBuiltInAgentKey: "briefs", featureKeys: ["briefs"] },
decisionNote: null,
decidedByUserId: null,
decidedAt: null,
updatedAt: new Date(),
});
return { olderId, newerId, approvalId: approval!.id };
}
it("self-heals duplicate active instances, keeping the oldest and cancelling the newer's approval", async () => {
const companyId = await seedCompany();
const { olderId, newerId, approvalId } = await seedLegacyDuplicateBriefs(companyId);
// A plain read resolves the duplicate rather than throwing.
const state = await builtInAgentService(db).get(companyId, "briefs");
expect(state.agentId).toBe(olderId);
const rows = await db.select().from(agents).where(eq(agents.companyId, companyId));
const byId = new Map(rows.map((row) => [row.id, row]));
expect(byId.get(olderId)?.status).toBe("idle");
expect(byId.get(newerId)?.status).toBe("terminated");
expect(
rows.filter((row) => row.status !== "terminated" && readBuiltInAgentMarker(row.metadata)?.key === "briefs"),
).toHaveLength(1);
const [approval] = await db.select().from(approvals).where(eq(approvals.id, approvalId));
expect(approval?.status).toBe("cancelled");
});
it("makes concurrent provisioning lose cleanly instead of creating duplicates", async () => {
const companyId = await seedCompany({ requireApproval: false });
const svc = builtInAgentService(db);
const [first, second] = await Promise.all([
svc.ensure(companyId, "briefs"),
svc.ensure(companyId, "briefs"),
]);
expect(first.agentId).toBeTruthy();
expect(second.agentId).toBe(first.agentId);
const rows = await db.select().from(agents).where(eq(agents.companyId, companyId));
expect(
rows.filter((row) => row.status !== "terminated" && readBuiltInAgentMarker(row.metadata)?.key === "briefs"),
).toHaveLength(1);
// The partial unique index now rejects any further active marked row.
await expect(
db.insert(agents).values({
id: randomUUID(),
companyId,
name: "Briefs Dupe",
role: "general",
status: "idle",
adapterType: "codex_local",
adapterConfig: { model: "gpt-5.4" },
runtimeConfig: {},
permissions: {},
metadata: withBuiltInAgentMarker({}, { key: "briefs", featureKeys: ["briefs"] }),
}),
).rejects.toMatchObject({
cause: {
code: "23505",
constraint_name: BUILT_IN_MARKER_UNIQUE_INDEX,
},
});
});
it("makes concurrent board-gated provisioning resolve to a single pending agent and approval", async () => {
const companyId = await seedCompany({ requireApproval: true });
const svc = builtInAgentService(db);
const results = await Promise.all([
svc.provision(companyId, "briefs"),
svc.provision(companyId, "briefs"),
]);
await expect(builtInAgentService(db).ensure(companyId, "briefs")).rejects.toMatchObject({
status: 409,
details: {
code: "built_in_agent_duplicate_instance",
key: "briefs",
},
} satisfies Partial<HttpError>);
const agentIds = new Set(results.map((result) => result.state.agentId));
expect(agentIds.size).toBe(1);
const rows = await db.select().from(agents).where(eq(agents.companyId, companyId));
expect(
rows.filter((row) => row.status !== "terminated" && readBuiltInAgentMarker(row.metadata)?.key === "briefs"),
).toHaveLength(1);
const openApprovals = await db
.select()
.from(approvals)
.where(and(eq(approvals.companyId, companyId), eq(approvals.status, "pending")));
const briefsApprovals = openApprovals.filter(
(row) => (row.payload as { sourceBuiltInAgentKey?: string } | null)?.sourceBuiltInAgentKey === "briefs",
);
expect(briefsApprovals).toHaveLength(1);
// The winner returns its freshly created approval; the loser returns the
// same approval or null (if it re-resolves before that row commits), never
// a second one.
expect(results.some((result) => result.approval?.id === briefsApprovals[0]!.id)).toBe(true);
for (const result of results) {
if (result.approval) {
expect(result.approval.id).toBe(briefsApprovals[0]!.id);
}
}
});
it("self-heals duplicates during startup reconciliation without aborting later companies", async () => {
const affectedCompanyId = await seedCompany({ requireApproval: false });
const { olderId, newerId } = await seedLegacyDuplicateBriefs(affectedCompanyId);
// A second company created after the affected one — previously skipped
// entirely because the duplicate error escaped the reconciliation loop.
const healthyCompanyId = await seedCompany({ requireApproval: false });
const result = await reconcileBuiltInAgentsOnStartup(db);
expect(result.companyFailures).toBe(0);
const affectedRows = await db.select().from(agents).where(eq(agents.companyId, affectedCompanyId));
const affectedById = new Map(affectedRows.map((row) => [row.id, row]));
expect(affectedById.get(olderId)?.status).toBe("idle");
expect(affectedById.get(newerId)?.status).toBe("terminated");
expect(
affectedRows.filter(
(row) => row.status !== "terminated" && readBuiltInAgentMarker(row.metadata)?.key === "briefs",
),
).toHaveLength(1);
// The company after the affected one still had its bundled agents provisioned.
const healthyCoach = await builtInAgentService(db).get(healthyCompanyId, "reflection-coach");
expect(healthyCoach.agentId).toBeTruthy();
});
it("automatically materializes the Reflection Coach bundle without enabling background work", async () => {

8
server/src/index.ts
View File

@ -829,7 +829,13 @@ export async function startServer(): Promise<StartedServer> {
void reconcileBuiltInAgentsOnStartup(db as any)
.then((result) => {
if (result.reconciled > 0 || result.unknown > 0 || result.duplicates > 0 || result.autoEnsured > 0) {
if (
result.reconciled > 0
|| result.unknown > 0
|| result.duplicates > 0
|| result.autoEnsured > 0
|| result.companyFailures > 0
) {
logger.warn(
result,
"startup reconciliation of built-in agents complete",

19
server/src/services/approvals.ts
View File

@ -121,6 +121,25 @@ export function approvalService(db: Db) {
.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,

186
server/src/services/built-in-agents.ts
View File

@ -800,6 +800,26 @@ function rowIsBuiltInAgent(row: typeof agents.$inferSelect, key: string) {
return marker?.key === key;
}
// Partial unique index (migration 0192) that guarantees one active built-in
// agent per (company, marker key). A losing provisioning race raises this as a
// 23505; provision()/ensure() catch it and re-resolve to the winning row.
const BUILT_IN_AGENT_MARKER_UNIQUE_INDEX = "agents_company_built_in_agent_key_unique_idx";
function isBuiltInAgentMarkerConflict(error: unknown): boolean {
const seen = new Set<unknown>();
let current: unknown = error;
while (typeof current === "object" && current !== null && !seen.has(current)) {
seen.add(current);
const maybe = current as { code?: string; constraint?: string; constraint_name?: string; cause?: unknown };
const constraint = maybe.constraint ?? maybe.constraint_name;
if (maybe.code === "23505" && constraint === BUILT_IN_AGENT_MARKER_UNIQUE_INDEX) {
return true;
}
current = maybe.cause;
}
return false;
}
export function builtInAgentService(db: Db) {
const agentSvc = agentService(db);
const accessSvc = accessService(db);
@ -1526,17 +1546,50 @@ export function builtInAgentService(db: Db) {
.sort((left, right) => left.createdAt.getTime() - right.createdAt.getTime() || left.id.localeCompare(right.id));
}
async function findSingleAgent(companyId: string, definition: BuiltInAgentDefinition) {
const markedRows = await findMarkedRows(companyId, definition.key);
if (markedRows.length > 1) {
throw conflict(`Multiple built-in agents found for ${definition.key}`, {
code: "built_in_agent_duplicate_instance",
key: definition.key,
agentIds: markedRows.map((row) => row.id),
// Self-heal duplicate built-in agents. `findMarkedRows` already sorts oldest
// first, so the first row is authoritative: keep it, terminate the rest, and
// cancel each duplicate's orphan pending hire_agent approval. Idempotent, so
// concurrent callers converge on the same surviving row.
async function resolveDuplicateMarkedRows(
companyId: string,
definition: BuiltInAgentDefinition,
markedRows: Array<typeof agents.$inferSelect>,
) {
const [keep, ...duplicates] = markedRows;
for (const duplicate of duplicates) {
const openApproval = await approvalSvc.findOpenHireApprovalForAgent(companyId, duplicate.id);
await agentSvc.terminate(duplicate.id);
if (openApproval) {
await approvalSvc.cancel(
openApproval.id,
`Cancelled: duplicate built-in ${definition.key} agent resolved during reconciliation.`,
);
}
await logActivity(db, {
companyId,
actorType: "system",
actorId: "built-in-agents",
action: "built_in_agent.duplicate_resolved",
entityType: "agent",
entityId: duplicate.id,
details: {
key: definition.key,
keptAgentId: keep!.id,
terminatedAgentId: duplicate.id,
cancelledApprovalId: openApproval?.id ?? null,
},
});
}
return keep!;
}
async function findSingleAgent(companyId: string, definition: BuiltInAgentDefinition) {
const markedRows = await findMarkedRows(companyId, definition.key);
if (markedRows.length === 0) return null;
const agent = await agentSvc.getById(markedRows[0]!.id);
const survivor = markedRows.length > 1
? await resolveDuplicateMarkedRows(companyId, definition, markedRows)
: markedRows[0]!;
const agent = await agentSvc.getById(survivor.id);
return agent as Agent | null;
}
@ -1561,7 +1614,12 @@ export function builtInAgentService(db: Db) {
return state(definition, await findSingleAgent(companyId, definition));
}
async function ensure(companyId: string, key: string, input: BuiltInAgentProvisionInput = {}) {
async function ensure(
companyId: string,
key: string,
input: BuiltInAgentProvisionInput = {},
options: { isRaceRetry?: boolean } = {},
) {
const definition = requireBuiltInAgentDefinition(key);
await ensureCompany(companyId);
const existing = await findSingleAgent(companyId, definition);
@ -1619,20 +1677,31 @@ export function builtInAgentService(db: Db) {
const reportsTo = definition.defaultManager === "single_root_agent"
? await findSingleRootManager(companyId)
: null;
const created = await agentSvc.create(companyId, {
...definitionPatch(definition, resolvedInput),
status: definition.defaultStatus ?? "idle",
pauseReason: definition.defaultStatus === "paused"
? `Built-in ${definition.displayName} is disabled until explicitly configured.`
: null,
pausedAt: definition.defaultStatus === "paused" ? new Date() : null,
reportsTo,
metadata: builtInMetadata(definition),
runtimeConfig: definition.defaultRuntimeConfig ?? {},
permissions: definition.defaultPermissions ?? {},
spentMonthlyCents: 0,
lastHeartbeatAt: null,
}, { allowBuiltInAgentMetadata: true }) as Agent;
let created: Agent;
try {
created = await agentSvc.create(companyId, {
...definitionPatch(definition, resolvedInput),
status: definition.defaultStatus ?? "idle",
pauseReason: definition.defaultStatus === "paused"
? `Built-in ${definition.displayName} is disabled until explicitly configured.`
: null,
pausedAt: definition.defaultStatus === "paused" ? new Date() : null,
reportsTo,
metadata: builtInMetadata(definition),
runtimeConfig: definition.defaultRuntimeConfig ?? {},
permissions: definition.defaultPermissions ?? {},
spentMonthlyCents: 0,
lastHeartbeatAt: null,
}, { allowBuiltInAgentMetadata: true }) as Agent;
} catch (error) {
// Lost the provisioning race: a concurrent writer inserted the row first
// and the partial unique index rejected ours. Re-run once; the winning
// row now exists, so we take the update path instead of inserting again.
if (!options.isRaceRetry && isBuiltInAgentMarkerConflict(error)) {
return ensure(companyId, key, input, { isRaceRetry: true });
}
throw error;
}
await logActivity(db, {
companyId,
@ -1713,16 +1782,34 @@ export function builtInAgentService(db: Db) {
const reportsTo = definition.defaultManager === "single_root_agent"
? await findSingleRootManager(companyId)
: null;
const pending = await agentSvc.create(companyId, {
...definitionPatch(definition, input),
status: "pending_approval",
reportsTo,
metadata: builtInMetadata(definition),
runtimeConfig: definition.defaultRuntimeConfig ?? {},
permissions: definition.defaultPermissions ?? {},
spentMonthlyCents: 0,
lastHeartbeatAt: null,
}, { allowBuiltInAgentMetadata: true }) as Agent;
let pending: Agent;
try {
pending = await agentSvc.create(companyId, {
...definitionPatch(definition, input),
status: "pending_approval",
reportsTo,
metadata: builtInMetadata(definition),
runtimeConfig: definition.defaultRuntimeConfig ?? {},
permissions: definition.defaultPermissions ?? {},
spentMonthlyCents: 0,
lastHeartbeatAt: null,
}, { allowBuiltInAgentMetadata: true }) as Agent;
} catch (error) {
// Lost the provisioning race: a concurrent writer inserted the row (and
// its own hire approval) first, and the partial unique index rejected
// ours before we created a paired approval. Re-resolve to the winner and
// return its pending state + open approval instead of surfacing the 23505.
if (isBuiltInAgentMarkerConflict(error)) {
const winner = await findSingleAgent(companyId, definition);
if (winner) {
const winnerApproval = winner.status === "pending_approval"
? await approvalSvc.findOpenHireApprovalForAgent(companyId, winner.id)
: null;
return { state: await state(definition, winner), approval: winnerApproval as Approval | null };
}
}
throw error;
}
const approval = await approvalSvc.create(companyId, {
type: "hire_agent",
@ -1906,11 +1993,22 @@ export async function reconcileBuiltInAgentsOnStartup(db: Db) {
let autoEnsured = 0;
let pendingApprovals = 0;
let defaultGrantsEnsured = 0;
// Isolate per-company failures so one bad company (e.g. an unresolvable data
// problem) can no longer abort reconciliation for every company after it.
let companyFailures = 0;
for (const company of companyRows) {
const result = await svc.autoProvisionBundledAgents(company.id);
autoEnsured += result.autoEnsured;
pendingApprovals += result.pendingApprovals;
defaultGrantsEnsured += result.defaultGrantsEnsured;
try {
const result = await svc.autoProvisionBundledAgents(company.id);
autoEnsured += result.autoEnsured;
pendingApprovals += result.pendingApprovals;
defaultGrantsEnsured += result.defaultGrantsEnsured;
} catch (err) {
companyFailures += 1;
console.error(
`built-in agent auto-provisioning failed for company ${company.id}; continuing with remaining companies`,
err,
);
}
}
const rows = await db
.select({
@ -1940,9 +2038,17 @@ export async function reconcileBuiltInAgentsOnStartup(db: Db) {
continue;
}
seen.add(instanceKey);
await svc.reconcileDefinitionDefaults(row.companyId, marker.key);
reconciled += 1;
try {
await svc.reconcileDefinitionDefaults(row.companyId, marker.key);
reconciled += 1;
} catch (err) {
companyFailures += 1;
console.error(
`built-in agent default reconciliation failed for company ${row.companyId} key ${marker.key}; continuing`,
err,
);
}
}
return { scanned, reconciled, unknown, duplicates, autoEnsured, pendingApprovals, defaultGrantsEnsured };
return { scanned, reconciled, unknown, duplicates, autoEnsured, pendingApprovals, defaultGrantsEnsured, companyFailures };
}