diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index af54d259ef9..9733401c99a 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -4,7 +4,7 @@ import { hasRequiredWorkspaceRecovery } from "../../../services/workspace-restor import { readNativeWorkspaceSyncReference } from "../../../services/native-runtime/native-workspace-sync.js"; import { isPreDispatchExecutionWaitVerified } from "../../../services/pre-dispatch-execution-wait.js"; import { isUserCommentContinuationReason } from "../../../services/explicit-native-continuation.js"; -import { queuedCommentIdsFromRunContext, withQueuedCommentIdsInWakePayload } from "../../../services/issue-queued-comment-queue.js"; +import { queuedCommentIdsFromRunContext, withQueuedCommentIdsInRunContext, withQueuedCommentIdsInWakePayload } from "../../../services/issue-queued-comment-queue.js"; import { notifyExternalAgentSettlement } from "../../external-agents/index.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../../../services/cancelled-workspace-preservation-notifications.js"; @@ -12,6 +12,7 @@ import { notifyInstructionCollection } from "../../../services/instruction-colle import { notifyScheduledRetryCancellation } from "../../../services/scheduled-retry-cancellation-notifications.js"; import { readClaimedNativeStopRecovery } from "../../../services/native-runtime/native-cancellation-request.js"; import { hasConversationContinuationPolicy } from "../../../services/conversation-continuation.js"; +import { userStopBoundaryAt } from "../../../services/run-cancellation.js"; import { getExecutionBlocker } from "../../../services/execution-blocker.js"; import { validateReconstructedWorkspaceDispatch } from "../../../services/native-runtime/native-workspace-reconstruction.js"; import { getNativeReviewAssignment } from "../../../services/native-runtime/native-review-participant.js"; @@ -67,6 +68,7 @@ import type { } from "../domain/policy.js"; import { MAX_TURN_CONTINUATION_RETRY_REASON, + WORKSPACE_BUSY_RETRY_REASON, allowsIssueInteractionWake, deriveCommentId, isNonAssigneeWorkspaceBusyRetry, @@ -949,6 +951,42 @@ export function createPostgresRunDispatchAdapter( return transactionResult.outcome; } + /** + * A system retry of an unstarted user turn (a workspace wait or an + * admission retry) still carries that user's message, but its own receipt + * is system-owned. Find the originating user receipt through a bounded chain + * of retries that never reached the provider, so the message can be saved + * under the user's authority instead of being dropped with the retry. + */ + async function unstartedUserTurnOrigin(tx: Db, run: HeartbeatRun, issueId: string) { + let current = run; + // Comments can coalesce into any retry in the chain while it waits. + const chainRunIds: string[] = [], chainWakeIds: string[] = []; + for (let hop = 0; hop < 8 && current.retryOfRunId; hop++) { + const [parent] = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId), + eq(heartbeatRuns.id, current.retryOfRunId), eq(heartbeatRuns.status, "cancelled"), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`, + )).for("update"); + if (!parent?.wakeupRequestId) return null; + const recovery = parseObject(parent.resultJson?.executionRecovery); + const unstarted = parent.errorCode === WORKSPACE_BUSY_RETRY_REASON + ? recovery.kind === "workspace_wait" && recovery.providerWorkStarted === false + : parent.errorCode === "execution_reconciliation_required" && await isPreDispatchExecutionWaitVerified(tx, parent); + if (!unstarted) return null; + const [wake] = await tx.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), + eq(agentWakeupRequests.id, parent.wakeupRequestId), eq(agentWakeupRequests.runId, parent.id), + )).for("update"); + if (!wake) return null; + chainRunIds.push(parent.id); chainWakeIds.push(wake.id); + if (wake.requestedByActorType === "user") + return ["cancelled", "skipped"].includes(wake.status) ? { run: parent, wake, chainRunIds, chainWakeIds } : null; + current = parent; + } + return null; + } + async function deferUnstartedUserComment(tx: Db, run: HeartbeatRun, issueId: string, now: Date) { if (!run.wakeupRequestId || !await isPreDispatchExecutionWaitVerified(tx, run)) return null; const wait = parseObject(run.resultJson?.executionWait); @@ -967,18 +1005,27 @@ export function createPostgresRunDispatchAdapter( // a message sent after the Stop is saved; admission still applies all gates. const copybackReplay = hasRequiredWorkspaceRecovery(source.resultJson) && !!baseline && baseline.state !== "finalized" && preservation.fingerprint === `native-workspace:${source.id}`; - const context = parseObject(run.contextSnapshot), commentId = deriveCommentId(context); - const commentIds = queuedCommentIdsFromRunContext(context); - if (!commentId || !commentIds.includes(commentId) || !isUserCommentContinuationReason(readNonEmptyString(context.wakeReason))) return null; - const [wake] = await tx.select().from(agentWakeupRequests).where(and( + const runContext = parseObject(run.contextSnapshot), commentId = deriveCommentId(runContext); + const commentIds = queuedCommentIdsFromRunContext(runContext); + if (!commentId || !commentIds.includes(commentId)) return null; + const [ownWake] = await tx.select().from(agentWakeupRequests).where(and( eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.id, run.wakeupRequestId), eq(agentWakeupRequests.runId, run.id), inArray(agentWakeupRequests.status, ["queued", "claimed"]), - eq(agentWakeupRequests.requestedByActorType, "user"), )).for("update"); - const payload = parseObject(wake?.payload); + if (!ownWake) return null; + const origin = ownWake.requestedByActorType === "system" ? await unstartedUserTurnOrigin(tx, run, issueId) : null; + if (ownWake.requestedByActorType !== "user" && !origin) return null; + const wake = origin?.wake ?? ownWake; + const receiptRunId = origin?.run.id ?? run.id; + const runIds = [run.id, ...(origin?.chainRunIds ?? [])]; + // A retry's context names the retry reason. Save the user's original wake + // context together with every message the retry carried. + const context = origin ? withQueuedCommentIdsInRunContext(parseObject(origin.run.contextSnapshot), commentIds) : runContext; + if (!isUserCommentContinuationReason(readNonEmptyString(context.wakeReason))) return null; + const payload = parseObject(wake.payload); const originalCommentId = readNonEmptyString(payload.commentId); - if (!wake?.requestedByActorId || payload.issueId !== issueId || + if (!wake.requestedByActorId || payload.issueId !== issueId || !originalCommentId || !commentIds.includes(originalCommentId)) return null; const comments = await tx.select().from(issueComments).where(and( eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId), @@ -989,24 +1036,25 @@ export function createPostgresRunDispatchAdapter( !comment.body.trim() || !comment.authorUserId) || comments.find(comment => comment.id === originalCommentId)?.authorUserId !== wake.requestedByActorId) return null; const latest = comments.find(comment => comment.id === commentId); - if (!copybackReplay && (!source.finishedAt || !latest || latest.createdAt <= source.finishedAt)) return null; - const sourceIds = context.queuedCommentSourceWakeIds; + const stoppedAt = userStopBoundaryAt(source); + if (!copybackReplay && (!stoppedAt || !latest || latest.createdAt <= stoppedAt)) return null; + const sourceIds = runContext.queuedCommentSourceWakeIds; if (sourceIds !== undefined && (!Array.isArray(sourceIds) || sourceIds.some(id => typeof id !== "string" || !isUuidLike(id)))) return null; const sourceWakeIds = Array.isArray(sourceIds) ? [...new Set(sourceIds as string[])] : []; const adopted = sourceWakeIds.length ? await tx.select().from(agentWakeupRequests).where(and( eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), - eq(agentWakeupRequests.runId, run.id), eq(agentWakeupRequests.status, "coalesced"), + inArray(agentWakeupRequests.runId, runIds), eq(agentWakeupRequests.status, "coalesced"), inArray(agentWakeupRequests.id, sourceWakeIds), sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, )).for("update") : []; if (adopted.length !== sourceWakeIds.length) return null; - const mergeTargets = [wake.id, ...adopted.map(receipt => receipt.id)]; + const mergeTargets = [...new Set([wake.id, ownWake.id, ...(origin?.chainWakeIds ?? []), ...adopted.map(receipt => receipt.id)])]; const coalesced = await tx.select().from(agentWakeupRequests).where(and( eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.status, "coalesced"), eq(agentWakeupRequests.requestedByActorType, "user"), sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, - or(and(eq(agentWakeupRequests.runId, run.id), eq(agentWakeupRequests.reason, "issue_execution_same_name")), - and(or(isNull(agentWakeupRequests.runId), eq(agentWakeupRequests.runId, run.id)), + or(and(inArray(agentWakeupRequests.runId, runIds), eq(agentWakeupRequests.reason, "issue_execution_same_name")), + and(or(isNull(agentWakeupRequests.runId), inArray(agentWakeupRequests.runId, runIds)), inArray(sql`${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId'`, mergeTargets))), )).for("update"); const authors = new Map(comments.map(comment => [comment.id, comment.authorUserId])); @@ -1024,8 +1072,9 @@ export function createPostgresRunDispatchAdapter( executionWait: { ...parseObject(run.resultJson?.executionWait), reason: "execution_recovery", message: "Waiting for execution recovery. Your message is saved." } }, commentIds), updatedAt: now, }).where(and(eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.companyId, run.companyId), - eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.runId, run.id), - inArray(agentWakeupRequests.status, ["queued", "claimed"]))).returning({ id: agentWakeupRequests.id }); + eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.runId, receiptRunId), + inArray(agentWakeupRequests.status, origin ? ["cancelled", "skipped"] : ["queued", "claimed"]))) + .returning({ id: agentWakeupRequests.id }); return deferred?.id ?? null; } @@ -1083,7 +1132,9 @@ export function createPostgresRunDispatchAdapter( const deferredWakeRequestId = expectedStatus === "queued" && decision.errorCode === "execution_reconciliation_required" ? await deferUnstartedUserComment(tx, row, issueId, now) : null; - if (row.wakeupRequestId && !deferredWakeRequestId) { + // A retry's system receipt still settles when its user's message was + // saved on the originating user receipt. + if (row.wakeupRequestId && deferredWakeRequestId !== row.wakeupRequestId) { await tx .update(agentWakeupRequests) .set({ status: "skipped", finishedAt: now, error: decision.reason, updatedAt: now }) diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 4f0283f019c..13075ff1f59 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -12,7 +12,7 @@ import { readNativeWorkspaceSyncReference } from "./native-runtime/native-worksp import { preserveNativeFailedWorkspace, preserveTaskRemoteWorkspaceWriter } from "./native-runtime/native-failed-workspace-recovery.js"; import { createHash } from "node:crypto"; import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; -import { canContinueCancelledRun } from "./run-cancellation.js"; +import { canContinueCancelledRun, userStopBoundaryAt } from "./run-cancellation.js"; import { readQueuedInteractionResponse } from "./queued-interaction-response.js"; import { isCancelledNativeStartup, isFailedNativePreparation, isFailedNativePreparationRetryAncestor } from "./cancelled-native-startup.js"; import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession, recordedLocalExecutionStopped } from "./native-local-process-stop.js"; @@ -104,6 +104,11 @@ export async function undeliveredLegacyUserCommentIds( // native identity/process metadata is not proof that its prompt was sent. // Admission separately verifies process termination before a new turn. sql`not (${heartbeatRuns.status} = 'cancelled' and ${heartbeatRuns.startedAt} is null)`, + // A claimed turn parked for a busy workspace stopped before provider + // dispatch; the host records that it sent no prompt. + sql`not coalesce(${heartbeatRuns.status} = 'cancelled' and ${heartbeatRuns.errorCode} = 'workspace_busy' + and ${heartbeatRuns.resultJson}#>>'{executionRecovery,kind}' = 'workspace_wait' + and ${heartbeatRuns.resultJson}#>'{executionRecovery,providerWorkStarted}' = 'false'::jsonb, false)`, or(...commentIds.map(id => or( sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId' = ${id}`, sql`${heartbeatRuns.contextSnapshot}->'wakeCommentIds' @> ${JSON.stringify([id])}::jsonb`, @@ -295,7 +300,7 @@ export async function admitExplicitNativeContinuation(input: { if (!run || run.agentId !== agentId || !terminal.includes(run.status) || (run.nativeIssueId ?? run.issueId ?? run.contextSnapshot?.issueId) !== issueId || !run.finishedAt) return blocked("source_unavailable", "The previous execution has not finished or its owner changed. Your message is saved."); - if (!queuedInterrupt && !queuedRequest && authorizedAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue."); + if (!queuedInterrupt && !queuedRequest && authorizedAt <= userStopBoundaryAt(run)!) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue."); if (adapterExecutionControls.has(run.id)) return blocked("execution_settling", "Waiting for the previous run to stop. Your message will start automatically."); const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, run.id), @@ -306,6 +311,9 @@ export async function admitExplicitNativeContinuation(input: { if (!lockedRun || lockedRun.status !== run.status || lockedRun.agentId !== run.agentId || lockedRun.finishedAt?.getTime() !== run.finishedAt.getTime()) return null; run = lockedRun; + // A user's message written after their Stop request is new direction even + // when the stopped native run settled later. + const stoppedAt = userStopBoundaryAt(run)!; const failedPreparation = await isFailedNativePreparation(db, run, coordinator); const unusedAdmission = run.status === "cancelled" && !run.startedAt && run.errorCode === "execution_reconciliation_required" && @@ -378,14 +386,14 @@ export async function admitExplicitNativeContinuation(input: { if (partiallyDeliveredQueue && run.runtimeMode !== "native" && !unusedAdmission && !cancelledStartup && !failedPreparation) return null; if ((queuedInterrupt || queuedRequest) && !legacyUserTurn && !unusedAdmission && !(queuedRequest && run.runtimeMode === "native" && - (run.status !== "cancelled" || authorizedAt > run.finishedAt!)) && + (run.status !== "cancelled" || authorizedAt > stoppedAt)) && !(queuedRequest && cancelledStartup && authorizedAt > run.finishedAt!) && !(nativeQueuedSend && (run.runtimeMode === "native" || cancelledStartup) && authorizedAt > run.finishedAt!) && !(queuedInterrupt && response?.source.requiresFreshSession && run.runtimeMode === "native")) return null; // A saved native message sent after Stop is a new user direction. Older // input cannot restart a cancelled run merely because it is still queued. if (queuedRequest && !queuedInterrupt && run.status === "cancelled" && !unusedAdmission && - !canContinueCancelledRun(run) && !(run.runtimeMode === "native" && authorizedAt > run.finishedAt!) && + !canContinueCancelledRun(run) && !(run.runtimeMode === "native" && authorizedAt > stoppedAt) && !(cancelledStartup && authorizedAt > run.finishedAt!)) return null; if (retry && run.status === "cancelled" && !canContinueCancelledRun(run) && !cancelledStartup) return blocked("cancelled_by_operator", "Inspect the stopped run and send a new message to continue."); diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index 9f609e40432..8c61d48fbbd 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -34,6 +34,7 @@ import { canContinueCancelledRun, readRunCancellation, requestedRunCancellation, + userStopBoundaryAt, } from "../run-cancellation.js"; import { settleSlackConversation } from "../slack-conversation-lifecycle.js"; import { readQueuedInteractionResponse } from "../queued-interaction-response.js"; @@ -665,7 +666,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont sql`${issueComments.id}::text = ${commentId}`, eq(issueComments.authorType, "user"), eq(issueComments.authorUserId, requestedByActorId), isNull(issueComments.deletedAt), isNull(issueComments.createdByRunId), - gt(issueComments.createdAt, run.finishedAt), + gt(issueComments.createdAt, userStopBoundaryAt(run)!), )); if (!comment?.body.trim()) continue; } else { diff --git a/server/src/services/run-cancellation.ts b/server/src/services/run-cancellation.ts index ab2687e8066..009735273e3 100644 --- a/server/src/services/run-cancellation.ts +++ b/server/src/services/run-cancellation.ts @@ -135,6 +135,21 @@ export function canContinueCancelledRun(run: Run): boolean { run.resultJson?.acpToolInventoryComplete === true && run.resultJson?.acpPendingToolCount === 0; } +/** When a user's Stop or interrupt took effect for message ordering. A native + * run finishes only after its provider stop and workspace save settle, which + * can take many seconds. A message written after the Stop request is a new + * direction even when that settlement finished later, so compare against the + * recorded request rather than `finishedAt`. Other cancellations keep + * `finishedAt`. */ +export function userStopBoundaryAt(run: Pick): Date | null { + if (!run.finishedAt) return null; + const cancellation = run.status === "cancelled" ? readRunCancellation(run.resultJson) : null; + if (!cancellation || cancellation.initiator.type !== "user" || + !["operator", "queued_message"].includes(cancellation.source)) return run.finishedAt; + const requestedAt = new Date(cancellation.recordedAt); + return Number.isFinite(requestedAt.getTime()) && requestedAt < run.finishedAt ? requestedAt : run.finishedAt; +} + export function isUnexpectedRunCancellation(run: Run): boolean { const bootstrap = run.resultJson?.executionRecovery as Record | undefined; return run.status === "cancelled" && Boolean(run.startedAt) && diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index b99542a255b..95629876651 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -4747,9 +4747,14 @@ describe("automatic retained workspace recovery", () => { const { f, original, reset: redirect, wake, queued, deps, control, execution, queueDeps, queue } = t; await db.update(issueComments).set({ body: "Actually, stop the essay. Instead give me a 5-bullet summary." }) .where(eq(issueComments.id, redirect.id)); - // Older queued input is not a new direction and must not undo the Stop. - if (!sentAfterStop) await db.update(heartbeatRuns).set({ finishedAt: new Date(redirect.createdAt.getTime() + 1_000) }) - .where(eq(heartbeatRuns.id, f.runId)); + // Older queued input is not a new direction and must not undo the Stop: + // the user requested Stop only after writing it. + if (!sentAfterStop) { + const stoppedAt = new Date(redirect.createdAt.getTime() + 1_000), stopped = await run(f.runId); + await db.update(heartbeatRuns).set({ finishedAt: stoppedAt, resultJson: { ...stopped.resultJson, + cancellation: { ...(stopped.resultJson?.cancellation as Record), recordedAt: stoppedAt.toISOString() } } }) + .where(eq(heartbeatRuns.id, f.runId)); + } try { expect(await queue.claimQueuedRun(queued)).toBeNull(); expect((await run(queued.id)).status).toBe("queued"); @@ -4802,6 +4807,108 @@ describe("automatic retained workspace recovery", () => { } finally { deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); } }, 30_000); + it.each([ + { retried: false, title: "continues a message written during Stop settlement once, without a second message, after recovery already resolved" }, + { retried: true, title: "continues a message carried by a workspace-busy retry once, without a second message, after recovery already resolved" }, + ])("$title", async ({ retried }) => { + // S47: the user stops (or interrupts) a native run and sends a message. + // The native source finishes only after its stop and workspace save + // settle, so its finishedAt is later than the message. Recovery resolves + // first, then the dispatch gate withholds the unstarted follow-up behind + // the recovered hold ("Saved work recovered. Send a message to continue."). + // The message the user already sent must be saved and continued by the + // ordinary deferred-receipt worker, not dropped. + const t = await queuedResetFixture(false); + const { f, original, reset: message, wake, queued, deps, control, execution, queueDeps, queue } = t; + queueDeps.resumeExecutionWaitComments.mockImplementation(control.resumeExecutionWaitComments); + queueDeps.releaseIssueExecutionAndPromote.mockImplementation(control.releaseIssueExecutionAndPromote); + await db.update(issueComments).set({ body: "Actually, give me a 5-bullet summary instead." }).where(eq(issueComments.id, message.id)); + // The Stop was requested before the message (fixture); settlement finished after it. + await db.update(heartbeatRuns).set({ finishedAt: new Date(message.createdAt.getTime() + 1_000) }) + .where(eq(heartbeatRuns.id, f.runId)); + let pending = queued; + let systemWakeId: string | null = null; + let expectedIds = [message.id]; + if (retried) { + // The first follow-up was claimed, met the stopped run's unsettled + // workspace save, and was parked as a system-owned workspace-busy retry + // carrying the message. While that retry waited, the user added a second + // message, which coalesced into it; a further retry carried both. + const busy = () => ({ status: "cancelled" as const, startedAt: new Date(), finishedAt: new Date(), errorCode: "workspace_busy", + resultJson: { executionRecovery: { kind: "workspace_wait", providerWorkStarted: false }, + cancellation: { source: "control_plane", expected: true, initiator: { type: "system" }, reason: "Waiting for the shared execution workspace", recordedAt: new Date().toISOString() } } }); + await db.update(heartbeatRuns).set(busy()).where(eq(heartbeatRuns.id, queued.id)); + await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, wake.id)); + const retry = async (parentId: string, context: Record, status: "queued" | "cancelled") => { + const [systemWake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: original.agentId, + source: "automation", triggerDetail: "system", reason: "workspace_busy_retry", status: status === "queued" ? "claimed" : "cancelled", + claimedAt: new Date(), requestedByActorType: "system", payload: { issueId: f.issueId, retryOfRunId: parentId } }).returning(); + const [retryRun] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: original.agentId, + invocationSource: "automation", triggerDetail: "system", status: "queued", runtimeMode: "legacy", responsibleUserId: "operator", + wakeupRequestId: systemWake.id, retryOfRunId: parentId, scheduledRetryReason: "workspace_busy", scheduledRetryAttempt: 1, + contextSnapshot: { ...context, wakeReason: "workspace_busy_retry", retryOfRunId: parentId, retryReason: "workspace_busy" } }).returning(); + await db.update(agentWakeupRequests).set({ runId: retryRun.id }).where(eq(agentWakeupRequests.id, systemWake.id)); + return { retryRun, systemWake }; + }; + const first = await retry(queued.id, queued.contextSnapshot ?? {}, "cancelled"); + const [second] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: "operator", body: "Also keep it under 100 words." }).returning(); + await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: original.agentId, + source: "on_demand", triggerDetail: "manual", reason: "issue_commented", status: "coalesced", runId: first.retryRun.id, + requestedByActorType: "user", requestedByActorId: "operator", + payload: { issueId: f.issueId, commentId: second.id, mutation: "comment", coalescedIntoWakeupRequestId: first.systemWake.id } }); + await db.update(heartbeatRuns).set(busy()).where(eq(heartbeatRuns.id, first.retryRun.id)); + expectedIds = [message.id, second.id]; + const carried = { ...queued.contextSnapshot, wakeCommentId: second.id, wakeCommentIds: expectedIds }; + const last = await retry(first.retryRun.id, carried, "queued"); + pending = last.retryRun; + await db.update(issues).set({ executionRunId: pending.id }).where(eq(issues.id, f.issueId)); + systemWakeId = last.systemWake.id; + } + try { + expect(await queue.claimQueuedRun(pending)).toBeNull(); + await recordRequiredWorkspaceRestoreFailure(db, await run(f.runId), { workspaceRestoreFailure: "restore_failed" }, f.lease); + const [hold] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, f.companyId), + eq(issueRecoveryActions.sourceIssueId, f.issueId), eq(issueRecoveryActions.fingerprint, `native-workspace:${f.runId}`))); + f.actionId = hold.id; f.input.actionId = hold.id; + await recoverRetainedWorkspace(f.input); + await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success", metadata: { ...f.lease.metadata, + remoteExecutionTermination: remoteTerminationReceipt(f.lease, { providerLeaseId: f.providerLeaseId, state: "stopped" }) } }).where(eq(environmentLeases.id, f.lease.id)); + await control.acknowledgeRemoteStop(f.runId, f.companyId); + deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); + // Recovery resolves and its completion hooks run before the follow-up is withheld. + await control.resumeRemoteStopComments(await run(f.runId)); + await vi.waitFor(async () => { + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, async current => { + if (current.id === f.runId) await control.resumeRemoteStopComments(current); + }) ?? []); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + }, { timeout: 10_000, interval: 10 }); + expect(await queue.claimQueuedRun(await run(pending.id))).toBeNull(); + expect(await run(pending.id)).toMatchObject({ status: "cancelled", errorCode: "execution_reconciliation_required", startedAt: null }); + const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + expect(saved).toMatchObject({ status: "deferred_issue_execution", runId: null, + payload: { executionWait: { reason: "execution_recovery", recoveryActionId: hold.id } } }); + expect(saved.payload?._paperclipWakeContext).toMatchObject({ wakeReason: "issue_commented", wakeCommentIds: expectedIds }); + if (systemWakeId) { + const [systemWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, systemWakeId)); + expect(systemWake.status).toBe("skipped"); + } + + // Only the deferred-receipt worker runs (its 30-second wait has elapsed). + await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 31_000) }).where(eq(agentWakeupRequests.id, wake.id)); + await queue.resumeDeferredExecutionReceipts({ companyId: f.companyId, requestId: wake.id }); + await queue.resumeDeferredExecutionReceipts({ companyId: f.companyId, requestId: wake.id }); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${f.issueId}`, eq(heartbeatRuns.status, "queued"))); + expect(successors).toHaveLength(1); + expect(successors[0].contextSnapshot?.wakeCommentIds).toEqual(expectedIds); + expect(deps.enqueueWakeup).toHaveBeenCalledWith(original.agentId, + expect.objectContaining({ requestedByActorType: "user", requestedByActorId: "operator" }), wake.id); + expect(queueDeps.executeRun).not.toHaveBeenCalled(); + } finally { deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); } + }, 30_000); + it("reaches its queued snapshot-only reset after earlier recovery candidates without starting agent work", async () => { const previous = db; db = createDb(connectionString);