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 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 committed 2026-11-11 10:57:52 -06:00
1 parent 3ded7bbd91
commit a244ad3b73
3 files changed
+52 -14

No files matched your search

+4 -1
View File
@@ -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 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 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 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 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 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 message to continue." An intentional Stop keeps "send a message" as its only
@@ -36,6 +36,17 @@ export function isProvenTurnSettlement(event: { eventType: string; payload: unkn
asRecord(asRecord(payload.prpEvent).payload).providerTerminalObserved !== false; 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. */ /** Automatic continuation is requested at most once per chain of runs. */
export const RESTART_RETRY_KEY_PREFIX = "restart-interrupted-turn:"; 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, * 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 * 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 * 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. * including stop proof and saved-work recovery before the next turn.
*/ */
export async function requestRestartInterruptedTurnRetry(db: Db, claim: NativeRestartRecoveryClaim): Promise<Outcome> { export async function requestRestartInterruptedTurnRetry(db: Db, claim: NativeRestartRecoveryClaim): Promise<Outcome> {
@@ -71,11 +84,17 @@ export async function requestInterruptedNativeTurnRetry(db: Db, runId: string):
const [wake] = await db.select().from(agentWakeupRequests).where(and( const [wake] = await db.select().from(agentWakeupRequests).where(and(
eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId),
eq(agentWakeupRequests.id, run.wakeupRequestId), eq(agentWakeupRequests.runId, run.id))); 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; const reason = typeof context.wakeReason === "string" ? context.wakeReason : wake?.reason;
// System wakes keep their policy: only a new user turn continues them. let actorId = userActorId;
if (!actorId) return { requested: false, reason: "not_user_request" }; if (!userActorId) {
if (reason === "task_retry") { // 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. // A Retry this hook already requested gets no second automatic attempt.
const retryId = typeof context.taskRetryRequestId === "string" ? context.taskRetryRequestId : null; const retryId = typeof context.taskRetryRequestId === "string" ? context.taskRetryRequestId : null;
const [receipt] = retryId ? await db.select({ key: agentWakeupRequests.idempotencyKey }).from(agentWakeupRequests).where(and( 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( const [comment] = commentId ? await db.select({ id: issueComments.id }).from(issueComments).where(and(
eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId), eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId),
sql`${issueComments.id}::text = ${commentId}`, eq(issueComments.authorType, "user"), 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" }; if (!comment && reason !== "issue_assigned") return { requested: false, reason: "message_unavailable" };
} else return { requested: false, reason: "not_user_request" }; } 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) const [issue] = await db.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId }).from(issues)
.where(and(eq(issues.companyId, run.companyId), eq(issues.id, issueId))); .where(and(eq(issues.companyId, run.companyId), eq(issues.id, issueId)));
if (!issue || issue.assigneeAgentId !== run.agentId || ["done", "cancelled"].includes(issue.status)) if (!issue || issue.assigneeAgentId !== run.agentId || ["done", "cancelled"].includes(issue.status))
@@ -114,6 +114,12 @@ describe("automatic retained workspace recovery", () => {
} }
const queuedSuccessors = async (companyId: string) => db.select().from(heartbeatRuns) const queuedSuccessors = async (companyId: string) => db.select().from(heartbeatRuns)
.where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "queued"))); .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "queued")));
async function systemWake(f: Awaited<ReturnType<typeof restartAllocationLostSource>>) {
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([ it.each([
{ title: "the first attachment found the previous runner gone", turn: "none" as const, error: "runner_remote_process_identity_unavailable", initial: true }, { 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 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 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 }, { 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); const f = await restartAllocationLostSource(turn, error);
if (system) await systemWake(f);
const service = heartbeatService(db, { environmentRuntime: f.environmentRuntime }); const service = heartbeatService(db, { environmentRuntime: f.environmentRuntime });
// Keep the agent's only slot busy so the successor stays observable as queued. // 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" }); await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });
@@ -164,18 +176,21 @@ describe("automatic retained workspace recovery", () => {
it.each([ it.each([
{ title: "its provider turn was still in flight", change: "turn" }, { title: "its provider turn was still in flight", change: "turn" },
{ title: "its provider turn ended without an observed provider terminal", change: "unproven" }, { 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" }, { 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 // 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 // provider session. The Runner now names it, so the run ends with the
// permanent code and Retry continues in a fresh provider session. // permanent code and Retry continues in a fresh provider session.
{ title: "the Runner had no saved state to resume the provider session", change: "unrestorable" }, { 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 }) => { ])("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"); const f = await restartAllocationLostSource(change === "turn" ? "in_flight" : change === "unproven" ? "unproven"
if (change === "system") await db.update(agentWakeupRequests).set({ requestedByActorType: "system", requestedByActorId: null, : change === "system" ? "settled" : "none");
reason: "issue_children_completed" }).where(eq(agentWakeupRequests.id, f.wake.id)); if (change.startsWith("system")) await systemWake(f);
if (change === "system") await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, wakeReason: "issue_children_completed" } }) if (change === "system_tool") await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId,
.where(eq(heartbeatRuns.id, 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 === "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", 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" }) error: "native_session_recovery_failed: NativeRunnerResumeStateUnavailableError: PRP provider resume state is unavailable" })