mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-22 16:51:28 +02:00
Fix sandbox bridge credentials and stalled review recovery (#8844)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The local adapter and heartbeat recovery systems decide whether an agent has a real control-plane mutation path. > - Sandboxed local adapters split execution between the trusted host process and the sandbox shell/tool surface. > - A host-side adapter can still reach Paperclip while the sandbox shell surface cannot, which leaves agents thinking no endpoint or credentials are configured even though the host can still post comments. > - Execution-policy review stages can also remain pending after a reviewer run finishes without recording a decision. > - This pull request makes the sandbox bridge available to the actual shell mutation surface and adds bounded recovery for terminal-but-still-pending review participants. > - The benefit is that agents get a real reachable Paperclip API path where they need it, and stalled review stages become visible recovery work instead of silently drifting. ## Linked Issues or Issue Description No exact public GitHub issue matched this combined failure. I searched for exact and related terms including `cannot reach the Paperclip control plane`, `execution_review_participant_recovery`, `sandbox callback bridge`, `review participant in_review`, and `control plane sandbox`. Related public issues: - Refs #8482 for `in_review` liveness invariant recovery. - Refs #863 for prior agent API-key reachability confusion. - Refs #248 for the broader sandboxed agent execution model. Bug summary: - What happened: a sandboxed local-adapter run could have host-side Paperclip access while the sandbox Bash/tool surface lacked a reachable API endpoint or usable run credentials. Separately, a reviewer run could finish while its execution-review stage remained pending, leaving the source issue in `in_review` with no decision and no live participant run. - Expected behavior: the mutation surface that agents actually use should receive a run-scoped Paperclip bridge, and pending review participants should get one bounded normal-model recovery wake before moving to explicit blocked/source-scoped recovery. - Steps to reproduce: run a sandbox-backed local adapter that needs Bash/curl/tooling to call Paperclip from inside the sandbox, or finish an execution-policy reviewer run without submitting the pending review decision. - Deployment mode: local/authenticated private development instance with sandbox-backed local adapters. ## What Changed - Changed sandbox callback bridge startup so bridge credentials are passed through the sandbox runner environment instead of embedded in the visible `nohup env ...` command string. - Added adapter-utils coverage proving the sandbox shell can call Paperclip through the bridge, forwards the host run JWT with `X-Paperclip-Run-Id`, and does not leak host or bridge tokens into stdout/stderr, runner command text, or runtime files. - Added one bounded execution-review participant recovery path for terminal reviewer runs whose `executionState` remains pending. - Escalated exhausted or non-invokable review participant recovery to blocked/source-scoped recovery with dedicated evidence, activity, and next-action text. - Documented the mutation-surface reachability contract in `doc/execution-semantics.md` and updated the Paperclip skill authentication guidance for sandbox bridge env vars. ## Verification - `pnpm exec vitest run packages/adapter-utils/src/execution-target-sandbox.test.ts` - `pnpm exec vitest run server/src/__tests__/heartbeat-process-recovery.test.ts --no-file-parallelism --maxWorkers=1` - `pnpm --filter @paperclipai/adapter-utils typecheck` - `pnpm --filter @paperclipai/server typecheck` - `git diff --check` - `curl -fsS $PAPERCLIP_API_URL/api/health` returned `status: ok` on the local instance. ## Risks - Medium behavioral risk: more `in_review` issues with terminal-but-pending reviewer runs will now be retried once and then blocked explicitly instead of remaining quiet. - Low sandbox bridge risk: credential delivery moved from command text to the runner environment, which is less leaky but depends on sandbox providers honoring the env payload for startup commands. - No database migration is included. - Full repo build and CI were not run locally before opening the PR; targeted server/adapter tests and typechecks passed. ## Model Used OpenAI GPT-5 via the Codex local agent, with repository tool use and shell-based code execution. The runtime did not expose a precise context-window value to the agent. ## 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 --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip authored and GitHub committed
1 parent
fb2b760915
commit
2eba718bef
8 files changed
+1142
-57
No files matched your search
@@ -63,7 +63,8 @@ async function seedGraph(db: Db, input: {
|
||||
projectSourceType?: string;
|
||||
targetProjectSourceType?: string;
|
||||
}): Promise<TestGraph> {
|
||||
const suffix = crypto.randomUUID().slice(0, 8);
|
||||
const suffix = crypto.randomUUID().replace(/-/g, "").slice(0, 12);
|
||||
const prefixSuffix = suffix.toUpperCase();
|
||||
const companyId = crypto.randomUUID();
|
||||
const otherCompanyId = crypto.randomUUID();
|
||||
const goalId = crypto.randomUUID();
|
||||
@@ -79,8 +80,8 @@ async function seedGraph(db: Db, input: {
|
||||
const otherIssueId = crypto.randomUUID();
|
||||
|
||||
await db.insert(companies).values([
|
||||
{ id: companyId, name: `Company ${suffix}`, issuePrefix: `F${suffix.slice(0, 4).toUpperCase()}` },
|
||||
{ id: otherCompanyId, name: `Other ${suffix}`, issuePrefix: `G${suffix.slice(0, 4).toUpperCase()}` },
|
||||
{ id: companyId, name: `Company ${suffix}`, issuePrefix: `F${prefixSuffix}` },
|
||||
{ id: otherCompanyId, name: `Other ${suffix}`, issuePrefix: `G${prefixSuffix}` },
|
||||
]);
|
||||
await db.insert(goals).values([
|
||||
{ id: goalId, companyId, title: "Goal", level: "company", status: "active" },
|
||||
|
||||
@@ -644,7 +644,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
async function seedStrandedIssueFixture(input: {
|
||||
status: "todo" | "in_progress";
|
||||
runStatus: "failed" | "timed_out" | "cancelled" | "succeeded";
|
||||
retryReason?: "assignment_recovery" | "issue_continuation_needed" | null;
|
||||
retryReason?: "assignment_recovery" | "issue_continuation_needed" | "execution_review_participant_recovery" | null;
|
||||
runSource?: string | null;
|
||||
assignToUser?: boolean;
|
||||
activePauseHold?: boolean;
|
||||
@@ -769,6 +769,103 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
return { companyId, agentId, runId, wakeupRequestId, issueId, rootIssueId };
|
||||
}
|
||||
|
||||
async function seedInReviewParticipantRunFixture(input?: {
|
||||
wakeReason?: string;
|
||||
retryReason?: string | null;
|
||||
}) {
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const runId = randomUUID();
|
||||
const wakeupRequestId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
const stageId = randomUUID();
|
||||
const now = new Date("2026-03-19T00:00:00.000Z");
|
||||
const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`;
|
||||
const wakeReason = input?.wakeReason ?? "execution_review_requested";
|
||||
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix,
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
|
||||
await db.insert(agents).values({
|
||||
id: agentId,
|
||||
companyId,
|
||||
name: "CodexReviewer",
|
||||
role: "engineer",
|
||||
status: "idle",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
|
||||
await db.insert(agentWakeupRequests).values({
|
||||
id: wakeupRequestId,
|
||||
companyId,
|
||||
agentId,
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: wakeReason,
|
||||
payload: {
|
||||
issueId,
|
||||
...(input?.retryReason ? { retryReason: input.retryReason } : {}),
|
||||
},
|
||||
status: "queued",
|
||||
runId,
|
||||
requestedAt: now,
|
||||
updatedAt: now,
|
||||
});
|
||||
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runId,
|
||||
companyId,
|
||||
agentId,
|
||||
invocationSource: "automation",
|
||||
triggerDetail: "system",
|
||||
status: "queued",
|
||||
wakeupRequestId,
|
||||
contextSnapshot: {
|
||||
issueId,
|
||||
taskId: issueId,
|
||||
wakeReason,
|
||||
...(input?.retryReason ? { retryReason: input.retryReason } : {}),
|
||||
},
|
||||
updatedAt: now,
|
||||
});
|
||||
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
title: "Review participant stayed pending",
|
||||
status: "in_review",
|
||||
priority: "medium",
|
||||
assigneeAgentId: agentId,
|
||||
assigneeUserId: null,
|
||||
executionRunId: runId,
|
||||
executionAgentNameKey: "codexreviewer",
|
||||
executionLockedAt: now,
|
||||
issueNumber: 1,
|
||||
identifier: `${issuePrefix}-1`,
|
||||
executionState: {
|
||||
status: "pending",
|
||||
currentStageId: stageId,
|
||||
currentStageIndex: 0,
|
||||
currentStageType: "review",
|
||||
currentParticipant: { type: "agent", agentId, userId: null },
|
||||
returnAssignee: { type: "agent", agentId, userId: null },
|
||||
reviewRequest: null,
|
||||
completedStageIds: [],
|
||||
lastDecisionId: null,
|
||||
lastDecisionOutcome: null,
|
||||
},
|
||||
});
|
||||
|
||||
return { companyId, agentId, runId, wakeupRequestId, issueId, stageId };
|
||||
}
|
||||
|
||||
async function seedAssignedTodoNoRunFixture(input?: {
|
||||
agentStatus?: "paused" | "idle" | "running";
|
||||
}) {
|
||||
@@ -816,10 +913,12 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
agentId: string;
|
||||
issueId: string;
|
||||
runId: string;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
retryReason?: "assignment_recovery" | "issue_continuation_needed" | null;
|
||||
previousStatus: "todo" | "in_progress" | "in_review";
|
||||
retryReason?: "assignment_recovery" | "issue_continuation_needed" | "execution_review_participant_recovery" | null;
|
||||
cause?: string;
|
||||
kind?: string;
|
||||
previousOwnerAgentId?: string | null;
|
||||
returnOwnerAgentId?: string | null;
|
||||
}) {
|
||||
const action = await waitForValue(async () =>
|
||||
db.select().from(issueRecoveryActions).where(
|
||||
@@ -839,8 +938,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
status: "active",
|
||||
ownerType: "agent",
|
||||
ownerAgentId: input.agentId,
|
||||
previousOwnerAgentId: input.agentId,
|
||||
returnOwnerAgentId: input.agentId,
|
||||
previousOwnerAgentId: input.previousOwnerAgentId ?? input.agentId,
|
||||
returnOwnerAgentId: input.returnOwnerAgentId ?? input.agentId,
|
||||
cause: input.cause ?? "stranded_assigned_issue",
|
||||
attemptCount: 1,
|
||||
maxAttempts: null,
|
||||
@@ -851,9 +950,13 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
latestRunId: input.runId,
|
||||
retryReason: input.retryReason ?? null,
|
||||
});
|
||||
expect(action.nextAction).toContain(
|
||||
input.kind === "missing_disposition" ? "valid issue disposition" : "Restore a live execution path",
|
||||
);
|
||||
if (input.cause === "execution_review_participant_recovery") {
|
||||
expect(action.nextAction).toContain("failed review participant path");
|
||||
} else {
|
||||
expect(action.nextAction).toContain(
|
||||
input.kind === "missing_disposition" ? "valid issue disposition" : "Restore a live execution path",
|
||||
);
|
||||
}
|
||||
|
||||
const recoveryIssues = await db
|
||||
.select()
|
||||
@@ -2554,6 +2657,458 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("re-enqueues an already stranded execution-review participant during reconciliation", async () => {
|
||||
const { agentId, issueId, runId, wakeupRequestId, stageId } = await seedInReviewParticipantRunFixture();
|
||||
const finishedAt = new Date("2026-03-19T00:05:00.000Z");
|
||||
await db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
status: "succeeded",
|
||||
startedAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
finishedAt,
|
||||
updatedAt: finishedAt,
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, runId));
|
||||
await db
|
||||
.update(agentWakeupRequests)
|
||||
.set({
|
||||
status: "completed",
|
||||
finishedAt,
|
||||
updatedAt: finishedAt,
|
||||
})
|
||||
.where(eq(agentWakeupRequests.id, wakeupRequestId));
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.reviewParticipantRequeued).toBe(1);
|
||||
expect(result.escalated).toBe(0);
|
||||
expect(result.issueIds).toEqual([issueId]);
|
||||
|
||||
const runs = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, agentId));
|
||||
const retryRun = runs.find((row) => row.id !== runId);
|
||||
expect(["queued", "running"]).toContain(retryRun?.status);
|
||||
expect(retryRun).toMatchObject({
|
||||
retryOfRunId: runId,
|
||||
});
|
||||
expect(retryRun?.contextSnapshot).toMatchObject({
|
||||
issueId,
|
||||
taskId: issueId,
|
||||
wakeReason: "execution_review_participant_recovery",
|
||||
retryReason: "execution_review_participant_recovery",
|
||||
source: "issue.execution_review_recovery",
|
||||
retryOfRunId: runId,
|
||||
currentStageId: stageId,
|
||||
currentStageType: "review",
|
||||
reviewRecoveryInstruction: expect.stringContaining("Submit the review decision now"),
|
||||
});
|
||||
expect(retryRun?.contextSnapshot as Record<string, unknown>).not.toHaveProperty("modelProfile");
|
||||
});
|
||||
|
||||
it("re-enqueues a stranded execution-review participant when another agent has the latest issue run", async () => {
|
||||
const { companyId, agentId, issueId, runId, wakeupRequestId, stageId } =
|
||||
await seedInReviewParticipantRunFixture();
|
||||
const otherAgentId = randomUUID();
|
||||
const otherRunId = randomUUID();
|
||||
const finishedAt = new Date("2026-03-19T00:05:00.000Z");
|
||||
|
||||
await db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
status: "succeeded",
|
||||
startedAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
finishedAt,
|
||||
updatedAt: finishedAt,
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, runId));
|
||||
await db
|
||||
.update(agentWakeupRequests)
|
||||
.set({
|
||||
status: "completed",
|
||||
finishedAt,
|
||||
updatedAt: finishedAt,
|
||||
})
|
||||
.where(eq(agentWakeupRequests.id, wakeupRequestId));
|
||||
|
||||
await db.insert(agents).values({
|
||||
id: otherAgentId,
|
||||
companyId,
|
||||
name: "CodexImplementor",
|
||||
role: "engineer",
|
||||
status: "idle",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: otherRunId,
|
||||
companyId,
|
||||
agentId: otherAgentId,
|
||||
invocationSource: "automation",
|
||||
triggerDetail: "system",
|
||||
status: "succeeded",
|
||||
contextSnapshot: {
|
||||
issueId,
|
||||
taskId: issueId,
|
||||
wakeReason: "issue_commented",
|
||||
},
|
||||
startedAt: new Date("2026-03-19T00:10:00.000Z"),
|
||||
finishedAt: new Date("2026-03-19T00:15:00.000Z"),
|
||||
createdAt: new Date(Date.now() + 1_000),
|
||||
updatedAt: new Date("2026-03-19T00:15:00.000Z"),
|
||||
});
|
||||
|
||||
const heartbeat = heartbeatService(db);
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.reviewParticipantRequeued).toBe(1);
|
||||
expect(result.escalated).toBe(0);
|
||||
expect(result.issueIds).toEqual([issueId]);
|
||||
|
||||
const retryRun = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, agentId))
|
||||
.then((runs) =>
|
||||
runs.find((row) =>
|
||||
row.id !== runId &&
|
||||
(row.contextSnapshot as Record<string, unknown> | null)?.retryReason ===
|
||||
"execution_review_participant_recovery"
|
||||
) ?? null
|
||||
);
|
||||
expect(retryRun).toMatchObject({
|
||||
retryOfRunId: runId,
|
||||
});
|
||||
expect(retryRun?.contextSnapshot).toMatchObject({
|
||||
issueId,
|
||||
currentStageId: stageId,
|
||||
currentStageType: "review",
|
||||
});
|
||||
});
|
||||
|
||||
it("re-enqueues a stranded execution-review participant when another agent has a queued issue wake", async () => {
|
||||
const { companyId, agentId, issueId, runId, wakeupRequestId } =
|
||||
await seedInReviewParticipantRunFixture();
|
||||
const otherAgentId = randomUUID();
|
||||
const otherWakeId = randomUUID();
|
||||
const finishedAt = new Date("2026-03-19T00:05:00.000Z");
|
||||
|
||||
await db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
status: "succeeded",
|
||||
startedAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
finishedAt,
|
||||
updatedAt: finishedAt,
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, runId));
|
||||
await db
|
||||
.update(agentWakeupRequests)
|
||||
.set({
|
||||
status: "completed",
|
||||
finishedAt,
|
||||
updatedAt: finishedAt,
|
||||
})
|
||||
.where(eq(agentWakeupRequests.id, wakeupRequestId));
|
||||
|
||||
await db.insert(agents).values({
|
||||
id: otherAgentId,
|
||||
companyId,
|
||||
name: "CodexImplementor",
|
||||
role: "engineer",
|
||||
status: "idle",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
await db.insert(agentWakeupRequests).values({
|
||||
id: otherWakeId,
|
||||
companyId,
|
||||
agentId: otherAgentId,
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: "issue_commented",
|
||||
payload: { issueId },
|
||||
status: "queued",
|
||||
});
|
||||
|
||||
const heartbeat = heartbeatService(db);
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.reviewParticipantRequeued).toBe(1);
|
||||
expect(result.escalated).toBe(0);
|
||||
expect(result.issueIds).toEqual([issueId]);
|
||||
|
||||
const wakeups = await db
|
||||
.select()
|
||||
.from(agentWakeupRequests)
|
||||
.where(eq(agentWakeupRequests.agentId, agentId));
|
||||
expect(wakeups.some((wakeup) =>
|
||||
wakeup.reason === "execution_review_participant_recovery" &&
|
||||
wakeup.status !== "skipped"
|
||||
)).toBe(true);
|
||||
});
|
||||
|
||||
it("retries a pending execution-review participant when another agent has an active issue run", async () => {
|
||||
const { companyId, agentId, issueId, runId } = await seedInReviewParticipantRunFixture();
|
||||
const otherAgentId = randomUUID();
|
||||
const otherRunId = randomUUID();
|
||||
await db.insert(agents).values({
|
||||
id: otherAgentId,
|
||||
companyId,
|
||||
name: "CodexImplementor",
|
||||
role: "engineer",
|
||||
status: "idle",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: otherRunId,
|
||||
companyId,
|
||||
agentId: otherAgentId,
|
||||
invocationSource: "automation",
|
||||
triggerDetail: "system",
|
||||
status: "running",
|
||||
contextSnapshot: {
|
||||
issueId,
|
||||
taskId: issueId,
|
||||
wakeReason: "issue_commented",
|
||||
},
|
||||
startedAt: new Date("2026-03-19T00:01:00.000Z"),
|
||||
updatedAt: new Date("2026-03-19T00:01:00.000Z"),
|
||||
});
|
||||
|
||||
const heartbeat = heartbeatService(db);
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
const reviewRecoveryRun = await waitForValue(async () => {
|
||||
const runs = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, agentId));
|
||||
return runs.find((row) =>
|
||||
(row.contextSnapshot as Record<string, unknown> | null)?.retryReason ===
|
||||
"execution_review_participant_recovery"
|
||||
) ?? null;
|
||||
}, 8_000);
|
||||
|
||||
expect(reviewRecoveryRun).toMatchObject({
|
||||
companyId,
|
||||
agentId,
|
||||
retryOfRunId: runId,
|
||||
});
|
||||
});
|
||||
|
||||
it("does not immediately recover a generic on-demand run used for an in-review agent API update", async () => {
|
||||
const { agentId, issueId, runId } = await seedInReviewParticipantRunFixture({
|
||||
wakeReason: "manual",
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
const settledRun = await waitForRunToSettle(heartbeat, runId, 8_000);
|
||||
expect(settledRun?.status).toBe("succeeded");
|
||||
|
||||
const runs = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, agentId));
|
||||
expect(runs.some((row) =>
|
||||
(row.contextSnapshot as Record<string, unknown> | null)?.retryReason ===
|
||||
"execution_review_participant_recovery"
|
||||
)).toBe(false);
|
||||
|
||||
const issue = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(eq(issues.id, issueId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
expect(issue?.status).toBe("in_review");
|
||||
expect(issue?.assigneeAgentId).toBe(agentId);
|
||||
});
|
||||
|
||||
it("retries a pending execution-review participant once before blocking with a recovery action", async () => {
|
||||
const { companyId, agentId, issueId, runId, stageId } = await seedInReviewParticipantRunFixture();
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
const reviewRecoveryRun = await waitForValue(async () => {
|
||||
const runs = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, agentId));
|
||||
return runs.find((row) =>
|
||||
(row.contextSnapshot as Record<string, unknown> | null)?.retryReason ===
|
||||
"execution_review_participant_recovery" &&
|
||||
row.status !== "queued" &&
|
||||
row.status !== "running"
|
||||
) ?? null;
|
||||
}, 8_000);
|
||||
expect(reviewRecoveryRun).toBeTruthy();
|
||||
expect(reviewRecoveryRun).toMatchObject({
|
||||
companyId,
|
||||
agentId,
|
||||
retryOfRunId: runId,
|
||||
status: "succeeded",
|
||||
});
|
||||
expect(reviewRecoveryRun?.contextSnapshot).toMatchObject({
|
||||
issueId,
|
||||
taskId: issueId,
|
||||
wakeReason: "execution_review_participant_recovery",
|
||||
retryReason: "execution_review_participant_recovery",
|
||||
source: "issue.execution_review_recovery",
|
||||
retryOfRunId: runId,
|
||||
currentStageId: stageId,
|
||||
currentStageType: "review",
|
||||
reviewRecoveryInstruction: expect.stringContaining("Submit the review decision now"),
|
||||
});
|
||||
expect(reviewRecoveryRun?.contextSnapshot as Record<string, unknown>).not.toHaveProperty("modelProfile");
|
||||
|
||||
const sourceIssue = await waitForValue(async () => {
|
||||
const row = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(eq(issues.id, issueId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return row?.status === "blocked" ? row : null;
|
||||
}, 8_000);
|
||||
expect(sourceIssue).toMatchObject({
|
||||
status: "blocked",
|
||||
assigneeAgentId: agentId,
|
||||
executionRunId: null,
|
||||
});
|
||||
|
||||
const recoveryAction = await expectSourceScopedStrandedRecoveryAction({
|
||||
companyId,
|
||||
agentId,
|
||||
issueId,
|
||||
runId: reviewRecoveryRun!.id,
|
||||
previousStatus: "in_review",
|
||||
retryReason: "execution_review_participant_recovery",
|
||||
cause: "execution_review_participant_recovery",
|
||||
});
|
||||
expect(recoveryAction.evidence).toMatchObject({
|
||||
latestRunId: reviewRecoveryRun?.id,
|
||||
latestRunStatus: "succeeded",
|
||||
latestRunErrorCode: null,
|
||||
recoveryCause: "execution_review_participant_recovery",
|
||||
});
|
||||
|
||||
const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId));
|
||||
const recoveryComment = comments.find((comment) =>
|
||||
comment.body.includes("pending execution-review participant once") &&
|
||||
comment.body.includes(`Recovery action: \`${recoveryAction.id}\``),
|
||||
);
|
||||
expect(recoveryComment).toBeTruthy();
|
||||
|
||||
const activity = await db.select().from(activityLog).where(eq(activityLog.entityId, issueId));
|
||||
expect(activity.some((event) =>
|
||||
(event.details as Record<string, unknown> | null)?.source ===
|
||||
"recovery.reconcile_execution_review_participant",
|
||||
)).toBe(true);
|
||||
});
|
||||
|
||||
it("blocks failed execution-review recovery under the reviewer when the source assignee differs", async () => {
|
||||
const { companyId, agentId, issueId, runId, wakeupRequestId, stageId } =
|
||||
await seedInReviewParticipantRunFixture({
|
||||
wakeReason: "execution_review_participant_recovery",
|
||||
retryReason: "execution_review_participant_recovery",
|
||||
});
|
||||
const sourceAssigneeAgentId = randomUUID();
|
||||
const finishedAt = new Date("2026-03-19T00:05:00.000Z");
|
||||
|
||||
await db.insert(agents).values({
|
||||
id: sourceAssigneeAgentId,
|
||||
companyId,
|
||||
name: "CodexImplementor",
|
||||
role: "engineer",
|
||||
status: "idle",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
assigneeAgentId: sourceAssigneeAgentId,
|
||||
executionState: {
|
||||
status: "pending",
|
||||
currentStageId: stageId,
|
||||
currentStageIndex: 0,
|
||||
currentStageType: "review",
|
||||
currentParticipant: { type: "agent", agentId, userId: null },
|
||||
returnAssignee: { type: "agent", agentId: sourceAssigneeAgentId, userId: null },
|
||||
reviewRequest: null,
|
||||
completedStageIds: [],
|
||||
lastDecisionId: null,
|
||||
lastDecisionOutcome: null,
|
||||
},
|
||||
})
|
||||
.where(eq(issues.id, issueId));
|
||||
await db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
status: "failed",
|
||||
startedAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
finishedAt,
|
||||
updatedAt: finishedAt,
|
||||
errorCode: "adapter_failed",
|
||||
error: "review recovery failed before submitting a decision",
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, runId));
|
||||
await db
|
||||
.update(agentWakeupRequests)
|
||||
.set({
|
||||
status: "failed",
|
||||
claimedAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
finishedAt,
|
||||
updatedAt: finishedAt,
|
||||
error: "review recovery failed before submitting a decision",
|
||||
})
|
||||
.where(eq(agentWakeupRequests.id, wakeupRequestId));
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.reviewParticipantRequeued).toBe(0);
|
||||
expect(result.escalated).toBe(1);
|
||||
expect(result.issueIds).toEqual([issueId]);
|
||||
|
||||
const sourceIssue = await waitForValue(async () => {
|
||||
const row = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(eq(issues.id, issueId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return row?.status === "blocked" ? row : null;
|
||||
});
|
||||
expect(sourceIssue).toMatchObject({
|
||||
status: "blocked",
|
||||
assigneeAgentId: agentId,
|
||||
});
|
||||
|
||||
const recoveryAction = await expectSourceScopedStrandedRecoveryAction({
|
||||
companyId,
|
||||
agentId,
|
||||
issueId,
|
||||
runId,
|
||||
previousStatus: "in_review",
|
||||
retryReason: "execution_review_participant_recovery",
|
||||
cause: "execution_review_participant_recovery",
|
||||
previousOwnerAgentId: sourceAssigneeAgentId,
|
||||
returnOwnerAgentId: sourceAssigneeAgentId,
|
||||
});
|
||||
expect(recoveryAction.evidence).toMatchObject({
|
||||
latestRunId: runId,
|
||||
latestRunStatus: "failed",
|
||||
latestRunErrorCode: "adapter_failed",
|
||||
recoveryCause: "execution_review_participant_recovery",
|
||||
});
|
||||
});
|
||||
|
||||
it.each([
|
||||
["failed", "adapter_failed"],
|
||||
["failed", "process_lost"],
|
||||
|
||||
@@ -279,6 +279,9 @@ const WORKSPACE_VALIDATION_FAILURE_CODE = "workspace_validation_failed";
|
||||
const WORKSPACE_VALIDATION_RECOVERY_CAUSE = "workspace_validation_failed";
|
||||
const CONFIGURATION_INCOMPLETE_FAILURE_CODE = "configuration_incomplete";
|
||||
const CONFIGURATION_INCOMPLETE_RECOVERY_CAUSE = "configuration_incomplete";
|
||||
const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON = "execution_review_participant_recovery";
|
||||
const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON = "execution_review_participant_recovery";
|
||||
const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE = "execution_review_participant_recovery";
|
||||
const GITHUB_PR_WORKFLOW_SKILL_KEY = "paperclipai/bundled/software-development/github-pr-workflow";
|
||||
const GITHUB_PR_WORKFLOW_SKILL_SLUG = "github-pr-workflow";
|
||||
const PUSH_CAPABILITY_ENV_KEYS = ["GH_TOKEN", "GITHUB_TOKEN"] as const;
|
||||
@@ -2035,7 +2038,10 @@ function summarizeRunFailureForIssueComment(
|
||||
|
||||
function didAutomaticRecoveryFail(
|
||||
latestRun: Pick<typeof heartbeatRuns.$inferSelect, "status" | "contextSnapshot"> | null,
|
||||
expectedRetryReason: "assignment_recovery" | "issue_continuation_needed",
|
||||
expectedRetryReason:
|
||||
| "assignment_recovery"
|
||||
| "issue_continuation_needed"
|
||||
| typeof EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON,
|
||||
) {
|
||||
if (!latestRun) return false;
|
||||
|
||||
@@ -2049,6 +2055,27 @@ function didAutomaticRecoveryFail(
|
||||
);
|
||||
}
|
||||
|
||||
function isExecutionReviewParticipantRecoveryRun(
|
||||
run: Pick<typeof heartbeatRuns.$inferSelect, "contextSnapshot"> | null,
|
||||
) {
|
||||
if (!run) return false;
|
||||
const context = parseObject(run.contextSnapshot);
|
||||
return readNonEmptyString(context.retryReason) === EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON;
|
||||
}
|
||||
|
||||
function isExecutionReviewParticipantRecoveryEligibleRun(
|
||||
run: Pick<typeof heartbeatRuns.$inferSelect, "contextSnapshot"> | null,
|
||||
) {
|
||||
if (!run) return false;
|
||||
const context = parseObject(run.contextSnapshot);
|
||||
const wakeReason = readNonEmptyString(context.wakeReason);
|
||||
return (
|
||||
wakeReason === "execution_review_requested" ||
|
||||
wakeReason === "execution_approval_requested" ||
|
||||
isExecutionReviewParticipantRecoveryRun(run)
|
||||
);
|
||||
}
|
||||
|
||||
function normalizeLedgerBillingType(value: unknown): BillingType {
|
||||
const raw = readNonEmptyString(value);
|
||||
switch (raw) {
|
||||
@@ -2397,6 +2424,7 @@ export function shouldResetTaskSessionForWake(
|
||||
if (
|
||||
wakeReason === "issue_assigned" ||
|
||||
wakeReason === "execution_review_requested" ||
|
||||
wakeReason === EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON ||
|
||||
wakeReason === "execution_approval_requested" ||
|
||||
wakeReason === "execution_changes_requested" ||
|
||||
// PF-4: timer-driven wakes are exploratory ("any new work?"). They do not
|
||||
@@ -2509,6 +2537,9 @@ export function describeSessionResetReason(
|
||||
const wakeReason = readNonEmptyString(contextSnapshot?.wakeReason);
|
||||
if (wakeReason === "issue_assigned") return "wake reason is issue_assigned";
|
||||
if (wakeReason === "execution_review_requested") return "wake reason is execution_review_requested";
|
||||
if (wakeReason === EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON) {
|
||||
return `wake reason is ${EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON}`;
|
||||
}
|
||||
if (wakeReason === "execution_approval_requested") return "wake reason is execution_approval_requested";
|
||||
if (wakeReason === "execution_changes_requested") return "wake reason is execution_changes_requested";
|
||||
// PF-4: paired with shouldResetTaskSessionForWake — keep the reason wording
|
||||
@@ -11458,6 +11489,18 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
);
|
||||
}
|
||||
|
||||
function buildExecutionReviewParticipantRecoveryComment(input: {
|
||||
latestRun: Pick<typeof heartbeatRuns.$inferSelect, "error" | "errorCode"> | null | undefined;
|
||||
}) {
|
||||
const failureSummary = summarizeRunFailureForIssueComment(input.latestRun);
|
||||
return (
|
||||
"Paperclip retried the pending execution-review participant once, but the review stage still has no completed decision " +
|
||||
`or live reviewer run.${failureSummary ?? ""} ` +
|
||||
"Moving it to `blocked` with a source-scoped recovery action so the recovery owner can repair the reviewer runtime, " +
|
||||
"restore the review stage, or record an intentional manual resolution."
|
||||
);
|
||||
}
|
||||
|
||||
async function releaseIssueExecutionAndPromote(
|
||||
run: typeof heartbeatRuns.$inferSelect,
|
||||
options: { suppressImmediateRecovery?: boolean } = {},
|
||||
@@ -11839,6 +11882,149 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
};
|
||||
}
|
||||
|
||||
const findExistingExecutionPath = (agentId?: string | null) =>
|
||||
tx
|
||||
.select({ id: heartbeatRuns.id })
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.companyId, issue.companyId),
|
||||
inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]),
|
||||
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issue.id}`,
|
||||
sql`${heartbeatRuns.id} <> ${run.id}`,
|
||||
agentId ? eq(heartbeatRuns.agentId, agentId) : sql`true`,
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
const issueHasScheduledMonitor =
|
||||
issue.monitorNextCheckAt instanceof Date &&
|
||||
issue.monitorNextCheckAt.getTime() > Date.now();
|
||||
const executionState = parseIssueExecutionState(issue.executionState);
|
||||
const currentParticipant = executionState?.status === "pending"
|
||||
? executionState.currentParticipant
|
||||
: null;
|
||||
const issueNeedsReviewParticipantRecovery =
|
||||
issue.status === "in_review" &&
|
||||
!issue.assigneeUserId &&
|
||||
currentParticipant?.type === "agent" &&
|
||||
currentParticipant.agentId === run.agentId &&
|
||||
isExecutionReviewParticipantRecoveryEligibleRun(run) &&
|
||||
HEARTBEAT_RUN_TERMINAL_STATUSES.includes(
|
||||
run.status as (typeof HEARTBEAT_RUN_TERMINAL_STATUSES)[number],
|
||||
);
|
||||
|
||||
if (issueNeedsReviewParticipantRecovery) {
|
||||
const existingReviewParticipantExecutionPath = await findExistingExecutionPath(currentParticipant.agentId);
|
||||
if (
|
||||
options.suppressImmediateRecovery ||
|
||||
existingReviewParticipantExecutionPath ||
|
||||
issueHasScheduledMonitor ||
|
||||
await isAutomaticRecoverySuppressedByPauseHold(db, issue.companyId, issue.id, treeControlSvc)
|
||||
) {
|
||||
return { kind: "released" as const };
|
||||
}
|
||||
|
||||
if (issue.originKind === RECOVERY_ORIGIN_KINDS.strandedIssueRecovery) {
|
||||
return {
|
||||
kind: "blocked_recovery_in_place" as const,
|
||||
issue,
|
||||
previousStatus: issue.status,
|
||||
};
|
||||
}
|
||||
|
||||
const shouldBlockReviewRecovery =
|
||||
!recoveryAgentInvokable ||
|
||||
!recoveryAgent ||
|
||||
isExecutionReviewParticipantRecoveryRun(run);
|
||||
if (shouldBlockReviewRecovery) {
|
||||
return {
|
||||
kind: "blocked" as const,
|
||||
issue,
|
||||
previousStatus: issue.status,
|
||||
comment: buildExecutionReviewParticipantRecoveryComment({ latestRun: run }),
|
||||
recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE,
|
||||
recoveryOwnerAgentId: currentParticipant.agentId,
|
||||
};
|
||||
}
|
||||
|
||||
const now = new Date();
|
||||
const wakeupRequest = await tx
|
||||
.insert(agentWakeupRequests)
|
||||
.values({
|
||||
companyId: issue.companyId,
|
||||
agentId: recoveryAgent.id,
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON,
|
||||
payload: withRecoveryModelProfileHint({
|
||||
issueId: issue.id,
|
||||
retryOfRunId: run.id,
|
||||
retryReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON,
|
||||
currentStageId: executionState?.currentStageId ?? null,
|
||||
currentStageType: executionState?.currentStageType ?? null,
|
||||
}, "normal_model"),
|
||||
status: "queued",
|
||||
requestedByActorType: "system",
|
||||
requestedByActorId: null,
|
||||
updatedAt: now,
|
||||
})
|
||||
.returning()
|
||||
.then((rows) => rows[0]);
|
||||
|
||||
const queuedRun = await tx
|
||||
.insert(heartbeatRuns)
|
||||
.values({
|
||||
companyId: issue.companyId,
|
||||
agentId: recoveryAgent.id,
|
||||
invocationSource: "automation",
|
||||
triggerDetail: "system",
|
||||
status: "queued",
|
||||
wakeupRequestId: wakeupRequest.id,
|
||||
contextSnapshot: withRecoveryModelProfileHint({
|
||||
issueId: issue.id,
|
||||
taskId: issue.id,
|
||||
wakeReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON,
|
||||
retryReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON,
|
||||
source: "issue.execution_review_recovery",
|
||||
retryOfRunId: run.id,
|
||||
currentStageId: executionState?.currentStageId ?? null,
|
||||
currentStageType: executionState?.currentStageType ?? null,
|
||||
reviewRecoveryInstruction:
|
||||
"The previous reviewer run ended while this execution-review stage was still pending. Submit the review decision now, or mark the issue blocked with the exact unblock action.",
|
||||
}, "normal_model"),
|
||||
sessionIdBefore: recoverySessionBefore,
|
||||
retryOfRunId: run.id,
|
||||
updatedAt: now,
|
||||
})
|
||||
.returning()
|
||||
.then((rows) => rows[0]);
|
||||
|
||||
await tx
|
||||
.update(agentWakeupRequests)
|
||||
.set({
|
||||
runId: queuedRun.id,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(eq(agentWakeupRequests.id, wakeupRequest.id));
|
||||
|
||||
await tx
|
||||
.update(issues)
|
||||
.set({
|
||||
executionRunId: queuedRun.id,
|
||||
executionAgentNameKey: recoveryAgentNameKey,
|
||||
executionLockedAt: now,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(eq(issues.id, issue.id));
|
||||
|
||||
return {
|
||||
kind: "queued_recovery" as const,
|
||||
run: queuedRun,
|
||||
};
|
||||
}
|
||||
|
||||
const issueNeedsImmediateRecovery =
|
||||
(issue.status === "todo" || issue.status === "in_progress") &&
|
||||
!issue.assigneeUserId &&
|
||||
@@ -11852,19 +12038,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
return { kind: "released" as const };
|
||||
}
|
||||
|
||||
const existingExecutionPath = await tx
|
||||
.select({ id: heartbeatRuns.id })
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.companyId, issue.companyId),
|
||||
inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]),
|
||||
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issue.id}`,
|
||||
sql`${heartbeatRuns.id} <> ${run.id}`,
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const existingExecutionPath = await findExistingExecutionPath();
|
||||
if (existingExecutionPath) {
|
||||
return { kind: "released" as const };
|
||||
}
|
||||
@@ -11987,7 +12161,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
if (promotionResult?.kind === "blocked") {
|
||||
await recovery.escalateStrandedAssignedIssue({
|
||||
issue: promotionResult.issue,
|
||||
previousStatus: promotionResult.previousStatus as "todo" | "in_progress",
|
||||
previousStatus: promotionResult.previousStatus as "todo" | "in_progress" | "in_review",
|
||||
latestRun: run,
|
||||
comment: promotionResult.comment,
|
||||
recoveryCause:
|
||||
@@ -11995,7 +12169,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
? WORKSPACE_VALIDATION_RECOVERY_CAUSE
|
||||
: promotionResult.recoveryCause === CONFIGURATION_INCOMPLETE_RECOVERY_CAUSE
|
||||
? CONFIGURATION_INCOMPLETE_RECOVERY_CAUSE
|
||||
: promotionResult.recoveryCause === EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE
|
||||
? EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE
|
||||
: undefined,
|
||||
recoveryOwnerAgentId: promotionResult.recoveryOwnerAgentId,
|
||||
});
|
||||
return;
|
||||
}
|
||||
@@ -12003,7 +12180,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
if (promotionResult?.kind === "blocked_recovery_in_place") {
|
||||
await recovery.escalateStrandedRecoveryIssueInPlace({
|
||||
issue: promotionResult.issue,
|
||||
previousStatus: promotionResult.previousStatus as "todo" | "in_progress",
|
||||
previousStatus: promotionResult.previousStatus as "todo" | "in_progress" | "in_review",
|
||||
latestRun: run,
|
||||
});
|
||||
return;
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { and, asc, desc, eq, gt, gte, inArray, isNull, notInArray, sql } from "drizzle-orm";
|
||||
import { and, asc, desc, eq, gt, gte, inArray, isNull, notInArray, or, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
DEFAULT_ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_LOOKBACK_HOURS,
|
||||
@@ -37,6 +37,7 @@ import { instanceSettingsService } from "../instance-settings.js";
|
||||
import { issueRecoveryActionService } from "../issue-recovery-actions.js";
|
||||
import { issueTreeControlService } from "../issue-tree-control.js";
|
||||
import { TERMINAL_HEARTBEAT_RUN_STATUSES, issueService } from "../issues.js";
|
||||
import { parseIssueExecutionState } from "../issue-execution-policy.js";
|
||||
import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js";
|
||||
import { getRunLogStore } from "../run-log-store.js";
|
||||
import {
|
||||
@@ -72,6 +73,7 @@ const ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES = 8 * 1024;
|
||||
const STRANDED_ISSUE_RECOVERY_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.strandedIssueRecovery;
|
||||
const STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.staleActiveRunEvaluation;
|
||||
const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext";
|
||||
const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON = "execution_review_participant_recovery";
|
||||
const SESSIONED_LOCAL_ADAPTERS = new Set([
|
||||
"claude_local",
|
||||
"codex_local",
|
||||
@@ -122,8 +124,11 @@ type StrandedRecoveryCause =
|
||||
| "stranded_assigned_issue"
|
||||
| "workspace_validation_failed"
|
||||
| "configuration_incomplete"
|
||||
| "execution_review_participant_recovery"
|
||||
| typeof SUCCESSFUL_RUN_MISSING_STATE_REASON;
|
||||
|
||||
type StrandedPreviousStatus = "todo" | "in_progress" | "in_review";
|
||||
|
||||
type SuccessfulRunHandoffRecoveryEvidence = {
|
||||
sourceRunId: string | null;
|
||||
correctiveRunId: string;
|
||||
@@ -175,9 +180,29 @@ function summarizeRunFailureForIssueComment(run: LatestIssueRun) {
|
||||
return null;
|
||||
}
|
||||
|
||||
function buildExecutionReviewParticipantRecoveryComment(latestRun: LatestIssueRun) {
|
||||
const failureSummary = summarizeRunFailureForIssueComment(latestRun);
|
||||
return (
|
||||
"Paperclip retried the pending execution-review participant once, but the review stage still has no completed decision " +
|
||||
`or live reviewer run.${failureSummary ?? ""} ` +
|
||||
"Moving it to `blocked` with a source-scoped recovery action so the recovery owner can repair the reviewer runtime, " +
|
||||
"restore the review stage, or record an intentional manual resolution."
|
||||
);
|
||||
}
|
||||
|
||||
function buildExecutionReviewParticipantUnavailableComment(latestRun: LatestIssueRun) {
|
||||
const failureSummary = summarizeRunFailureForIssueComment(latestRun);
|
||||
return (
|
||||
"Paperclip cannot continue the pending execution-review participant because the participant is not invokable " +
|
||||
`and the review stage has no completed decision or live reviewer run.${failureSummary ?? ""} ` +
|
||||
"Moving it to `blocked` with a source-scoped recovery action so the recovery owner can repair the reviewer runtime, " +
|
||||
"restore the review stage, or record an intentional manual resolution."
|
||||
);
|
||||
}
|
||||
|
||||
function didAutomaticRecoveryFail(
|
||||
latestRun: LatestIssueRun,
|
||||
expectedRetryReason: "assignment_recovery" | "issue_continuation_needed",
|
||||
expectedRetryReason: "assignment_recovery" | "issue_continuation_needed" | typeof EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON,
|
||||
) {
|
||||
if (!latestRun) return false;
|
||||
|
||||
@@ -189,6 +214,11 @@ function didAutomaticRecoveryFail(
|
||||
);
|
||||
}
|
||||
|
||||
function isTerminalIssueRun(latestRun: LatestIssueRun) {
|
||||
if (!latestRun) return false;
|
||||
return TERMINAL_HEARTBEAT_RUN_STATUSES.has(latestRun.status);
|
||||
}
|
||||
|
||||
const TRANSIENT_INFRA_CONTINUATION_ERROR_CODES = new Set<string>([
|
||||
"adapter_failed",
|
||||
"codex_transient_upstream",
|
||||
@@ -523,6 +553,35 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function getLatestIssueRunForAgent(
|
||||
companyId: string,
|
||||
issueId: string,
|
||||
agentId: string,
|
||||
): Promise<LatestIssueRun> {
|
||||
return db
|
||||
.select({
|
||||
id: heartbeatRuns.id,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
status: heartbeatRuns.status,
|
||||
error: heartbeatRuns.error,
|
||||
errorCode: heartbeatRuns.errorCode,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot,
|
||||
livenessState: heartbeatRuns.livenessState,
|
||||
resultJson: heartbeatRuns.resultJson,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
eq(heartbeatRuns.agentId, agentId),
|
||||
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,
|
||||
),
|
||||
)
|
||||
.orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function summarizeRecentContinuationRetries(
|
||||
companyId: string,
|
||||
issueId: string,
|
||||
@@ -571,7 +630,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
return { consecutive, latestFinishedAt };
|
||||
}
|
||||
|
||||
async function hasActiveExecutionPath(companyId: string, issueId: string) {
|
||||
async function hasActiveExecutionPath(companyId: string, issueId: string, agentId?: string | null) {
|
||||
const [run, deferredWake] = await Promise.all([
|
||||
db
|
||||
.select({ id: heartbeatRuns.id })
|
||||
@@ -581,6 +640,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]),
|
||||
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,
|
||||
agentId ? eq(heartbeatRuns.agentId, agentId) : sql`true`,
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
@@ -593,6 +653,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
eq(agentWakeupRequests.companyId, companyId),
|
||||
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`,
|
||||
agentId ? eq(agentWakeupRequests.agentId, agentId) : sql`true`,
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
@@ -618,7 +679,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
.then((rows) => Boolean(rows[0]));
|
||||
}
|
||||
|
||||
async function hasQueuedIssueWake(companyId: string, issueId: string) {
|
||||
async function hasQueuedIssueWake(companyId: string, issueId: string, agentId?: string | null) {
|
||||
return db
|
||||
.select({ id: agentWakeupRequests.id })
|
||||
.from(agentWakeupRequests)
|
||||
@@ -627,6 +688,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
eq(agentWakeupRequests.companyId, companyId),
|
||||
eq(agentWakeupRequests.status, "queued"),
|
||||
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`,
|
||||
agentId ? eq(agentWakeupRequests.agentId, agentId) : sql`true`,
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
@@ -677,10 +739,11 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
async function enqueueStrandedIssueRecovery(input: {
|
||||
issueId: string;
|
||||
agentId: string;
|
||||
reason: "issue_assignment_recovery" | "issue_continuation_needed";
|
||||
retryReason: "assignment_recovery" | "issue_continuation_needed";
|
||||
reason: "issue_assignment_recovery" | "issue_continuation_needed" | typeof EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON;
|
||||
retryReason: "assignment_recovery" | "issue_continuation_needed" | typeof EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON;
|
||||
source: string;
|
||||
retryOfRunId?: string | null;
|
||||
extraContext?: Record<string, unknown>;
|
||||
}) {
|
||||
const queued = await deps.enqueueWakeup(input.agentId, {
|
||||
source: "automation",
|
||||
@@ -689,6 +752,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
payload: withRecoveryModelProfileHint({
|
||||
issueId: input.issueId,
|
||||
...(input.retryOfRunId ? { retryOfRunId: input.retryOfRunId } : {}),
|
||||
...(input.extraContext ?? {}),
|
||||
}, "normal_model"),
|
||||
requestedByActorType: "system",
|
||||
requestedByActorId: null,
|
||||
@@ -699,6 +763,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
retryReason: input.retryReason,
|
||||
source: input.source,
|
||||
...(input.retryOfRunId ? { retryOfRunId: input.retryOfRunId } : {}),
|
||||
...(input.extraContext ?? {}),
|
||||
}, "normal_model"),
|
||||
});
|
||||
|
||||
@@ -2052,8 +2117,12 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
async function resolveStrandedIssueRecoveryOwnerAgentId(issue: typeof issues.$inferSelect) {
|
||||
async function resolveStrandedIssueRecoveryOwnerAgentId(
|
||||
issue: typeof issues.$inferSelect,
|
||||
preferredOwnerAgentId?: string | null,
|
||||
) {
|
||||
const candidateIds: string[] = [];
|
||||
if (preferredOwnerAgentId) candidateIds.push(preferredOwnerAgentId);
|
||||
if (issue.assigneeAgentId) {
|
||||
const assignee = await getAgent(issue.assigneeAgentId);
|
||||
if (assignee?.reportsTo) candidateIds.push(assignee.reportsTo);
|
||||
@@ -2091,7 +2160,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
function buildStrandedIssueRecoveryDescription(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
latestRun: LatestIssueRun;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
previousStatus: StrandedPreviousStatus;
|
||||
prefix: string;
|
||||
recoveryCause?: StrandedRecoveryCause;
|
||||
successfulRunHandoffEvidence?: SuccessfulRunHandoffRecoveryEvidence | null;
|
||||
@@ -2134,9 +2203,26 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
|
||||
const retryReason = readNonEmptyString(parseObject(input.latestRun?.contextSnapshot)?.retryReason) ?? "unknown";
|
||||
const failureSummary = summarizeRunFailureForIssueComment(input.latestRun);
|
||||
const isReviewParticipantRecovery = input.recoveryCause === "execution_review_participant_recovery";
|
||||
const detectedInvariant = isReviewParticipantRecovery
|
||||
? "execution_review_participant_recovery"
|
||||
: "stranded_assigned_issue";
|
||||
const requiredAction = isReviewParticipantRecovery
|
||||
? [
|
||||
"- Inspect the latest reviewer run and the pending execution-review stage.",
|
||||
"- Fix the reviewer runtime, restore the source issue to `in_review` with a live participant, or record an intentional manual resolution.",
|
||||
"- When the source issue has a live review path or has been intentionally resolved, mark this recovery issue done.",
|
||||
]
|
||||
: [
|
||||
"- Inspect the latest run and source issue state.",
|
||||
"- Fix the runtime/adapter problem, reassign the source issue, or convert the source issue into a clear manual-review state.",
|
||||
"- When the source issue has a live execution path or has been intentionally resolved, mark this recovery issue done.",
|
||||
];
|
||||
|
||||
return [
|
||||
"Paperclip exhausted automatic recovery for an assigned issue and created this explicit recovery task.",
|
||||
isReviewParticipantRecovery
|
||||
? "Paperclip exhausted automatic recovery for a pending execution-review participant and created this explicit recovery task."
|
||||
: "Paperclip exhausted automatic recovery for an assigned issue and created this explicit recovery task.",
|
||||
"",
|
||||
"## Source",
|
||||
"",
|
||||
@@ -2144,7 +2230,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
`- Previous source status: \`${input.previousStatus}\``,
|
||||
`- Latest retry run: ${runLink}`,
|
||||
`- Latest retry status: \`${input.latestRun?.status ?? "unknown"}\``,
|
||||
`- Detected invariant: \`stranded_assigned_issue\``,
|
||||
`- Detected invariant: \`${detectedInvariant}\``,
|
||||
`- Retry reason: \`${retryReason}\``,
|
||||
failureSummary ? `- Failure: ${failureSummary.trim()}` : "- Failure: none recorded",
|
||||
"",
|
||||
@@ -2154,16 +2240,14 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
"",
|
||||
"## Required Action",
|
||||
"",
|
||||
"- Inspect the latest run and source issue state.",
|
||||
"- Fix the runtime/adapter problem, reassign the source issue, or convert the source issue into a clear manual-review state.",
|
||||
"- When the source issue has a live execution path or has been intentionally resolved, mark this recovery issue done.",
|
||||
...requiredAction,
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
async function ensureStrandedIssueRecoveryIssue(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
latestRun: LatestIssueRun;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
previousStatus: StrandedPreviousStatus;
|
||||
recoveryCause?: StrandedRecoveryCause;
|
||||
successfulRunHandoffEvidence?: SuccessfulRunHandoffRecoveryEvidence | null;
|
||||
}) {
|
||||
@@ -2284,7 +2368,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
function buildStrandedRecoveryActionEvidence(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
latestRun: LatestIssueRun;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
previousStatus: StrandedPreviousStatus;
|
||||
recoveryCause: StrandedRecoveryCause;
|
||||
successfulRunHandoffEvidence?: SuccessfulRunHandoffRecoveryEvidence | null;
|
||||
}) {
|
||||
@@ -2314,12 +2398,16 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
async function ensureSourceScopedStrandedRecoveryAction(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
latestRun: LatestIssueRun;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
previousStatus: StrandedPreviousStatus;
|
||||
recoveryCause?: StrandedRecoveryCause;
|
||||
recoveryOwnerAgentId?: string | null;
|
||||
successfulRunHandoffEvidence?: SuccessfulRunHandoffRecoveryEvidence | null;
|
||||
}) {
|
||||
const recoveryCause = input.recoveryCause ?? "stranded_assigned_issue";
|
||||
const ownerAgentId = await resolveStrandedIssueRecoveryOwnerAgentId(input.issue);
|
||||
const ownerAgentId = await resolveStrandedIssueRecoveryOwnerAgentId(
|
||||
input.issue,
|
||||
input.recoveryOwnerAgentId,
|
||||
);
|
||||
const now = new Date();
|
||||
const action = await recoveryActionsSvc.upsertSourceScoped({
|
||||
companyId: input.issue.companyId,
|
||||
@@ -2350,6 +2438,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
: "Repair the source issue workspace link, project workspace cwd, or git checkout before resuming adapter execution."
|
||||
: recoveryCause === "configuration_incomplete"
|
||||
? "Bind the missing secret(s) named in the run failure to the agent/project/routine env before resuming adapter execution."
|
||||
: recoveryCause === "execution_review_participant_recovery"
|
||||
? "Repair the failed review participant path, restore the source issue to in_review with a live reviewer, or record an intentional manual resolution."
|
||||
: "Restore a live execution path, fix the runtime/adapter failure, or record an intentional manual resolution.",
|
||||
wakePolicy: recoveryCause === "workspace_validation_failed" || recoveryCause === "configuration_incomplete"
|
||||
? {
|
||||
@@ -2414,7 +2504,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
|
||||
function buildRecoveryIssueInPlaceEscalationComment(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
previousStatus: StrandedPreviousStatus;
|
||||
latestRun: LatestIssueRun;
|
||||
prefix: string;
|
||||
}) {
|
||||
@@ -2441,7 +2531,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
|
||||
async function escalateStrandedRecoveryIssueInPlace(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
previousStatus: StrandedPreviousStatus;
|
||||
latestRun: LatestIssueRun;
|
||||
}) {
|
||||
const updated = await issuesSvc.update(input.issue.id, { status: "blocked" });
|
||||
@@ -2574,10 +2664,11 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
|
||||
async function escalateStrandedAssignedIssue(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
previousStatus: StrandedPreviousStatus;
|
||||
latestRun: LatestIssueRun;
|
||||
comment?: string;
|
||||
recoveryCause?: StrandedRecoveryCause;
|
||||
recoveryOwnerAgentId?: string | null;
|
||||
successfulRunHandoffEvidence?: SuccessfulRunHandoffRecoveryEvidence | null;
|
||||
}) {
|
||||
if (isStrandedIssueRecoveryIssue(input.issue)) {
|
||||
@@ -2594,6 +2685,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
previousStatus: input.previousStatus,
|
||||
latestRun: input.latestRun,
|
||||
recoveryCause,
|
||||
recoveryOwnerAgentId: input.recoveryOwnerAgentId,
|
||||
successfulRunHandoffEvidence: input.successfulRunHandoffEvidence,
|
||||
});
|
||||
const blockerIds = await existingUnresolvedBlockerIssueIds(input.issue.companyId, input.issue.id);
|
||||
@@ -2697,6 +2789,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
? "recovery.reconcile_workspace_validation_failed"
|
||||
: input.recoveryCause === "configuration_incomplete"
|
||||
? "recovery.reconcile_configuration_incomplete"
|
||||
: input.recoveryCause === "execution_review_participant_recovery"
|
||||
? "recovery.reconcile_execution_review_participant"
|
||||
: "recovery.reconcile_stranded_assigned_issue",
|
||||
recoveryCause: input.recoveryCause ?? "stranded_assigned_issue",
|
||||
latestRunId: input.latestRun?.id ?? null,
|
||||
@@ -2750,8 +2844,11 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
.where(
|
||||
and(
|
||||
isNull(issues.assigneeUserId),
|
||||
inArray(issues.status, ["todo", "in_progress"]),
|
||||
sql`${issues.assigneeAgentId} is not null`,
|
||||
inArray(issues.status, ["todo", "in_progress", "in_review"]),
|
||||
or(
|
||||
sql`${issues.assigneeAgentId} is not null`,
|
||||
eq(issues.status, "in_review"),
|
||||
),
|
||||
),
|
||||
);
|
||||
|
||||
@@ -2763,6 +2860,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
successfulContinuationObserved: 0,
|
||||
orphanBlockersAssigned: 0,
|
||||
successfulRunHandoffEscalated: 0,
|
||||
reviewParticipantRequeued: 0,
|
||||
escalated: 0,
|
||||
waitingOnReviewResolved: 0,
|
||||
recentProgressExempted: 0,
|
||||
@@ -2771,19 +2869,36 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
};
|
||||
|
||||
for (const issue of candidates) {
|
||||
const agentId = issue.assigneeAgentId;
|
||||
const executionState = issue.status === "in_review"
|
||||
? parseIssueExecutionState(issue.executionState)
|
||||
: null;
|
||||
const pendingExecutionState = executionState?.status === "pending" ? executionState : null;
|
||||
const currentParticipant = pendingExecutionState
|
||||
? pendingExecutionState.currentParticipant
|
||||
: null;
|
||||
const participantAgentId = currentParticipant?.type === "agent" ? currentParticipant.agentId : null;
|
||||
const agentId = issue.status === "in_review" && participantAgentId
|
||||
? participantAgentId
|
||||
: issue.assigneeAgentId;
|
||||
if (!agentId) {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
const agent = await getAgent(agentId);
|
||||
if (!agent || agent.companyId !== issue.companyId || !(await isAgentInvokable(agent))) {
|
||||
const agentInvokable = agent && agent.companyId === issue.companyId
|
||||
? await isAgentInvokable(agent)
|
||||
: false;
|
||||
if (issue.status !== "in_review" && !agentInvokable) {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (await hasActiveExecutionPath(issue.companyId, issue.id)) {
|
||||
if (await hasActiveExecutionPath(
|
||||
issue.companyId,
|
||||
issue.id,
|
||||
issue.status === "in_review" ? agentId : null,
|
||||
)) {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
@@ -2802,7 +2917,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
if (isStrandedIssueRecoveryIssue(issue) && isUnsuccessfulTerminalIssueRun(latestRun)) {
|
||||
const updated = await escalateStrandedRecoveryIssueInPlace({
|
||||
issue,
|
||||
previousStatus: issue.status as "todo" | "in_progress",
|
||||
previousStatus: issue.status as StrandedPreviousStatus,
|
||||
latestRun,
|
||||
});
|
||||
if (updated) {
|
||||
@@ -2814,6 +2929,108 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
continue;
|
||||
}
|
||||
|
||||
if (issue.status === "in_review") {
|
||||
if (!participantAgentId || !pendingExecutionState) {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
const participantLatestRun = await getLatestIssueRunForAgent(
|
||||
issue.companyId,
|
||||
issue.id,
|
||||
participantAgentId,
|
||||
);
|
||||
|
||||
if (!participantLatestRun || !isTerminalIssueRun(participantLatestRun)) {
|
||||
if (!agentInvokable) {
|
||||
const updated = await escalateStrandedAssignedIssue({
|
||||
issue,
|
||||
previousStatus: "in_review",
|
||||
latestRun: participantLatestRun,
|
||||
comment: buildExecutionReviewParticipantUnavailableComment(participantLatestRun),
|
||||
recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON,
|
||||
recoveryOwnerAgentId: participantAgentId,
|
||||
});
|
||||
if (updated) {
|
||||
result.escalated += 1;
|
||||
result.issueIds.push(issue.id);
|
||||
} else {
|
||||
result.skipped += 1;
|
||||
}
|
||||
} else {
|
||||
result.skipped += 1;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!agentInvokable) {
|
||||
const updated = await escalateStrandedAssignedIssue({
|
||||
issue,
|
||||
previousStatus: "in_review",
|
||||
latestRun: participantLatestRun,
|
||||
comment: buildExecutionReviewParticipantUnavailableComment(participantLatestRun),
|
||||
recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON,
|
||||
recoveryOwnerAgentId: participantAgentId,
|
||||
});
|
||||
if (updated) {
|
||||
result.escalated += 1;
|
||||
result.issueIds.push(issue.id);
|
||||
} else {
|
||||
result.skipped += 1;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (didAutomaticRecoveryFail(participantLatestRun, EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON)) {
|
||||
const updated = await escalateStrandedAssignedIssue({
|
||||
issue,
|
||||
previousStatus: "in_review",
|
||||
latestRun: participantLatestRun,
|
||||
comment: buildExecutionReviewParticipantRecoveryComment(participantLatestRun),
|
||||
recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON,
|
||||
recoveryOwnerAgentId: participantAgentId,
|
||||
});
|
||||
if (updated) {
|
||||
result.escalated += 1;
|
||||
result.issueIds.push(issue.id);
|
||||
} else {
|
||||
result.skipped += 1;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (await hasQueuedIssueWake(issue.companyId, issue.id, participantAgentId)) {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (await isInvocationBudgetBlocked(issue, participantAgentId)) {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
const queued = await enqueueStrandedIssueRecovery({
|
||||
issueId: issue.id,
|
||||
agentId: participantAgentId,
|
||||
reason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON,
|
||||
retryReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON,
|
||||
source: "issue.execution_review_recovery",
|
||||
retryOfRunId: participantLatestRun.id,
|
||||
extraContext: {
|
||||
currentStageId: pendingExecutionState.currentStageId ?? null,
|
||||
currentStageType: pendingExecutionState.currentStageType ?? null,
|
||||
reviewRecoveryInstruction:
|
||||
"The previous reviewer run ended while this execution-review stage was still pending. Submit the review decision now, or mark the issue blocked with the exact unblock action.",
|
||||
},
|
||||
});
|
||||
if (queued) {
|
||||
result.reviewParticipantRequeued += 1;
|
||||
result.issueIds.push(issue.id);
|
||||
} else {
|
||||
result.skipped += 1;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (issue.status === "todo") {
|
||||
if (!latestRun) {
|
||||
if (await hasQueuedIssueWake(issue.companyId, issue.id)) {
|
||||
|
||||
Reference in new issue
Block a user