mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-23 17:33:51 +02:00
Guard stranded recovery recursion
This commit is contained in:
1 parent
6e8966a954
commit
4850ae158b
3 files changed
+212
No files matched your search
@@ -921,6 +921,61 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
expect(comments[0]?.body).toContain(`Recovery issue: [${recovery.identifier}]`);
|
||||
});
|
||||
|
||||
it("blocks failed recovery work in place during immediate terminal-run cleanup", async () => {
|
||||
const sourceIssueId = randomUUID();
|
||||
const { companyId, agentId, runId, issueId } = await seedRunFixture({
|
||||
agentStatus: "idle",
|
||||
processPid: 999_999_999,
|
||||
processLossRetryCount: 1,
|
||||
});
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
title: "Recover stalled issue PAP-1",
|
||||
originKind: "stranded_issue_recovery",
|
||||
originId: sourceIssueId,
|
||||
})
|
||||
.where(eq(issues.id, issueId));
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reapOrphanedRuns();
|
||||
expect(result.reaped).toBe(1);
|
||||
expect(result.runIds).toEqual([runId]);
|
||||
|
||||
const runs = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, agentId));
|
||||
expect(runs).toHaveLength(1);
|
||||
expect(runs[0]?.status).toBe("failed");
|
||||
|
||||
const recoveryIssue = await waitForValue(async () =>
|
||||
db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => {
|
||||
const issue = rows[0] ?? null;
|
||||
return issue?.status === "blocked" ? issue : null;
|
||||
})
|
||||
);
|
||||
expect(recoveryIssue?.assigneeAgentId).toBe(agentId);
|
||||
expect(recoveryIssue?.originKind).toBe("stranded_issue_recovery");
|
||||
expect(recoveryIssue?.originId).toBe(sourceIssueId);
|
||||
expect(recoveryIssue?.executionRunId).toBeNull();
|
||||
|
||||
const nestedRecoveries = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery"), eq(issues.originId, issueId)));
|
||||
expect(nestedRecoveries).toHaveLength(0);
|
||||
|
||||
const comments = await waitForValue(async () => {
|
||||
const rows = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId));
|
||||
return rows.length > 0 ? rows : null;
|
||||
});
|
||||
expect(comments).toHaveLength(1);
|
||||
expect(comments[0]?.body).toContain("stopped automatic stranded-work recovery");
|
||||
expect(comments[0]?.body).toContain("recovery issues do not create nested `stranded_issue_recovery` issues");
|
||||
expect(comments[0]?.body).toContain("Latest retry failure: `process_lost`");
|
||||
});
|
||||
|
||||
it("does not block paused-tree work when immediate continuation recovery is suppressed by the hold", async () => {
|
||||
const { companyId, agentId, runId, issueId } = await seedRunFixture({
|
||||
agentStatus: "idle",
|
||||
@@ -1441,6 +1496,51 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
expect(comments[0]?.body).toContain(`Recovery issue: [${recovery.identifier}]`);
|
||||
});
|
||||
|
||||
it("blocks stranded recovery issues in place instead of creating nested recovery issues", async () => {
|
||||
const sourceIssueId = randomUUID();
|
||||
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
|
||||
status: "in_progress",
|
||||
runStatus: "failed",
|
||||
});
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
title: "Recover stalled issue PAP-1",
|
||||
originKind: "stranded_issue_recovery",
|
||||
originId: sourceIssueId,
|
||||
})
|
||||
.where(eq(issues.id, issueId));
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.dispatchRequeued).toBe(0);
|
||||
expect(result.continuationRequeued).toBe(0);
|
||||
expect(result.escalated).toBe(1);
|
||||
expect(result.issueIds).toEqual([issueId]);
|
||||
|
||||
const recoveryIssue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null);
|
||||
expect(recoveryIssue?.status).toBe("blocked");
|
||||
expect(recoveryIssue?.assigneeAgentId).toBe(agentId);
|
||||
expect(recoveryIssue?.originKind).toBe("stranded_issue_recovery");
|
||||
expect(recoveryIssue?.originId).toBe(sourceIssueId);
|
||||
|
||||
const nestedRecoveries = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery"), eq(issues.originId, issueId)));
|
||||
expect(nestedRecoveries).toHaveLength(0);
|
||||
|
||||
const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId));
|
||||
expect(runs).toHaveLength(1);
|
||||
expect(runs[0]?.id).toBe(runId);
|
||||
|
||||
const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId));
|
||||
expect(comments).toHaveLength(1);
|
||||
expect(comments[0]?.body).toContain("stopped automatic stranded-work recovery");
|
||||
expect(comments[0]?.body).toContain("Latest retry failure: `process_lost` - run failed before issue advanced.");
|
||||
expect(comments[0]?.body).toContain("recovery issues do not create nested `stranded_issue_recovery` issues");
|
||||
});
|
||||
|
||||
it("does not escalate paused-tree recovery when the automatic continuation retry was cancelled by the hold", async () => {
|
||||
const { companyId, agentId, issueId } = await seedStrandedIssueFixture({
|
||||
status: "in_progress",
|
||||
|
||||
@@ -100,6 +100,7 @@ import {
|
||||
} from "./execution-workspace-policy.js";
|
||||
import { instanceSettingsService } from "./instance-settings.js";
|
||||
import {
|
||||
RECOVERY_ORIGIN_KINDS,
|
||||
RUN_LIVENESS_CONTINUATION_REASON,
|
||||
buildRunLivenessContinuationIdempotencyKey,
|
||||
decideRunLivenessContinuation,
|
||||
@@ -6030,6 +6031,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
}
|
||||
|
||||
const shouldBlockImmediately =
|
||||
issue.originKind === RECOVERY_ORIGIN_KINDS.strandedIssueRecovery ||
|
||||
!recoveryAgentInvokable ||
|
||||
!recoveryAgent ||
|
||||
didAutomaticRecoveryFail(run, issue.status === "todo" ? "assignment_recovery" : "issue_continuation_needed");
|
||||
|
||||
@@ -178,6 +178,19 @@ function isAgentInvokable(agent: typeof agents.$inferSelect | null | undefined)
|
||||
return Boolean(agent && !["paused", "terminated", "pending_approval"].includes(agent.status));
|
||||
}
|
||||
|
||||
function isStrandedIssueRecoveryIssue(issue: Pick<typeof issues.$inferSelect, "originKind">) {
|
||||
return issue.originKind === STRANDED_ISSUE_RECOVERY_ORIGIN_KIND;
|
||||
}
|
||||
|
||||
function isUnsuccessfulTerminalIssueRun(latestRun: LatestIssueRun) {
|
||||
return Boolean(
|
||||
latestRun &&
|
||||
UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES.includes(
|
||||
latestRun.status as (typeof UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES)[number],
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
function parseLivenessIncidentKey(incidentKey: string | null | undefined) {
|
||||
if (!incidentKey) return null;
|
||||
return parseIssueGraphLivenessIncidentKey(incidentKey);
|
||||
@@ -1248,6 +1261,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
latestRun: LatestIssueRun;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
}) {
|
||||
if (isStrandedIssueRecoveryIssue(input.issue)) return null;
|
||||
|
||||
const existing = await findOpenStrandedIssueRecoveryIssue(input.issue.companyId, input.issue.id);
|
||||
if (existing) return existing;
|
||||
|
||||
@@ -1306,6 +1321,78 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
return recovery;
|
||||
}
|
||||
|
||||
function buildRecoveryIssueInPlaceEscalationComment(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
latestRun: LatestIssueRun;
|
||||
prefix: string;
|
||||
}) {
|
||||
const runLink = input.latestRun
|
||||
? runUiLink({ id: input.latestRun.id, agentId: input.latestRun.agentId }, input.prefix)
|
||||
: "none";
|
||||
const retryReason = readNonEmptyString(parseObject(input.latestRun?.contextSnapshot)?.retryReason) ?? "none";
|
||||
const failureSummary = summarizeRunFailureForIssueComment(input.latestRun);
|
||||
|
||||
return [
|
||||
"Paperclip stopped automatic stranded-work recovery for this recovery issue.",
|
||||
"",
|
||||
`- Recovery issue: ${issueUiLink({ identifier: input.issue.identifier, id: input.issue.id }, input.prefix)}`,
|
||||
`- Previous status: \`${input.previousStatus}\``,
|
||||
`- Latest run: ${runLink}`,
|
||||
`- Latest run status: \`${input.latestRun?.status ?? "unknown"}\``,
|
||||
`- Retry reason: \`${retryReason}\``,
|
||||
failureSummary ? `- Failure: ${failureSummary.trim()}` : "- Failure: none recorded",
|
||||
"- Guard: recovery issues do not create nested `stranded_issue_recovery` issues.",
|
||||
"",
|
||||
"Next action: the current recovery owner should inspect the failed run evidence, restore a live execution path or record the manual resolution, then move this recovery issue out of `blocked`.",
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
async function escalateStrandedRecoveryIssueInPlace(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
latestRun: LatestIssueRun;
|
||||
}) {
|
||||
const updated = await issuesSvc.update(input.issue.id, { status: "blocked" });
|
||||
if (!updated) return null;
|
||||
|
||||
const prefix = await getCompanyIssuePrefix(input.issue.companyId);
|
||||
await issuesSvc.addComment(
|
||||
input.issue.id,
|
||||
buildRecoveryIssueInPlaceEscalationComment({
|
||||
issue: input.issue,
|
||||
previousStatus: input.previousStatus,
|
||||
latestRun: input.latestRun,
|
||||
prefix,
|
||||
}),
|
||||
{},
|
||||
);
|
||||
|
||||
await logActivity(db, {
|
||||
companyId: input.issue.companyId,
|
||||
actorType: "system",
|
||||
actorId: "system",
|
||||
agentId: null,
|
||||
runId: null,
|
||||
action: "issue.updated",
|
||||
entityType: "issue",
|
||||
entityId: input.issue.id,
|
||||
details: {
|
||||
identifier: input.issue.identifier,
|
||||
status: "blocked",
|
||||
previousStatus: input.previousStatus,
|
||||
source: "recovery.reconcile_stranded_recovery_issue",
|
||||
latestRunId: input.latestRun?.id ?? null,
|
||||
latestRunStatus: input.latestRun?.status ?? null,
|
||||
latestRunErrorCode: input.latestRun?.errorCode ?? null,
|
||||
originKind: input.issue.originKind,
|
||||
originId: input.issue.originId,
|
||||
},
|
||||
});
|
||||
|
||||
return updated;
|
||||
}
|
||||
|
||||
async function existingBlockerIssueIds(companyId: string, issueId: string) {
|
||||
return db
|
||||
.select({ blockerIssueId: issueRelations.issueId })
|
||||
@@ -1348,6 +1435,14 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
latestRun: LatestIssueRun;
|
||||
comment: string;
|
||||
}) {
|
||||
if (isStrandedIssueRecoveryIssue(input.issue)) {
|
||||
return escalateStrandedRecoveryIssueInPlace({
|
||||
issue: input.issue,
|
||||
previousStatus: input.previousStatus,
|
||||
latestRun: input.latestRun,
|
||||
});
|
||||
}
|
||||
|
||||
const recoveryIssue = await ensureStrandedIssueRecoveryIssue({
|
||||
issue: input.issue,
|
||||
previousStatus: input.previousStatus,
|
||||
@@ -1448,6 +1543,21 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
}
|
||||
|
||||
const latestRun = await getLatestIssueRun(issue.companyId, issue.id);
|
||||
if (isStrandedIssueRecoveryIssue(issue) && isUnsuccessfulTerminalIssueRun(latestRun)) {
|
||||
const updated = await escalateStrandedRecoveryIssueInPlace({
|
||||
issue,
|
||||
previousStatus: issue.status as "todo" | "in_progress",
|
||||
latestRun,
|
||||
});
|
||||
if (updated) {
|
||||
result.escalated += 1;
|
||||
result.issueIds.push(issue.id);
|
||||
} else {
|
||||
result.skipped += 1;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (issue.status === "todo") {
|
||||
if (!latestRun || latestRun.status === "succeeded") {
|
||||
result.skipped += 1;
|
||||
|
||||
Reference in new issue
Block a user