mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-13 07:12:00 +01:00
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:
1 parent
3ded7bbd91
commit
a244ad3b73
3 files changed
+52
-14
No files matched your search
@@ -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
|
||||
|
||||
@@ -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<Outcome> {
|
||||
@@ -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))
|
||||
|
||||
@@ -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<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([
|
||||
{ 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" })
|
||||
|
||||
Reference in new issue
Block a user