diff --git a/server/src/__tests__/dot-runner.test.ts b/server/src/__tests__/dot-runner.test.ts index ab1d3b3a693..0aedc8c99e7 100644 --- a/server/src/__tests__/dot-runner.test.ts +++ b/server/src/__tests__/dot-runner.test.ts @@ -2,7 +2,7 @@ import { applyPendingMigrations } from "@paperclipai/db"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { createGrokRoutineDispatcher } from "../services/grok-bot/dispatcher.js"; import { grokRoutineService } from "../services/grok-bot/routine.js"; -import { assertNoExternalOverlap } from "../modules/external-agents/admission.js"; +import { assertNoExternalOverlap, withExternalAdmissionGuard } from "../modules/external-agents/admission.js"; import { accessService } from "../services/access.js"; import { Readable } from "node:stream"; import { createPublicMcpTransfers } from "../services/public-mcp/file-transfers.js"; @@ -718,6 +718,48 @@ describe("durable Dot Runner integration", () => { await f.events.unsubscribe(f.principal, f.subscription); await f.events.stop(); }, 30000); + it.each(["accept", "finish"] as const)("settles %s under the admission guard without waiting on a run that admission holds", async action => { + const f = await fixture(); + const { run, assignment } = await offeredWork(f); + const requestId = randomUUID(); + await db.insert(dotRunnerOperations).values({ companyId: f.company.id, assignmentId: assignment.id, requestId, + digest: "test", command: { action }, status: "admitted" }); + let release!: () => void; + let held!: () => void; + const locked = new Promise(resolve => { held = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + // Execution admission locks the run before it takes the guard. A settle + // that holds the guard must not then need that run row, or the two + // transactions wait on each other. + const admission = db.transaction(async tx => { + await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)).for("update"); + held(); await gate; + await withExternalAdmissionGuard(tx, f.company.id, f.agent.id, async () => {}); + }); + await locked; + const port = f.broker.port({ binding: { companyId: f.company.id, agentId: f.agent.id, runId: run.id }, provider: { binding: f.snapshot } }); + // Capture each outcome now so a failure surfaces as itself, not as the wait timeout. + let settled = false; + let settleError: unknown; + const settle = port.settle({ sourceEventId: randomUUID(), payload: { requestId, binding: { ...f.snapshot, runId: run.id, + normalizedSessionId: assignment.normalizedSessionId, turnId: assignment.turnId, assignmentRevision: assignment.revision }, + outcome: { status: action === "accept" ? "accepted" : "completed" } } } as Parameters[0]) + .then(() => { settled = true; }, error => { settleError = error; }); + const admitted = admission.then(() => undefined, error => error as unknown); + try { + await vi.waitFor(() => { if (settleError) throw settleError; expect(settled).toBe(true); }, { timeout: 5000, interval: 20 }); + } finally { + release(); + await settle; + const admissionError = await admitted; + if (settleError) throw settleError; + if (admissionError) throw admissionError; + } + expect((await db.select().from(dotRunnerAssignments).where(eq(dotRunnerAssignments.id, assignment.id)))[0]?.status) + .toBe(action === "accept" ? "accepted" : "settled"); + await f.events.unsubscribe(f.principal, f.subscription); await f.events.stop(); + }, 30000); + it.each(["succeeded", "failed", "cancelled", "timed_out"] as const)("fences a stranded %s assignment before new admission without claiming external stop", async status => { const f = await fixture(); const { run, assignment } = await offeredWork(f); diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 1560c6e3695..fb8bbbbfb78 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1010,18 +1010,25 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { try { await executorHeartbeat.resumeQueuedRuns(); - await Promise.race([ - adapterStarted, - new Promise((_, reject) => { - setTimeout( - () => - reject( - new Error("Timed out waiting for adapter execution to start"), - ), - 3_000, - ); - }), - ]); + // Starting the run includes real runtime skill preparation for this + // fresh company, which alone takes several seconds. A fixed 3s budget + // raced that work, so wait for the adapter itself and fail fast only + // when the run settles without reaching it. The test timeout bounds a + // start that hangs. + let started = false; + void adapterStarted.then(() => { started = true; }); + const settledBeforeStart = (async () => { + while (!started) { + const run = await executorHeartbeat.getRun(runId); + if (!run || (run.status !== "queued" && run.status !== "running")) { + throw new Error( + `Run settled as ${run?.status ?? "missing"} (${run?.errorCode ?? "no error code"}) before adapter execution started`, + ); + } + await new Promise((resolve) => setTimeout(resolve, 50)); + } + })(); + await Promise.race([adapterStarted, settledBeforeStart]); await db .update(heartbeatRuns) @@ -1057,7 +1064,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { releaseAdapter(); await executorHeartbeat.drainActiveRunExecutions(); } - }); + // Start alone measures 3-6s locally, before the 5s settle wait. + }, 30_000); async function seedStrandedIssueFixture(input: { status: "todo" | "in_progress"; diff --git a/server/src/__tests__/tool-access-service.test.ts b/server/src/__tests__/tool-access-service.test.ts index a6b26a1a78a..1cbf9c20570 100644 --- a/server/src/__tests__/tool-access-service.test.ts +++ b/server/src/__tests__/tool-access-service.test.ts @@ -2154,6 +2154,15 @@ describeEmbeddedPostgres("tool access service", () => { let replacementFinished: Promise | undefined; try { await waitForFirstReplacementLock(cleanupPid); + releaseFirstReplacement(); + await firstReplacement; + await membershipIsLocked; + + // Start the empty replacement only once cleanup owns the membership + // row. Had it queued behind the first replacement too, both waiters + // would wake on that commit and race for the row; a replacement that + // wins it legitimately widens before cleanup, which made the + // fail-closed outcome below depend on scheduling. let replacementSettled = false; const replacement = replacementService .replaceConnectionGrantMembers(connection.id, grant.id, []) @@ -2165,12 +2174,10 @@ describeEmbeddedPostgres("tool access service", () => { replacementSettled = true; }); replacementFinished = replacement; - await waitForFirstReplacementLock(replacementPid); - expect(replacementSettled).toBe(false); - - releaseFirstReplacement(); - await firstReplacement; - await membershipIsLocked; + await vi.waitFor(async () => { + const [waiting] = await db.execute<{ blockers: number[] }>(sql`select pg_blocking_pids(${replacementPid}) as blockers`); + expect(waiting!.blockers).toContain(cleanupPid); + }, { timeout: 5000 }); expect(replacementSettled).toBe(false); releaseCleanup(); await cleanup; diff --git a/server/src/__tests__/voice-session-store.integration.test.ts b/server/src/__tests__/voice-session-store.integration.test.ts index 8c0bb1f1d6c..bc46af63f93 100644 --- a/server/src/__tests__/voice-session-store.integration.test.ts +++ b/server/src/__tests__/voice-session-store.integration.test.ts @@ -56,9 +56,14 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); }); + // Issue prefixes are unique across companies. About a hundred fixtures share + // this database, so seven random hex digits alone could collide; the + // sequence keeps each prefix in this run distinct within the ten characters + // an issue mention allows. + let companySequence = 0; async function fixture() { const companyId = randomUUID(), endpointId = randomUUID(), agentId = randomUUID(), applicationId = randomUUID(), connectionId = randomUUID(), issueId = randomUUID(), conversationId = randomUUID(), callerId = randomUUID(); - await db.insert(companies).values({ id: companyId, name: "Voice test", issuePrefix: `V${companyId.slice(0, 7).toUpperCase()}` }); + await db.insert(companies).values({ id: companyId, name: "Voice test", issuePrefix: `V${(++companySequence).toString(36).padStart(3, "0")}${companyId.slice(0, 6)}`.toUpperCase() }); await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: callerId, status: "active", membershipRole: "member" }); await db.insert(agents).values({ id: agentId, companyId, name: "Existing agent", role: "operator", status: "idle", adapterType: "paperclip_runner" }); await db.insert(issues).values({ id: issueId, companyId, title: "Existing task", assigneeAgentId: agentId, status: "in_progress" }); diff --git a/server/src/modules/external-agents/native-broker.ts b/server/src/modules/external-agents/native-broker.ts index 6523156f995..4ccbcda72b3 100644 --- a/server/src/modules/external-agents/native-broker.ts +++ b/server/src/modules/external-agents/native-broker.ts @@ -312,10 +312,15 @@ export function createNativeExternalBroker(db:Db,provider:Provider,identity:I if (provider === "openai_dot") await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `binding:${a.bindingId}`); await tx.execute(sql`update ${operationTable} set status=${provider==="muse"&&outcome.status==="rejected"?"rejected":store.completed},outcome=${JSON.stringify(outcome)}::jsonb,source_event_id=${event.sourceEventId},updated_at=now() where id=${op.id}::uuid`); const action=op.command.action; - await tx.execute(sql`update ${assignmentTable} set last_activity_at=now() where id=${a.id}::uuid`); - if(action==="renew"&&outcome.status==="renewed")await tx.execute(sql`update ${assignmentTable} set expires_at=${new Date(Number(outcome.expiresAtUnixMs)).toISOString()}::timestamptz where id=${a.id}::uuid`); - if(a.status!=="fenced"&&outcome.status!=="rejected"&&action==="accept")await tx.execute(sql`update ${assignmentTable} set status='accepted',${sql.raw(store.acceptedAt)}=now() where id=${a.id}::uuid`); - if(a.status!=="fenced"&&action==="finish"&&(provider==="muse"?outcome.status!=="rejected":outcome.status==="completed"))await tx.execute(sql`update ${assignmentTable} set status='settled',${sql.raw(provider==="muse"?"accepted_result_at":"settled_at")}=now() where id=${a.id}::uuid`); + // Write the assignment once. PostgreSQL re-runs foreign-key checks + // when a transaction updates a row version it already wrote, so a + // second UPDATE takes KEY SHARE on the run while this guard is held. + // Admission locks that run before the guard, and the two deadlock. + const assignmentChanges=[sql`last_activity_at=now()`]; + if(action==="renew"&&outcome.status==="renewed")assignmentChanges.push(sql`expires_at=${new Date(Number(outcome.expiresAtUnixMs)).toISOString()}::timestamptz`); + if(a.status!=="fenced"&&outcome.status!=="rejected"&&action==="accept")assignmentChanges.push(sql`status='accepted',${sql.raw(store.acceptedAt)}=now()`); + if(a.status!=="fenced"&&action==="finish"&&(provider==="muse"?outcome.status!=="rejected":outcome.status==="completed"))assignmentChanges.push(sql`status='settled',${sql.raw(provider==="muse"?"accepted_result_at":"settled_at")}=now()`); + await tx.execute(sql`update ${assignmentTable} set ${sql.join(assignmentChanges,sql`,`)} where id=${a.id}::uuid`); if(binding?.providerKind==="grok_bot"&&a.status!=="fenced"&&action==="finish"&&outcome.status==="completed")await grokRoutineService(tx as unknown as Db).settleWorkNotifications(a.companyId,a.bindingId,a.id); if(provider==="muse") { if(action==="consume_input"&&outcome.status==="consumed") {