From dc12197cce0d7c0b849c560b4ab1d7dd920b4bfd Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Tue, 28 Jul 2026 11:12:58 -0700 Subject: [PATCH] fix: prevent duplicate built-in agents and self-heal reconciliation (#10223) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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. --- ...t-in-agent-unique-marker-migration.test.ts | 119 +++++++++++ .../0195_built_in_agent_unique_marker.sql | 53 +++++ packages/db/src/migrations/meta/_journal.json | 7 + server/src/__tests__/built-in-agents.test.ts | 169 ++++++++++++++-- server/src/index.ts | 8 +- server/src/services/approvals.ts | 19 ++ server/src/services/built-in-agents.ts | 186 ++++++++++++++---- 7 files changed, 507 insertions(+), 54 deletions(-) create mode 100644 packages/db/src/built-in-agent-unique-marker-migration.test.ts create mode 100644 packages/db/src/migrations/0195_built_in_agent_unique_marker.sql diff --git a/packages/db/src/built-in-agent-unique-marker-migration.test.ts b/packages/db/src/built-in-agent-unique-marker-migration.test.ts new file mode 100644 index 0000000000..9364149f7e --- /dev/null +++ b/packages/db/src/built-in-agent-unique-marker-migration.test.ts @@ -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> = []; +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); +}); diff --git a/packages/db/src/migrations/0195_built_in_agent_unique_marker.sql b/packages/db/src/migrations/0195_built_in_agent_unique_marker.sql new file mode 100644 index 0000000000..107638289c --- /dev/null +++ b/packages/db/src/migrations/0195_built_in_agent_unique_marker.sql @@ -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'; diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index f74509bb73..a62b0c9359 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -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 } ] } diff --git a/server/src/__tests__/built-in-agents.test.ts b/server/src/__tests__/built-in-agents.test.ts index b276d846e1..5ab8b6dde0 100644 --- a/server/src/__tests__/built-in-agents.test.ts +++ b/server/src/__tests__/built-in-agents.test.ts @@ -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); + 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 () => { diff --git a/server/src/index.ts b/server/src/index.ts index 3908d1cfc1..a21de6b67c 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -829,7 +829,13 @@ export async function startServer(): Promise { 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", diff --git a/server/src/services/approvals.ts b/server/src/services/approvals.ts index a8a2ebac0f..a21845fa4e 100644 --- a/server/src/services/approvals.ts +++ b/server/src/services/approvals.ts @@ -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, diff --git a/server/src/services/built-in-agents.ts b/server/src/services/built-in-agents.ts index 3a6ffa22e8..253686f7ba 100644 --- a/server/src/services/built-in-agents.ts +++ b/server/src/services/built-in-agents.ts @@ -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(); + 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, + ) { + 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 }; }