diff --git a/packages/db/src/migrations/0228_violet_scorpion.sql b/packages/db/src/migrations/0228_violet_scorpion.sql index 59c19802dff..0a154cbc620 100644 --- a/packages/db/src/migrations/0228_violet_scorpion.sql +++ b/packages/db/src/migrations/0228_violet_scorpion.sql @@ -140,6 +140,8 @@ CREATE UNIQUE INDEX IF NOT EXISTS "issue_question_response_deliveries_interactio CREATE UNIQUE INDEX IF NOT EXISTS "issue_question_response_deliveries_correlation_uq" ON "issue_question_response_deliveries" USING btree ("correlation_id");--> statement-breakpoint CREATE INDEX IF NOT EXISTS "issue_question_response_deliveries_pending_idx" ON "issue_question_response_deliveries" USING btree ("status","created_at");--> statement-breakpoint CREATE INDEX IF NOT EXISTS "issue_question_response_deliveries_company_issue_idx" ON "issue_question_response_deliveries" USING btree ("company_id","issue_id","created_at");--> statement-breakpoint +-- paperclip:migration-safety-ignore large-create-index-not-concurrently: This partial index covers a new question-response key prefix, so existing deployments have no matching rows; Drizzle applies migrations transactionally and cannot use CONCURRENTLY. +CREATE UNIQUE INDEX IF NOT EXISTS "agent_wakeup_requests_question_response_delivery_idempotency_uq" ON "agent_wakeup_requests" USING btree ("company_id","idempotency_key") WHERE "agent_wakeup_requests"."idempotency_key" LIKE 'question-response:%' AND "agent_wakeup_requests"."status" NOT IN ('skipped', 'failed', 'cancelled');--> statement-breakpoint CREATE INDEX IF NOT EXISTS "managed_agent_profiles_company_idx" ON "managed_agent_profiles" USING btree ("company_id");--> statement-breakpoint CREATE UNIQUE INDEX IF NOT EXISTS "managed_agent_profiles_company_key_uq" ON "managed_agent_profiles" USING btree ("company_id","profile_key");--> statement-breakpoint CREATE UNIQUE INDEX IF NOT EXISTS "managed_agent_profiles_company_resource_uq" ON "managed_agent_profiles" USING btree ("company_id","anthropic_agent_id","agent_version","environment_id");--> statement-breakpoint diff --git a/packages/db/src/migrations/meta/0228_snapshot.json b/packages/db/src/migrations/meta/0228_snapshot.json index aea6b30b74b..fe82ad78ffd 100644 --- a/packages/db/src/migrations/meta/0228_snapshot.json +++ b/packages/db/src/migrations/meta/0228_snapshot.json @@ -1605,6 +1605,28 @@ "method": "btree", "with": {} }, + "agent_wakeup_requests_question_response_delivery_idempotency_uq": { + "name": "agent_wakeup_requests_question_response_delivery_idempotency_uq", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "idempotency_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"agent_wakeup_requests\".\"idempotency_key\" LIKE 'question-response:%' AND \"agent_wakeup_requests\".\"status\" NOT IN ('skipped', 'failed', 'cancelled')", + "concurrently": false, + "method": "btree", + "with": {} + }, "agent_wakeup_requests_company_payload_issue_idx": { "name": "agent_wakeup_requests_company_payload_issue_idx", "columns": [ diff --git a/packages/db/src/schema/agent_wakeup_requests.ts b/packages/db/src/schema/agent_wakeup_requests.ts index a0888d9700b..06cde63ec45 100644 --- a/packages/db/src/schema/agent_wakeup_requests.ts +++ b/packages/db/src/schema/agent_wakeup_requests.ts @@ -43,6 +43,11 @@ export const agentWakeupRequests = pgTable( dispositionRepairIdempotencyUq: uniqueIndex("agent_wakeup_requests_disposition_repair_idempotency_uq") .on(table.companyId, table.idempotencyKey) .where(sql`${table.idempotencyKey} LIKE 'issue_disposition_repair:%' AND ${table.status} <> 'skipped'`), + questionResponseDeliveryIdempotencyUq: uniqueIndex( + "agent_wakeup_requests_question_response_delivery_idempotency_uq", + ) + .on(table.companyId, table.idempotencyKey) + .where(sql`${table.idempotencyKey} LIKE 'question-response:%' AND ${table.status} NOT IN ('skipped', 'failed', 'cancelled')`), companyPayloadIssueIdx: index("agent_wakeup_requests_company_payload_issue_idx").on( table.companyId, sql`(${table.payload} ->> 'issueId')`, diff --git a/server/src/__tests__/question-response-delivery.test.ts b/server/src/__tests__/question-response-delivery.test.ts index ff893a932d4..208024dd6f9 100644 --- a/server/src/__tests__/question-response-delivery.test.ts +++ b/server/src/__tests__/question-response-delivery.test.ts @@ -244,7 +244,7 @@ describeEmbeddedPostgres("question response delivery", () => { expect(steer).not.toHaveBeenCalled(); expect(wakeup).toHaveBeenCalledTimes(1); expect(wakeup.mock.calls[0]?.[1]).toMatchObject({ - idempotencyKey: `interaction:${seeded.interaction.id}:answered`, + idempotencyKey: `question-response:${seeded.interaction.id}`, contextSnapshot: { interactionId: seeded.interaction.id, interactionStatus: "answered", @@ -365,6 +365,95 @@ describeEmbeddedPostgres("question response delivery", () => { }); }); + it("enforces one durable wake per question-response idempotency key", async () => { + const seeded = await seed({ sourceStatus: "running" }); + const idempotencyKey = `question-response:${seeded.interaction.id}`; + const request = { + companyId: seeded.companyId, + agentId: seeded.agentId, + source: "automation", + triggerDetail: "system", + reason: "issue_commented", + idempotencyKey, + } as const; + + await db.insert(agentWakeupRequests).values({ ...request, status: "queued" }); + await expect(db.insert(agentWakeupRequests).values({ + ...request, + status: "coalesced", + })).rejects.toMatchObject({ cause: { code: "23505" } }); + + // Suppression receipts are intentionally outside the fence so the outbox + // can retry after scheduling is enabled again. + await expect(db.insert(agentWakeupRequests).values({ + ...request, + status: "skipped", + finishedAt: new Date(), + })).resolves.toBeDefined(); + }); + + it("reuses the winning wake when a concurrent insert hits the idempotency fence", async () => { + const seeded = await seed({ sourceStatus: "running" }); + const fallbackRunId = randomUUID(); + const wakeup = vi.fn().mockImplementation(async ( + _agentId: string, + options: { idempotencyKey?: string | null }, + ) => { + const request = { + companyId: seeded.companyId, + agentId: seeded.agentId, + source: "automation", + triggerDetail: "system", + reason: "issue_commented", + idempotencyKey: options.idempotencyKey, + } as const; + const [winner] = await db.insert(agentWakeupRequests).values({ + ...request, + status: "queued", + }).returning(); + await db.insert(heartbeatRuns).values({ + id: fallbackRunId, + companyId: seeded.companyId, + agentId: seeded.agentId, + invocationSource: "automation", + status: "queued", + runtimeMode: "legacy", + driverKind: "codex", + wakeupRequestId: winner!.id, + contextSnapshot: { issueId: seeded.issueId }, + }); + await db.update(agentWakeupRequests).set({ runId: fallbackRunId }) + .where(eq(agentWakeupRequests.id, winner!.id)); + + // Model the losing claimant reaching the same transactional insert after + // the winner commits. The service must recover the winner's receipt. + await db.insert(agentWakeupRequests).values({ + ...request, + status: "coalesced", + }); + throw new Error("unreachable"); + }); + + const outcome = await questionResponseDeliveryService(db, { + heartbeat: { wakeup } as never, + steer: vi.fn(), + }).deliver(seeded.interaction.id); + + expect(outcome).toMatchObject({ + status: "fallback_queued", + mode: "wake_fallback", + targetRunId: fallbackRunId, + }); + expect(wakeup).toHaveBeenCalledTimes(1); + const [delivery] = await db.select().from(issueQuestionResponseDeliveries); + expect(delivery).toMatchObject({ + status: "fallback_queued", + attemptCount: 1, + errorCount: 0, + targetRunId: fallbackRunId, + }); + }); + it("reuses a durable wake receipt instead of issuing a duplicate continuation", async () => { const seeded = await seed({ sourceStatus: "running" }); const [wakeRequest] = await db.insert(agentWakeupRequests).values({ @@ -374,7 +463,7 @@ describeEmbeddedPostgres("question response delivery", () => { triggerDetail: "system", reason: "issue_commented", status: "queued", - idempotencyKey: `interaction:${seeded.interaction.id}:answered`, + idempotencyKey: `question-response:${seeded.interaction.id}`, }).returning(); const [wakeRun] = await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, @@ -518,7 +607,7 @@ describeEmbeddedPostgres("question response delivery", () => { expect(steer).not.toHaveBeenCalled(); expect(wakeup).toHaveBeenCalledTimes(1); expect(wakeup.mock.calls[0]?.[1]).toMatchObject({ - idempotencyKey: `interaction:${seeded.interaction.id}:answered`, + idempotencyKey: `question-response:${seeded.interaction.id}`, contextSnapshot: { issueId: seeded.issueId, interactionId: seeded.interaction.id, diff --git a/server/src/services/question-response-delivery.ts b/server/src/services/question-response-delivery.ts index 3608415ec82..1e75b87c798 100644 --- a/server/src/services/question-response-delivery.ts +++ b/server/src/services/question-response-delivery.ts @@ -15,6 +15,7 @@ import type { import type { PaperclipQuestionResponse, } from "../vendor/paperclip-runner/index.js"; +import { isUniqueViolation } from "../db-errors.js"; import { getTelemetryClient } from "../telemetry.js"; import { logger } from "../middleware/logger.js"; import { logActivity } from "./activity-log.js"; @@ -29,6 +30,8 @@ const DELIVERY_CLAIM_STALE_MS = 30_000; const DELIVERY_CLAIM_REFRESH_MS = 10_000; const MAX_DELIVERY_ATTEMPTS = 5; const DELIVERY_CORRELATION_PREFIX = "question-response:"; +const QUESTION_RESPONSE_WAKE_IDEMPOTENCY_CONSTRAINT = + "agent_wakeup_requests_question_response_delivery_idempotency_uq"; class DeliveryClaimUnavailableError extends Error { constructor() { @@ -38,6 +41,7 @@ class DeliveryClaimUnavailableError extends Error { } const DURABLE_WAKE_REQUEST_STATUSES = [ "queued", + "claimed", "running", "succeeded", "coalesced", @@ -590,7 +594,10 @@ export function questionResponseDeliveryService( } const actor = actorForInteraction(interaction); - const wakeIdempotencyKey = `interaction:${interaction.id}:${interaction.status}`; + // This is a new, migration-fenced namespace. The partial unique index on + // agent_wakeup_requests makes the wake transaction itself idempotent, so a + // reclaimed stale worker cannot create a second continuation run. + const wakeIdempotencyKey = `question-response:${interaction.id}`; try { const existingWake = await findDurableWakeRequest({ companyId: interaction.companyId, @@ -654,6 +661,29 @@ export function questionResponseDeliveryService( }); } catch (error) { if (error instanceof DeliveryClaimUnavailableError) return terminalOutcome(interactionId); + if (isUniqueViolation(error, QUESTION_RESPONSE_WAKE_IDEMPOTENCY_CONSTRAINT)) { + // A concurrent claimant won the transactional wake fence after our + // preflight lookup. Reuse its committed receipt instead of consuming + // an error retry or issuing another continuation. + const durableWake = await findDurableWakeRequest({ + companyId: interaction.companyId, + agentId: assigneeAgentId, + idempotencyKey: wakeIdempotencyKey, + }); + if (durableWake) { + const targetRun = durableWake.run ?? queuedSuccessor ?? null; + const coalesced = Boolean(queuedSuccessor && targetRun?.id === queuedSuccessor.id); + return recordTerminal({ + delivery: claimed, + interaction, + status: coalesced ? "delivered" : "fallback_queued", + mode: coalesced ? "coalesced" : "wake_fallback", + targetRunId: targetRun?.id ?? null, + adapter: targetRun?.driverKind ?? adapter, + errorCode: steeringErrorCode, + }); + } + } const errorCode = error instanceof Error && compactLine(error.message) ? compactLine(error.message)!.slice(0, 160) : "question_response_wake_failed";