diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 485cb5e9ad1..dc423774d43 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1582,6 +1582,14 @@ delta when attaching a new run. Other stale-event and authority checks remain. ### Explicit user continuation after execution failure +A deferred chat Retry retains its authorized source batch when it resumes. +Separate board queues do not join that batch. For a historical mixed batch +refused before provider startup, a new task Retry can restore the undelivered +receipts only after certifying the exact bootstrap failure, source authority, +unchanged board comments and absence of execution, allocation or prior delivery. +The restoration is audited. Historical runs and budget-skipped receipts remain +unchanged; each input still passes its ordinary admission and dispatch checks. + When a user's comment also assigns the task back to its agent, the assignment wake retains that comment's continuation intent. Admission validates the saved comment, its author, task, timing and delivery receipt through the same checks as diff --git a/doc/task-recovery.md b/doc/task-recovery.md index cede3a25dc2..22ad527af6e 100644 --- a/doc/task-recovery.md +++ b/doc/task-recovery.md @@ -50,6 +50,14 @@ current access, task eligibility, pause and budget gates must still pass before one successor starts. Restarts reuse the receipt; they do not create another Retry generation or another external publication. +Resuming a chat Retry preserves its exact authorized input batch. Independent +board messages keep their own queue receipts and enter a separate turn. If an +older queue merged those messages and dispatch refused the batch before any +provider work, a new deliberate Retry can restore the undelivered receipts. +Restoration requires the exact bootstrap failure, no execution or allocation, +current chat authority, unchanged board input and no prior delivery. It records +an audit event and retains the failed run and original skipped receipts. + Several retained recovery holds can describe the same failed execution. A historical hold without a system recovery record uses the current exact Retry's workspace strategy for that run. It does not borrow another run's strategy or diff --git a/maps/agent-access-map/chat-channels.md b/maps/agent-access-map/chat-channels.md index ce1399bb7b4..b023fd376f7 100644 --- a/maps/agent-access-map/chat-channels.md +++ b/maps/agent-access-map/chat-channels.md @@ -1019,3 +1019,10 @@ comment, session generation, and admitted instruction snapshot. A separate retry receipt preserves the original skipped receipt and older failed native result. Current access and normal native stop/cleanup guards still govern execution; board wake metadata alone cannot manufacture chat authority or replay old work. + +Deferred chat Retry excludes independent board queues from its pinned batch. +An explicit board Retry can restore a historical mixed batch only with exact +never-started evidence, current source authority, unchanged board comments and +no prior delivery or allocation. The server records the restoration in activity; +failed runs and original skipped receipts remain history. This grants no new +agent retry capability and does not weaken execution-time source validation. diff --git a/maps/feature-map/chat-channels.md b/maps/feature-map/chat-channels.md index 41b6bb81dc2..414af23e762 100644 --- a/maps/feature-map/chat-channels.md +++ b/maps/feature-map/chat-channels.md @@ -272,6 +272,13 @@ After successful admission, reload and verify that the prior skipped receipt doe not strand a later real provider reply or tree resume. Provider fixtures support this behavior; live fixed-source runtime and destination proof remain separate. +Also defer Retry behind cleanup, queue an independent board message, then +complete cleanup. Verify one run receives the exact GitHub batch and the board +message enters its own turn. For an older mixed batch rejected before provider +startup, ordinary task Retry must recover only certified undelivered receipts. +Changed input, active ownership, allocations and prior delivery must keep the +refusal effective. Check the restoration audit and unchanged failed-run history. + ### `delegated-configuration` Endpoint activity hides private-task events while retaining a next-page cursor across private-only windows, so older readable history stays reachable. Actual-JWT fixtures verify this separately from live provider delivery. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index d8d9720f2ca..6226c069956 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -61898,9 +61898,10 @@ 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_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_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"); const legacyIngress = mode.startsWith("legacy_"); const realBudgetIngress = mode === "real_budget_ingress" || legacyIngress; const executionRevalidation = mode.startsWith("legacy_execution"); @@ -61983,6 +61984,11 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { endpointId: context.endpoint.id, provider, thread: context.thread.thread, message: makeMessage({ id: "second-saved-after-budget-hold", text: "@maya include the follow-up", userId: "U-SAFE-PROGRESS", mentioned: true }), trigger: "mention" }); + if (queuedBoard) for (let index = 1; index < 10; index++) await deliverMessage({ + callbacks: context.runtime.configurations.get(context.endpoint.id)!.callbacks, + endpointId: context.endpoint.id, provider, thread: context.thread.thread, + message: makeMessage({ id: `saved-followup-${index}`, text: `@maya review follow-up ${index}`, + userId: "U-SAFE-PROGRESS", mentioned: true }), trigger: "mention" }); const savedInputs = await db.select().from(chatActions).where(and( eq(chatActions.companyId, companyId), eq(chatActions.conversationId, context.conversation.id), eq(chatActions.kind, "inbound_wakeup"), sql`${chatActions.id} <> ${context.action.id}`)) @@ -62008,7 +62014,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(await unadmitted()).toHaveLength(1); // The budget was cleared. Retry must reconsider this independent saved // request through chat admission, not replay the older failed turn. - const held = mode === "unreconciled" || mode === "committed_delayed_cleanup" || mode === "committed_unknown_cleanup" || mode === "committed_live_controller"; + const held = mode === "unreconciled" || mode === "committed_delayed_cleanup" || queuedBoard || mode === "committed_unknown_cleanup" || mode === "committed_live_controller"; if (mode === "unreconciled" || mode.startsWith("committed_") || executionRevalidation) { await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: context.issue.id, nativePhase: "terminal_failure", runnerProfileJson: {} }).where(eq(heartbeatRuns.id, context.runId)); @@ -62035,7 +62041,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { provider: "daytona", providerLeaseId: "owned-saved-input-sandbox" }; await db.insert(environmentLeases).values({ ...lease, issueId: context.issue.id, status: "expired", cleanupStatus: "success", releasedAt: new Date(), leasePolicy: "ephemeral", - metadata: mode === "committed_unknown_cleanup" || mode === "committed_delayed_cleanup" ? {} : { + metadata: mode === "committed_unknown_cleanup" || mode === "committed_delayed_cleanup" || queuedBoard ? {} : { remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "destroyed" }) } }); } } @@ -62309,18 +62315,169 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect((await db.select().from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, context.issue.id))))) .toEqual(expect.arrayContaining([expect.objectContaining({ status: "active" })])); - if (mode === "committed_delayed_cleanup") { + if (mode === "committed_delayed_cleanup" || queuedBoard) { + const [boardComment] = queuedBoard ? await db.insert(issueComments).values({ + companyId, issueId: context.issue.id, authorType: "user", authorUserId: "owner-user", + body: "Keep this independent continuation for its own authorized turn.", + }).returning() : []; + const [boardWake] = boardComment ? await db.insert(agentWakeupRequests).values({ companyId, + agentId: context.fixture.assignedAgentId, source: "automation", triggerDetail: "system", + reason: "issue_commented", status: "deferred_issue_execution", + requestedByActorType: "user", requestedByActorId: "owner-user", + payload: { issueId: context.issue.id, commentId: boardComment.id, wakeCommentIds: [boardComment.id], + _paperclipWakeContext: { issueId: context.issue.id, taskKey: context.issue.id, + source: "issue.comment.reopen", wakeReason: "issue_commented", + wakeCommentId: boardComment.id, wakeCommentIds: [boardComment.id] } }, + }).returning() : []; const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, context.runId)); await db.update(environmentLeases).set({ metadata: { ...lease.metadata, remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "destroyed" }) }, }).where(eq(environmentLeases.id, lease.id)); await reconcileTaskChatRetryReceipts(db); await context.service.processFailedChatRunRetry(receipt.chatActionId!); - const [resumed] = await readAgentWakeupRequestsById(receipt.chatActionId!); + let [resumed] = await readAgentWakeupRequestsById(receipt.chatActionId!); expect(resumed).toMatchObject({ id: admitted.id, status: "queued" }); expect(resumed.runId).toBeTruthy(); expect((await readHeartbeatRunsById(resumed.runId!))[0]).toMatchObject({ status: "queued", contextSnapshot: { wakeCommentId: saved.payload.commentId } }); + if (boardWake) { + const heartbeat = heartbeatService(db); + if (mode.startsWith("committed_board_existing")) { + // Persist the historical producer's merged batch, then let real + // dispatch reject it before any identity or provider starts. + const [queued] = await readHeartbeatRunsById(resumed.runId!); + if (mode === "committed_board_existing_coalesced") { + await db.update(chatActions).set({ status: "processed" }).where(inArray(chatActions.id, savedInputs.map(source => source.id))); + await db.update(agentWakeupRequests).set({ status: "coalesced", runId: queued.id, + payload: sql`${agentWakeupRequests.payload} || ${JSON.stringify({ coalescedIntoWakeupRequestId: resumed.id })}::jsonb`, + }).where(inArray(agentWakeupRequests.id, savedInputs.map(source => source.id))); + } + const mixedIds = [boardComment!.id, ...expectedComments]; + // Replaying this test with the queue fix reverted uses the old + // producer's actual persisted merge. With the fixed producer, + // retain historical state using its shared serializers. + if (!(queued.contextSnapshot?.wakeCommentIds as string[])?.includes(boardComment!.id)) { + const { withQueuedCommentIdsInRunContext, withQueuedCommentIdsInWakePayload } = + await import("../services/issue-queued-comment-queue.js"); + await db.update(heartbeatRuns).set({ contextSnapshot: + withQueuedCommentIdsInRunContext(queued.contextSnapshot, mixedIds, [boardWake.id]) }) + .where(eq(heartbeatRuns.id, queued.id)); + await db.update(agentWakeupRequests).set({ payload: + withQueuedCommentIdsInWakePayload(resumed.payload, mixedIds) }) + .where(eq(agentWakeupRequests.id, resumed.id)); + await db.update(agentWakeupRequests).set({ status: "coalesced", runId: queued.id, + finishedAt: new Date() }).where(eq(agentWakeupRequests.id, boardWake.id)); + } + expect((await readHeartbeatRunsById(queued.id))[0].contextSnapshot).toMatchObject({ + wakeCommentId: expectedComments.at(-1), wakeCommentIds: mixedIds, + queuedCommentSourceWakeIds: [boardWake.id], + }); + expect((await readAgentWakeupRequestsById(boardWake.id))[0]).toMatchObject({ + status: "coalesced", runId: queued.id, payload: { issueId: context.issue.id }, + }); + expect((await readAgentWakeupRequestsById(boardWake.id))[0].payload) + .not.toHaveProperty("coalescedIntoWakeupRequestId"); + await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(and( + eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "running"), isNull(heartbeatRuns.wakeupRequestId))); + await db.update(agents).set({ status: "idle" }).where(eq(agents.id, context.fixture.assignedAgentId)); + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + const [failed] = await readHeartbeatRunsById(queued.id); + expect(failed).toMatchObject({ status: "failed", errorCode: "chat_failed_run_retry_not_authorized", + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }); + await db.insert(heartbeatRuns).values({ companyId, agentId: context.fixture.assignedAgentId, + status: "running", startedAt: new Date() }); + await db.update(agents).set({ status: "running" }).where(eq(agents.id, context.fixture.assignedAgentId)); + const nextInput = { ...request, observedRunId: failed.id, idempotencyKey: randomUUID() }; + if (mode.endsWith("source_edited")) await db.update(issueComments).set({ body: "Changed external source" }) + .where(eq(issueComments.id, String(saved.payload.commentId))); + if (mode.endsWith("board_edited")) await db.update(issueComments).set({ body: "Changed board source", + updatedAt: new Date(boardComment!.createdAt.getTime() + 1) }) + .where(eq(issueComments.id, boardComment!.id)); + if (mode.endsWith("controller")) await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(heartbeatRuns.id, failed.id)); + if (mode.endsWith("allocation")) await db.insert(environmentLeases).values({ companyId, + heartbeatRunId: failed.id, issueId: context.issue.id, provider: "daytona", providerLeaseId: "unknown-executor", + status: "active", leasePolicy: "ephemeral" }); + if (mode.endsWith("receipt")) await db.update(agentWakeupRequests).set({ status: "claimed" }) + .where(eq(agentWakeupRequests.id, boardWake.id)); + if (mode.endsWith("source_owner")) await db.update(agentWakeupRequests).set({ + status: "coalesced", runId: context.runId, + payload: sql`${agentWakeupRequests.payload} || ${JSON.stringify({ coalescedIntoWakeupRequestId: resumed.id })}::jsonb`, + }).where(eq(agentWakeupRequests.id, saved.id)); + if (mode.endsWith("delivered")) await db.insert(heartbeatRuns).values({ companyId, + agentId: context.fixture.assignedAgentId, status: "succeeded", finishedAt: new Date(), + createdAt: new Date(failed.createdAt.getTime() - 1), + contextSnapshot: { issueId: context.issue.id, wakeCommentIds: + [mode.endsWith("source_delivered") ? saved.payload.commentId : boardComment!.id] }, + }); + 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") { + await expect(requestTaskRetry(nextInput)).rejects.toMatchObject({ + details: { code: "chat_failed_run_retry_not_authorized" }, + }); + expect(await db.select().from(agentWakeupRequests).where(inArray(agentWakeupRequests.id, inputIds)).orderBy(asc(agentWakeupRequests.id))) + .toEqual(before); + return; + } + 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: context.service })); + app.use(errorHandler); + const httpRequest = (await import("supertest")).default; + 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; + expect(next).toMatchObject({ accepted: true }); + expect((await retryHttp()).body.requestId).toBe(next.requestId); + 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(); + if (mode === "committed_board_existing_coalesced") { + const restored = await db.select().from(agentWakeupRequests) + .where(inArray(agentWakeupRequests.id, savedInputs.map(source => source.id))); + expect(restored).toHaveLength(10); + expect(restored.every(wake => wake.status === "coalesced" && wake.runId === resumed.runId)).toBe(true); + } + expect((await readHeartbeatRunsById(failed.id))[0].contextSnapshot).toEqual(failed.contextSnapshot); + } + const realInitialize = runIdentityModule.initializeRunIdentity; + const initialized: Array<{ runId: string; messageId: string | null }> = []; + expect((await readAgentWakeupRequestsById(boardWake.id))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + const probe = vi.spyOn(runIdentityModule, "initializeRunIdentity").mockImplementation(async (database, input) => { + const identity = await realInitialize(database, input); + if (input.companyId !== companyId) return identity; + initialized.push({ runId: input.runId, messageId: identity.messageId }); + throw new Error("certified_batch_identity_probe_complete"); + }); + try { + await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(and( + eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "running"), isNull(heartbeatRuns.wakeupRequestId))); + await db.update(agents).set({ status: "idle" }).where(eq(agents.id, context.fixture.assignedAgentId)); + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + const [successor] = await readHeartbeatRunsById(resumed.runId!); + expect(successor).toMatchObject({ error: "certified_batch_identity_probe_complete", + contextSnapshot: { wakeCommentIds: expectedComments } }); + expect(successor.contextSnapshot?.queuedCommentSourceWakeIds).toBeUndefined(); + const [boardReceipt] = await readAgentWakeupRequestsById(boardWake.id); + expect(boardReceipt.runId).toBeTruthy(); + expect(boardReceipt.runId).not.toBe(resumed.runId); + expect(initialized).toEqual([ + { runId: resumed.runId, messageId: saved.payload.commentId }, + { runId: boardReceipt.runId, messageId: boardComment!.id }, + ]); + expect((await readHeartbeatRunsById(boardReceipt.runId!))[0]).toMatchObject({ + contextSnapshot: { wakeCommentIds: [boardComment!.id] }, + }); + } finally { probe.mockRestore(); await heartbeat.drainActiveRunExecutions(); } + } await context.service.processFailedChatRunRetry(receipt.chatActionId!); await reconcileTaskChatRetryReceipts(db); expect((await readHeartbeatRunsByCompanyId(companyId)).filter(run => @@ -62354,7 +62511,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } } expect((await readAgentWakeupRequestsById(saved.id))[0]).toMatchObject(mode === "source_claimed" - ? { status: "queued", runId: context.runId } : { status: "skipped", runId: null }); + ? { status: "queued", runId: context.runId } : mode === "committed_board_existing_coalesced" + ? { status: "coalesced", runId: expect.any(String) } : { status: "skipped", runId: null }); expect((await readHeartbeatRunsById(context.runId))[0]).toMatchObject({ status: "failed" }); expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(previousPublications); } finally { await context.service.shutdown(); } diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index 0792b140497..cd5cf7c0bf5 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -7,6 +7,7 @@ import { notifyChatActionWork, notifyChatPublicationWork } from "../chat-work-no 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 type { TaskRetryWakeAuthority } from "./github/task-retry-authorization.js"; import { @@ -76,6 +77,7 @@ import { import { logActivity, + persistActivity, } from "../activity-log.js"; import { type IssueAssignmentWakeupDeps, @@ -1361,6 +1363,82 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti } }); + /** A historical queue merge can consume independent board input before chat + * dispatch rejects the changed batch. A new board Retry may restore those + * never-delivered receipts, after certifying both their original authorities. + * 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; + }) { + const [run] = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId), + eq(heartbeatRuns.id, input.failedRunId), + )).for("share", { noWait: true }); + const queueIds = run?.contextSnapshot?.queuedCommentSourceWakeIds; + if (run?.errorCode !== "chat_failed_run_retry_not_authorized" || !Array.isArray(queueIds) || !queueIds.length) return; + if (queueIds.length > 50 || queueIds.some(id => typeof id !== "string" || !isUuidLike(id)) || + new Set(queueIds).size !== queueIds.length || !await isPreProviderChatFailure(tx, run)) throw failedChatRetryDenied(); + const source = await authorizeFailedChatRunRetry(tx, { ...input, retryActionId: run.wakeupRequestId!, runId: run.id }); + const sourceIds = source.taskRetryInput?.sourceActionIds; + if (source.provider !== "github" || !sourceIds?.length || !source.intent.payload.taskRetryIntentId) + throw failedChatRetryDenied(); + const queues = await tx.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.agentId, input.agentId), + inArray(agentWakeupRequests.id, queueIds as string[]), + )).for("update"); + const byId = new Map(queues.map(queue => [queue.id, queue])); + if (queues.length !== queueIds.length) throw failedChatRetryDenied(); + const boardIds = (queueIds as string[]).flatMap(id => { + const queue = byId.get(id)!; + const ids = queuedCommentIdsFromWakePayload(queue.payload); + if (queue.status !== "coalesced" || queue.runId !== run.id || queue.payload?.issueId !== input.issueId || + queue.requestedByActorType !== "user" || !queue.requestedByActorId || + !["issue_commented", "issue_reopened_via_comment"].includes(queue.reason ?? "") || + queue.idempotencyKey?.startsWith("chat-inbound:") || queue.payload?.queuedCommentInterrupt || + queue.payload?.coalescedIntoWakeupRequestId || !ids.length || ids.some(id => !isUuidLike(id))) + throw failedChatRetryDenied(); + return ids; + }); + if (new Set(boardIds).size !== boardIds.length || boardIds.some(id => source.commentIds.includes(id)) || + JSON.stringify(run.contextSnapshot?.wakeCommentIds) !== JSON.stringify([...boardIds, ...source.commentIds]) || + run.contextSnapshot?.wakeCommentId !== source.commentIds.at(-1)) throw failedChatRetryDenied(); + const comments = await tx.select().from(issueComments).where(and( + eq(issueComments.companyId, input.companyId), eq(issueComments.issueId, input.issueId), + inArray(issueComments.id, boardIds), + )).for("share", { noWait: true }); + const commentsById = new Map(comments.map(comment => [comment.id, comment])); + for (const queue of queues) for (const id of queuedCommentIdsFromWakePayload(queue.payload)) { + const comment = commentsById.get(id); + if (!comment || comment.authorType !== "user" || comment.authorUserId !== queue.requestedByActorId || + comment.createdByRunId || comment.deletedAt || !comment.body.trim() || + comment.updatedAt.getTime() !== comment.createdAt.getTime()) throw failedChatRetryDenied(); + } + const [delivered] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId), ne(heartbeatRuns.id, run.id), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${input.issueId}`, + sql`not (${heartbeatRuns.status} = 'cancelled' and ${heartbeatRuns.startedAt} is null)`, + // Budget-skipped source receipts remain immutable even after a later + // Retry delivers them. Check run history for both input authorities. + or(...[...boardIds, ...source.commentIds].map(id => or(sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId' = ${id}`, + sql`${heartbeatRuns.contextSnapshot}->'wakeCommentIds' @> ${JSON.stringify([id])}::jsonb`))), + )).limit(1); + if (delivered) throw failedChatRetryDenied(); + // Source authorization above checked every external receipt and its owner. + // Reset only this certified unstarted batch under the task's admission lock. + const skippedIds = source.taskRetryInput?.skippedActionIds ?? []; + await tx.update(agentWakeupRequests).set({ status: "deferred_issue_execution", runId: null, finishedAt: null, + payload: sql`${agentWakeupRequests.payload} - 'coalescedIntoWakeupRequestId'`, updatedAt: new Date(), + }).where(and(eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.agentId, input.agentId), + inArray(agentWakeupRequests.id, [...sourceIds, ...queueIds as string[]]), + ne(agentWakeupRequests.status, "skipped"))); + await persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "user", actorId: input.initiatedByUserId, + action: "issue.retry_inputs_restored", entityType: "issue", entityId: input.issueId, runId: run.id, + details: { sourceRetryActionId: source.intent.id, externalRequestIds: sourceIds, boardRequestIds: queueIds } }); + // The old skipped receipts stay unchanged. The certified failed attempt + // proves this batch was never delivered even though it predates that run. + return { sourceActionIds: sourceIds, skippedActionIds: skippedIds } satisfies TaskRetryInput; + } + async function prepareFailedChatRunRetry( tx: DbOrTransaction, input: { @@ -1372,7 +1450,9 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti retryIntentId?: string; }, ): Promise<{ actionId: string; issueId: string }> { - const taskRetryInput = input.retryIntentId ? await selectTaskRetryInput(tx, input) : undefined; + 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 }); const providerActionId = input.retryIntentId ? `task-retry:${input.retryIntentId}` : `failed-run-retry:${source.failedRunId}`; diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 17590f7be12..9cfffbc857f 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -4247,7 +4247,10 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende inArray(agentWakeupRequests.agentId, handoffSource ? [agentId, handoffSource.agentId] : [agentId]), eq(agentWakeupRequests.status, "deferred_issue_execution"), sql`${agentWakeupRequests.payload}->>'issueId' = ${issue.id}`, - canCoalesceComments || opts.taskRetryRequestId ? undefined : eq(agentWakeupRequests.id, executionWaitRequestId!), + // Chat Retry already pins its complete authorized source + // batch. Keep separate board input for its own admission. + canCoalesceComments || (opts.taskRetryRequestId && !failedChatRetry) + ? undefined : eq(agentWakeupRequests.id, executionWaitRequestId!), ), ) .orderBy(asc(agentWakeupRequests.requestedAt))