From a244ad3b739add3219ea0c3d84e01cea85ba9b2e Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 11 Nov 2026 10:57:52 -0600 Subject: [PATCH] fix(recovery): retry a system-woken run once when it lost its runner before any turn A system wake whose remote run lost its Runner before any turn or tool event rested on a manual Retry, although nothing it did could be repeated. requestInterruptedNativeTurnRetry now requests the Retry once under the run's responsible user in exactly that case; every other system wake keeps the visible Retry. Co-Authored-By: Claude Opus 5.5 --- doc/execution-semantics.md | 5 ++- .../heartbeat/restart-interrupted-turn.ts | 32 +++++++++++++++---- .../workspace-recovery-retry-copyback.test.ts | 29 +++++++++++++---- 3 files changed, 52 insertions(+), 14 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 40903c46bee..db74b625c1d 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1711,7 +1711,10 @@ assigned the task, or clicked Retry. It does this only when the run recorded no provider turn or its last provider turn is settled. Activity records the request as `restart-recovery` on behalf of that user. Admission keeps every gate: saved-work recovery, stop proof, approvals, pauses, budget and ownership. A Retry that the -server requested gets no second automatic attempt. System wakes and runs with an +server requested gets no second automatic attempt. A system wake (for example a +monitor or an automatic assignment) continues this way only when its run recorded +no turn event and no tool event at all, and only under the run's responsible user; +nothing such a run did can be repeated. Other system wakes and runs with an in-flight provider turn do not continue automatically. For these failed runs, a recovered workspace hold offers Retry (`canRetry`) and says "Retry or send a message to continue." An intentional Stop keeps "send a message" as its only diff --git a/server/src/services/heartbeat/restart-interrupted-turn.ts b/server/src/services/heartbeat/restart-interrupted-turn.ts index a1458a27826..0be97b12c02 100644 --- a/server/src/services/heartbeat/restart-interrupted-turn.ts +++ b/server/src/services/heartbeat/restart-interrupted-turn.ts @@ -36,6 +36,17 @@ export function isProvenTurnSettlement(event: { eventType: string; payload: unkn asRecord(asRecord(payload.prpEvent).payload).providerTerminalObserved !== false; } +/** Any turn or tool event: the run may have done provider work that a new turn could repeat. */ +const PROVIDER_TURN_ACTIVITY_EVENTS = [...TURN_STARTED_EVENTS, ...TURN_SETTLED_EVENTS, + "tool.execution.started", "tool.execution.completed"]; + +async function hasProviderTurnActivity(db: Db, run: { companyId: string; id: string }) { + const [event] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + inArray(heartbeatRunEvents.eventType, PROVIDER_TURN_ACTIVITY_EVENTS))).limit(1); + return Boolean(event); +} + /** Automatic continuation is requested at most once per chain of runs. */ export const RESTART_RETRY_KEY_PREFIX = "restart-interrupted-turn:"; @@ -47,7 +58,9 @@ type Outcome = { requested: true; requestId: string } | { requested: false; reas * request the same Retry a user would click, once and idempotently per run, * under the requesting user's authority. Only when the run's user input was not * yet acted on, or its last provider turn is settled; an in-flight turn's - * outcome is unknown, so it keeps the visible Retry. Admission keeps every gate, + * outcome is unknown, so it keeps the visible Retry. A system wake continues + * this way only when its run recorded no turn or tool event at all, under the + * run's responsible user. Admission keeps every gate, * including stop proof and saved-work recovery before the next turn. */ export async function requestRestartInterruptedTurnRetry(db: Db, claim: NativeRestartRecoveryClaim): Promise { @@ -71,11 +84,17 @@ export async function requestInterruptedNativeTurnRetry(db: Db, runId: string): const [wake] = await db.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))); - const actorId = wake?.requestedByActorType === "user" ? wake.requestedByActorId : null; + const userActorId = wake?.requestedByActorType === "user" ? wake.requestedByActorId : null; const reason = typeof context.wakeReason === "string" ? context.wakeReason : wake?.reason; - // System wakes keep their policy: only a new user turn continues them. - if (!actorId) return { requested: false, reason: "not_user_request" }; - if (reason === "task_retry") { + let actorId = userActorId; + if (!userActorId) { + // System wakes keep their policy (only a new user turn continues them), + // except when the run never started a provider turn: nothing it did can be + // repeated, so the Retry runs once under the run's responsible user. + if (!wake || reason === "task_retry" || !run.responsibleUserId || await hasProviderTurnActivity(db, run)) + return { requested: false, reason: "not_user_request" }; + actorId = run.responsibleUserId; + } else if (reason === "task_retry") { // A Retry this hook already requested gets no second automatic attempt. const retryId = typeof context.taskRetryRequestId === "string" ? context.taskRetryRequestId : null; const [receipt] = retryId ? await db.select({ key: agentWakeupRequests.idempotencyKey }).from(agentWakeupRequests).where(and( @@ -88,10 +107,11 @@ export async function requestInterruptedNativeTurnRetry(db: Db, runId: string): const [comment] = commentId ? await db.select({ id: issueComments.id }).from(issueComments).where(and( eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId), sql`${issueComments.id}::text = ${commentId}`, eq(issueComments.authorType, "user"), - eq(issueComments.authorUserId, actorId), isNull(issueComments.deletedAt), isNull(issueComments.createdByRunId), + eq(issueComments.authorUserId, userActorId), isNull(issueComments.deletedAt), isNull(issueComments.createdByRunId), )) : []; if (!comment && reason !== "issue_assigned") return { requested: false, reason: "message_unavailable" }; } else return { requested: false, reason: "not_user_request" }; + if (!actorId) return { requested: false, reason: "not_user_request" }; const [issue] = await db.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId }).from(issues) .where(and(eq(issues.companyId, run.companyId), eq(issues.id, issueId))); if (!issue || issue.assigneeAgentId !== run.agentId || ["done", "cancelled"].includes(issue.status)) diff --git a/server/src/services/workspace-recovery-retry-copyback.test.ts b/server/src/services/workspace-recovery-retry-copyback.test.ts index 39bbad5f77b..41040f959c4 100644 --- a/server/src/services/workspace-recovery-retry-copyback.test.ts +++ b/server/src/services/workspace-recovery-retry-copyback.test.ts @@ -114,6 +114,12 @@ describe("automatic retained workspace recovery", () => { } const queuedSuccessors = async (companyId: string) => db.select().from(heartbeatRuns) .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "queued"))); + async function systemWake(f: Awaited>) { + await db.update(agentWakeupRequests).set({ requestedByActorType: "system", requestedByActorId: null, + reason: "issue_monitor_due" }).where(eq(agentWakeupRequests.id, f.wake.id)); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, wakeReason: "issue_monitor_due" } }) + .where(eq(heartbeatRuns.id, f.runId)); + } it.each([ { title: "the first attachment found the previous runner gone", turn: "none" as const, error: "runner_remote_process_identity_unavailable", initial: true }, @@ -122,8 +128,14 @@ describe("automatic retained workspace recovery", () => { { title: "a same-run retry found the settled turn's remote runner gone", turn: "settled" as const, error: "runner_remote_process_identity_unavailable", initial: false }, { title: "a same-run retry found the settled turn's sandbox lease gone", turn: "settled" as const, error: "native_remote_recovery_lease_mismatch", initial: false }, { title: "a same-run retry named the lease check that failed", turn: "settled" as const, error: "native_remote_recovery_lease_mismatch: stage=environment_acquire field=lease_status", initial: false }, - ])("continues a user turn once, without a second message, after $title and recovery saved the work", async ({ turn, error, initial }) => { + // Staging class T (c14bab19): a monitor wake's session snapshot timed out + // before any turn, then the same-run retry found the Runner gone. No turn or + // tool event means nothing can be repeated, so the run's responsible user's + // Retry continues it once. + { title: "a system wake's run found its runner gone before any provider turn", turn: "none" as const, error: "runner_remote_process_identity_unavailable", initial: false, system: true }, + ])("continues a user turn once, without a second message, after $title and recovery saved the work", async ({ turn, error, initial, system }) => { const f = await restartAllocationLostSource(turn, error); + if (system) await systemWake(f); const service = heartbeatService(db, { environmentRuntime: f.environmentRuntime }); // Keep the agent's only slot busy so the successor stays observable as queued. await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); @@ -164,18 +176,21 @@ describe("automatic retained workspace recovery", () => { it.each([ { title: "its provider turn was still in flight", change: "turn" }, { title: "its provider turn ended without an observed provider terminal", change: "unproven" }, - { title: "the source was a system wake", change: "system" }, + { title: "the source was a system wake whose provider turn settled", change: "system" }, + { title: "the source was a system wake whose run used a tool before its turn event", change: "system_tool" }, + { title: "the source was a system wake whose run had no responsible user", change: "system_unowned" }, { title: "the failure was not a lost runner or allocation", change: "error" }, // Staging class P: the new sandbox had no Runner state to resume the saved // provider session. The Runner now names it, so the run ends with the // permanent code and Retry continues in a fresh provider session. { title: "the Runner had no saved state to resume the provider session", change: "unrestorable" }, ])("does not continue automatically when $title, and offers a Retry that works on first use", async ({ change }) => { - const f = await restartAllocationLostSource(change === "turn" ? "in_flight" : change === "unproven" ? "unproven" : "none"); - if (change === "system") await db.update(agentWakeupRequests).set({ requestedByActorType: "system", requestedByActorId: null, - reason: "issue_children_completed" }).where(eq(agentWakeupRequests.id, f.wake.id)); - if (change === "system") await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, wakeReason: "issue_children_completed" } }) - .where(eq(heartbeatRuns.id, f.runId)); + const f = await restartAllocationLostSource(change === "turn" ? "in_flight" : change === "unproven" ? "unproven" + : change === "system" ? "settled" : "none"); + if (change.startsWith("system")) await systemWake(f); + if (change === "system_tool") await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, + agentId: f.agentId, eventType: "tool.execution.started", stream: "system", level: "info", message: "tool" }); + if (change === "system_unowned") await db.update(heartbeatRuns).set({ responsibleUserId: null }).where(eq(heartbeatRuns.id, f.runId)); if (change === "error") await db.update(heartbeatRuns).set({ error: "native_provider_terminal_failed" }).where(eq(heartbeatRuns.id, f.runId)); if (change === "unrestorable") await db.update(heartbeatRuns).set({ errorCode: "native_provider_state_incompatible", error: "native_session_recovery_failed: NativeRunnerResumeStateUnavailableError: PRP provider resume state is unavailable" })