diff --git a/server/src/__tests__/heartbeat-native-predispatch-retry.test.ts b/server/src/__tests__/heartbeat-native-predispatch-retry.test.ts index bb42b3bf482..7ffd99c0dd4 100644 --- a/server/src/__tests__/heartbeat-native-predispatch-retry.test.ts +++ b/server/src/__tests__/heartbeat-native-predispatch-retry.test.ts @@ -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, options: { failDescriptorOnce?: { failed: boolean }; - unretiredPredecessorOnce?: { failed: boolean; environmentId: string } } = {}): + unretiredPredecessorOnce?: { failed: boolean; environmentId: string }; turnMessages?: Map } = {}): NonNullable { return execution => { let identity: NativeRunIdentity; @@ -45,7 +47,8 @@ function nativeBackend(db: ReturnType, 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(); + 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 }); diff --git a/server/src/services/execution-continuation.test.ts b/server/src/services/execution-continuation.test.ts index 24e108e18ac..947eb1d42ef 100644 --- a/server/src/services/execution-continuation.test.ts +++ b/server/src/services/execution-continuation.test.ts @@ -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", diff --git a/server/src/services/execution-continuation.ts b/server/src/services/execution-continuation.ts index 5e2dc7aa6c8..816822956b1 100644 --- a/server/src/services/execution-continuation.ts +++ b/server/src/services/execution-continuation.ts @@ -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, 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; +}): Promise { + 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(); + 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 } : {}),