diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 98efb8a882f..eb80643cf22 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -944,6 +944,36 @@ same preparation retains its directory and receipt check. A published workspace on the same allocation retains its recorded directory. Automatic bootstrap timeout retries still require a fresh allocation under the policy above. +### Saved chat Retry preparation + +Task Retry can meet a busy chat identity or endpoint lock before its source is authorized. +The canonical Retry and its linked chat action then retain one unadmitted request. +The existing chat retry worker prepares that request after the lock is available. +It checks the current board actor, source, principal, conversation, task owner, +and canonical intent before normal admission. Stop, changed input, revoked access, +or newer execution can reject it without adopting input or publishing a notice. +Restoration of a certified unstarted input batch stays in the same transaction as +source preparation. Contention cannot commit a partial restoration. + +Pending preparation and a prepared but unsettled action show the existing +retry-scheduled state and the chat outbox's actual retry or stale-claim deadline. +They do not imply that a provider started. Source authorization contention has a separate +60-second deadline saved with the pending request, or at the first contention +for an older action. Waiting on these locks does not consume the eight delivery +attempts. Expiry is checked before source preparation commits and before a +prepared action can leave its authorization wait. A lock that becomes available +after expiry cannot complete a still-pending wait. The authorization transaction +saves completion of that wait under the current claim before wake delivery. +A restart rechecks current source and execution gates; elapsed time alone does +not expire a completed wait. New contention still uses the original deadline. +Generic delivery failure cannot put a completed wait back into the pending +state. A lost commit acknowledgment preserves the stored phase and does not +authorize a wake. +Duplicate clicks and background sweeps cannot renew either limit. Preparation +does not create a generic board wake. Only the endpoint query's lock-unavailable +error can defer through its explicit Retry callback. Other database and +authorization errors keep their existing behavior. + ### Runner authentication and ACP startup handshake bounds The native Runner waits up to the ten-second connection setup budget for the diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 4b3786861b9..e4e6fc9e798 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -20,7 +20,7 @@ import { nativeChatProgressWork } from "../services/chat-native-progress-work.js import { cloudActorHeaderSourceFromHeaders, resolveCloudTenantActor } from "../middleware/auth.js"; import { cloudTenantPrimaryCompanyId } from "../services/cloud-instance.js"; import { accessService } from "../services/access.js"; -import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; +import { databaseWorkPending, reconcileDatabaseWork, signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; import { claimTaskRetryInput, releaseUnadmittedTaskRetryInput } from "../services/chat-channels/task-retry-input.js"; import { restoreNativeChatReviewPresentationInTransaction } from "../services/native-runtime/native-chat-review-presentation.js"; import { notifyChatEndpointConfiguration, notifyChatCompanyConfiguration } from "../services/chat-channels/configuration-work-notifications.js"; @@ -40,6 +40,7 @@ import { createChatRuntime } from "../services/chat-runtime.js"; import { createChatSdkRuntime } from "../services/chat-sdk-runtime.js"; import { readFreshWorkspaceReadiness } from "../services/fresh-workspace-continuation.js"; import { requestTaskRetry, reconcileTaskChatRetryReceipts } from "../services/task-retry.js"; +import * as taskRetryModule from "../services/task-retry.js"; import { taskInterruptionPresentation } from "../services/task-interruption-presentation.js"; import { withQueuedCommentIdsInWakePayload } from "../services/issue-queued-comment-queue.js"; import { createSystemRecovery } from "../services/system-recovery-state.js"; @@ -64479,6 +64480,580 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { return { ...f, pending, originalDelivery }; } + async function busyChatRetryHttpFixture(withPending = false) { + const f = await currentGitHubTaskRetryFixture(withPending); + const originalPublications = await readChatPublicationsByEndpointId(f.endpoint.id); + const { issueRoutes } = await import("../routes/issues.js"); + const companyId = f.fixture.companyId, issueId = f.issue.id; + const app = express(); + app.use(express.json()); + app.use((req, _res, next) => { req.actor = boardActor(companyId); next(); }); + app.use("/api", issueRoutes(db, createStorageService().storage, { chatRunRetries: f.service })); + app.use(errorHandler); + let release = () => {}; + const held = new Promise(resolve => { release = resolve; }); + let acquired = () => {}; + const locked = new Promise(resolve => { acquired = resolve; }); + const holding = db.transaction(async tx => { + await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`chat-identity:${companyId}:${f.action.principalId}`}, 0))`); + acquired(); + await held; + }); + await locked; + f.wakeup.mockClear(); + return { ...f, app, companyId, issueId, originalPublications, + releaseIdentity: async () => { release(); await holding; } }; + } + + it("retains one HTTP task Retry intent while saved chat identity authorization is busy", async () => { + const f = await busyChatRetryHttpFixture(); + let restarted: ReturnType | undefined; + try { + const { companyId, issueId, app } = f; + const original = (await readHeartbeatRunsById(f.runId))[0]; + const originalInput = (await readAgentWakeupRequestsById(f.action.id))[0]; + const body = { observedRunId: f.runId, idempotencyKey: randomUUID() }; + const responses = await Promise.all([body, body, { ...body, idempotencyKey: randomUUID() }] + .map(value => request(app).post(`/api/issues/${issueId}/retry`).send(value))); + for (const response of responses) expect(response.status, JSON.stringify(response.body)).toBe(202); + expect(responses[0].body.presentation).toMatchObject({ state: "retry_scheduled", canRetry: false }); + expect(new Set(responses.map(response => response.body.requestId)).size).toBe(1); + expect(new Set(responses.map(response => response.body.chatActionId)).size).toBe(1); + const response = responses[0]; + const [intent] = await readAgentWakeupRequestsById(response.body.requestId); + expect(intent).toMatchObject({ reason: "task_retry", status: "deferred_issue_execution", runId: null }); + expect((await request(app).post(`/api/issues/${issueId}/retry`).send(body)).body.requestId).toBe(intent.id); + const [action] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + expect(action.payload).toMatchObject({ sourceAuthorizationPending: true, taskRetryIntentId: intent.id }); + expect(action.payload.sourceScopeSha256).toBeUndefined(); + await f.service.processFailedChatRunRetry(action.id); + const [waiting] = await db.select().from(chatActions).where(eq(chatActions.id, action.id)); + expect(await taskInterruptionPresentation(db, (await readIssuesById(issueId))[0])) + .toMatchObject({ state: "retry_scheduled", nextRetryAt: waiting.result!.retryAt, canRetry: false }); + expect((await readAgentWakeupRequestsById(f.action.id))[0]).toEqual(originalInput); + await heartbeatService(db).resumeTaskRetry(companyId, intent.id); + expect(f.wakeup).not.toHaveBeenCalled(); + expect(await readChatPublicationsByEndpointId(f.endpoint.id)).toEqual(f.originalPublications); + expect((await readHeartbeatRunsByCompanyId(companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(0); + expect((await db.select().from(chatActions).where(and(eq(chatActions.companyId, companyId), + eq(chatActions.kind, "failed_run_retry"))))).toHaveLength(1); + await f.service.shutdown(); + restarted = createService(new FakeChatSdkRuntime(), f.providerFetch, { wakeup: heartbeatService(db).wakeup }); + await f.releaseIdentity(); + await vi.waitFor(async () => { + await restarted!.service.processFailedChatRunRetries(); + await reconcileTaskChatRetryReceipts(db, { companyId, requestId: intent.id }); + expect((await readAgentWakeupRequestsById(intent.id))[0]?.runId).toBeTruthy(); + }, { timeout: 5_000, interval: 50 }); + const [settled] = await readAgentWakeupRequestsById(intent.id); + const [successor] = await readHeartbeatRunsById(settled.runId!); + expect(successor).toMatchObject({ status: "queued", retryOfRunId: f.runId, + contextSnapshot: { source: "chat:github", taskRetryRequestId: intent.id, + wakeCommentIds: [f.action.payload.commentId] } }); + expect(restarted.wakeup).toHaveBeenCalledOnce(); + expect((await readHeartbeatRunsByCompanyId(companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(1); + expect((await readHeartbeatRunsById(f.runId))[0]).toEqual(original); + const [authorized] = await db.select().from(chatActions).where(eq(chatActions.id, action.id)); + expect(authorized.payload.sourceAuthorizationPending).toBeUndefined(); + expect(authorized.payload.sourceScopeSha256).toMatch(/^[a-f0-9]{64}$/); + expect((await request(app).post(`/api/issues/${issueId}/retry`).send(body)).body.requestId).toBe(intent.id); + } finally { + await f.releaseIdentity(); + await restarted?.service.shutdown(); + await f.service.shutdown(); + } + }); + + it.each(["unchanged", "source_edited", "principal_revoked", "unknown_source", "actor_revoked_before"] as const)("retains HTTP Retry under the endpoint authorization row lock: %s", async mode => { + const f = await busyChatRetryHttpFixture(true); + let release = () => {}; + let holding: Promise | undefined; + try { + await f.releaseIdentity(); + const sourceIds = [f.action.id, f.pending!.id]; + const originalInputs = await db.select().from(agentWakeupRequests).where(inArray(agentWakeupRequests.id, sourceIds)) + .orderBy(asc(agentWakeupRequests.id)); + if (mode === "unknown_source") { + const [source] = await readHeartbeatRunsById(f.runId); + await db.update(heartbeatRuns).set({ contextSnapshot: { ...source.contextSnapshot, source: "chat:unknown" } }) + .where(eq(heartbeatRuns.id, f.runId)); + } + if (mode === "actor_revoked_before") await db.update(companyMemberships).set({ status: "suspended", updatedAt: new Date() }) + .where(and(eq(companyMemberships.companyId, f.companyId), eq(companyMemberships.principalId, "owner-user"))); + let acquired = () => {}; + const locked = new Promise(resolve => { acquired = resolve; }); + const held = new Promise(resolve => { release = resolve; }); + holding = db.transaction(async tx => { + await tx.select({ id: chatEndpoints.id }).from(chatEndpoints).where(and(eq(chatEndpoints.companyId, f.companyId), + eq(chatEndpoints.id, f.endpoint.id))).for("no key update"); + acquired(); await held; + }); + await locked; + const response = await request(f.app).post(`/api/issues/${f.issueId}/retry`) + .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }); + if (mode === "unknown_source" || mode === "actor_revoked_before") { + expect(response.status, JSON.stringify(response.body)).toBe(mode === "unknown_source" ? 409 : 404); + expect(await db.select().from(chatActions).where(and(eq(chatActions.companyId, f.companyId), + eq(chatActions.kind, "failed_run_retry")))).toEqual([]); + expect(await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), + eq(agentWakeupRequests.reason, "task_retry")))).toEqual([]); + expect(f.wakeup).not.toHaveBeenCalled(); + expect(await readChatPublicationsByEndpointId(f.endpoint.id)).toEqual(f.originalPublications); + expect(await db.select().from(agentWakeupRequests).where(inArray(agentWakeupRequests.id, sourceIds)) + .orderBy(asc(agentWakeupRequests.id))).toEqual(originalInputs); + expect((await readHeartbeatRunsByCompanyId(f.companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(0); + return; + } + expect(response.status, JSON.stringify(response.body)).toBe(202); + await f.service.processFailedChatRunRetry(response.body.chatActionId); + expect((await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)))[0]) + .toMatchObject({ status: "issued", payload: { sourceAuthorizationPending: true }, result: { attemptCount: 0 } }); + expect(f.wakeup).not.toHaveBeenCalled(); + expect(await db.select().from(agentWakeupRequests).where(inArray(agentWakeupRequests.id, sourceIds)) + .orderBy(asc(agentWakeupRequests.id))).toEqual(originalInputs); + expect(await readChatPublicationsByEndpointId(f.endpoint.id)).toEqual(f.originalPublications); + if (mode === "source_edited") await db.update(issueComments).set({ body: "Changed saved request", + updatedAt: new Date(Date.now() + 1) }).where(eq(issueComments.id, String(f.pending!.payload.commentId))); + if (mode === "principal_revoked") await db.update(companyMemberships).set({ status: "suspended", updatedAt: new Date() }) + .where(and(eq(companyMemberships.companyId, f.companyId), eq(companyMemberships.principalId, "owner-user"))); + release(); await holding; + await vi.waitFor(async () => { + await f.service.processFailedChatRunRetries(); + const [action] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + expect(action.status).toBe(mode === "unchanged" ? "processed" : "failed"); + }, { timeout: 5_000, interval: 50 }); + if (mode === "unchanged") { + await reconcileTaskChatRetryReceipts(db, { companyId: f.companyId, requestId: response.body.requestId }); + const [intent] = await readAgentWakeupRequestsById(response.body.requestId); + expect((await readHeartbeatRunsById(intent.runId!))[0]).toMatchObject({ status: "queued", retryOfRunId: f.runId, + contextSnapshot: { wakeCommentIds: [f.pending!.payload.commentId] } }); + expect(f.wakeup).toHaveBeenCalledOnce(); + expect((await readHeartbeatRunsByCompanyId(f.companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(1); + } else { + expect(f.wakeup).not.toHaveBeenCalled(); + expect((await readAgentWakeupRequestsById(response.body.requestId))[0]).toMatchObject({ status: "failed", runId: null }); + expect(await db.select().from(agentWakeupRequests).where(inArray(agentWakeupRequests.id, sourceIds)) + .orderBy(asc(agentWakeupRequests.id))).toEqual(originalInputs); + expect(await readChatPublicationsByEndpointId(f.endpoint.id)).toEqual(f.originalPublications); + expect((await readHeartbeatRunsByCompanyId(f.companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(0); + } + } finally { release(); await holding; await f.releaseIdentity(); await f.service.shutdown(); } + }); + + it("keeps a saved HTTP Retry alive through a healthy nine-second identity transaction", async () => { + const f = await busyChatRetryHttpFixture(); + try { + const original = (await readHeartbeatRunsById(f.runId))[0]; + const originalInput = (await readAgentWakeupRequestsById(f.action.id))[0]; + const response = await request(f.app).post(`/api/issues/${f.issueId}/retry`) + .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }); + expect(response.status, JSON.stringify(response.body)).toBe(202); + const started = performance.now(); + await vi.waitFor(async () => { + await f.service.processFailedChatRunRetries(); + expect(performance.now() - started).toBeGreaterThanOrEqual(9_000); + }, { timeout: 12_000, interval: 50 }); + const [waiting] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + expect(waiting.status).toBe("issued"); + expect(waiting.result?.attemptCount).toBe(0); + expect(waiting.payload.sourceAuthorizationPending).toBe(true); + expect(f.wakeup).not.toHaveBeenCalled(); + expect(await readChatPublicationsByEndpointId(f.endpoint.id)).toEqual(f.originalPublications); + expect((await readAgentWakeupRequestsById(f.action.id))[0]).toEqual(originalInput); + expect((await readHeartbeatRunsByCompanyId(f.companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(0); + await f.releaseIdentity(); + await vi.waitFor(async () => { + await f.service.processFailedChatRunRetries(); + await reconcileTaskChatRetryReceipts(db, { companyId: f.companyId, requestId: response.body.requestId }); + expect((await readAgentWakeupRequestsById(response.body.requestId))[0]?.runId).toBeTruthy(); + }, { timeout: 5_000, interval: 50 }); + expect(f.wakeup).toHaveBeenCalledOnce(); + expect((await readHeartbeatRunsByCompanyId(f.companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(1); + expect((await readHeartbeatRunsById(f.runId))[0]).toEqual(original); + } finally { await f.releaseIdentity(); await f.service.shutdown(); } + }, 30_000); + + it("projects prepared but unsettled task Retry using its actual stale-claim deadline", async () => { + const f = await busyChatRetryHttpFixture(); + let releaseWake = () => {}; + const heldWake = new Promise(resolve => { releaseWake = resolve; }); + let processing: ReturnType | undefined; + try { + const response = await request(f.app).post(`/api/issues/${f.issueId}/retry`) + .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }); + expect(response.status, JSON.stringify(response.body)).toBe(202); + await f.service.processFailedChatRunRetry(response.body.chatActionId); + const [waiting] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + await f.releaseIdentity(); + const wake = heartbeatService(db).wakeup; + f.wakeup.mockImplementation(async (...args) => { await heldWake; return wake(...args); }); + await new Promise(resolve => setTimeout(resolve, Math.max(0, Date.parse(String(waiting.result?.retryAt)) - Date.now()) + 1)); + processing = f.service.processFailedChatRunRetry(response.body.chatActionId); + await vi.waitFor(() => expect(f.wakeup).toHaveBeenCalledOnce()); + const [prepared] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + expect(prepared.status).toBe("processing"); + expect(prepared.payload.sourceAuthorizationPending).toBeUndefined(); + expect((await readAgentWakeupRequestsById(response.body.requestId))[0]?.runId).toBeNull(); + const deadline = await f.service.nextChatActionAt("failed_run_retry"); + expect(deadline).not.toBeNull(); + expect(await taskInterruptionPresentation(db, (await readIssuesById(f.issueId))[0])) + .toMatchObject({ state: "retry_scheduled", canRetry: false, nextRetryAt: new Date(deadline!).toISOString() }); + releaseWake(); + await processing; + await reconcileTaskChatRetryReceipts(db, { companyId: f.companyId, requestId: response.body.requestId }); + expect((await readAgentWakeupRequestsById(response.body.requestId))[0]?.runId).toBeTruthy(); + expect(await taskInterruptionPresentation(db, (await readIssuesById(f.issueId))[0])) + .toMatchObject({ state: "working", nextRetryAt: null }); + } finally { releaseWake(); await processing; await f.releaseIdentity(); await f.service.shutdown(); } + }); + + it.each(["source_edited", "principal_revoked", "stopped", "mixed_publications", "newer_queued_run"] as const)( + "does not adopt input or deliver a pending identity-contended Retry after %s", async mode => { + const f = await busyChatRetryHttpFixture(); + try { + let expectedPublications = f.originalPublications; + const originalInput = (await readAgentWakeupRequestsById(f.action.id))[0]; + const response = await request(f.app).post(`/api/issues/${f.issueId}/retry`) + .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }); + expect(response.status, JSON.stringify(response.body)).toBe(202); + await f.service.processFailedChatRunRetry(response.body.chatActionId); + if (mode === "source_edited") await db.update(issueComments).set({ body: "Changed saved request", + updatedAt: new Date(Date.now() + 1) }).where(eq(issueComments.id, String(f.action.payload.commentId))); + if (mode === "principal_revoked") await db.update(companyMemberships).set({ status: "suspended", updatedAt: new Date() }) + .where(and(eq(companyMemberships.companyId, f.companyId), eq(companyMemberships.principalId, "owner-user"))); + if (mode === "stopped") await heartbeatService(db).cancelInvocationsForAgents([f.fixture.assignedAgentId], "operator_stop"); + if (mode === "newer_queued_run") { + const [source] = await readHeartbeatRunsById(f.runId); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.fixture.assignedAgentId, + issueId: f.issueId, nativeIssueId: f.issueId, runtimeMode: "native", status: "queued", + createdAt: new Date(Math.max(Date.now(), source.createdAt.getTime() + 1)), + contextSnapshot: { issueId: f.issueId } }); + } + if (mode === "mixed_publications") { + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorAgentId: f.fixture.assignedAgentId, createdByRunId: f.runId, body: "Saved failed result" }).returning(); + await db.insert(chatPublications).values((["failed", "delivery_unknown"] as const).map(state => ({ + companyId: f.companyId, endpointId: f.endpoint.id, conversationId: f.conversation.id, + issueId: f.issueId, commentId: comment.id, idempotencyKey: `run:${f.runId}:${state}`, + payload: { text: comment.body }, state, + }))); + expectedPublications = await readChatPublicationsByEndpointId(f.endpoint.id); + } + const expectedRunIds = (await readHeartbeatRunsByCompanyId(f.companyId)).map(run => run.id).sort(); + await f.releaseIdentity(); + await vi.waitFor(async () => { + await f.service.processFailedChatRunRetries(); + const [action] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + expect(action.status).toBe("failed"); + }, { timeout: 5_000, interval: 50 }); + expect(f.wakeup).not.toHaveBeenCalled(); + expect(await readChatPublicationsByEndpointId(f.endpoint.id)).toEqual(expectedPublications); + expect(await taskInterruptionPresentation(db, (await readIssuesById(f.issueId))[0])) + .toMatchObject({ state: mode === "newer_queued_run" ? "working" : "failed", nextRetryAt: null }); + expect((await readAgentWakeupRequestsById(f.action.id))[0]).toEqual(originalInput); + expect((await readAgentWakeupRequestsById(response.body.requestId))[0]?.runId).toBeNull(); + if (mode === "stopped") expect((await readAgentWakeupRequestsById(response.body.requestId))[0]?.status).toBe("cancelled"); + expect((await readHeartbeatRunsByCompanyId(f.companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(0); + expect((await readHeartbeatRunsByCompanyId(f.companyId)).map(run => run.id).sort()).toEqual(expectedRunIds); + } finally { await f.releaseIdentity(); await f.service.shutdown(); } + }); + + it.each([false, true])("expires the saved identity wait without renewal when the lock is released before expiry processing: %s", async released => { + const f = await busyChatRetryHttpFixture(); + let restarted: ReturnType | undefined; + try { + const originalInput = (await readAgentWakeupRequestsById(f.action.id))[0]; + const body = { observedRunId: f.runId, idempotencyKey: randomUUID() }; + const response = await request(f.app).post(`/api/issues/${f.issueId}/retry`).send(body); + expect(response.status, JSON.stringify(response.body)).toBe(202); + const [saved] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + const deadline = String(saved.result?.identityContentionDeadlineAt); + expect(Date.parse(deadline)).toBe(saved.createdAt.getTime() + 60_000); + await f.service.processFailedChatRunRetry(saved.id); + const [waiting] = await db.select().from(chatActions).where(eq(chatActions.id, saved.id)); + expect(waiting.result).toMatchObject({ attemptCount: 0, identityContentionCount: 1, + identityContentionDeadlineAt: deadline }); + expect(Date.parse(String(waiting.result?.retryAt))).toBeLessThanOrEqual(Date.parse(deadline)); + for (const value of [body, { ...body, idempotencyKey: randomUUID() }]) { + const duplicate = await request(f.app).post(`/api/issues/${f.issueId}/retry`).send(value); + expect(duplicate.status, JSON.stringify(duplicate.body)).toBe(202); + expect(duplicate.body.requestId).toBe(response.body.requestId); + } + expect((await db.select().from(chatActions).where(eq(chatActions.id, saved.id)))[0].result) + .toEqual(waiting.result); + await f.service.shutdown(); + restarted = createService(new FakeChatSdkRuntime(), f.providerFetch, { wakeup: heartbeatService(db).wakeup }); + const expiredDeadline = new Date(Date.now() - 1).toISOString(); + await db.update(chatActions).set({ result: { ...waiting.result, + identityContentionDeadlineAt: expiredDeadline, retryAt: expiredDeadline } }).where(eq(chatActions.id, saved.id)); + if (released) await f.releaseIdentity(); + await restarted.service.processFailedChatRunRetries(); + const [failed] = await db.select().from(chatActions).where(eq(chatActions.id, saved.id)); + expect(failed).toMatchObject({ status: "failed", payload: { sourceAuthorizationPending: true }, + result: { code: "failed_chat_retry_contention_expired", attemptCount: 0, + identityContentionDeadlineAt: expiredDeadline } }); + expect(failed.result?.retryAt).toBeUndefined(); + await f.releaseIdentity(); + await restarted.service.processFailedChatRunRetries(); + await restarted.service.processFailedChatRunRetry(saved.id); + await heartbeatService(db).resumeTaskRetry(f.companyId, response.body.requestId); + expect((await db.select().from(chatActions).where(eq(chatActions.id, saved.id)))[0]).toEqual(failed); + expect((await readAgentWakeupRequestsById(response.body.requestId))[0]).toMatchObject({ status: "failed", runId: null }); + expect((await readAgentWakeupRequestsById(f.action.id))[0]).toEqual(originalInput); + expect(restarted.wakeup).not.toHaveBeenCalled(); + expect(f.wakeup).not.toHaveBeenCalled(); + expect(await readChatPublicationsByEndpointId(f.endpoint.id)).toEqual(f.originalPublications); + expect((await readHeartbeatRunsByCompanyId(f.companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(0); + } finally { await f.releaseIdentity(); await restarted?.service.shutdown(); await f.service.shutdown(); } + }); + + it("persists the first identity wait deadline for an older pending action", async () => { + const f = await busyChatRetryHttpFixture(); + let restarted: ReturnType | undefined; + try { + const response = await request(f.app).post(`/api/issues/${f.issueId}/retry`) + .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }); + expect(response.status, JSON.stringify(response.body)).toBe(202); + await f.service.processFailedChatRunRetry(response.body.chatActionId); + await db.update(chatActions).set({ result: { code: "failed_chat_retry_contended", attemptCount: 0 } }) + .where(eq(chatActions.id, response.body.chatActionId)); + const startedAt = Date.now(); + await f.service.processFailedChatRunRetry(response.body.chatActionId); + const [waiting] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + const deadline = Date.parse(String(waiting.result?.identityContentionDeadlineAt)); + expect(deadline).toBeGreaterThanOrEqual(startedAt + 60_000); + expect(deadline).toBeLessThanOrEqual(Date.now() + 60_000); + await f.service.shutdown(); + restarted = createService(new FakeChatSdkRuntime(), f.providerFetch, { wakeup: heartbeatService(db).wakeup }); + await new Promise(resolve => setTimeout(resolve, Math.max(0, Date.parse(String(waiting.result?.retryAt)) - Date.now()) + 1)); + await restarted.service.processFailedChatRunRetries(); + const [again] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + expect(again.result).toMatchObject({ attemptCount: 0, identityContentionCount: 2, + identityContentionDeadlineAt: waiting.result?.identityContentionDeadlineAt }); + expect(restarted.wakeup).not.toHaveBeenCalled(); + } finally { await f.releaseIdentity(); await restarted?.service.shutdown(); await f.service.shutdown(); } + }); + + it.each(["early_release", "expired_release", "authorization_expired"] as const)("honors a prepared Retry identity wait after %s", async mode => { + const f = await busyChatRetryHttpFixture(); + let release = () => {}; + let holding: Promise | undefined; + let releaseAuthorization = () => {}; + let authorizationHolding: Promise | undefined; + let processing: Promise | undefined; + try { + const originalInput = (await readAgentWakeupRequestsById(f.action.id))[0]; + const response = await request(f.app).post(`/api/issues/${f.issueId}/retry`) + .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }); + expect(response.status, JSON.stringify(response.body)).toBe(202); + await f.service.processFailedChatRunRetry(response.body.chatActionId); + const processDue = async () => { + const [current] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + await new Promise(resolve => setTimeout(resolve, Math.max(0, Date.parse(String(current.result?.retryAt)) - Date.now()) + 1)); + await f.service.processFailedChatRunRetry(current.id); + }; + await f.releaseIdentity(); + f.wakeup.mockRejectedValueOnce(new Error("temporary_wakeup_unavailable")); + await processDue(); + const [prepared] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + expect(prepared.status).toBe("issued"); + expect(prepared.payload.sourceAuthorizationPending).toBeUndefined(); + expect(prepared.result?.identityContentionPending).toBeUndefined(); + let acquired = () => {}; + const locked = new Promise(resolve => { acquired = resolve; }); + const held = new Promise(resolve => { release = resolve; }); + holding = db.transaction(async tx => { + await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`chat-identity:${f.companyId}:${f.action.principalId}`}, 0))`); + acquired(); await held; + }); + await locked; + await processDue(); + const [waiting] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + expect(waiting).toMatchObject({ status: "issued", result: { attemptCount: 1, identityContentionCount: 2, + identityContentionDeadlineAt: prepared.result?.identityContentionDeadlineAt } }); + expect(f.wakeup).toHaveBeenCalledOnce(); + if (mode === "expired_release") { + const expired = new Date(Date.now() - 1).toISOString(); + await db.update(chatActions).set({ result: { ...waiting.result, + identityContentionDeadlineAt: expired, retryAt: expired } }).where(eq(chatActions.id, waiting.id)); + } + let authorizationDeadline = 0, authorizationPid = 0; + if (mode === "authorization_expired") { + authorizationDeadline = Date.now() + 5_000; + await db.update(chatActions).set({ result: { ...waiting.result, + identityContentionDeadlineAt: new Date(authorizationDeadline).toISOString(), retryAt: new Date(0).toISOString() } }) + .where(eq(chatActions.id, waiting.id)); + let acquiredAuthorization = () => {}; + const authorizationLocked = new Promise(resolve => { acquiredAuthorization = resolve; }); + const heldAuthorization = new Promise(resolve => { releaseAuthorization = resolve; }); + authorizationHolding = db.transaction(async tx => { + await tx.select({ id: chatConversations.id }).from(chatConversations).where(and( + eq(chatConversations.companyId, f.companyId), eq(chatConversations.id, f.conversation.id))).for("update"); + const [{ pid }] = await tx.execute(sql`select pg_backend_pid() as pid`); + authorizationPid = Number(pid); + acquiredAuthorization(); await heldAuthorization; + }); + await authorizationLocked; + } + release(); await holding; + if (mode === "authorization_expired") { + processing = processDue(); + await vi.waitFor(async () => { + const [{ blocked }] = await db.execute(sql`select exists(select 1 from pg_stat_activity + where ${authorizationPid} = any(pg_blocking_pids(pid))) as blocked`); + expect(blocked).toBe(true); + }, { timeout: 5_000, interval: 25 }); + await new Promise(resolve => setTimeout(resolve, Math.max(0, authorizationDeadline - Date.now()) + 1)); + releaseAuthorization(); await authorizationHolding; + await processing; + } else await processDue(); + const [settled] = await db.select().from(chatActions).where(eq(chatActions.id, waiting.id)); + expect(settled.status).toBe(mode === "early_release" ? "processed" : "failed"); + expect(settled.result?.identityContentionPending).toBeUndefined(); + if (mode !== "early_release") { + expect(settled.result).toMatchObject({ code: "failed_chat_retry_contention_expired", attemptCount: 1 }); + expect(f.wakeup).toHaveBeenCalledOnce(); + expect((await readAgentWakeupRequestsById(response.body.requestId))[0]).toMatchObject({ status: "failed", runId: null }); + expect((await readAgentWakeupRequestsById(f.action.id))[0]).toEqual(originalInput); + expect(await readChatPublicationsByEndpointId(f.endpoint.id)).toEqual(f.originalPublications); + expect((await readHeartbeatRunsByCompanyId(f.companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(0); + } else { + await reconcileTaskChatRetryReceipts(db, { companyId: f.companyId, requestId: response.body.requestId }); + expect((await readAgentWakeupRequestsById(response.body.requestId))[0]?.runId).toBeTruthy(); + expect(f.wakeup).toHaveBeenCalledTimes(2); + expect((await readHeartbeatRunsByCompanyId(f.companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(1); + } + } finally { + releaseAuthorization(); release(); await authorizationHolding; await processing; + await holding; await f.releaseIdentity(); await f.service.shutdown(); + } + }); + + it.each(["restart", "stopped", "ack_lost"] as const)("recovers the committed authorization phase after %s without renewing its deadline", async mode => { + const f = await busyChatRetryHttpFixture(); + let restarted: ReturnType | undefined; + let releaseWake = () => {}; + const heldWake = new Promise(resolve => { releaseWake = resolve; }); + let oldProcessing: ReturnType | undefined; + let oldShutdown: Promise | undefined; + let restoreBegin = () => {}; + try { + const original = (await readHeartbeatRunsById(f.runId))[0]; + const originalInput = (await readAgentWakeupRequestsById(f.action.id))[0]; + const response = await request(f.app).post(`/api/issues/${f.issueId}/retry`) + .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }); + expect(response.status, JSON.stringify(response.body)).toBe(202); + await f.service.processFailedChatRunRetry(response.body.chatActionId); + const [pending] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + const deadline = new Date(Date.now() + 5_000).toISOString(); + await db.update(chatActions).set({ result: { ...pending.result, + identityContentionDeadlineAt: deadline, retryAt: new Date(0).toISOString() } }).where(eq(chatActions.id, pending.id)); + let lostAck = false; + if (mode === "ack_lost") { + const begin = db.$client.begin.bind(db.$client); + const fault = vi.spyOn(db.$client, "begin").mockImplementation((async (...args: unknown[]) => { + const value = await Reflect.apply(begin, db.$client, args); + if (!lostAck && value?.intent?.id === pending.id) { + const [committed] = await db.select().from(chatActions).where(eq(chatActions.id, pending.id)); + expect(committed.result?.identityContentionPending).toBeUndefined(); + const expectedResult = { ...value.intent.result }; + delete expectedResult.identityContentionPending; + expect(committed.result).toEqual(expectedResult); + expect(committed.updatedAt).toEqual(value.intent.updatedAt); + lostAck = true; + throw new Error("authorization_commit_ack_lost"); + } + return value; + }) as typeof db.$client.begin); + restoreBegin = () => fault.mockRestore(); + } else f.wakeup.mockImplementation(async () => { await heldWake; throw new Error("owner_lost_before_wakeup"); }); + await f.releaseIdentity(); + oldProcessing = f.service.processFailedChatRunRetry(pending.id); + if (mode === "ack_lost") { + await oldProcessing; + restoreBegin(); + expect(lostAck).toBe(true); + expect(f.wakeup).not.toHaveBeenCalled(); + await reconcileDatabaseWork(db, CHAT_ACTION_QUEUES.failed_run_retry); + } else await vi.waitFor(() => expect(f.wakeup).toHaveBeenCalledOnce()); + const [authorized] = await db.select().from(chatActions).where(eq(chatActions.id, pending.id)); + if (mode !== "ack_lost") expect(authorized.status).toBe("processing"); + expect(authorized.payload.sourceAuthorizationPending).toBeUndefined(); + expect((await readAgentWakeupRequestsById(pending.id))[0]).toBeUndefined(); + oldShutdown = f.service.shutdown(); + await new Promise(resolve => setTimeout(resolve, Math.max(0, Date.parse(deadline) - Date.now()) + 1)); + await db.update(chatActions).set({ updatedAt: new Date(Date.now() - 60_001) }).where(eq(chatActions.id, pending.id)); + if (mode === "stopped") await heartbeatService(db).cancelInvocationsForAgents([f.fixture.assignedAgentId], "operator_stop"); + restarted = createService(new FakeChatSdkRuntime(), f.providerFetch, { wakeup: heartbeatService(db).wakeup }); + await restarted.service.processFailedChatRunRetry(pending.id); + await reconcileTaskChatRetryReceipts(db, { companyId: f.companyId, requestId: response.body.requestId }); + const [settled] = await db.select().from(chatActions).where(eq(chatActions.id, pending.id)); + const [intent] = await readAgentWakeupRequestsById(response.body.requestId); + if (mode !== "stopped") { + expect(settled.status).toBe("processed"); + expect(intent.runId).toBeTruthy(); + expect(restarted.wakeup).toHaveBeenCalledOnce(); + expect((await readHeartbeatRunsByCompanyId(f.companyId)).filter(run => run.retryOfRunId === f.runId) + .map(run => run.id)).toEqual([intent.runId]); + } else { + expect(["issued", "failed"]).toContain(settled.status); + expect(intent).toMatchObject({ status: "cancelled", runId: null }); + expect(restarted.wakeup).toHaveBeenCalledOnce(); + await expect(restarted.wakeup.mock.results[0].value).resolves.toBeNull(); + expect((await readAgentWakeupRequestsById(pending.id))[0]).toBeUndefined(); + expect((await readHeartbeatRunsByCompanyId(f.companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(0); + } + expect(authorized.result?.identityContentionPending).toBeUndefined(); + expect(settled.result?.identityContentionDeadlineAt).toBe(deadline); + expect(settled.result?.identityContentionCount).toBe(pending.result?.identityContentionCount); + if (mode === "ack_lost") expect(settled.result?.attemptCount).toBe(2); + expect(await readChatPublicationsByEndpointId(f.endpoint.id)).toEqual(f.originalPublications); + expect((await readHeartbeatRunsById(f.runId))[0]).toEqual(original); + expect((await readAgentWakeupRequestsById(f.action.id))[0]).toEqual(originalInput); + releaseWake(); await oldProcessing; await oldShutdown; + expect((await db.select().from(chatActions).where(eq(chatActions.id, pending.id)))[0]).toEqual(settled); + } finally { + restoreBegin(); + releaseWake(); await oldProcessing; await oldShutdown; + await f.releaseIdentity(); await restarted?.service.shutdown(); await f.service.shutdown(); + } + }); + + it("exhausts the existing delivery Retry budget after identity authorization without background renewal", async () => { + const f = await busyChatRetryHttpFixture(); + try { + const originalInput = (await readAgentWakeupRequestsById(f.action.id))[0]; + const response = await request(f.app).post(`/api/issues/${f.issueId}/retry`) + .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }); + expect(response.status, JSON.stringify(response.body)).toBe(202); + await f.service.processFailedChatRunRetry(response.body.chatActionId); + const [contended] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + await f.releaseIdentity(); + f.wakeup.mockRejectedValue(new Error("temporary_wakeup_unavailable")); + for (let attempt = 0; attempt < 8; attempt++) { + const [current] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + if (current.status === "failed") break; + const retryAt = Date.parse(String(current.result?.retryAt)); + expect(Number.isFinite(retryAt)).toBe(true); + await new Promise(resolve => setTimeout(resolve, Math.max(0, retryAt - Date.now()) + 1)); + await f.service.processFailedChatRunRetries(); + } + const [failed] = await db.select().from(chatActions).where(eq(chatActions.id, response.body.chatActionId)); + expect(failed).toMatchObject({ status: "failed", result: { attemptCount: 8, identityContentionCount: 1, + identityContentionDeadlineAt: contended.result?.identityContentionDeadlineAt } }); + expect(failed.payload.sourceAuthorizationPending).toBeUndefined(); + expect(f.wakeup).toHaveBeenCalledTimes(8); + await f.service.processFailedChatRunRetries(); + await f.service.processFailedChatRunRetry(failed.id); + await heartbeatService(db).resumeTaskRetry(f.companyId, response.body.requestId); + expect((await db.select().from(chatActions).where(eq(chatActions.id, failed.id)))[0]).toEqual(failed); + expect((await readAgentWakeupRequestsById(response.body.requestId))[0]).toMatchObject({ status: "failed", runId: null }); + expect((await readAgentWakeupRequestsById(f.action.id))[0]).toEqual(originalInput); + expect(f.wakeup).toHaveBeenCalledTimes(8); + expect(await readChatPublicationsByEndpointId(f.endpoint.id)).toEqual(f.originalPublications); + } finally { await f.releaseIdentity(); await f.service.shutdown(); } + }, 20_000); + async function successfulGitHubQueuedInputFixture() { const f = await currentGitHubTaskRetryFixture(true); const companyId = f.fixture.companyId, issueId = f.issue.id; @@ -65578,7 +66153,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }); it.each([ - ...(["real_budget_ingress", "legacy_budget_ingress", "legacy_prior_response", "legacy_execution", "legacy_execution_log", "legacy_execution_lease", "legacy_execution_profile", "legacy_execution_event", "legacy_execution_controller", "legacy_execution_intent", "legacy_no_prior", "legacy_prior_late", "legacy_prior_unprocessed", "legacy_prior_delivery_unprocessed", "legacy_prior_numeric_app", "legacy_response_conflict", "legacy_prior_wrong_app", "legacy_prior_wrong_agent", "legacy_prior_unlinked", "legacy_prior_retargeted", "legacy_explicit_null", "legacy_explicit_empty", "legacy_explicit_malformed", "legacy_explicit_mismatch", "legacy_current_app_changed", "source_app_changed", "source_destination_changed", "accepted", "multiple_saved", "real_queue", "committed_succeeded", "committed_consumed", "committed_failure", "committed_failed_publication", "committed_unknown_publication", "committed_pending_publication", "committed_mixed_publication", "committed_consumed_publication", "committed_succeeded_publication", "committed_delayed_cleanup", "committed_board", "committed_board_existing", "committed_board_existing_coalesced", "committed_board_existing_source_edited", "committed_board_existing_board_edited", "committed_board_existing_controller", "committed_board_existing_allocation", "committed_board_existing_receipt", "committed_board_existing_delivered", "committed_board_existing_source_owner", "committed_board_existing_source_delivered", "committed_unknown_cleanup", "committed_live_controller", "unreconciled", "source_edited", "source_claimed", "malformed_previous", "endpoint_paused", "principal_revoked", "still_budget_blocked"] as const) + ...(["real_budget_ingress", "legacy_budget_ingress", "legacy_prior_response", "legacy_execution", "legacy_execution_log", "legacy_execution_lease", "legacy_execution_profile", "legacy_execution_event", "legacy_execution_controller", "legacy_execution_intent", "legacy_no_prior", "legacy_prior_late", "legacy_prior_unprocessed", "legacy_prior_delivery_unprocessed", "legacy_prior_numeric_app", "legacy_response_conflict", "legacy_prior_wrong_app", "legacy_prior_wrong_agent", "legacy_prior_unlinked", "legacy_prior_retargeted", "legacy_explicit_null", "legacy_explicit_empty", "legacy_explicit_malformed", "legacy_explicit_mismatch", "legacy_current_app_changed", "source_app_changed", "source_destination_changed", "accepted", "multiple_saved", "real_queue", "committed_succeeded", "committed_consumed", "committed_failure", "committed_failed_publication", "committed_unknown_publication", "committed_pending_publication", "committed_mixed_publication", "committed_consumed_publication", "committed_succeeded_publication", "committed_delayed_cleanup", "committed_board", "committed_board_existing", "committed_board_existing_coalesced", "committed_board_existing_identity_contended", "committed_board_existing_identity_expired", "committed_board_existing_source_edited", "committed_board_existing_board_edited", "committed_board_existing_controller", "committed_board_existing_allocation", "committed_board_existing_receipt", "committed_board_existing_delivered", "committed_board_existing_source_owner", "committed_board_existing_source_delivered", "committed_unknown_cleanup", "committed_live_controller", "unreconciled", "source_edited", "source_claimed", "malformed_previous", "endpoint_paused", "principal_revoked", "still_budget_blocked"] as const) .map(mode => ({ provider: "github" as const, mode })), ])("retries the newer saved $provider request instead of the failed request after a budget hold: $mode", async ({ provider, mode }) => { const queuedBoard = mode === "committed_board" || mode.startsWith("committed_board_existing"); @@ -66095,7 +66670,38 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }); const inputIds = [boardWake.id, ...savedInputs.map(source => source.id)]; const before = await db.select().from(agentWakeupRequests).where(inArray(agentWakeupRequests.id, inputIds)).orderBy(asc(agentWakeupRequests.id)); - if (mode !== "committed_board_existing" && mode !== "committed_board_existing_coalesced") { + if (mode === "committed_board_existing_identity_contended") { + const events: string[] = []; + const topics = [DELIVERY_QUEUES.deferredExecutionReceipts, CHAT_ACTION_QUEUES.failed_run_retry]; + const unsubscribes = topics.map(topic => subscribeDatabaseWork(db, topic, event => events.push(event))); + const rollback = new Error("rollback restored Retry preparation"); + let rolledBackActionId: string | undefined; + try { + await expect(db.transaction(async tx => { + events.length = 0; + const staged = await requestTaskRetry({ ...nextInput, db: tx as unknown as Db }); + rolledBackActionId = staged.chatActionId ?? undefined; + expect(rolledBackActionId).toBeTruthy(); + expect(events).toContain("intent"); + expect(events).not.toContain("settled"); + expect(await db.select().from(agentWakeupRequests).where(inArray(agentWakeupRequests.id, inputIds)) + .orderBy(asc(agentWakeupRequests.id))).toEqual(before); + expect(await tx.select().from(activityLog).where(and(eq(activityLog.companyId, companyId), + eq(activityLog.runId, failed.id), eq(activityLog.action, "issue.retry_inputs_restored")))).toHaveLength(1); + throw rollback; + })).rejects.toBe(rollback); + for (const topic of topics) { + await reconcileDatabaseWork(db, topic); + expect(databaseWorkPending(db, topic)).toBe(false); + } + expect(await db.select().from(chatActions).where(eq(chatActions.id, rolledBackActionId!))).toEqual([]); + expect(await db.select().from(activityLog).where(and(eq(activityLog.companyId, companyId), + eq(activityLog.runId, failed.id), eq(activityLog.action, "issue.retry_inputs_restored")))).toEqual([]); + expect(await db.select().from(agentWakeupRequests).where(inArray(agentWakeupRequests.id, inputIds)) + .orderBy(asc(agentWakeupRequests.id))).toEqual(before); + } finally { unsubscribes.forEach(unsubscribe => unsubscribe()); } + } + if (mode !== "committed_board_existing" && mode !== "committed_board_existing_coalesced" && mode !== "committed_board_existing_identity_contended" && mode !== "committed_board_existing_identity_expired") { await expect(requestTaskRetry(nextInput)).rejects.toMatchObject({ details: { code: "chat_failed_run_retry_not_authorized" }, }); @@ -66112,14 +66718,98 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const retryHttp = () => httpRequest(app).post(`/api/issues/${context.issue.id}/retry`).send({ idempotencyKey: nextInput.idempotencyKey, observedRunId: nextInput.observedRunId, }).expect(202); - const next = (await retryHttp()).body; + const restoredRetryWake = vi.fn(heartbeat.wakeup); + const retryWithIdentityContention = async () => { + if (mode !== "committed_board_existing_identity_contended" && mode !== "committed_board_existing_identity_expired") return retryHttp(); + let release = () => {}; + const held = new Promise(resolve => { release = resolve; }); + let acquired = () => {}; + const locked = new Promise(resolve => { acquired = resolve; }); + const holding = db.transaction(async tx => { + await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`chat-identity:${companyId}:${saved.principalId}`}, 0))`); + acquired(); await held; + }); + try { + await locked; + const response = await retryHttp(); + if (mode === "committed_board_existing_identity_expired" || mode === "committed_board_existing_identity_contended") + await context.service.processFailedChatRunRetry(response.body.chatActionId); + if (mode === "committed_board_existing_identity_contended") context.wakeup + .mockImplementation(restoredRetryWake).mockRejectedValueOnce(new Error("retry_admission_unavailable_before_receipt")); + expect(await db.select().from(agentWakeupRequests).where(inArray(agentWakeupRequests.id, inputIds)) + .orderBy(asc(agentWakeupRequests.id))).toEqual(before); + return response; + } finally { release(); await holding; } + }; + const next = (await retryWithIdentityContention()).body; expect(next).toMatchObject({ accepted: true }); + if (mode === "committed_board_existing_identity_expired") { + const [pending] = await db.select().from(chatActions).where(eq(chatActions.id, next.chatActionId!)); + expect(pending.payload).toMatchObject({ sourceAuthorizationPending: true, restoreUnstartedInput: true }); + const deadline = Date.now() + 5_000; + await db.update(chatActions).set({ result: { ...pending.result, + identityContentionDeadlineAt: new Date(deadline).toISOString(), retryAt: new Date(0).toISOString() } }) + .where(eq(chatActions.id, pending.id)); + const realPermission = taskRetryModule.taskRetryMutationAllowed; + const permission = vi.spyOn(taskRetryModule, "taskRetryMutationAllowed").mockImplementation(async (database, input) => { + const allowed = await realPermission(database, input); + if (input.companyId === companyId) { + expect(await database.select().from(activityLog).where(and(eq(activityLog.companyId, companyId), + eq(activityLog.runId, failed.id), eq(activityLog.action, "issue.retry_inputs_restored")))).toHaveLength(1); + await new Promise(resolve => setTimeout(resolve, Math.max(0, deadline - Date.now()) + 1)); + } + return allowed; + }); + const publications = await readChatPublicationsByEndpointId(context.endpoint.id); + const runIds = (await readHeartbeatRunsByCompanyId(companyId)).map(run => run.id).sort(); + context.wakeup.mockClear(); + try { + await context.service.processFailedChatRunRetry(pending.id); + expect(permission).toHaveBeenCalled(); + expect((await db.select().from(chatActions).where(eq(chatActions.id, pending.id)))[0]) + .toMatchObject({ status: "failed", payload: { sourceAuthorizationPending: true }, + result: { code: "failed_chat_retry_contention_expired" } }); + expect(await db.select().from(activityLog).where(and(eq(activityLog.companyId, companyId), + eq(activityLog.runId, failed.id), eq(activityLog.action, "issue.retry_inputs_restored")))).toEqual([]); + expect(await db.select().from(agentWakeupRequests).where(inArray(agentWakeupRequests.id, inputIds)) + .orderBy(asc(agentWakeupRequests.id))).toEqual(before); + expect(context.wakeup).not.toHaveBeenCalled(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(publications); + expect((await readAgentWakeupRequestsById(next.requestId!))[0]).toMatchObject({ status: "failed", runId: null }); + expect((await readHeartbeatRunsByCompanyId(companyId)).map(run => run.id).sort()).toEqual(runIds); + } finally { permission.mockRestore(); } + return; + } expect((await retryHttp()).body.requestId).toBe(next.requestId); + if (mode === "committed_board_existing_identity_contended") await vi.waitFor(async () => { + await context.service.processFailedChatRunRetries(); + expect(await db.select().from(activityLog).where(and(eq(activityLog.companyId, companyId), + eq(activityLog.runId, failed.id), eq(activityLog.action, "issue.retry_inputs_restored")))).toHaveLength(1); + }, { timeout: 5_000, interval: 50 }); expect(await db.select().from(activityLog).where(and(eq(activityLog.companyId, companyId), eq(activityLog.runId, failed.id), eq(activityLog.action, "issue.retry_inputs_restored")))).toHaveLength(1); - expect(await context.service.processFailedChatRunRetry(next.chatActionId!)).toMatchObject({ status: "queued" }); - [resumed] = await readAgentWakeupRequestsById(next.chatActionId!); - expect(resumed.runId).toBeTruthy(); + await vi.waitFor(async () => { + expect(await context.service.processFailedChatRunRetry(next.chatActionId!)).toMatchObject({ status: "queued" }); + [resumed] = await readAgentWakeupRequestsById(next.chatActionId!); + expect(resumed).toMatchObject({ companyId, agentId: context.fixture.assignedAgentId, + requestedByActorType: "user", requestedByActorId: "owner-user", + payload: { taskRetryRequestId: next.requestId, retryOfRunId: failed.id } }); + expect(resumed.runId).toBeTruthy(); + await reconcileTaskChatRetryReceipts(db, { companyId, requestId: next.requestId! }); + expect((await readAgentWakeupRequestsById(next.requestId!))[0]?.runId).toBe(resumed.runId); + }, { timeout: 5_000, interval: 50 }); + if (mode === "committed_board_existing_identity_contended") { + expect(restoredRetryWake).toHaveBeenCalledOnce(); + expect(restoredRetryWake).toHaveBeenCalledWith(context.fixture.assignedAgentId, expect.objectContaining({ + taskRetryRequestId: next.requestId, requestedByActorType: "user", requestedByActorId: "owner-user", + contextSnapshot: expect.objectContaining({ taskRetryRequestId: next.requestId, + chatFailedRunRetry: { version: 1, actionId: next.chatActionId, failedRunId: failed.id } }), + })); + expect((await readHeartbeatRunsByCompanyId(companyId)).filter(run => + run.contextSnapshot?.taskRetryRequestId === next.requestId).map(run => run.id)).toEqual([resumed.runId]); + expect(await db.select().from(activityLog).where(and(eq(activityLog.companyId, companyId), + eq(activityLog.runId, failed.id), eq(activityLog.action, "issue.retry_inputs_restored")))).toHaveLength(1); + } if (mode === "committed_board_existing_coalesced") { const restored = await db.select().from(agentWakeupRequests) .where(inArray(agentWakeupRequests.id, savedInputs.map(source => source.id))); diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index aec7ce9a3fb..d90c7a6c989 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -13,6 +13,8 @@ import { FAILED_RUN_RETRY_STALE_MS, dueChatAction } from "../chat-action-work.js import { readChatAuthorizationRun } from "./authorization-run-read.js"; import { selectTaskRetryInput, claimTaskRetryInput, settleTaskRetryInput, releaseUnadmittedTaskRetryInput, taskRetryInputSchema, type TaskRetryInput } from "./task-retry-input.js"; import { queuedCommentIdsFromWakePayload } from "../issue-queued-comment-queue.js"; +import { taskRetryMutationAllowed } from "../task-retry.js"; +import { z } from "zod"; import type { TaskRetryWakeAuthority } from "./github/task-retry-authorization.js"; import { @@ -21,6 +23,7 @@ import { import { and, asc, + desc, eq, gte, inArray, @@ -97,6 +100,7 @@ import { CHAT_CONVERSATION_ID_KEY } from "../../modules/run-dispatch/index.js"; type DbOrTransaction = Db | Parameters[0]>[0]; type DeliveryRow = typeof chatDeliveries.$inferSelect; type RetryRuntimeFence = { credentialFingerprint: string; generation: number }; +const FAILED_RUN_RETRY_IDENTITY_CONTENTION_MS = 60_000; export type ChatFailedRetrySource = Awaited["failedChatRetrySource"]>>; type ChatFailedRunRetryOptions = { @@ -184,6 +188,14 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti taskRetryInput?: TaskRetryInput; sourceScopeSha256: string; }; + type PendingRetryBinding = Pick; + const pendingRetrySourceSchema = z.object({ + version: z.literal(1), sourceAuthorizationPending: z.literal(true), + failedRunId: z.string().uuid(), issueId: z.string().uuid(), agentId: z.string().uuid(), + initiatedByUserId: z.string().min(1), taskRetryIntentId: z.string().uuid(), + taskRetryInput: taskRetryInputSchema.optional(), + restoreUnstartedInput: z.literal(true).optional(), + }).strict(); function hashFailedRetrySource( scope: Omit, ): FailedChatRetrySource { @@ -751,6 +763,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti publication?: boolean; taskRetryIntentId?: string; taskRetryInput?: TaskRetryInput; + onIdentityContention?: (binding: PendingRetryBinding) => void; taskRetryReauthorization?: TaskRetryWakeAuthority; /** Internal selector for already committed output, never retry admission. */ committedResponse?: { @@ -1148,11 +1161,17 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti ) ) throw failedChatRetryDenied(); + const binding = { endpointId: first.endpointId, conversationId: first.conversationId, principalId: first.principalId }; + const deferSourceAuthorization = (): never => { + input.onIdentityContention?.(binding); + throw new NativeChatReviewPresentationContentionError(); + }; + const endpointContention = input.taskRetryIntentId && !input.publication && !input.committedResponse + ? { onEndpointContention: deferSourceAuthorization } : {}; const identity = await tx.execute( sql`select pg_try_advisory_xact_lock(hashtextextended(${`chat-identity:${input.companyId}:${first.principalId}`}, 0)) as locked`, ); - if (!identity[0]?.locked) - throw new NativeChatReviewPresentationContentionError(); + if (!identity[0]?.locked) deferSourceAuthorization(); const actionIds = ordered.map(action => action.id); const currentActions = await tx.select().from(chatActions).where(and(eq(chatActions.companyId, input.companyId), inArray(chatActions.id, actionIds))).orderBy(asc(chatActions.id)).for("update", { noWait: true }); @@ -1182,6 +1201,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti const invalidated = new Set(lifecycleRows.map(row => lifecycleKey(row.target, row.context))); const batch = reauthorization ? await authorizeGitHubTaskRetryBatch(tx, ordered, { nonblocking: true, terminal: input.publication === true, taskRetry: reauthorization, + ...endpointContention, }).catch(error => { if (isExternalActionAuthorizationChange(error)) throw failedChatRetryDenied(input.taskRetryIntentId ? "source_authorization" : undefined, safeChatWakeAuthorityReason(error)); throw error; }) : null; const canonicalSources: Array> = []; let destination: Awaited> | null = @@ -1190,6 +1210,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti destination = batch?.get(action.id) ?? await authorizeInboundWakeup(tx, action, { nonblocking: true, terminal: input.publication === true, + ...endpointContention, ...(reauthorization ? { taskRetry: reauthorization } : {}), }).catch((error) => { if (isExternalActionAuthorizationChange(error)) @@ -1373,6 +1394,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti * The failed run and its canonical Retry remain unchanged history. */ async function restoreUnstartedRetryInput(tx: DbOrTransaction, input: { companyId: string; issueId: string; agentId: string; failedRunId: string; initiatedByUserId: string; + onIdentityContention?: (binding: PendingRetryBinding) => void; }) { const [run] = await tx.select().from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId), @@ -1455,18 +1477,40 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti retryIntentId?: string; }, ): Promise<{ actionId: string; issueId: string }> { - let taskRetryInput = input.retryIntentId ? await selectTaskRetryInput(tx, input) : undefined; - if (input.retryIntentId && taskRetryInput && !taskRetryInput.sourceActionIds.length) - taskRetryInput = await restoreUnstartedRetryInput(tx, input) ?? taskRetryInput; - const source = await failedChatRetrySource(tx, { ...input, taskRetryIntentId: input.retryIntentId, taskRetryInput }); + let taskRetryInput: TaskRetryInput | undefined; + let source: FailedChatRetrySource | null = null; + const pending: { binding: PendingRetryBinding | null; restoreUnstartedInput: boolean } = { + binding: null, restoreUnstartedInput: false, + }; + const onIdentityContention = (binding: PendingRetryBinding) => { pending.binding = binding; }; + try { + source = await tx.transaction(async sourceTx => { + taskRetryInput = input.retryIntentId ? await selectTaskRetryInput(sourceTx, input) : undefined; + if (input.retryIntentId && taskRetryInput && !taskRetryInput.sourceActionIds.length) { + pending.restoreUnstartedInput = true; + taskRetryInput = await restoreUnstartedRetryInput(sourceTx, { ...input, onIdentityContention }) ?? taskRetryInput; + } + return failedChatRetrySource(sourceTx, { ...input, taskRetryIntentId: input.retryIntentId, + taskRetryInput, onIdentityContention }); + }); + } catch (error) { + if (!(error instanceof NativeChatReviewPresentationContentionError) || !input.retryIntentId || !pending.binding) throw error; + } + const binding = source ?? pending.binding!; + const payload = source ? { version: 1, ...source, initiatedByUserId: input.initiatedByUserId, + ...(input.retryIntentId ? { taskRetryIntentId: input.retryIntentId } : {}) } + : pendingRetrySourceSchema.parse({ version: 1, sourceAuthorizationPending: true, + failedRunId: input.failedRunId, issueId: input.issueId, agentId: input.agentId, + initiatedByUserId: input.initiatedByUserId, taskRetryIntentId: input.retryIntentId, taskRetryInput, + ...(pending.restoreUnstartedInput ? { restoreUnstartedInput: true } : {}) }); const providerActionId = input.retryIntentId - ? `task-retry:${input.retryIntentId}` : `failed-run-retry:${source.failedRunId}`; + ? `task-retry:${input.retryIntentId}` : `failed-run-retry:${source!.failedRunId}`; const [existing] = await tx .select() .from(chatActions) .where( and( - eq(chatActions.endpointId, source.endpointId), + eq(chatActions.endpointId, binding.endpointId), eq(chatActions.providerActionId, providerActionId), ), ) @@ -1475,12 +1519,16 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti if ( existing.kind !== "failed_run_retry" || existing.companyId !== input.companyId || - existing.payload.sourceScopeSha256 !== source.sourceScopeSha256 + (source ? existing.payload.sourceScopeSha256 !== source.sourceScopeSha256 + : JSON.stringify(existing.payload) !== JSON.stringify(payload)) ) throw failedChatRetryDenied(); return { actionId: existing.id, issueId: input.issueId }; } - if (!input.retryIntentId) await assertFailedNativeRetryState(tx, source, input.companyId); + if (!input.retryIntentId) { + if (!source) throw failedChatRetryDenied(); + await assertFailedNativeRetryState(tx, source, input.companyId); + } const [busy] = await tx .select({ id: heartbeatRuns.id }) .from(heartbeatRuns) @@ -1503,19 +1551,16 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti .insert(chatActions) .values({ companyId: input.companyId, - endpointId: source.endpointId, - conversationId: source.conversationId, - principalId: source.principalId, + endpointId: binding.endpointId, + conversationId: binding.conversationId, + principalId: binding.principalId, kind: "failed_run_retry", providerActionId, - payload: { - version: 1, - ...source, - initiatedByUserId: input.initiatedByUserId, - ...(input.retryIntentId ? { taskRetryIntentId: input.retryIntentId } : {}), - }, + payload, status: "issued", - result: { code: "failed_chat_retry_queued", attemptCount: 0 }, + result: { code: source ? "failed_chat_retry_queued" : "failed_chat_retry_contended", attemptCount: 0, + ...(!source ? { retryAt: now.toISOString(), identityContentionCount: 0, identityContentionPending: true, + identityContentionDeadlineAt: new Date(now.getTime() + FAILED_RUN_RETRY_IDENTITY_CONTENTION_MS).toISOString() } : {}) }, createdAt: now, updatedAt: now, }) @@ -1530,12 +1575,56 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti details: { retryActionId: intent.id, issueId: input.issueId, - commentCount: source.commentIds.length, + ...(source ? { commentCount: source.commentIds.length } : { sourceAuthorizationPending: true }), }, }); return { actionId: intent.id, issueId: input.issueId }; } + function assertIdentityContentionDeadline(action: typeof chatActions.$inferSelect) { + const deadline = action.result?.identityContentionDeadlineAt; + if (deadline !== undefined && !(typeof deadline === "string" && Date.parse(deadline) > Date.now())) { + throw new NativeChatReviewPresentationContentionError(); + } + } + + async function preparePendingRetrySource(action: typeof chatActions.$inferSelect) { + const pending = pendingRetrySourceSchema.safeParse(action.payload); + if (!pending.success) throw failedChatRetryDenied(); + return db.transaction(async tx => { + const input = pending.data; + const restored = input.restoreUnstartedInput ? await restoreUnstartedRetryInput(tx, { + companyId: action.companyId, issueId: input.issueId, agentId: input.agentId, + failedRunId: input.failedRunId, initiatedByUserId: input.initiatedByUserId, + }) : undefined; + const source = await failedChatRetrySource(tx, { companyId: action.companyId, + issueId: input.issueId, agentId: input.agentId, failedRunId: input.failedRunId, + taskRetryIntentId: input.taskRetryIntentId, taskRetryInput: restored ?? input.taskRetryInput }); + const retry = await readTaskChatRetryIntent(tx as unknown as Db, { companyId: action.companyId, + issueId: input.issueId, agentId: input.agentId, sourceRunId: input.failedRunId, + intentId: input.taskRetryIntentId, chatActionId: action.id }); + if (!retry || retry.wake.status !== "deferred_issue_execution" || retry.wake.runId || + retry.intent.actorId !== input.initiatedByUserId || + !await taskRetryMutationAllowed(tx as unknown as Db, { companyId: action.companyId, + issueId: input.issueId, actorId: retry.intent.actorId, actorSource: retry.intent.actorSource })) throw failedChatRetryDenied(); + const [latest] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, action.companyId), or(eq(heartbeatRuns.issueId, input.issueId), + eq(heartbeatRuns.nativeIssueId, input.issueId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${input.issueId}`), + )).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1); + if (latest?.id !== input.failedRunId || source.endpointId !== action.endpointId || + source.conversationId !== action.conversationId || source.principalId !== action.principalId) throw failedChatRetryDenied(); + await notifyChatActionWork(tx, "failed_run_retry"); + assertIdentityContentionDeadline(action); + const [prepared] = await tx.update(chatActions).set({ payload: { version: 1, ...source, + initiatedByUserId: input.initiatedByUserId, taskRetryIntentId: input.taskRetryIntentId } }).where(and( + eq(chatActions.id, action.id), eq(chatActions.companyId, action.companyId), eq(chatActions.status, "processing"), + eq(chatActions.updatedAt, action.updatedAt), sql`${chatActions.payload}->'sourceAuthorizationPending' = 'true'::jsonb`, + )).returning(); + if (!prepared) throw failedChatRetryDenied(); + return prepared; + }); + } + async function authorizeFailedChatRunRetry( tx: DbOrTransaction, input: { @@ -1545,6 +1634,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti retryActionId: string; runId?: string; publication?: boolean; + onIdentityContention?: (binding: PendingRetryBinding) => void; }, ) { const [snapshot] = await tx @@ -1564,6 +1654,10 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti snapshot.payload.agentId !== input.agentId ) throw failedChatRetryDenied(); + if (snapshot.payload.sourceAuthorizationPending === true) { + if (["failed", "cancelled"].includes(snapshot.status)) throw failedChatRetryDenied(); + throw new NativeChatReviewPresentationContentionError(); + } const source = await failedChatRetrySource(tx, { ...input, failedRunId: snapshot.payload.failedRunId, @@ -1773,7 +1867,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti Date.parse(candidate.result.retryAt) > now.getTime()) ) return failedChatRetryReceipt(candidate); - const [claimed] = await db.transaction(async (tx) => { + const [claimedCandidate] = await db.transaction(async (tx) => { await notifyChatActionWork(tx, "failed_run_retry"); return tx .update(chatActions) @@ -1787,9 +1881,26 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti ) .returning(); }); - if (!claimed) return failedChatRetryReceipt(candidate); - const attemptCount = Number(candidate.result?.attemptCount ?? 0) + 1; + if (!claimedCandidate) return failedChatRetryReceipt(candidate); + let claimed = claimedCandidate; + const previousAttemptCount = Number(candidate.result?.attemptCount ?? 0); + const attemptCount = previousAttemptCount + 1; + let observedIdentityContention = false; + let contentionResult: Record = { + ...(candidate.result?.identityContentionDeadlineAt !== undefined + ? { identityContentionDeadlineAt: candidate.result.identityContentionDeadlineAt } : {}), + ...(candidate.result?.identityContentionCount !== undefined + ? { identityContentionCount: candidate.result.identityContentionCount } : {}), + ...(candidate.result?.identityContentionPending === true ? { identityContentionPending: true } : {}), + }; const settle = async (status: string, result: Record) => { + const settledResult = { ...contentionResult, ...result }; + if (["processed", "failed", "cancelled"].includes(status)) delete settledResult.identityContentionPending; + const durableResult = !observedIdentityContention && settledResult.identityContentionPending === true + ? sql`(${JSON.stringify(settledResult)}::jsonb - 'identityContentionPending') || case + when ${chatActions.result}->'identityContentionPending' = 'true'::jsonb + then jsonb_build_object('identityContentionPending', true) else '{}'::jsonb end` + : settledResult; const [current] = await db.transaction(async (tx) => { if (typeof claimed.payload.taskRetryIntentId === "string") await notifyTaskRetrySettlement(tx, { companyId: claimed.companyId, requestId: claimed.payload.taskRetryIntentId, @@ -1800,7 +1911,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti }); return tx .update(chatActions) - .set({ status, result, updatedAt: new Date() }) + .set({ status, result: durableResult, updatedAt: new Date() }) .where( and( eq(chatActions.id, claimed.id), @@ -1813,15 +1924,31 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti return failedChatRetryReceipt(await reconcileFailedTaskRetry(current ?? claimed)); }; try { + if (claimed.payload.sourceAuthorizationPending === true || contentionResult.identityContentionPending === true) + assertIdentityContentionDeadline(claimed); + if (claimed.payload.sourceAuthorizationPending === true) { + claimed = await preparePendingRetrySource(claimed); + } const scope = { companyId: claimed.companyId, issueId: String(claimed.payload.issueId), agentId: String(claimed.payload.agentId), retryActionId: claimed.id, }; - const source = await db.transaction((tx) => - authorizeFailedChatRunRetry(tx, scope), - ); + const source = await db.transaction(async tx => { + const authorized = await authorizeFailedChatRunRetry(tx, scope); + if (contentionResult.identityContentionPending === true) { + await notifyChatActionWork(tx, "failed_run_retry"); + assertIdentityContentionDeadline(claimed); + const [advanced] = await tx.update(chatActions).set({ + result: sql`coalesce(${chatActions.result}, '{}'::jsonb) - 'identityContentionPending'`, + }).where(and(eq(chatActions.companyId, claimed.companyId), eq(chatActions.id, claimed.id), + eq(chatActions.status, "processing"), eq(chatActions.updatedAt, now))).returning({ id: chatActions.id }); + if (!advanced) throw failedChatRetryDenied(); + } + return authorized; + }); + delete contentionResult.identityContentionPending; const existing = await db .select() .from(agentWakeupRequests) @@ -1945,6 +2072,24 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti code: "failed_chat_retry_durable", attemptCount, }); } + if (error instanceof NativeChatReviewPresentationContentionError) { + observedIdentityContention = true; + const contendedAt = Date.now(); + const identityContentionDeadlineAt = contentionResult.identityContentionDeadlineAt + ?? new Date(contendedAt + FAILED_RUN_RETRY_IDENTITY_CONTENTION_MS).toISOString(); + const deadline = typeof identityContentionDeadlineAt === "string" ? Date.parse(identityContentionDeadlineAt) : NaN; + const identityContentionCount = Number(contentionResult.identityContentionCount ?? 0) + 1; + contentionResult = { identityContentionDeadlineAt, identityContentionCount, identityContentionPending: true }; + const retryable = Number.isSafeInteger(previousAttemptCount) && previousAttemptCount >= 0 && previousAttemptCount < 8 + && Number.isSafeInteger(identityContentionCount) && identityContentionCount > 0 + && Number.isFinite(deadline) && contendedAt < deadline; + return settle(retryable ? "issued" : "failed", { + code: retryable ? "failed_chat_retry_contended" : "failed_chat_retry_contention_expired", + attemptCount: previousAttemptCount, + ...(retryable ? { retryAt: new Date(Math.min(deadline, + contendedAt + Math.min(5_000, identityContentionCount * 250))).toISOString() } : {}), + }); + } const retryable = !denied && attemptCount < 8; return settle(retryable ? "issued" : "failed", { code: retryable diff --git a/server/src/services/chat-channels/github/task-retry-authorization.ts b/server/src/services/chat-channels/github/task-retry-authorization.ts index d27edc958cf..42a6ae9ab6c 100644 --- a/server/src/services/chat-channels/github/task-retry-authorization.ts +++ b/server/src/services/chat-channels/github/task-retry-authorization.ts @@ -116,10 +116,12 @@ type GitHubRetryBinding = { export function createGitHubTaskRetryBatchAuthority(authorize: ( tx: DbOrTransaction, action: typeof chatActions.$inferSelect, - notice: { nonblocking: true; terminal: boolean; taskRetry: TaskRetryWakeAuthority }, + notice: { nonblocking: true; terminal: boolean; taskRetry: TaskRetryWakeAuthority; + onEndpointContention?: () => never }, ) => Promise) { async function authorizeGitHubTaskRetryBatch(tx: DbOrTransaction, actions: Array, - notice: { nonblocking: true; terminal: boolean; taskRetry: TaskRetryWakeAuthority }) { + notice: { nonblocking: true; terminal: boolean; taskRetry: TaskRetryWakeAuthority; + onEndpointContention?: () => never }) { const deny = chatWakeAuthorityDenied; const first = actions[0]; if (!first) throw deny("batch_scope"); diff --git a/server/src/services/chat-channels/inbound/wakeup-authorization.ts b/server/src/services/chat-channels/inbound/wakeup-authorization.ts index 0d45fcb7606..8c231e0822f 100644 --- a/server/src/services/chat-channels/inbound/wakeup-authorization.ts +++ b/server/src/services/chat-channels/inbound/wakeup-authorization.ts @@ -43,7 +43,8 @@ export function createInboundWakeAuthority(db: Db, options: WakeAuthorityOptions async function authorizeInboundWakeup( tx: DbOrTransaction, action: typeof chatActions.$inferSelect, - notice?: { nonblocking: true; terminal: boolean; taskRetry?: TaskRetryWakeAuthority }, + notice?: { nonblocking: true; terminal: boolean; taskRetry?: TaskRetryWakeAuthority; + onEndpointContention?: () => never }, ) { const payload = action.payload; const deny = chatWakeAuthorityDenied; @@ -91,7 +92,17 @@ export function createInboundWakeAuthority(db: Db, options: WakeAuthorityOptions ) .for("no key update", { noWait: true }) .limit(1) - .then((rows) => rows[0] ?? null); + .then((rows) => rows[0] ?? null) + .catch(error => { + if (notice?.onEndpointContention) { + let cause: unknown = error; + for (let depth = 0; depth < 4 && cause && typeof cause === "object"; depth += 1) { + if ("code" in cause && cause.code === "55P03") notice.onEndpointContention(); + cause = "cause" in cause ? cause.cause : undefined; + } + } + throw error; + }); const delivery = await tx .select() .from(chatDeliveries) diff --git a/server/src/services/task-interruption-presentation.test.ts b/server/src/services/task-interruption-presentation.test.ts index f6cd4681ddc..0b491a1d008 100644 --- a/server/src/services/task-interruption-presentation.test.ts +++ b/server/src/services/task-interruption-presentation.test.ts @@ -7,6 +7,38 @@ const run = { id: "run", agentId: "agent", status: "failed", scheduledRetryAt: n resultJson: { projectAccessDenied: { projectId: "22222222-2222-4222-8222-222222222222" } } }; const agent = { status: "idle", pauseReason: null }; +describe("unadmitted chat Retry presentation", () => { + const now = Date.parse("2026-01-01T12:00:00Z"); + const failed = { ...run, errorCode: "native_provider_terminal_failed", resultJson: null }; + const wait = { id: "retry", requestedAt: new Date(now), sourceRunId: run.id, + preparationStatus: "issued", preparationRetryAt: now + 250 }; + const input = { run: failed, agent, wait, paused: false, approval: false }; + + it.each(["issued", "processing"])("uses the persisted %s outbox deadline without implying provider work", status => { + for (const offset of [0, 20_000]) expect(projectTaskInterruption(issue, { + ...input, wait: { ...wait, preparationStatus: status }, + }, now + offset)).toMatchObject({ state: "retry_scheduled", canRetry: false, + nextRetryAt: new Date(wait.preparationRetryAt).toISOString() }); + }); + + it.each(["failed", "cancelled"])("keeps terminal %s preparation out of the working state", status => { + expect(projectTaskInterruption(issue, { ...input, wait: { ...wait, preparationStatus: status, + preparationRetryAt: null } }, now)).toMatchObject({ state: "failed", nextRetryAt: null }); + }); + + it("keeps current gates and admitted execution authoritative", () => { + expect(projectTaskInterruption(issue, { ...input, paused: true }, now)) + .toMatchObject({ state: "waiting_for_user", userAction: "resume" }); + expect(projectTaskInterruption(issue, { ...input, approval: true }, now)) + .toMatchObject({ state: "waiting_for_user", userAction: "approval" }); + for (const status of ["queued", "running"]) expect(projectTaskInterruption(issue, { + ...input, run: { ...failed, id: "successor", status }, + }, now)).toMatchObject({ state: "working", nextRetryAt: null }); + expect(projectTaskInterruption(issue, { ...input, wait: { ...wait, sourceRunId: "old-run" } }, now)) + .toMatchObject({ state: "working", nextRetryAt: null }); + }); +}); + it("explains a persisted Retry wait without exposing the saved payload", () => { const failed = { ...run, errorCode: "native_provider_terminal_failed", resultJson: null }; const wait = { id: "retry", requestedAt: new Date(), sourceRunId: failed.id, diff --git a/server/src/services/task-interruption-presentation.ts b/server/src/services/task-interruption-presentation.ts index 31e7901538d..91e074fd518 100644 --- a/server/src/services/task-interruption-presentation.ts +++ b/server/src/services/task-interruption-presentation.ts @@ -1,5 +1,5 @@ import { and, desc, eq, inArray, isNull, sql, type SQL } from "drizzle-orm"; -import { agents, agentWakeupRequests, approvals, companies, heartbeatRuns, issueApprovals, issueRecoveryActions, +import { agents, agentWakeupRequests, approvals, chatActions, companies, heartbeatRuns, issueApprovals, issueRecoveryActions, issues, issueTreeHoldMembers, issueTreeHolds, projects, type Db } from "@paperclipai/db"; import { unionAll } from "drizzle-orm/pg-core"; import { HEARTBEAT_RUN_STATUSES, readProviderFailure, type TaskInterruptionPresentation } from "@paperclipai/shared"; @@ -14,12 +14,15 @@ import { createRunSecretRedactionRegistry } from "./run-secret-redaction.js"; import { instanceSettingsService } from "./instance-settings.js"; import { redactCurrentUserText } from "../log-redaction.js"; import { logger } from "../middleware/logger.js"; +import { canonicalExecutionUuid } from "./execution-identity.js"; +import { chatActionDeadline } from "./chat-action-work.js"; type ProjectionIssue = Pick; type ProjectionRun = Pick & Partial>; -type ProjectionWait = { id: string; requestedAt: Date; sourceRunId?: string | null; executionWaitReason?: string | null }; +type ProjectionWait = { id: string; requestedAt: Date; sourceRunId?: string | null; executionWaitReason?: string | null; + preparationStatus?: string | null; preparationRetryAt?: number | string | null }; type ProjectionState = { run?: ProjectionRun; action?: typeof issueRecoveryActions.$inferSelect; @@ -105,6 +108,12 @@ export function projectTaskInterruption(issue: ProjectionIssue, input: Projectio if (!ready) return { ...base, responsibility: "user", state: "waiting_for_user", userAction: "project_access", projectAccess }; } + if (retryWait?.sourceRunId === run.id && retryWait.preparationStatus) { + const deadline = retryWait.preparationRetryAt == null ? NaN : Number(retryWait.preparationRetryAt); + if (["issued", "processing"].includes(retryWait.preparationStatus) && Number.isFinite(deadline)) + return { ...base, state: "retry_scheduled", nextRetryAt: new Date(deadline).toISOString() }; + return { ...base, state: "failed", canRetry: retryWait.preparationStatus !== "cancelled" }; + } const retryAt = system?.nextAttemptAt ?? run.scheduledRetryAt?.getTime(); if (retryAt && retryAt > now) return { ...base, state: "retry_scheduled", nextRetryAt: new Date(retryAt).toISOString(), canRetry: true }; @@ -181,8 +190,16 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i // A new message must not hide an existing Retry's admission explanation. db.selectDistinctOn([wakeIssue, isRetryWait], { issueId: wakeIssue, id: agentWakeupRequests.id, requestedAt: agentWakeupRequests.requestedAt, sourceRunId: boundedJsonString(sql`${agentWakeupRequests.payload}#>'{taskRetry,sourceRunId}'`, 160), - executionWaitReason: boundedJsonString(sql`${agentWakeupRequests.payload}#>'{executionWait,reason}'`, 80) }) - .from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), inArray(wakeIssue, ids), + executionWaitReason: boundedJsonString(sql`${agentWakeupRequests.payload}#>'{executionWait,reason}'`, 80), + preparationStatus: chatActions.status, preparationRetryAt: chatActionDeadline("failed_run_retry") }) + .from(agentWakeupRequests).leftJoin(chatActions, and(eq(chatActions.companyId, agentWakeupRequests.companyId), + eq(chatActions.id, canonicalExecutionUuid(sql`${agentWakeupRequests.payload}#>>'{taskRetry,chatActionId}'`)), + eq(chatActions.kind, "failed_run_retry"), sql`${chatActions.payload}->>'taskRetryIntentId' = ${agentWakeupRequests.id}::text`, + sql`${chatActions.payload}->>'issueId' = ${wakeIssue}`, + sql`${chatActions.payload}->>'agentId' = ${agentWakeupRequests.agentId}::text`, + sql`${chatActions.payload}->>'failedRunId' = ${agentWakeupRequests.payload}#>>'{taskRetry,sourceRunId}'`, + inArray(chatActions.status, ["issued", "processing", "failed", "cancelled"]))) + .where(and(eq(agentWakeupRequests.companyId, companyId), inArray(wakeIssue, ids), eq(agentWakeupRequests.status, "deferred_issue_execution"), eq(agentWakeupRequests.requestedByActorType, "user"))) .orderBy(wakeIssue, isRetryWait, desc(agentWakeupRequests.requestedAt), desc(agentWakeupRequests.id)), agentIds.length ? db.select({ id: agents.id, status: agents.status, pauseReason: agents.pauseReason }).from(agents)