mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
Fix source-scoped recovery escalation race
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
f54aba63fc
commit
4980babc2d
6 files changed
+191
-135
No files matched your search
@@ -111,11 +111,15 @@ function formatEmbeddedPostgresError(error: unknown): string {
|
||||
}
|
||||
|
||||
async function probeEmbeddedPostgresSupport(): Promise<EmbeddedPostgresTestSupport> {
|
||||
const { dataDir, instance } = await createEmbeddedPostgresTestInstance(
|
||||
"paperclip-embedded-postgres-probe-",
|
||||
);
|
||||
let dataDir: string | null = null;
|
||||
let instance: EmbeddedPostgresInstance | null = null;
|
||||
|
||||
try {
|
||||
const created = await createEmbeddedPostgresTestInstance(
|
||||
"paperclip-embedded-postgres-probe-",
|
||||
);
|
||||
dataDir = created.dataDir;
|
||||
instance = created.instance;
|
||||
await instance.initialise();
|
||||
await instance.start();
|
||||
return { supported: true };
|
||||
@@ -125,8 +129,8 @@ async function probeEmbeddedPostgresSupport(): Promise<EmbeddedPostgresTestSuppo
|
||||
reason: formatEmbeddedPostgresError(error),
|
||||
};
|
||||
} finally {
|
||||
await instance.stop().catch(() => {});
|
||||
cleanupEmbeddedPostgresTestDirs(dataDir);
|
||||
await instance?.stop().catch(() => {});
|
||||
if (dataDir) cleanupEmbeddedPostgresTestDirs(dataDir);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -140,9 +144,14 @@ export async function getEmbeddedPostgresTestSupport(): Promise<EmbeddedPostgres
|
||||
export async function startEmbeddedPostgresTestDatabase(
|
||||
tempDirPrefix: string,
|
||||
): Promise<EmbeddedPostgresTestDatabase> {
|
||||
const { dataDir, port, instance } = await createEmbeddedPostgresTestInstance(tempDirPrefix);
|
||||
let dataDir: string | null = null;
|
||||
let instance: EmbeddedPostgresInstance | null = null;
|
||||
|
||||
try {
|
||||
const created = await createEmbeddedPostgresTestInstance(tempDirPrefix);
|
||||
dataDir = created.dataDir;
|
||||
instance = created.instance;
|
||||
const { port } = created;
|
||||
await instance.initialise();
|
||||
await instance.start();
|
||||
|
||||
@@ -154,13 +163,13 @@ export async function startEmbeddedPostgresTestDatabase(
|
||||
return {
|
||||
connectionString,
|
||||
cleanup: async () => {
|
||||
await instance.stop().catch(() => {});
|
||||
cleanupEmbeddedPostgresTestDirs(dataDir);
|
||||
await instance?.stop().catch(() => {});
|
||||
if (dataDir) cleanupEmbeddedPostgresTestDirs(dataDir);
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
await instance.stop().catch(() => {});
|
||||
cleanupEmbeddedPostgresTestDirs(dataDir);
|
||||
await instance?.stop().catch(() => {});
|
||||
if (dataDir) cleanupEmbeddedPostgresTestDirs(dataDir);
|
||||
throw new Error(
|
||||
`Failed to start embedded PostgreSQL test database: ${formatEmbeddedPostgresError(error)}`,
|
||||
);
|
||||
|
||||
@@ -20,6 +20,7 @@ import {
|
||||
heartbeatRuns,
|
||||
issueComments,
|
||||
issueDocuments,
|
||||
issueRecoveryActions,
|
||||
issueRelations,
|
||||
issueTreeHoldMembers,
|
||||
issueTreeHolds,
|
||||
@@ -328,6 +329,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
await db.delete(documentRevisions);
|
||||
await db.delete(documents);
|
||||
await db.delete(issueRelations);
|
||||
await db.delete(issueRecoveryActions);
|
||||
await db.delete(issueTreeHoldMembers);
|
||||
await db.delete(issueTreeHolds);
|
||||
for (let attempt = 0; attempt < 5; attempt += 1) {
|
||||
@@ -692,53 +694,59 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
return { companyId, agentId, issueId };
|
||||
}
|
||||
|
||||
async function expectStrandedRecoveryArtifacts(input: {
|
||||
async function expectSourceScopedStrandedRecoveryAction(input: {
|
||||
companyId: string;
|
||||
agentId: string;
|
||||
issueId: string;
|
||||
runId: string;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
retryReason: "assignment_recovery" | "issue_continuation_needed";
|
||||
retryReason?: "assignment_recovery" | "issue_continuation_needed" | null;
|
||||
cause?: string;
|
||||
kind?: string;
|
||||
}) {
|
||||
const recovery = await waitForValue(async () =>
|
||||
db.select().from(issues).where(
|
||||
const action = await waitForValue(async () =>
|
||||
db.select().from(issueRecoveryActions).where(
|
||||
and(
|
||||
eq(issues.companyId, input.companyId),
|
||||
eq(issues.originKind, "stranded_issue_recovery"),
|
||||
eq(issues.originId, input.issueId),
|
||||
eq(issueRecoveryActions.companyId, input.companyId),
|
||||
eq(issueRecoveryActions.sourceIssueId, input.issueId),
|
||||
),
|
||||
).then((rows) => rows[0] ?? null),
|
||||
);
|
||||
if (!recovery) throw new Error("Expected stranded issue recovery issue to be created");
|
||||
if (!action) throw new Error("Expected source-scoped stranded recovery action to be created");
|
||||
|
||||
expect(recovery).toMatchObject({
|
||||
expect(action).toMatchObject({
|
||||
companyId: input.companyId,
|
||||
parentId: input.issueId,
|
||||
assigneeAgentId: input.agentId,
|
||||
originKind: "stranded_issue_recovery",
|
||||
originId: input.issueId,
|
||||
originRunId: input.runId,
|
||||
priority: "medium",
|
||||
assigneeAdapterOverrides: { modelProfile: "cheap" },
|
||||
sourceIssueId: input.issueId,
|
||||
recoveryIssueId: null,
|
||||
kind: input.kind ?? "stranded_assigned_issue",
|
||||
status: "active",
|
||||
ownerType: "agent",
|
||||
ownerAgentId: input.agentId,
|
||||
previousOwnerAgentId: input.agentId,
|
||||
returnOwnerAgentId: input.agentId,
|
||||
cause: input.cause ?? "stranded_assigned_issue",
|
||||
attemptCount: 1,
|
||||
maxAttempts: 1,
|
||||
});
|
||||
expect(recovery.title).toContain("Recover stalled issue");
|
||||
expect(recovery.description).toContain(`Previous source status: \`${input.previousStatus}\``);
|
||||
expect(recovery.description).toContain(`Retry reason: \`${input.retryReason}\``);
|
||||
expect(recovery.description).toContain("Fix the runtime/adapter problem");
|
||||
expect(action.evidence).toMatchObject({
|
||||
sourceIssueId: input.issueId,
|
||||
previousStatus: input.previousStatus,
|
||||
latestRunId: input.runId,
|
||||
retryReason: input.retryReason ?? null,
|
||||
});
|
||||
expect(action.nextAction).toContain(
|
||||
input.kind === "missing_disposition" ? "valid issue disposition" : "Restore a live execution path",
|
||||
);
|
||||
|
||||
const relation = await db
|
||||
const recoveryIssues = await db
|
||||
.select()
|
||||
.from(issueRelations)
|
||||
.where(
|
||||
and(
|
||||
eq(issueRelations.companyId, input.companyId),
|
||||
eq(issueRelations.issueId, recovery.id),
|
||||
eq(issueRelations.relatedIssueId, input.issueId),
|
||||
eq(issueRelations.type, "blocks"),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
expect(relation).toBeTruthy();
|
||||
.from(issues)
|
||||
.where(and(
|
||||
eq(issues.companyId, input.companyId),
|
||||
eq(issues.originKind, "stranded_issue_recovery"),
|
||||
eq(issues.originId, input.issueId),
|
||||
));
|
||||
expect(recoveryIssues).toHaveLength(0);
|
||||
|
||||
const wakeups = await db
|
||||
.select()
|
||||
@@ -746,13 +754,14 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
.where(eq(agentWakeupRequests.agentId, input.agentId));
|
||||
const recoveryWakeup = wakeups.find((wakeup) => {
|
||||
const payload = wakeup.payload as Record<string, unknown> | null;
|
||||
return payload?.issueId === recovery.id &&
|
||||
return payload?.issueId === input.issueId &&
|
||||
payload?.sourceIssueId === input.issueId &&
|
||||
payload?.recoveryActionId === action.id &&
|
||||
payload?.strandedRunId === input.runId;
|
||||
});
|
||||
expect(recoveryWakeup).toMatchObject({
|
||||
companyId: input.companyId,
|
||||
reason: "issue_assigned",
|
||||
reason: "source_scoped_recovery_action",
|
||||
source: "assignment",
|
||||
payload: expect.objectContaining({ modelProfile: "cheap" }),
|
||||
});
|
||||
@@ -765,15 +774,23 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
expect(recoveryRun?.contextSnapshot).toMatchObject({
|
||||
issueId: recovery.id,
|
||||
taskId: recovery.id,
|
||||
source: "stranded_issue_recovery",
|
||||
issueId: input.issueId,
|
||||
taskId: input.issueId,
|
||||
source: "issue_recovery_action",
|
||||
recoveryActionId: action.id,
|
||||
sourceIssueId: input.issueId,
|
||||
strandedRunId: input.runId,
|
||||
modelProfile: "cheap",
|
||||
});
|
||||
await waitForHeartbeatIdle(db);
|
||||
const sourceIssue = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(eq(issues.id, input.issueId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
expect(sourceIssue?.status).toBe("blocked");
|
||||
|
||||
return recovery;
|
||||
return action;
|
||||
}
|
||||
|
||||
async function sourceBlockerIssueIds(companyId: string, sourceIssueId: string) {
|
||||
@@ -1056,7 +1073,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
expect(blockedIssue?.checkoutRunId).toBeNull();
|
||||
if (!continuationRun?.id) throw new Error("Expected continuation recovery run to exist");
|
||||
|
||||
const recovery = await expectStrandedRecoveryArtifacts({
|
||||
const recoveryAction = await expectSourceScopedStrandedRecoveryAction({
|
||||
companyId,
|
||||
agentId,
|
||||
issueId,
|
||||
@@ -1065,17 +1082,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
retryReason: "issue_continuation_needed",
|
||||
});
|
||||
|
||||
const blockerRelations = await db
|
||||
.select()
|
||||
.from(issueRelations)
|
||||
.where(
|
||||
and(
|
||||
eq(issueRelations.companyId, companyId),
|
||||
eq(issueRelations.relatedIssueId, issueId),
|
||||
eq(issueRelations.type, "blocks"),
|
||||
),
|
||||
);
|
||||
expect(blockerRelations.map((relation) => relation.issueId)).toEqual([recovery.id]);
|
||||
await expect(sourceBlockerIssueIds(companyId, issueId)).resolves.toEqual([]);
|
||||
|
||||
const comments = await waitForValue(async () => {
|
||||
const rows = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId));
|
||||
@@ -1083,7 +1090,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
});
|
||||
expect(comments).toHaveLength(1);
|
||||
expect(comments[0]?.body).toContain("retried continuation");
|
||||
expect(comments[0]?.body).toContain(`Recovery issue: [${recovery.identifier}]`);
|
||||
expect(comments[0]?.body).toContain(`Recovery action: \`${recoveryAction.id}\``);
|
||||
expect(comments[0]?.body).toContain("Recovery owner: [CodexCoder]");
|
||||
});
|
||||
|
||||
it("blocks failed recovery work in place during immediate terminal-run cleanup", async () => {
|
||||
@@ -1600,27 +1608,28 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
expect(result.successfulRunHandoffEscalated).toBe(1);
|
||||
expect(result.issueIds).toEqual([issueId]);
|
||||
|
||||
const recovery = await waitForValue(async () =>
|
||||
db.select().from(issues).where(
|
||||
and(
|
||||
eq(issues.companyId, companyId),
|
||||
eq(issues.originKind, "stranded_issue_recovery"),
|
||||
eq(issues.originId, issueId),
|
||||
),
|
||||
).then((rows) => rows[0] ?? null),
|
||||
);
|
||||
expect(recovery?.assigneeAgentId).toBe(agentId);
|
||||
expect(recovery?.title).toContain("Recover missing next step");
|
||||
expect(recovery?.description).toContain("Normalized cause: `successful_run_missing_state`");
|
||||
expect(recovery?.description).toContain("not a runtime/adapter crash report");
|
||||
expect(recovery?.description).toContain(`Source run: [\`${sourceRunId}\`]`);
|
||||
expect(recovery?.description).toContain("Missing disposition: `clear_next_step`");
|
||||
expect(recovery?.description).toContain("Source assignee: [CodexCoder]");
|
||||
expect(recovery?.description).not.toContain("sk-test-successful-handoff-secret");
|
||||
const recoveryAction = await expectSourceScopedStrandedRecoveryAction({
|
||||
companyId,
|
||||
agentId,
|
||||
issueId,
|
||||
runId,
|
||||
previousStatus: "in_progress",
|
||||
retryReason: null,
|
||||
cause: SUCCESSFUL_RUN_MISSING_STATE_REASON,
|
||||
kind: "missing_disposition",
|
||||
});
|
||||
expect(recoveryAction.evidence).toMatchObject({
|
||||
sourceRunId,
|
||||
missingDisposition: "clear_next_step",
|
||||
latestRunStatus: "failed",
|
||||
latestRunErrorCode: "adapter_failed",
|
||||
recoveryCause: SUCCESSFUL_RUN_MISSING_STATE_REASON,
|
||||
});
|
||||
expect(JSON.stringify(recoveryAction.evidence)).not.toContain("sk-test-successful-handoff-secret");
|
||||
|
||||
const sourceIssue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null);
|
||||
expect(sourceIssue?.status).toBe("blocked");
|
||||
await expect(sourceBlockerIssueIds(companyId, issueId)).resolves.toEqual([recovery?.id]);
|
||||
await expect(sourceBlockerIssueIds(companyId, issueId)).resolves.toEqual([]);
|
||||
|
||||
const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId));
|
||||
expect(comments[0]?.body).toBe(SUCCESSFUL_RUN_HANDOFF_EXHAUSTED_NOTICE_BODY);
|
||||
@@ -1636,7 +1645,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
expect.objectContaining({
|
||||
title: "Recovery owner",
|
||||
rows: expect.arrayContaining([
|
||||
expect.objectContaining({ type: "issue_link", identifier: recovery?.identifier }),
|
||||
expect.objectContaining({ type: "key_value", label: "Recovery action", value: recoveryAction.id }),
|
||||
expect.objectContaining({ type: "agent_link", label: "Recovery owner", name: "CodexCoder" }),
|
||||
]),
|
||||
}),
|
||||
@@ -1657,7 +1666,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
});
|
||||
|
||||
it("escalates an exhausted successful handoff run that still leaves no disposition", async () => {
|
||||
const { companyId, runId, issueId } = await seedStrandedIssueFixture({
|
||||
const { companyId, agentId, runId, issueId } = await seedStrandedIssueFixture({
|
||||
status: "in_progress",
|
||||
runStatus: "succeeded",
|
||||
livenessState: "advanced",
|
||||
@@ -1687,17 +1696,21 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
expect(result.successfulContinuationObserved).toBe(0);
|
||||
expect(result.successfulRunHandoffEscalated).toBe(1);
|
||||
|
||||
const recovery = await waitForValue(async () =>
|
||||
db.select().from(issues).where(
|
||||
and(
|
||||
eq(issues.companyId, companyId),
|
||||
eq(issues.originKind, "stranded_issue_recovery"),
|
||||
eq(issues.originId, issueId),
|
||||
),
|
||||
).then((rows) => rows[0] ?? null),
|
||||
);
|
||||
expect(recovery?.description).toContain("Latest handoff run status: `succeeded`");
|
||||
expect(recovery?.description).toContain("Suggested");
|
||||
const recoveryAction = await expectSourceScopedStrandedRecoveryAction({
|
||||
companyId,
|
||||
agentId,
|
||||
issueId,
|
||||
runId,
|
||||
previousStatus: "in_progress",
|
||||
retryReason: null,
|
||||
cause: SUCCESSFUL_RUN_MISSING_STATE_REASON,
|
||||
kind: "missing_disposition",
|
||||
});
|
||||
expect(recoveryAction.evidence).toMatchObject({
|
||||
sourceRunId,
|
||||
latestRunStatus: "succeeded",
|
||||
missingDisposition: "clear_next_step",
|
||||
});
|
||||
});
|
||||
|
||||
it("clears the detached warning when the run reports activity again", async () => {
|
||||
@@ -2063,7 +2076,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null);
|
||||
expect(issue?.status).toBe("blocked");
|
||||
|
||||
const recovery = await expectStrandedRecoveryArtifacts({
|
||||
const recoveryAction = await expectSourceScopedStrandedRecoveryAction({
|
||||
companyId,
|
||||
agentId,
|
||||
issueId,
|
||||
@@ -2071,13 +2084,14 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
previousStatus: "todo",
|
||||
retryReason: "assignment_recovery",
|
||||
});
|
||||
expect(recovery.description ?? "").not.toContain("sk-test-recovery-secret");
|
||||
expect(JSON.stringify(recoveryAction.evidence)).not.toContain("sk-test-recovery-secret");
|
||||
|
||||
const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId));
|
||||
expect(comments).toHaveLength(1);
|
||||
expect(comments[0]?.body).toContain("retried dispatch");
|
||||
expect(comments[0]?.body).toContain("Latest retry failure details were withheld from the issue thread");
|
||||
expect(comments[0]?.body).toContain(`Recovery issue: [${recovery.identifier}]`);
|
||||
expect(comments[0]?.body).toContain(`Recovery action: \`${recoveryAction.id}\``);
|
||||
expect(comments[0]?.body).toContain("Recovery owner: [CodexCoder]");
|
||||
});
|
||||
|
||||
it("blocks an already stranded recovery issue without creating a recovery child", async () => {
|
||||
@@ -2457,7 +2471,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null);
|
||||
expect(issue?.status).toBe("blocked");
|
||||
|
||||
const recovery = await expectStrandedRecoveryArtifacts({
|
||||
const recoveryAction = await expectSourceScopedStrandedRecoveryAction({
|
||||
companyId,
|
||||
agentId,
|
||||
issueId,
|
||||
@@ -2470,7 +2484,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
expect(comments).toHaveLength(1);
|
||||
expect(comments[0]?.body).toContain("retried continuation");
|
||||
expect(comments[0]?.body).toContain("Latest retry failure details were withheld from the issue thread");
|
||||
expect(comments[0]?.body).toContain(`Recovery issue: [${recovery.identifier}]`);
|
||||
expect(comments[0]?.body).toContain(`Recovery action: \`${recoveryAction.id}\``);
|
||||
expect(comments[0]?.body).toContain("Recovery owner: [CodexCoder]");
|
||||
});
|
||||
|
||||
it("redacts error-code-only stranded recovery failures in issue copy", async () => {
|
||||
@@ -2486,7 +2501,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.escalated).toBe(1);
|
||||
|
||||
const recovery = await expectStrandedRecoveryArtifacts({
|
||||
const recoveryAction = await expectSourceScopedStrandedRecoveryAction({
|
||||
companyId,
|
||||
agentId,
|
||||
issueId,
|
||||
@@ -2494,8 +2509,10 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
previousStatus: "in_progress",
|
||||
retryReason: "issue_continuation_needed",
|
||||
});
|
||||
expect(recovery.description).toContain("Latest retry failure details were withheld from the issue thread");
|
||||
expect(recovery.description).not.toContain("- Failure: none recorded");
|
||||
expect(recoveryAction.evidence).toMatchObject({
|
||||
latestRunErrorCode: "adapter_exit_code",
|
||||
});
|
||||
expect(JSON.stringify(recoveryAction.evidence)).not.toContain("- Failure: none recorded");
|
||||
|
||||
const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId));
|
||||
expect(comments).toHaveLength(1);
|
||||
@@ -2516,6 +2533,15 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
);
|
||||
expect(results.every((result) => result.status === "fulfilled")).toBe(true);
|
||||
|
||||
const actions = await db
|
||||
.select()
|
||||
.from(issueRecoveryActions)
|
||||
.where(and(
|
||||
eq(issueRecoveryActions.companyId, companyId),
|
||||
eq(issueRecoveryActions.sourceIssueId, issueId),
|
||||
));
|
||||
expect(actions).toHaveLength(1);
|
||||
expect(actions[0]?.attemptCount).toBeGreaterThanOrEqual(1);
|
||||
const recoveries = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
@@ -2524,8 +2550,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
eq(issues.originKind, "stranded_issue_recovery"),
|
||||
eq(issues.originId, issueId),
|
||||
));
|
||||
expect(recoveries).toHaveLength(1);
|
||||
await expect(sourceBlockerIssueIds(companyId, issueId)).resolves.toEqual([recoveries[0]?.id]);
|
||||
expect(recoveries).toHaveLength(0);
|
||||
await expect(sourceBlockerIssueIds(companyId, issueId)).resolves.toEqual([]);
|
||||
});
|
||||
|
||||
it("blocks stranded recovery issues in place instead of creating nested recovery issues", async () => {
|
||||
@@ -2783,7 +2809,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null);
|
||||
expect(issue?.status).toBe("blocked");
|
||||
|
||||
const recovery = await expectStrandedRecoveryArtifacts({
|
||||
const recoveryAction = await expectSourceScopedStrandedRecoveryAction({
|
||||
companyId,
|
||||
agentId,
|
||||
issueId,
|
||||
@@ -2796,7 +2822,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
expect(comments).toHaveLength(1);
|
||||
expect(comments[0]?.body).toContain("automatically retried continuation");
|
||||
expect(comments[0]?.body).toContain("still has no live execution path");
|
||||
expect(comments[0]?.body).toContain(`Recovery issue: [${recovery.identifier}]`);
|
||||
expect(comments[0]?.body).toContain(`Recovery action: \`${recoveryAction.id}\``);
|
||||
expect(comments[0]?.body).toContain("Recovery owner: [CodexCoder]");
|
||||
});
|
||||
|
||||
it("allows one productive-terminal recovery after regular continuation recovery made progress", async () => {
|
||||
|
||||
@@ -1690,6 +1690,7 @@ function shouldAutoCheckoutIssueForWake(input: {
|
||||
const wakeReason = readNonEmptyString(input.contextSnapshot?.wakeReason);
|
||||
if (!wakeReason) return false;
|
||||
if (wakeReason === "issue_comment_mentioned") return false;
|
||||
if (wakeReason === "source_scoped_recovery_action") return false;
|
||||
if (wakeReason.startsWith("execution_")) return false;
|
||||
|
||||
return true;
|
||||
|
||||
@@ -1635,37 +1635,43 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
lastAttemptAt: now,
|
||||
});
|
||||
|
||||
if (ownerAgentId) {
|
||||
await deps.enqueueWakeup(ownerAgentId, {
|
||||
source: "assignment",
|
||||
triggerDetail: "system",
|
||||
reason: "source_scoped_recovery_action",
|
||||
idempotencyKey: `source_scoped_recovery_action:${action.id}:${action.attemptCount}`,
|
||||
payload: withRecoveryModelProfileHint({
|
||||
issueId: input.issue.id,
|
||||
sourceIssueId: input.issue.id,
|
||||
recoveryActionId: action.id,
|
||||
strandedRunId: input.latestRun?.id ?? null,
|
||||
recoveryCause,
|
||||
}),
|
||||
requestedByActorType: "system",
|
||||
requestedByActorId: null,
|
||||
contextSnapshot: withRecoveryModelProfileHint({
|
||||
issueId: input.issue.id,
|
||||
taskId: input.issue.id,
|
||||
wakeReason: "source_scoped_recovery_action",
|
||||
source: "issue_recovery_action",
|
||||
recoveryActionId: action.id,
|
||||
sourceIssueId: input.issue.id,
|
||||
strandedRunId: input.latestRun?.id ?? null,
|
||||
recoveryCause,
|
||||
}),
|
||||
});
|
||||
}
|
||||
|
||||
return action;
|
||||
}
|
||||
|
||||
async function enqueueSourceScopedStrandedRecoveryWake(input: {
|
||||
action: Awaited<ReturnType<typeof recoveryActionsSvc.upsertSourceScoped>>;
|
||||
issue: typeof issues.$inferSelect;
|
||||
latestRun: LatestIssueRun;
|
||||
recoveryCause: StrandedRecoveryCause;
|
||||
}) {
|
||||
if (!input.action.ownerAgentId) return;
|
||||
await deps.enqueueWakeup(input.action.ownerAgentId, {
|
||||
source: "assignment",
|
||||
triggerDetail: "system",
|
||||
reason: "source_scoped_recovery_action",
|
||||
idempotencyKey: `source_scoped_recovery_action:${input.action.id}:${input.action.attemptCount}`,
|
||||
payload: withRecoveryModelProfileHint({
|
||||
issueId: input.issue.id,
|
||||
sourceIssueId: input.issue.id,
|
||||
recoveryActionId: input.action.id,
|
||||
strandedRunId: input.latestRun?.id ?? null,
|
||||
recoveryCause: input.recoveryCause,
|
||||
}),
|
||||
requestedByActorType: "system",
|
||||
requestedByActorId: null,
|
||||
contextSnapshot: withRecoveryModelProfileHint({
|
||||
issueId: input.issue.id,
|
||||
taskId: input.issue.id,
|
||||
wakeReason: "source_scoped_recovery_action",
|
||||
source: "issue_recovery_action",
|
||||
recoveryActionId: input.action.id,
|
||||
sourceIssueId: input.issue.id,
|
||||
strandedRunId: input.latestRun?.id ?? null,
|
||||
recoveryCause: input.recoveryCause,
|
||||
}),
|
||||
});
|
||||
}
|
||||
|
||||
function buildRecoveryIssueInPlaceEscalationComment(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
@@ -1790,11 +1796,12 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
});
|
||||
}
|
||||
|
||||
const recoveryCause = input.recoveryCause ?? "stranded_assigned_issue";
|
||||
const recoveryAction = await ensureSourceScopedStrandedRecoveryAction({
|
||||
issue: input.issue,
|
||||
previousStatus: input.previousStatus,
|
||||
latestRun: input.latestRun,
|
||||
recoveryCause: input.recoveryCause,
|
||||
recoveryCause,
|
||||
successfulRunHandoffEvidence: input.successfulRunHandoffEvidence,
|
||||
});
|
||||
const blockerIds = await existingUnresolvedBlockerIssueIds(input.issue.companyId, input.issue.id);
|
||||
@@ -1818,6 +1825,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
correctiveRun: input.latestRun ? { id: input.latestRun.id, status: input.latestRun.status } : null,
|
||||
sourceAssignee,
|
||||
recoveryIssue: null,
|
||||
recoveryActionId: recoveryAction.id,
|
||||
recoveryOwner,
|
||||
latestIssueStatus: input.issue.status,
|
||||
latestHandoffRunStatus: input.latestRun?.status ?? "unknown",
|
||||
@@ -1878,6 +1886,13 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
},
|
||||
});
|
||||
|
||||
await enqueueSourceScopedStrandedRecoveryWake({
|
||||
action: recoveryAction,
|
||||
issue: input.issue,
|
||||
latestRun: input.latestRun,
|
||||
recoveryCause,
|
||||
});
|
||||
|
||||
return updated;
|
||||
}
|
||||
|
||||
|
||||
@@ -256,6 +256,7 @@ describe("successful run handoff decision", () => {
|
||||
title: "Recover missing next step PAP-1",
|
||||
status: "todo",
|
||||
} as any,
|
||||
recoveryActionId: "77777777-7777-4777-8777-777777777777",
|
||||
recoveryOwner: { id: "66666666-6666-4666-8666-666666666666", name: "CTO" } as any,
|
||||
latestIssueStatus: "in_progress",
|
||||
latestHandoffRunStatus: "failed",
|
||||
@@ -273,7 +274,7 @@ describe("successful run handoff decision", () => {
|
||||
expect.objectContaining({
|
||||
title: "Recovery owner",
|
||||
rows: expect.arrayContaining([
|
||||
expect.objectContaining({ type: "issue_link", identifier: "PAP-2" }),
|
||||
expect.objectContaining({ type: "key_value", label: "Recovery action", value: "77777777-7777-4777-8777-777777777777" }),
|
||||
expect.objectContaining({ type: "agent_link", label: "Recovery owner", name: "CTO" }),
|
||||
]),
|
||||
}),
|
||||
|
||||
@@ -181,6 +181,7 @@ export function buildSuccessfulRunHandoffExhaustedNotice(input: {
|
||||
correctiveRun: NullableNoticeRun;
|
||||
sourceAssignee: NullableNoticeAgent;
|
||||
recoveryIssue: NullableNoticeIssue;
|
||||
recoveryActionId?: string | null;
|
||||
recoveryOwner: NullableNoticeAgent;
|
||||
latestIssueStatus: string;
|
||||
latestHandoffRunStatus: string;
|
||||
@@ -200,7 +201,9 @@ export function buildSuccessfulRunHandoffExhaustedNotice(input: {
|
||||
title: "Recovery owner",
|
||||
rows: [
|
||||
issueLinkRow("Source issue", input.issue),
|
||||
issueLinkRow("Recovery issue", input.recoveryIssue),
|
||||
input.recoveryActionId
|
||||
? keyValueRow("Recovery action", input.recoveryActionId)
|
||||
: issueLinkRow("Recovery issue", input.recoveryIssue),
|
||||
agentLinkRow("Recovery owner", input.recoveryOwner),
|
||||
agentLinkRow("Source assignee", input.sourceAssignee),
|
||||
keyValueRow("Suggested action", "choose and record a valid issue disposition without copying transcript content"),
|
||||
|
||||
Reference in new issue
Block a user