diff --git a/doc/connections/GITHUB.md b/doc/connections/GITHUB.md index 74279d8fd5f..90f206b0a17 100644 --- a/doc/connections/GITHUB.md +++ b/doc/connections/GITHUB.md @@ -386,10 +386,15 @@ exact executor termination proof. Successful results and uncertain delivery do not grant a replay. A certified failed-chat Retry retains the exact accepted message identity when its predecessor stopped before identity initialization, so a published review can expose its authorized passive response wait. -If an inherited task-resume attempt already recorded a result, ordinary Retry -explains that exact replay is unavailable and directs the user to send a new -message to continue from saved work. That new turn still passes normal current -requester, access and executor-containment checks. +When the exact failed GitHub request cannot be repeated (for example, the +attempt already recorded a result, its output was already published, the run +did not start from a GitHub message, or the requester's GitHub access changed), +a person's Retry continues the task from its saved work instead, the same way a +new message on the task would. That continuation runs as the person who clicked +Retry in a fresh session, never with the GitHub requester's authority, and never +repeats the original request or its output. It still passes the normal current +access and executor-containment checks. Automatic Retry on a person's behalf +does not take this path; it continues only with the exact GitHub request. If that successor failed at Retry authorization before runtime or provider startup, the next normal Retry follows its canonical source receipt. The exact diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index ac155e36497..b83ae20c147 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1611,6 +1611,10 @@ 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. +### Connected task Retry without an exact chat replay + +A person's task Retry on a connected task first asks the chat retry owner to repeat the exact failed chat request. When the latest run is that person's own saved board turn that cannot be certified for replay, when no chat retry owner is available, or when that owner refuses (`chat_failed_run_retry_not_authorized`), the refusal and any writes it made are rolled back and the same Retry receipt becomes an ordinary board continuation: the clicking person is the responsible user, the run starts a fresh session from saved work, and no chat action, chat authority, or output replay is created. This is the same authority as a new message from that person. When the conversation still holds saved or unadmitted chat input for the task, Retry keeps the exact-replay path and its refusal, because that input continues through its own admission. Dispatch keeps every normal gate (current task access, executor containment, recovery holds, unadmitted chat input). Projection and the route therefore agree: a shown Retry on a connected task is accepted. Retry that Paperclip requests by itself on a person's behalf (restart recovery, provider limit reset) keeps the exact-replay requirement and still settles on the refusal. A budget resume continues the person's own Retry, so it takes the board continuation too. No migration is needed; tasks blocked by an earlier release become retryable at the next click. + ### Saved chat Retry preparation Task Retry can meet a busy chat identity or endpoint lock before its source is authorized. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index edee564c8b5..dd651dcacba 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -65471,9 +65471,14 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect((await readHeartbeatRunsByCompanyId(companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(1); } else { expect([202, 409], JSON.stringify(response.body)).toContain(response.status); - if (response.status === 202) { + if (response.status === 202 && response.body.chatActionId) { const outcome = await f.service.processFailedChatRunRetry(response.body.chatActionId); expect(outcome.status).not.toBe("queued"); + } else if (response.status === 202) { + const [intent] = await readAgentWakeupRequestsById(response.body.requestId); + expect(intent).toMatchObject({ reason: "task_retry", requestedByActorId: "owner-user", + payload: { issueId, taskRetry: { sourceRunId: f.runId } } }); + expect((intent.payload!.taskRetry as Record).chatActionId).toBeUndefined(); } expect((await readHeartbeatRunsByCompanyId(companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(0); } @@ -66290,7 +66295,12 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } const response = await send(); if (["principal_revoked", "accepted_result"].includes(mode)) { - expect(response.status).toBe(409); + expect(response.status, JSON.stringify(response.body)).toBe(202); + expect(response.body.chatActionId).toBeUndefined(); + const [intent] = await readAgentWakeupRequestsById(response.body.requestId); + expect(intent).toMatchObject({ reason: "task_retry", requestedByActorId: "owner-user", + payload: { issueId, taskRetry: { sourceRunId: f.runId } } }); + expect((intent.payload!.taskRetry as Record).chatActionId).toBeUndefined(); } else { expect(response.status).toBe(202); if (mode === "source_edited") await db.update(issueComments).set({ body: "Changed after Retry acceptance" }).where(eq(issueComments.id, String(f.action.payload.commentId))); @@ -66378,11 +66388,26 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const denySource = async (apply: () => Promise, restore: () => Promise, stale = false) => { try { await apply(); - const denied = await request(app).post(`/api/issues/${issueId}/retry`).send({ idempotencyKey: randomUUID(), observedRunId: current!.id }).expect(stale ? 200 : 409); - if (stale) expect(denied.body).toMatchObject({ accepted: false, stale: true }); - expect(await db.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and( - eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "task_retry")))).toHaveLength(0); - } finally { await restore(); } + const denied = await request(app).post(`/api/issues/${issueId}/retry`).send({ idempotencyKey: randomUUID(), observedRunId: current!.id }).expect(stale ? 200 : 202); + const retries = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "task_retry"))); + if (stale) { + expect(denied.body).toMatchObject({ accepted: false, stale: true }); + expect(retries).toHaveLength(0); + } else { + expect(denied.body.chatActionId).toBeUndefined(); + expect(retries).toHaveLength(1); + expect(retries[0]).toMatchObject({ id: denied.body.requestId, requestedByActorId: "owner-user", + payload: { issueId, taskRetry: { sourceRunId: current!.id } } }); + const taskRetry = retries[0].payload!.taskRetry as Record; + expect(taskRetry.boardMessageOrigin).toBeUndefined(); + expect(taskRetry.chatActionId).toBeUndefined(); + } + } finally { + await db.delete(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.reason, "task_retry"))); + await restore(); + } }; await denySource(() => db.update(issueComments).set({ body: "Edited after acceptance", updatedAt: new Date() }).where(eq(issueComments.id, comment.id)), () => db.update(issueComments).set({ body: comment.body, updatedAt: sql`${issueComments.createdAt}` }).where(eq(issueComments.id, comment.id))); @@ -66493,17 +66518,26 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const [firstRetry] = await readHeartbeatRunsById(successor.id); const nextRequest = () => request(app).post(`/api/issues/${issueId}/retry`).send({ idempotencyKey: randomUUID(), observedRunId: firstRetry.id }); + const continueWithoutReplay = async () => { + const continued = await nextRequest().expect(202); + expect(continued.body.chatActionId).toBeUndefined(); + const [intent] = await readAgentWakeupRequestsById(continued.body.requestId); + expect(intent).toMatchObject({ reason: "task_retry", requestedByActorId: "owner-user", + payload: { issueId, taskRetry: { sourceRunId: firstRetry.id } } }); + expect((intent.payload!.taskRetry as Record).boardMessageOrigin).toBeUndefined(); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, intent.id)); + }; await db.update(heartbeatRuns).set({ sessionIdAfter: "contradictory-provider-session" }).where(eq(heartbeatRuns.id, firstRetry.id)); - await nextRequest().expect(409); + await continueWithoutReplay(); await db.update(heartbeatRuns).set({ sessionIdAfter: null }).where(eq(heartbeatRuns.id, firstRetry.id)); await db.update(issueComments).set({ body: "Changed original accepted message", updatedAt: new Date() }).where(eq(issueComments.id, comment.id)); - await nextRequest().expect(409); + await continueWithoutReplay(); await db.update(issueComments).set({ body: comment.body, updatedAt: sql`${issueComments.createdAt}` }).where(eq(issueComments.id, comment.id)); const [firstIntent] = await readAgentWakeupRequestsById(retry.id); await db.update(agentWakeupRequests).set({ payload: { ...firstIntent.payload, taskRetry: { ...(firstIntent.payload!.taskRetry as object), sourceRunId: firstRetry.id } } }).where(eq(agentWakeupRequests.id, firstIntent.id)); await db.update(heartbeatRuns).set({ contextSnapshot: { ...firstRetry.contextSnapshot, previousRunId: firstRetry.id } }).where(eq(heartbeatRuns.id, firstRetry.id)); - await nextRequest().expect(409); + await continueWithoutReplay(); await db.update(agentWakeupRequests).set({ payload: firstIntent.payload }).where(eq(agentWakeupRequests.id, firstIntent.id)); await db.update(heartbeatRuns).set({ contextSnapshot: firstRetry.contextSnapshot }).where(eq(heartbeatRuns.id, firstRetry.id)); const second = await nextRequest().expect(202); @@ -66544,7 +66578,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } }); - it("distinguishes an inherited GitHub requester from permission to replay a committed provider failure", async () => { + it("continues an inherited GitHub task from its recorded result instead of replaying the committed provider failure", async () => { const f = await failedChatRetryFixture("github", "42", true, false, true); try { const companyId = f.fixture.companyId, agentId = f.fixture.assignedAgentId, issueId = f.issue.id; @@ -66593,19 +66627,30 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { app.use((req, _res, next) => { req.actor = boardActor(companyId); next(); }); app.use("/api", issueRoutes(db, createStorageService().storage, { chatRunRetries: f.service })); app.use(errorHandler); - const denied = await request(app).post(`/api/issues/${issueId}/retry`).send({ + const continued = await request(app).post(`/api/issues/${issueId}/retry`).send({ idempotencyKey: input.idempotencyKey, observedRunId: input.observedRunId, - }).expect(409); - expect(denied.body).toMatchObject({ error: "This attempt already recorded a result. Send a new message on the task to continue from the saved work.", - details: { code: "chat_failed_run_retry_not_authorized", reason: "accepted_result_without_new_input" } }); - // Isolate the later accepted-result guard using only owned fixture data. - // Resolving inherited provenance cannot by itself authorize replay. + }).expect(202); + expect(continued.body).toMatchObject({ accepted: true, stale: false, observedRunId: resumed!.id }); + expect(continued.body.chatActionId).toBeUndefined(); + const assertBoardContinuation = async (requestId: string) => { + const [intent] = await readAgentWakeupRequestsById(requestId); + expect(intent).toMatchObject({ reason: "task_retry", requestedByActorType: "user", requestedByActorId: "owner-user", + payload: { issueId, taskRetry: { sourceRunId: resumed!.id, actorId: "owner-user" } } }); + expect((intent.payload!.taskRetry as Record).chatActionId).toBeUndefined(); + expect(await db.select().from(chatActions).where(and(eq(chatActions.companyId, companyId), + eq(chatActions.kind, "failed_run_retry")))).toHaveLength(0); + await db.update(agentWakeupRequests).set({ status: "failed", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, requestId)); + }; + await assertBoardContinuation(continued.body.requestId); + // The later accepted-result guard also continues from the saved result + // instead of repeating the exact chat request. await db.update(heartbeatRuns).set({ contextSnapshot: { ...failedResume.contextSnapshot, source: "chat:github", wakeCommentId: f.action.payload.commentId, wakeCommentIds: [f.action.payload.commentId] } }) .where(eq(heartbeatRuns.id, failedResume.id)); try { - await expect(requestTaskRetry({ ...input, idempotencyKey: randomUUID() })).rejects.toMatchObject({ details: { - code: "chat_failed_run_retry_not_authorized", reason: "accepted_result_without_new_input" } }); + const receipt = await requestTaskRetry({ ...input, idempotencyKey: randomUUID() }); + expect(receipt).toMatchObject({ accepted: true, chatActionId: undefined }); + await assertBoardContinuation(receipt.requestId!); } finally { await db.update(heartbeatRuns).set({ contextSnapshot: failedResume.contextSnapshot }).where(eq(heartbeatRuns.id, failedResume.id)); } expect((await readHeartbeatRunsByCompanyId(companyId)).filter(run => run.retryOfRunId === resumed!.id)).toHaveLength(0); } finally { await f.service.shutdown(); } @@ -67164,9 +67209,12 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } 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" }, - }); + const continued = await requestTaskRetry(nextInput); + expect(continued).toMatchObject({ accepted: true, chatActionId: undefined }); + const [intent] = await readAgentWakeupRequestsById(continued.requestId!); + expect(intent).toMatchObject({ reason: "task_retry", requestedByActorId: "owner-user", + payload: { taskRetry: { sourceRunId: failed.id } } }); + expect((intent.payload!.taskRetry as Record).chatActionId).toBeUndefined(); expect(await db.select().from(agentWakeupRequests).where(inArray(agentWakeupRequests.id, inputIds)).orderBy(asc(agentWakeupRequests.id))) .toEqual(before); return; diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index 38e62ea7157..a4fa493a857 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -102,6 +102,9 @@ 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; +const FAILED_CHAT_RETRY_DIAGNOSTIC = "This chat request can't be repeated exactly as it was sent. Retry continues the task from its saved work instead."; +const FAILED_CHAT_RETRY_DIAGNOSTICS = [FAILED_CHAT_RETRY_DIAGNOSTIC, + "The exact failed chat request can no longer be retried safely. Check its current connection, access, and recovery state."]; export type ChatFailedRetrySource = Awaited["failedChatRetrySource"]>>; type ChatFailedRunRetryOptions = { @@ -139,8 +142,8 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti "prior_result_publication" | "prior_interaction" | "source_authorization", authorityReason?: ChatWakeAuthorityReason) => conflict( reason === "accepted_result_without_new_input" - ? "This attempt already recorded a result. Send a new message on the task to continue from the saved work." - : "The exact failed chat request can no longer be retried safely. Check its current connection, access, and recovery state.", + ? "This attempt already saved its result, so it is not repeated. Retry continues from that saved result." + : FAILED_CHAT_RETRY_DIAGNOSTIC, { code: "chat_failed_run_retry_not_authorized", ...(reason ? { reason } : {}), ...(reason === "source_authorization" && authorityReason ? { authorityReason } : {}) }, ); class FailedChatRetryPublicationReadError extends Error { @@ -227,8 +230,9 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti run: typeof heartbeatRuns.$inferSelect, allowWorkspacePreparation = false, ): Promise { - const retryDiagnostic = "The exact failed chat request can no longer be retried safely. Check its current connection, access, and recovery state."; - const failedRetry = run.errorCode === "chat_failed_run_retry_not_authorized" && run.error === retryDiagnostic; + const failedRetry = run.errorCode === "chat_failed_run_retry_not_authorized" && + FAILED_CHAT_RETRY_DIAGNOSTICS.includes(run.error ?? ""); + const retryDiagnostic = failedRetry ? run.error : null; const validation = run.resultJson?.workspaceValidation as Record | undefined; const recovery = run.resultJson?.executionRecovery as Record | undefined; const workspaceValidation = run.errorCode === "workspace_validation_failed" && diff --git a/server/src/services/task-retry.test.ts b/server/src/services/task-retry.test.ts index cbc567f6aa3..e9c6c3f9c08 100644 --- a/server/src/services/task-retry.test.ts +++ b/server/src/services/task-retry.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { agents, agentWakeupRequests, chatConversations, chatEndpoints, companies, budgetPolicies, costEvents, createDb, type Db, environmentLeases, environments, executionWorkspaces, executionWorkspaceSnapshots, workspaceOperations, heartbeatRunEvents, heartbeatRuns, issueComments, issues, approvals, issueApprovals, issueRecoveryActions, nativeRunFinalizations, toolApplications, toolConnections, toolInvocations, applyPendingMigrations } from "@paperclipai/db"; +import { agents, agentWakeupRequests, chatActions, chatConversations, chatEndpoints, companies, budgetPolicies, costEvents, createDb, type Db, environmentLeases, environments, executionWorkspaces, executionWorkspaceSnapshots, workspaceOperations, heartbeatRunEvents, heartbeatRuns, issueComments, issues, approvals, issueApprovals, issueRecoveryActions, nativeRunFinalizations, toolApplications, toolConnections, toolInvocations, applyPendingMigrations } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { priorRecoveryApiReceipt, unresolvedRecoveryToolInvocation } from "./recovery-effect-receipts.js"; import { requestTaskRetry } from "./task-retry.js"; @@ -937,6 +937,95 @@ describe("durable task Retry admission", () => { expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, input.companyId))).toHaveLength(0); }); + const refusedExactChatReplay = (reason?: string) => async (tx: Db, scope: { companyId: string; issueId: string; failedRunId: string }) => { + const [endpoint] = await tx.select().from(chatEndpoints).where(eq(chatEndpoints.companyId, scope.companyId)); + await tx.insert(chatActions).values({ companyId: scope.companyId, endpointId: endpoint.id, kind: "failed_run_retry", + status: "issued", providerActionId: randomUUID(), payload: { issueId: scope.issueId, failedRunId: scope.failedRunId } }); + throw new HttpError(409, "refused", { code: "chat_failed_run_retry_not_authorized", ...(reason ? { reason } : {}) }); + }; + + it.each([undefined, "accepted_result_without_new_input", "prior_result_publication", "source_authorization"])( + "continues a connected task from saved work when its exact chat request cannot be repeated: %s", async reason => { + const input = await seedChatTask(); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.observedRunId)); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: input.issueId, source: "issue.monitor", + wakeReason: "issue_monitor_due" } }).where(eq(heartbeatRuns.id, source.id)); + const receipt = await requestTaskRetry({ ...input, actorSource: "local_implicit", prepareChatRetry: refusedExactChatReplay(reason) }); + expect(receipt).toMatchObject({ accepted: true, stale: false, observedRunId: source.id }); + expect(receipt.chatActionId).toBeUndefined(); + expect(await db.select().from(chatActions).where(eq(chatActions.companyId, input.companyId))).toHaveLength(0); + const [intent] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!)); + expect(intent).toMatchObject({ status: "deferred_issue_execution", reason: "task_retry", requestedByActorId: input.actorId, + payload: { issueId: input.issueId, taskRetry: { sourceRunId: source.id, actorId: input.actorId } } }); + expect((intent.payload!.taskRetry as Record).chatActionId).toBeUndefined(); + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }).where(eq(agents.id, source.agentId)); + await db.insert(heartbeatRuns).values({ companyId: input.companyId, agentId: source.agentId, status: "running" }); + await heartbeatService(db).resumeTaskRetry(input.companyId, receipt.requestId!); + const queued = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.status, "queued"))); + expect(queued).toHaveLength(1); + expect(queued[0].contextSnapshot).toMatchObject({ issueId: input.issueId, previousRunId: source.id, + taskRetryRequestId: receipt.requestId, forceFreshSession: true, responsibleUserId: input.actorId }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!)))[0]) + .toMatchObject({ status: "coalesced", runId: queued[0].id }); + }); + + it.each(["person", "automatic"] as const)( + "continues a person's saved board turn on a connected task from saved work without replaying it: %s", async initiator => { + const input = await seedChatTask(); + const [source] = await db.update(heartbeatRuns).set({ runtimeMode: "legacy", errorCode: "adapter_failed", + error: "Provider stopped", responsibleUserId: input.actorId, sessionIdAfter: "provider-session", + contextSnapshot: { issueId: input.issueId, source: "issue.comment.reopen", wakeCommentId: randomUUID() } }) + .where(eq(heartbeatRuns.id, input.observedRunId)).returning(); + let calls = 0; + const request = { ...input, actorSource: "local_implicit" as const, + ...(initiator === "automatic" ? { initiatedBy: "restart_recovery" as const } : {}), + prepareChatRetry: async () => { calls += 1; return { actionId: randomUUID() }; } }; + if (initiator === "automatic") { + await expect(requestTaskRetry(request)).rejects.toMatchObject({ status: 409, + details: { code: "chat_failed_run_retry_not_authorized", reason: "saved_board_message_not_replayable" } }); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, input.companyId))).toHaveLength(0); + expect(calls).toBe(0); + return; + } + const receipt = await requestTaskRetry(request); + expect(receipt).toMatchObject({ accepted: true, stale: false, observedRunId: source.id }); + expect(receipt.chatActionId).toBeUndefined(); + expect(calls).toBe(0); + const [intent] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!)); + const taskRetry = intent.payload!.taskRetry as Record; + expect(taskRetry).toMatchObject({ sourceRunId: source.id, actorId: input.actorId }); + expect(taskRetry.chatActionId).toBeUndefined(); + expect(taskRetry.boardMessageOrigin).toBeUndefined(); + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }).where(eq(agents.id, source.agentId)); + await db.insert(heartbeatRuns).values({ companyId: input.companyId, agentId: source.agentId, status: "running" }); + await heartbeatService(db).resumeTaskRetry(input.companyId, receipt.requestId!); + const queued = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.status, "queued"))); + expect(queued).toHaveLength(1); + expect(queued[0].contextSnapshot).toMatchObject({ previousRunId: source.id, taskRetryRequestId: receipt.requestId, + forceFreshSession: true, responsibleUserId: input.actorId }); + expect(queued[0].contextSnapshot?.boardMessageRetry).toBeUndefined(); + expect(queued[0].contextSnapshot?.wakeCommentId).toBeUndefined(); + }); + + it("continues a person's Retry on a connected task when no chat retry owner is available", async () => { + const input = await seedChatTask(); + const receipt = await requestTaskRetry({ ...input, actorSource: "local_implicit" }); + expect(receipt).toMatchObject({ accepted: true, chatActionId: undefined }); + const automatic = await seedChatTask(); + await expect(requestTaskRetry({ ...automatic, initiatedBy: "provider_limit_reset" })).rejects.toMatchObject({ status: 409 }); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, automatic.companyId))).toHaveLength(0); + }); + + it("keeps an automatic Retry from continuing a connected task without the exact chat request", async () => { + const input = await seedChatTask(); + await expect(requestTaskRetry({ ...input, initiatedBy: "restart_recovery", prepareChatRetry: refusedExactChatReplay() })) + .rejects.toMatchObject({ status: 409, details: { code: "chat_failed_run_retry_not_authorized" } }); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, input.companyId))).toHaveLength(0); + expect(await db.select().from(chatActions).where(eq(chatActions.companyId, input.companyId))).toHaveLength(0); + }); + it("does not retry an authorization denial", async () => { const input = await seedChatTask(); let calls = 0; diff --git a/server/src/services/task-retry.ts b/server/src/services/task-retry.ts index 4525c5986c7..a6119c48f1a 100644 --- a/server/src/services/task-retry.ts +++ b/server/src/services/task-retry.ts @@ -18,13 +18,14 @@ import { preserveNativeFailedWorkspace, preserveTaskRemoteWorkspaceWriter } from import { NativeWorkspaceFinalizationBusyError } from "./native-runtime/native-workspace-finalization-ownership.js"; import { adapterExecutionControls } from "./adapter-execution-control.js"; import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; -import { deferredChatInputPredicate, resumeDeferredChatWake } from "./durable-chat-wakeup.js"; +import { deferredChatInputPredicate, resumeDeferredChatWake, unadmittedChatWakeupCondition } from "./durable-chat-wakeup.js"; import { notifyDeferredExecutionWork } from "./deferred-execution-notifications.js"; import { hasPendingRunCleanup } from "./execution-blocker.js"; import { PersistedExecutionWaitRefusal } from "./deferred-execution-wait.js"; import { rearmExhaustedNativeFinalization } from "./native-runtime/exhausted-finalization-retry.js"; import { NativeChatReviewPresentationContentionError } from "./native-runtime/native-chat-review-presentation.js"; import { isExternalChatWaitAuthorizationContention } from "./native-runtime/chat-attachment-reuse.js"; +import { selectTaskRetryInput } from "./chat-channels/task-retry-input.js"; export async function taskRetryMutationAllowed(db: Db, input: { companyId: string; issueId: string; actorId: string; actorSource: TaskRetryIntent["actorSource"]; @@ -241,6 +242,11 @@ function automaticRetryActor(input: Pick | undefined)?.code === "chat_failed_run_retry_not_authorized"; +} + function isTaskRetryContention(error: unknown) { return error instanceof NativeChatReviewPresentationContentionError || isExternalChatWaitAuthorizationContention(error); } @@ -418,9 +424,15 @@ function admitTaskRetry(input: TaskRetryRequest) { await notifyDeferredExecutionWork(tx, { companyId: issue.companyId, requestId: id }); } } - if (chat && !active && !busy && !boardMessageOrigin && !intent.queuedChatRequestId && (latest.contextSnapshot?.source === "issue.comment.reopen" || latest.contextSnapshot?.boardMessageRetry === true) && - latest.responsibleUserId === input.actorId) { - throw conflict("This saved message cannot be retried safely. Send a new message on the task to continue from saved work.", + const savedBoardTurn = Boolean(chat && !active && !busy && !boardMessageOrigin && !intent.queuedChatRequestId && + (latest.contextSnapshot?.source === "issue.comment.reopen" || latest.contextSnapshot?.boardMessageRetry === true) && + latest.responsibleUserId === input.actorId); + const hasUnadmittedChatInput = async () => (await tx.select({ id: issues.id }).from(issues).where(and(eq(issues.id, issue.id), + eq(issues.companyId, issue.companyId), unadmittedChatWakeupCondition(issues.id, issues.companyId))).limit(1)).length > 0 || + Boolean((await selectTaskRetryInput(tx, { companyId: issue.companyId, issueId: issue.id, + agentId: issue.assigneeAgentId!, failedRunId: latest.id }))?.sourceActionIds.length); + if (savedBoardTurn && (input.initiatedBy || await hasUnadmittedChatInput())) { + throw conflict("This saved message can't be repeated exactly as it was sent. Retry continues the task from its saved work instead.", { code: "chat_failed_run_retry_not_authorized", reason: "saved_board_message_not_replayable" }); } await tx.insert(agentWakeupRequests).values({ id, companyId: issue.companyId, agentId: issue.assigneeAgentId, @@ -429,12 +441,23 @@ function admitTaskRetry(input: TaskRetryRequest) { status: active || busy ? "coalesced" : "deferred_issue_execution", runId: active?.runId ?? (busy ? latest.id : null), payload: { issueId: issue.id, taskRetry: intent }, finishedAt: active || busy ? new Date() : null }); - if (chat && !boardMessageOrigin && !intent.queuedChatRequestId && !active && !busy) { - if (!input.prepareChatRetry) throw conflict("Retry is unavailable for this connected conversation."); - intent.chatActionId = (await input.prepareChatRetry(db, { companyId: issue.companyId, issueId: issue.id, - agentId: issue.assigneeAgentId, failedRunId: latest.id, initiatedByUserId: input.actorId, retryIntentId: id })).actionId; - await tx.update(agentWakeupRequests).set({ payload: { issueId: issue.id, taskRetry: intent } }) - .where(eq(agentWakeupRequests.id, id)); + if (chat && !boardMessageOrigin && !intent.queuedChatRequestId && !active && !busy && !savedBoardTurn) { + if (!input.prepareChatRetry && input.initiatedBy) throw conflict("Retry is unavailable for this connected conversation."); + const prepareChatRetry = input.prepareChatRetry ?? (() => { throw conflict("Retry is unavailable for this connected conversation.", + { code: "chat_failed_run_retry_not_authorized" }); }); + const scope = { companyId: issue.companyId, issueId: issue.id, agentId: issue.assigneeAgentId, + failedRunId: latest.id, initiatedByUserId: input.actorId, retryIntentId: id }; + const prepared = await tx.transaction(savepoint => prepareChatRetry(savepoint as unknown as Db, scope)) + .catch(async error => { + if (input.initiatedBy || !isExactChatReplayRefusal(error)) throw error; + if (await hasUnadmittedChatInput()) throw error; + return null; + }); + if (prepared) { + intent.chatActionId = prepared.actionId; + await tx.update(agentWakeupRequests).set({ payload: { issueId: issue.id, taskRetry: intent } }) + .where(eq(agentWakeupRequests.id, id)); + } } if (!busy) { for (const action of records) {