mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-14 00:26:28 +01:00
fix(retry): carry the original wake to every re-queued run (#20989)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A run serves one wake: a comment, an answer, an assignment, or a one-shot task monitor check (`issue_monitor_due`). > - When a run fails, Paperclip often starts a successor run by itself: the bounded transient retry, the uncharged `interrupted_preparation` re-queue, workspace or connection waits, missing-comment repair, capacity retries and recovery successors. Each successor links to the failed run through `retry_of_run_id`. > - Most of these re-queues replace the wake reason with their own (for example `transient_failure_retry`). The agent then does not know which wake it must handle. > - #20970 fixed this for the board Retry only. A no-fault chaos wave showed the same loss on an automatic retry: the monitor run failed before provider work, the automatic retry succeeded, and the agent did not treat it as the monitor check. The monitor was already consumed, so the task rested `in_progress` with nothing pending. > - This pull request covers all re-queue paths at one shared point: the continuation builder that every issue dispatch uses (`buildExecutionContinuation`, called from `heartbeat/task-context.ts`). It walks the re-queue chain to the first run that was not a re-queue and records that run's wake reason as `trigger.retriedReason`. The wake prompt already shows that field (#20970). > - The benefit is that every successor of a failed run handles the same wake as the run it replaces. ## Linked Issues or Issue Description **What happened?** No-fault restart-chaos wave on master `87fe4d9793`, volume-free mode (`w-20261112T010036-s6801`, Claude, task t01, workload W2). The monitor wake started run `78e27ef9`. It failed before provider work (`native_predecessor_retirement_unproven`). The `interrupted_preparation` re-queue (attempt 1/12) started run `0cd387a0` on a fresh sandbox, and that run succeeded. The agent reported that no board comment arrived and that the monitor wake was not visible. It yielded `response_wake` again. The monitor was `triggered` with no `nextCheckAt`. The issue rested `in_progress` with `canRetry=false`, no live run, no wake and no monitor. The harness graded `R1 rested_not_done` and `R3 missing_retry`. **Expected behavior** A successor of a failed run tells the agent which wake it re-delivers. A re-queued monitor check is handled as the monitor check. **Steps to reproduce** 1. Wake a native agent with `issue_monitor_due` on a task. 2. Make the run fail before provider work, for example when the previous run's sandbox session cannot be retired. 3. Let the `interrupted_preparation` re-queue run. 4. Before this change, the successor's continuation trigger has the re-queue reason only, and its prompt does not name the monitor check. **Additional context** Follow-up to #20970. Refs #20970 Refs #20935 Refs #20863 ## What Changed - `server/src/services/execution-continuation.ts`: new `requeuedWakeReason`. - It starts at the current run's `retry_of_run_id` column. If that is empty, it uses the context `retryOfRunId`, or the source run of a board Retry. - It follows the chain within the same company and issue, at most 32 links, until it finds a run with no re-queue link. It returns that run's wake reason. - A `trigger.retriedReason` already saved on a run in the chain is reused. - `trigger.retriedReason` is set only when the original reason differs from the current reason and is not itself a Retry reason. A re-queue that keeps the original wake reason (`preserveWakeContext`) is unchanged. - This one point covers every successor that links with `retry_of_run_id`: bounded transient retries, `interrupted_preparation` (#20863), workspace and AI-connection waits, missing-comment repair (`run-lifecycle.ts`), native capacity retries, safe replacement, and the board Retry. - Tests: - `server/src/__tests__/heartbeat-native-predispatch-retry.test.ts`: the assembled heartbeat service runs a monitor-due wake, a predecessor-retirement failure before provider work, and the automatic `interrupted_preparation` re-queue. It asserts the successor's saved continuation and wake prompt. - `server/src/services/execution-continuation.test.ts`: a two-link chain of automatic re-queues, and a re-queue that keeps the original reason. ## Verification - Red on master: `cd server && npx vitest run src/__tests__/heartbeat-native-predispatch-retry.test.ts -t "delivers the monitor check"`. The successor's trigger had `sourceRunId` = the failed monitor run, but no `retriedReason`. - Green: `heartbeat-native-predispatch-retry.test.ts`, `execution-continuation.test.ts` and `heartbeat/retries.test.ts` pass (78 tests). - `npx tsc --noEmit -p server` passes. - Chaos: the existing no-fault W2 cell in volume-free mode reproduces this when the monitor run fails before provider work. I did not rerun a wave. ## Risks - Low risk. The change adds one optional trigger field to more wakes. It changes no admission, status or authority path. - The walk costs at most one indexed primary-key read per chain link, only for runs that are re-queues. - The walk follows only same-company, same-issue runs. Old saved runs work, because they already store `wakeReason` and the `retry_of_run_id` column. - The resting state "agent yielded `response_wake` after the monitor was consumed" is still possible when an agent really waits. #20935 shows it on the run projection as "Waiting for a reply". The ownership reconciler's I7 check still counts it as `no_owner_no_wake`. This PR does not change I7. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file edits, local test execution). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
8818402e7a
commit
5d1bb5192f
3 files changed
+110
-7
No files matched your search
@@ -4,6 +4,8 @@ import { activityLog, agents, authUsers, companyMemberships, createDb, environme
|
||||
issueRecoveryActions, issues } from "@paperclipai/db";
|
||||
import { NativePredecessorRetirementUnprovenError } from "../services/native-runtime/predecessor-retirement.js";
|
||||
import type { NativeRunIdentity, NativeSession, NativeSessionBackend, PrpEvent } from "@paperclipai/paperclip-runner";
|
||||
import type { ExecutionContinuationEnvelope } from "@paperclipai/shared";
|
||||
import { renderPaperclipWakePrompt } from "@paperclipai/adapter-utils/server-utils";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { heartbeatService, type HeartbeatServiceOptions } from "../services/heartbeat.js";
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
@@ -34,7 +36,7 @@ vi.mock("../services/budget-reservations.js", async importOriginal => {
|
||||
const support = await getRecoveryPostgresSupport({});
|
||||
|
||||
function nativeBackend(db: ReturnType<typeof createDb>, options: { failDescriptorOnce?: { failed: boolean };
|
||||
unretiredPredecessorOnce?: { failed: boolean; environmentId: string } } = {}):
|
||||
unretiredPredecessorOnce?: { failed: boolean; environmentId: string }; turnMessages?: Map<string, string> } = {}):
|
||||
NonNullable<HeartbeatServiceOptions["nativeSessionBackendFactory"]> {
|
||||
return execution => {
|
||||
let identity: NativeRunIdentity;
|
||||
@@ -45,7 +47,8 @@ function nativeBackend(db: ReturnType<typeof createDb>, options: { failDescripto
|
||||
identity: () => identity,
|
||||
capabilities: async () => capabilities,
|
||||
attachRun: async input => { identity = input.identity; },
|
||||
startTurn: async () => {
|
||||
startTurn: async input => {
|
||||
options.turnMessages?.set(identity.runId, JSON.stringify(input.message));
|
||||
const turnId = randomUUID();
|
||||
turn.resolve(turnId);
|
||||
return { turnId };
|
||||
@@ -204,6 +207,51 @@ describe.skipIf(!support.supported)("heartbeat native pre-dispatch failure retry
|
||||
}
|
||||
}, 90_000);
|
||||
|
||||
it("delivers the monitor check to the automatic retry of a monitor run that failed before provider work", async () => {
|
||||
const { companyId, agentId } = fixture.current;
|
||||
const db = createDb(fixture.current.temporary.connectionString, { maxConnections: 4 });
|
||||
const [environment] = await db.insert(environments).values({ name: `reused-${randomUUID()}`, driver: "sandbox" }).returning();
|
||||
const turnMessages = new Map<string, string>();
|
||||
const heartbeat = heartbeatService(db, { nativeSessionBackendFactory: nativeBackend(db,
|
||||
{ unretiredPredecessorOnce: { failed: false, environmentId: environment!.id }, turnMessages }) });
|
||||
try {
|
||||
await db.update(agents).set({ status: "idle", adapterConfig: { provider: "codex" } }).where(eq(agents.id, agentId));
|
||||
await instanceSettingsService(db).updateExperimental({ enableNativeRunner: true });
|
||||
const userId = randomUUID();
|
||||
await db.insert(authUsers).values({ id: userId, name: "Task owner", email: `${userId}@example.invalid`,
|
||||
createdAt: new Date(), updatedAt: new Date() });
|
||||
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner" });
|
||||
const [issue] = await db.insert(issues).values({ companyId, title: "Phase 2 on the monitor check", status: "in_progress",
|
||||
assigneeAgentId: agentId, responsibleUserId: userId, monitorNotes: "Phase 2 check" }).returning();
|
||||
const run = await heartbeat.wakeup(agentId, { source: "automation", triggerDetail: "system", reason: "issue_monitor_due",
|
||||
payload: { issueId: issue!.id }, contextSnapshot: { issueId: issue!.id, taskKey: issue!.id,
|
||||
source: "issue.monitor", wakeReason: "issue_monitor_due", monitorNotes: "Phase 2 check" } });
|
||||
expect(run).not.toBeNull();
|
||||
await waitForCondition("the monitor run to settle", () => heartbeat.getTaskDrainStatus().quiescent);
|
||||
const [failed] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run!.id));
|
||||
expect(failed).toMatchObject({ status: "failed", resultJson: expect.objectContaining({
|
||||
failureCause: "native_predecessor_retirement_unproven" }) });
|
||||
const [successor] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId),
|
||||
eq(heartbeatRuns.retryOfRunId, run!.id)));
|
||||
expect(successor).toMatchObject({ scheduledRetryReason: "interrupted_preparation" });
|
||||
|
||||
expect(await heartbeat.promoteDueScheduledRetries()).toMatchObject({ promoted: 1 });
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
await waitForCondition("the retry to start its provider turn", () => turnMessages.has(successor!.id));
|
||||
await waitForCondition("the retry to settle", () => heartbeat.getTaskDrainStatus().quiescent);
|
||||
const [retried] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, successor!.id));
|
||||
const continuation = retried!.contextSnapshot?.executionContinuation as ExecutionContinuationEnvelope;
|
||||
expect(continuation.trigger).toMatchObject({ sourceRunId: run!.id, retriedReason: "issue_monitor_due" });
|
||||
expect(renderPaperclipWakePrompt(retried!.contextSnapshot?.paperclipWake)).toContain("- retried wake: issue_monitor_due");
|
||||
expect(turnMessages.get(successor!.id)).toContain("- retried wake: issue_monitor_due");
|
||||
expect(turnMessages.has(run!.id)).toBe(false);
|
||||
} finally {
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
await closeIdleWarmNativeSessionsForRestart();
|
||||
await db.$client.end();
|
||||
}
|
||||
}, 90_000);
|
||||
|
||||
it("keeps the hold when the failure follows entry into the native executor", async () => {
|
||||
const { companyId } = fixture.current;
|
||||
const db = createDb(fixture.current.temporary.connectionString, { maxConnections: 4 });
|
||||
|
||||
@@ -396,6 +396,33 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
}
|
||||
});
|
||||
|
||||
it("carries the original wake through a chain of automatic re-queues", async () => {
|
||||
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
const firstRetryId = randomUUID(), secondRetryId = randomUUID();
|
||||
await db.update(heartbeatRuns).set({ status: "failed", errorCode: "adapter_failed",
|
||||
contextSnapshot: { issueId, wakeReason: "issue_monitor_due", monitorNotes: "Phase 2 check" } })
|
||||
.where(eq(heartbeatRuns.id, runId));
|
||||
try {
|
||||
await db.insert(heartbeatRuns).values([
|
||||
{ id: firstRetryId, companyId, agentId, status: "failed", retryOfRunId: runId,
|
||||
contextSnapshot: { issueId, wakeReason: "transient_failure_retry", retryReason: "interrupted_preparation" } },
|
||||
{ id: secondRetryId, companyId, agentId, status: "running", retryOfRunId: firstRetryId,
|
||||
contextSnapshot: { issueId, wakeReason: "transient_failure_retry" } },
|
||||
]);
|
||||
const envelope = await buildExecutionContinuation({ db, companyId, issueId, agentId, runId: secondRetryId,
|
||||
context: { wakeReason: "transient_failure_retry" }, summary: null, exposeLowTrustRaw: false });
|
||||
expect(envelope.trigger).toMatchObject({ reason: "transient_failure_retry", retriedReason: "issue_monitor_due" });
|
||||
const preserved = await buildExecutionContinuation({ db, companyId, issueId, agentId, runId: firstRetryId,
|
||||
context: { wakeReason: "issue_monitor_due" }, summary: null, exposeLowTrustRaw: false });
|
||||
expect(preserved.trigger).not.toHaveProperty("retriedReason");
|
||||
} finally {
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, secondRetryId));
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, firstRetryId));
|
||||
await db.update(heartbeatRuns).set({ status: "failed", errorCode: null, contextSnapshot: source.contextSnapshot })
|
||||
.where(eq(heartbeatRuns.id, runId));
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps Local CLI run-authored comments as history without promoting them to human direction", async () => {
|
||||
const id = randomUUID();
|
||||
await db.insert(issueComments).values({ id, companyId, issueId, authorType: "user",
|
||||
|
||||
@@ -21,6 +21,7 @@ import { githubRecoveryReceipts } from "./chat-channels/github/recovery-receipts
|
||||
import { readTaskRetryIntent } from "./task-retry.js";
|
||||
|
||||
const retryWakeReasons = new Set(["task_retry", "retry_failed_run"]);
|
||||
const maxRequeueChain = 32;
|
||||
|
||||
const savedRecordOperations = new Set([
|
||||
"PUT /api/issues/{id}/documents/{key}",
|
||||
@@ -194,6 +195,35 @@ export async function currentContinuationOrigins(
|
||||
}
|
||||
|
||||
/** Re-read task scope at dispatch, including messages already delivered to an earlier provider session. */
|
||||
function requeueSourceRunId(context: Record<string, unknown>, retryOfRunId: string | null): string | null {
|
||||
if (retryOfRunId) return retryOfRunId;
|
||||
const retryOf = string(context.retryOfRunId);
|
||||
if (retryOf) return retryOf;
|
||||
if (!retryWakeReasons.has(string(context.wakeReason) ?? "")) return null;
|
||||
return string(object(context.explicitUserContinuation).previousRunId) ?? string(context.previousRunId);
|
||||
}
|
||||
|
||||
async function requeuedWakeReason(db: Db, input: {
|
||||
companyId: string; issueId: string; runId?: string; context: Record<string, unknown>;
|
||||
}): Promise<string | null> {
|
||||
const [current] = input.runId ? await db.select({ retryOfRunId: heartbeatRuns.retryOfRunId }).from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId))) : [];
|
||||
let next = requeueSourceRunId(input.context, current?.retryOfRunId ?? null);
|
||||
const visited = new Set<string>();
|
||||
while (next && !visited.has(next) && visited.size < maxRequeueChain) {
|
||||
visited.add(next);
|
||||
const [run] = await db.select({ context: heartbeatRuns.contextSnapshot, retryOfRunId: heartbeatRuns.retryOfRunId })
|
||||
.from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, next)));
|
||||
const context = object(run?.context);
|
||||
if (!run || context.issueId !== input.issueId) return null;
|
||||
const carried = string(object(object(context.executionContinuation).trigger).retriedReason);
|
||||
if (carried) return carried;
|
||||
next = requeueSourceRunId(context, run.retryOfRunId);
|
||||
if (!next) return string(context.wakeReason);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
export async function buildExecutionContinuation(input: {
|
||||
db: Db;
|
||||
companyId: string;
|
||||
@@ -537,11 +567,9 @@ export async function buildExecutionContinuation(input: {
|
||||
lastTerminal.status === "interrupted" || lastTerminal.errorCode === "process_lost")
|
||||
? lastTerminal.id : undefined);
|
||||
const triggerReason = string(input.context.wakeReason) ?? "task_execution";
|
||||
const sourceContext = object(sourceRun?.context);
|
||||
const retriedCandidate = retryWakeReasons.has(triggerReason) && sourceRun
|
||||
? string(object(object(sourceContext.executionContinuation).trigger).retriedReason) ?? string(sourceContext.wakeReason)
|
||||
: null;
|
||||
const retriedReason = retriedCandidate && !retryWakeReasons.has(retriedCandidate) ? retriedCandidate : null;
|
||||
const retriedCandidate = await requeuedWakeReason(db, { companyId, issueId, runId: input.runId, context: input.context });
|
||||
const retriedReason = retriedCandidate && retriedCandidate !== triggerReason && !retryWakeReasons.has(retriedCandidate)
|
||||
? retriedCandidate : null;
|
||||
return {
|
||||
...(interruptedRunId ? { interruptedRunId } : {}),
|
||||
...(resumeDelta ? { resumeDelta } : {}),
|
||||
|
||||
Reference in new issue
Block a user