diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 0e93aefda20..7c47d59f1ad 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -276,6 +276,14 @@ still checks the exact current binding and snapshot after provider settlement. The work is complete and terminal. +An authorized agent working on another task may add evidence or request a +follow-up on a completed task. That control-plane request does not restore the +completed task's closed workspace under the caller's run identity. When the +follow-up is admitted, its own executor restores the task's selected workspace +if task completion closed it. Workspace execution, company, project, and source +access checks still apply under that executor. An archive for another reason +does not become eligible for restoration through this path. + ### `cancelled` The work will not continue and is terminal. @@ -2212,6 +2220,11 @@ unknown, and they do not gain automatic replay eligibility. Admission validates the persisted comment's author, task, and time against every held predecessor. Retry validates the selected failed run's company, task, and agent and preserves that run's identity through admission and history loading. +When preparation failed before a source recovery-action row was created, the +admitted Retry receipt saves its continuation authorization. Existing action +records remain the authority when present. Dispatch checks the exact receipt, +successor, source run, and user. A caller-provided continuation object does not +establish authority. Duplicate Retry requests adopt the same successor. Only a validated, unconsumed human queue receipt can authorize saved-message continuation after a native failure; arbitrary historical input, an agent-authored comment, or a generic system wake cannot release a hold. The source task keeps its assignee. Process ownership, active controllers, cleanup leases, pause, approval, budget, and normal diff --git a/server/src/__tests__/heartbeat-git-restore-point-resume.test.ts b/server/src/__tests__/heartbeat-git-restore-point-resume.test.ts index 9620083333c..6ae1694a4f7 100644 --- a/server/src/__tests__/heartbeat-git-restore-point-resume.test.ts +++ b/server/src/__tests__/heartbeat-git-restore-point-resume.test.ts @@ -68,7 +68,7 @@ describe.skipIf(!support.supported)("heartbeat resume of an idle-expired workspa await fs.rm(root, { recursive: true, force: true }); }, 60_000); - async function seed(options: { closedParent?: boolean } = {}) { + async function seed(options: { closedParent?: boolean; terminalArchive?: boolean } = {}) { const home = path.join(root, randomUUID()); vi.stubEnv("PAPERCLIP_HOME", home); vi.stubEnv("PAPERCLIP_INSTANCE_ID", "default"); @@ -120,7 +120,11 @@ describe.skipIf(!support.supported)("heartbeat resume of an idle-expired workspa const unpushed = git(realized.cwd, "rev-parse", "HEAD"); await fs.writeFile(path.join(realized.cwd, "notes.md"), "uncommitted notes\n"); const workspace = () => db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId)).then(rows => rows[0]!); - const expiration = await expireReconstructableGitWorkspace(db, await workspace(), { + const expiration = options.terminalArchive ? await (async () => { + await db.update(executionWorkspaces).set({ status: "archived", cleanupReason: "issue_terminal", + closedAt: new Date("2020-01-01T00:00:00Z") }).where(eq(executionWorkspaces.id, workspaceId)); + return null; + })() : await expireReconstructableGitWorkspace(db, await workspace(), { dryRun: false, expireBase: false, now: new Date(), idleCutoff: options.closedParent ? undefined : new Date(Date.now() - 7 * DAY_MS), restorePoint: { storage: getStorageService(), notBefore: new Date(Date.now() - DAY_MS), keyDurable: true }, }); @@ -167,6 +171,28 @@ describe.skipIf(!support.supported)("heartbeat resume of an idle-expired workspa expect(issue.executionWorkspaceId).toBe(seeded.workspaceId); }); + it("starts a reopened task in its terminal workspace with the saved files intact", async () => { + const seeded = await seed({ terminalArchive: true }); + const archived = await seeded.workspace(); + expect(archived).toMatchObject({ status: "archived", cleanupReason: "issue_terminal" }); + expect(archived.metadata?.gitCacheExpiration).toBeUndefined(); + await seeded.queueRun(); + const heartbeat = heartbeatService(db); + await heartbeat.resumeQueuedRuns(); + const run = await waitForRun(heartbeat, seeded.runId); + await heartbeat.waitForRunExecutionDrain(seeded.runId); + + expect(run).toMatchObject({ status: "succeeded" }); + expect(adapterExecute).toHaveBeenCalledTimes(1); + expect(adapterExecute.mock.calls[0]?.[0]).toMatchObject({ context: { executionWorkspaceId: seeded.workspaceId, + paperclipWorkspace: { cwd: seeded.cwd } } }); + expect(git(seeded.cwd, "rev-parse", "HEAD")).toBe(seeded.unpushed); + expect(await fs.readFile(path.join(seeded.cwd, "notes.md"), "utf8")).toBe("uncommitted notes\n"); + expect(await seeded.workspace()).toMatchObject({ status: "active", closedAt: null }); + const [issue] = await db.select().from(issues).where(eq(issues.id, seeded.issueId)); + expect(issue.executionWorkspaceId).toBe(seeded.workspaceId); + }); + it("restores a closed parent's expired workspace when an open child binds to it", async () => { const seeded = await seed({ closedParent: true }); expect(seeded.expiration).toMatchObject({ expired: true, restorePoint: true }); diff --git a/server/src/__tests__/issue-comment-reopen-routes.test.ts b/server/src/__tests__/issue-comment-reopen-routes.test.ts index 737da478ef3..11c1d07efb0 100644 --- a/server/src/__tests__/issue-comment-reopen-routes.test.ts +++ b/server/src/__tests__/issue-comment-reopen-routes.test.ts @@ -42,6 +42,16 @@ const mockHeartbeatService = vi.hoisted(() => ({ cancelRun: vi.fn(async (): Promise => null), })); +const mockExecutionWorkspaceService = vi.hoisted(() => ({ + getStoredById: vi.fn(async (): Promise => null), + reopenClosedIsolatedExecutionWorkspaceForIssue: vi.fn(), +})); + +vi.mock("../services/execution-workspaces.js", async importOriginal => ({ + ...await importOriginal(), + executionWorkspaceService: () => mockExecutionWorkspaceService, +})); + const mockAgentService = vi.hoisted(() => ({ getById: vi.fn(), list: vi.fn(), @@ -197,7 +207,7 @@ vi.mock("../services/index.js", () => ({ remapOpenThreadsForDocument: async () => [], }), documentService: () => ({}), - executionWorkspaceService: () => ({}), + executionWorkspaceService: () => mockExecutionWorkspaceService, feedbackService: () => mockFeedbackService, goalService: () => ({}), heartbeatService: () => mockHeartbeatService, @@ -405,6 +415,8 @@ describe("issue comment reopen routes", () => { mockHeartbeatService.getRun.mockReset(); mockHeartbeatService.getActiveRunForAgent.mockReset(); mockHeartbeatService.cancelRun.mockReset(); + mockExecutionWorkspaceService.getStoredById.mockReset().mockResolvedValue(null); + mockExecutionWorkspaceService.reopenClosedIsolatedExecutionWorkspaceForIssue.mockReset(); mockAgentService.getById.mockReset(); mockAgentService.list.mockReset(); mockAgentService.resolveByReference.mockReset(); @@ -587,6 +599,33 @@ describe("issue comment reopen routes", () => { ); }); + it.each(["comment", "update"])("accepts an authorized cross-task %s without restoring the target's files", async operation => { + const issue = { ...makeIssue("done"), executionWorkspaceId: "33333333-3333-4333-8333-333333333333" }; + const actor = { ...agentActor(), source: "agent_jwt", runId: "66666666-6666-4666-8666-666666666666" }; + mockIssueService.getById.mockResolvedValue(issue); + mockIssueService.update.mockResolvedValue({ ...issue, status: "todo" }); + mockHeartbeatService.getRun.mockResolvedValue({ id: actor.runId, agentId: actor.agentId, + companyId: issue.companyId, status: "running", runtimeMode: "native", + nativeIssueId: "55555555-5555-4555-8555-555555555555", contextSnapshot: {} }); + mockExecutionWorkspaceService.getStoredById.mockResolvedValue({ id: issue.executionWorkspaceId, + companyId: issue.companyId, sourceIssueId: issue.id, mode: "isolated_workspace", status: "archived", + closedAt: new Date("2020-01-01T00:00:00Z"), cleanupReason: "issue_terminal" }); + // A scanner's run may change this task, but cannot execute in its workspace. + mockExecutionWorkspaceService.reopenClosedIsolatedExecutionWorkspaceForIssue.mockRejectedValue( + new HttpError(403, "Task workspace run identity is unavailable")); + const app = request(await installActor(createApp(), actor)); + const response = operation === "comment" + ? await app.post(`/api/issues/${issue.id}/comments`).send({ body: "New regression evidence", resume: true }) + : await app.patch(`/api/issues/${issue.id}`).send({ comment: "New regression evidence", resume: true }); + + expect(response.status, JSON.stringify(response.body)).toBe(operation === "comment" ? 201 : 200); + expect(mockIssueService.update.mock.calls[0]?.slice(0, 2)) + .toEqual([issue.id, expect.objectContaining({ status: "todo" })]); + expect(mockExecutionWorkspaceService.reopenClosedIsolatedExecutionWorkspaceForIssue).not.toHaveBeenCalled(); + await waitForWakeup(() => expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(issue.assigneeAgentId, + expect.objectContaining({ payload: expect.objectContaining({ issueId: issue.id }) }))); + }); + it("finishes missing reference and activity writes for an accepted comment", async () => { const issue = makeIssue("done"); mockIssueService.getById.mockResolvedValue(issue); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index e7ae7f20531..4a967876ed8 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -16,6 +16,7 @@ import { issueSummariesQuerySchema, type IssueSummary } from "@paperclipai/share import { listIssueSummaries } from "../services/issue-summaries.js"; import { assertNoCaseHandoffMutation } from "../services/heartbeat/case-handoff-lifecycle.js"; import { assertLegacyRunTaskOwnership } from "../legacy-run-task-ownership.js"; +import { runRetainsIssueIdentity } from "../services/task-repository-acquisition.js"; import { withConversationReadState } from "../services/agent-chat-read-state.js"; import { markIssueReadSchema } from "@paperclipai/shared"; import { enqueueFastResponse } from "../services/fast-responses.js"; @@ -7867,6 +7868,8 @@ export function issueRoutes( // this request did not set the flag. The caller continues but // must not install the guard, or it can clear the flag that // the other request still owns. + // "deferred" - an authorized cross-task comment/update leaves restoration + // to the target's executor. No workspace flag was changed. // null - this function sent an error response, so the caller stops. // it runs only after the route already authorized the request on the issue. async function reopenClosedIssueExecutionWorkspaceOrRespond( @@ -7874,11 +7877,22 @@ export function issueRoutes( res: Response, issue: { id: string; companyId: string; projectId?: string | null }, workspace: Pick, + options: { deferForOtherTask?: boolean } = {}, ): Promise<{ - outcome: "reopened" | "already-open"; - generation: number; + outcome: "reopened" | "already-open" | "deferred"; + generation: number | null; } | null> { const actor = getActorInfo(req); + if (options.deferForOtherTask && req.actor.type === "agent" && req.actor.runId) { + const run = await heartbeat.getRun(req.actor.runId); + if (run?.status === "running" && run.companyId === issue.companyId && run.agentId === req.actor.agentId + && !runRetainsIssueIdentity(run, issue.id)) { + // Task authority permits a scanner's comment or follow-up. Its run + // cannot restore another task's files. The target's admitted executor + // restores the workspace with its own run identity when work resumes. + return { outcome: "deferred", generation: null }; + } + } const result = await executionWorkspacesSvc.reopenClosedIsolatedExecutionWorkspaceForIssue( { @@ -14990,6 +15004,7 @@ export function issueRoutes( res, existing, closedExecutionWorkspace, + { deferForOtherTask: true }, ); if (reopenOutcome === null) { return; @@ -18997,6 +19012,7 @@ export function issueRoutes( res, issue, closedExecutionWorkspace, + { deferForOtherTask: true }, ); if (reopenOutcome === null) { return; diff --git a/server/src/services/execution-continuation.ts b/server/src/services/execution-continuation.ts index 616b08b3249..197de689434 100644 --- a/server/src/services/execution-continuation.ts +++ b/server/src/services/execution-continuation.ts @@ -18,6 +18,7 @@ import { queuedInteractionId } from "./queued-interaction-response.js"; import { childReviewOutcomes } from "./native-runtime/child-review-outcomes.js"; import { isCompletedOnboardingHandoffWake } from "./chat-completion-delivery.js"; import { githubRecoveryReceipts } from "./chat-channels/github/recovery-receipts.js"; +import { readTaskRetryIntent } from "./task-retry.js"; export class StaleExecutionContinuationError extends Error { constructor(readonly code: "continuation_task_ownership_changed") { @@ -424,8 +425,22 @@ export async function buildExecutionContinuation(input: { eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.status, "coalesced"), eq(agentWakeupRequests.runId, input.runId), sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, )) : []; + const taskRetryId = string(explicitContinuation.taskRetryRequestId); + const [taskRetryWake] = taskRetryId && z.string().guid().safeParse(taskRetryId).success + ? await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, taskRetryId), + eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.reason, "task_retry"), + eq(agentWakeupRequests.requestedByActorType, "user"), eq(agentWakeupRequests.status, "coalesced"), + sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, + input.runId ? eq(agentWakeupRequests.runId, input.runId) : undefined, + )) : []; + const taskRetry = readTaskRetryIntent(taskRetryWake?.payload); + const retryAuthorization = taskRetry && taskRetry.generationId === taskRetryWake?.id + && taskRetry.actorId === taskRetryWake.requestedByActorId + ? object(object(taskRetryWake.payload?.taskRetry).continuation) : {}; const continuationAuthorizations = [...reconciliations.map(row => object(row.evidence.explicitUserContinuation)), - ...(recordedQueue ? [object(object(recordedQueue.payload?.queuedCommentInterrupt).continuation)] : [])] + ...(recordedQueue ? [object(object(recordedQueue.payload?.queuedCommentInterrupt).continuation)] : []), + retryAuthorization] .filter(value => value.previousRunId === explicitUserSource && (!input.runId || value.runId === input.runId) && value.commentId === explicitContinuation.commentId && @@ -443,15 +458,6 @@ export async function buildExecutionContinuation(input: { sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is not null`, input.runId ? eq(agentWakeupRequests.runId, input.runId) : undefined, )) : []; - const taskRetryId = string(explicitContinuation.taskRetryRequestId); - const [taskRetryWake] = taskRetryId && z.string().guid().safeParse(taskRetryId).success - ? await db.select().from(agentWakeupRequests).where(and( - eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, taskRetryId), - eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.reason, "task_retry"), - eq(agentWakeupRequests.requestedByActorType, "user"), eq(agentWakeupRequests.status, "coalesced"), - sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, - input.runId ? eq(agentWakeupRequests.runId, input.runId) : undefined, - )) : []; const authorization = continuationAuthorizations.find(value => taskRetryId ? value.taskRetryRequestId === taskRetryId && taskRetryWake?.runId === value.runId && taskRetryWake.requestedByActorId === value.actorId && diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index d4b04366d78..3d747a325b2 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -540,10 +540,16 @@ export async function admitExplicitNativeContinuation(input: { if (readiness) await persistActivity(db, { companyId, actorType: "system", actorId: "task-recovery", action: "issue.fresh_workspace_selected", entityType: "issue", entityId: issueId, details: { ...readiness, runId: input.successorRunId } }); - if (input.taskRetryRequestId) await db.update(agentWakeupRequests).set({ - payload: sql`jsonb_set(${agentWakeupRequests.payload}, '{taskRetry,strategy}', - ${JSON.stringify(readiness?.mode === "fresh_isolated_workspace" ? "fresh_isolated_workspace" : "continue")}::jsonb)`, - }).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, input.taskRetryRequestId))); + if (input.taskRetryRequestId) { + const payload = sql`jsonb_set(${agentWakeupRequests.payload} #- '{taskRetry,continuation}', '{taskRetry,strategy}', + ${JSON.stringify(readiness?.mode === "fresh_isolated_workspace" ? "fresh_isolated_workspace" : "continue")}::jsonb)`; + // Preserve the action ledger as the authority when it exists. Preparation + // can fail before that ledger exists, so only that path needs a fallback. + const hasActionAuthorization = actions.some(action => (action.evidence.runId ?? action.evidence.sourceRunId) === previous.id); + await db.update(agentWakeupRequests).set({ payload: hasActionAuthorization ? payload + : sql`jsonb_set(${payload}, '{taskRetry,continuation}', ${JSON.stringify(authorization)}::jsonb)`, + }).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, input.taskRetryRequestId))); + } return continuation; } diff --git a/server/src/services/heartbeat/environment-selection.test.ts b/server/src/services/heartbeat/environment-selection.test.ts index cb250773b11..88456c80e7e 100644 --- a/server/src/services/heartbeat/environment-selection.test.ts +++ b/server/src/services/heartbeat/environment-selection.test.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { mkdtemp, mkdir, rm } from "node:fs/promises"; +import { mkdtemp, mkdir, rm, writeFile, readFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { agents, companies, createDb, environments, heartbeatRuns, issues, projects, projectWorkspaces, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, type Db } from "@paperclipai/db"; @@ -137,6 +137,25 @@ describe.skipIf(!support.supported)("heartbeat environment selection boundary", expect(result.requestedReusableExecutionWorkspaceConfig).toBeNull(); }); + it("restores a reopened task's terminal workspace under its own admitted run", async () => { + const { input, issue, workspace } = await fixture(); + const saved = await workspace(); + await writeFile(join(home, "saved-work.txt"), "Retain the completed investigation\n"); + await executionWorkspaceService(db).update(saved.id, { status: "archived", strategyType: "project_primary", + cleanupReason: "issue_terminal", closedAt: new Date("2020-01-01T00:00:00Z") }); + await db.update(issues).set({ executionWorkspaceId: saved.id, executionWorkspacePreference: "reuse_existing" }) + .where(eq(issues.id, issue.id)); + input.task.issueRef!.executionWorkspaceId = saved.id; + input.task.issueRef!.executionWorkspacePreference = "reuse_existing"; + + const result = await selectHeartbeatEnvironment(db, input); + + expect(result.reusableExistingExecutionWorkspace).toMatchObject({ id: saved.id, status: "active" }); + expect(await readFile(join(home, "saved-work.txt"), "utf8")).toBe("Retain the completed investigation\n"); + expect(await executionWorkspaceService(db).getStoredById(input.agent.companyId, saved.id)) + .toMatchObject({ status: "active", closedAt: null, sourceIssueId: issue.id }); + }); + describe("workspaces removed by git cache expiration", () => { async function expired() { const f = await fixture(); diff --git a/server/src/services/heartbeat/environment-selection.ts b/server/src/services/heartbeat/environment-selection.ts index 58c2713c9c2..534459f7001 100644 --- a/server/src/services/heartbeat/environment-selection.ts +++ b/server/src/services/heartbeat/environment-selection.ts @@ -122,7 +122,9 @@ export async function selectHeartbeatEnvironment(db: Db, input: HeartbeatEnviron && existingExecutionWorkspace.id === readNonEmptyString(issueRef?.executionWorkspaceId) && (existingExecutionWorkspace.status === "archived" || existingExecutionWorkspace.status === "cleanup_failed") && (readGitCacheExpiration(existingExecutionWorkspace as Parameters[0])?.workspaceRemoved - || readTaskDirectoryRetention(existingExecutionWorkspace) !== null)) { + || readTaskDirectoryRetention(existingExecutionWorkspace) !== null + || (existingExecutionWorkspace.sourceIssueId === issueId + && existingExecutionWorkspace.cleanupReason?.startsWith("issue_terminal") === true))) { const [boundIssue] = await db.select({ id: issues.id, companyId: issues.companyId, projectId: issues.projectId }) .from(issues).where(and(eq(issues.id, issueId), eq(issues.companyId, agent.companyId))); if (!boundIssue) throw new Error("Execution workspace issue is unavailable"); diff --git a/server/src/services/task-retry.test.ts b/server/src/services/task-retry.test.ts index a05fd80146c..a02c7eb5623 100644 --- a/server/src/services/task-retry.test.ts +++ b/server/src/services/task-retry.test.ts @@ -14,6 +14,7 @@ import { instanceSettingsService } from "./instance-settings.js"; import { heartbeatService } from "./heartbeat.js"; import { taskChatRetryCanAwaitRecovery, taskChatRetryExecutionStopped } from "./chat-channels/task-retry-authority.js"; import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js"; +import { buildExecutionContinuation } from "./execution-continuation.js"; describe("durable task Retry admission", () => { let postgres: Awaited>; @@ -38,6 +39,49 @@ describe("durable task Retry admission", () => { contextSnapshot: { issueId }, finishedAt: new Date() }); return { db, companyId, issueId, actorId: "board", observedRunId: runId, idempotencyKey: randomUUID() }; } + it.each(["valid", "wrong_run", "wrong_actor", "wrong_source", "wrong_generation", "unadmitted", "missing_authorization"])( + "dispatches a preparation Retry with no recovery-action row only from its exact admitted receipt: %s", async state => { + const input = await seed(); + await db.update(issues).set({ status: "todo" }).where(eq(issues.id, input.issueId)); + const [source] = await db.update(heartbeatRuns).set({ runtimeMode: "legacy", errorCode: "setup_failed", + executionStage: "preparing", runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + }).where(eq(heartbeatRuns.id, input.observedRunId)).returning(); + const receipt = await requestTaskRetry(input), successorRunId = randomUUID(); + const continuation = await db.transaction(async tx => { + await tx.select().from(issues).where(eq(issues.id, input.issueId)).for("update"); + const result = await admitExplicitNativeContinuation({ db: tx as unknown as Db, + companyId: input.companyId, issueId: input.issueId, agentId: source.agentId, + actorId: input.actorId, actorType: "user", reason: "task_retry", commentId: null, + successorRunId, taskRetryRequestId: receipt.requestId!, + }); + await tx.insert(heartbeatRuns).values({ id: successorRunId, companyId: input.companyId, agentId: source.agentId, + issueId: input.issueId, status: "queued", contextSnapshot: { issueId: input.issueId } }); + await tx.update(agentWakeupRequests).set({ status: "coalesced", runId: successorRunId }) + .where(eq(agentWakeupRequests.id, receipt.requestId!)); + return result; + }); + expect(continuation).toMatchObject({ previousRunId: source.id, taskRetryRequestId: receipt.requestId }); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, input.issueId))).toHaveLength(0); + const [admitted] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!)); + const taskRetry = admitted.payload!.taskRetry as Record; + if (state !== "valid") await db.update(agentWakeupRequests).set({ + ...(state === "wrong_run" ? { runId: source.id } : {}), + ...(state === "wrong_actor" ? { requestedByActorId: "other-user" } : {}), + ...(state === "unadmitted" ? { status: "deferred_issue_execution" } : {}), + ...(state === "wrong_source" ? { payload: { ...admitted.payload, taskRetry: { ...taskRetry, sourceRunId: randomUUID() } } } : {}), + ...(state === "wrong_generation" ? { payload: { ...admitted.payload, taskRetry: { ...taskRetry, generationId: randomUUID() } } } : {}), + ...(state === "missing_authorization" ? { payload: { ...admitted.payload, + taskRetry: Object.fromEntries(Object.entries(taskRetry).filter(([key]) => key !== "continuation")) } } : {}), + }).where(eq(agentWakeupRequests.id, admitted.id)); + const result = buildExecutionContinuation({ db, companyId: input.companyId, issueId: input.issueId, + agentId: source.agentId, runId: successorRunId, + context: { issueId: input.issueId, explicitUserContinuation: continuation }, summary: null, exposeLowTrustRaw: false }); + if (state === "valid") await expect(result).resolves.toMatchObject({ interruptedRunId: source.id }); + else await expect(result).rejects.toThrow("continuation_user_authorization_missing"); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, source.id)))[0]) + .toMatchObject({ status: "failed", resultJson: source.resultJson }); + }); it("serializes concurrent clicks and remembers each key after completion", async () => { const input = await seed(); const second = { ...input, idempotencyKey: randomUUID() };