diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index 964684dddfa..873381d5c3b 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -395,6 +395,8 @@ Side effects: - entering `done` sets `completed_at` - entering `cancelled` sets `cancelled_at` +Detailed ownership, execution, blocker, and crash-recovery semantics are documented in `doc/execution-semantics.md`. + ## 8.3 Approval Status - `pending -> approved | rejected | cancelled` diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md new file mode 100644 index 00000000000..1c841e0d595 --- /dev/null +++ b/doc/execution-semantics.md @@ -0,0 +1,252 @@ +# Execution Semantics + +Status: Current implementation guide +Date: 2026-04-13 +Audience: Product and engineering + +This document explains how Paperclip interprets issue assignment, issue status, execution runs, wakeups, parent/sub-issue structure, and blocker relationships. + +`doc/SPEC-implementation.md` remains the V1 contract. This document is the detailed execution model behind that contract. + +## 1. Core Model + +Paperclip separates four concepts that are easy to blur together: + +1. structure: parent/sub-issue relationships +2. dependency: blocker relationships +3. ownership: who is responsible for the issue now +4. execution: whether the control plane currently has a live path to move the issue forward + +The system works best when those are kept separate. + +## 2. Assignee Semantics + +An issue has at most one assignee. + +- `assigneeAgentId` means the issue is owned by an agent +- `assigneeUserId` means the issue is owned by a human board user +- both cannot be set at the same time + +This is a hard invariant. Paperclip is single-assignee by design. + +## 3. Status Semantics + +Paperclip issue statuses are not just UI labels. They imply different expectations about ownership and execution. + +### `backlog` + +The issue is not ready for active work. + +- no execution expectation +- no pickup expectation +- safe resting state for future work + +### `todo` + +The issue is actionable but not actively claimed. + +- it may be assigned or unassigned +- no checkout/execution lock is required yet +- for agent-assigned work, Paperclip may still need a wake path to ensure the assignee actually sees it + +### `in_progress` + +The issue is actively owned work. + +- requires an assignee +- for agent-owned issues, this is a strict execution-backed state +- for user-owned issues, this is a human ownership state and is not backed by heartbeat execution + +For agent-owned issues, `in_progress` should not be allowed to become a silent dead state. + +### `blocked` + +The issue cannot proceed until something external changes. + +This is the right state for: + +- waiting on another issue +- waiting on a human decision +- waiting on an external dependency or system +- work that automatic recovery could not safely continue + +### `in_review` + +Execution work is paused because the next move belongs to a reviewer or approver, not the current executor. + +### `done` + +The work is complete and terminal. + +### `cancelled` + +The work will not continue and is terminal. + +## 4. Agent-Owned vs User-Owned Execution + +The execution model differs depending on assignee type. + +### Agent-owned issues + +Agent-owned issues are part of the control plane's execution loop. + +- Paperclip can wake the assignee +- Paperclip can track runs linked to the issue +- Paperclip can recover some lost execution state after crashes/restarts + +### User-owned issues + +User-owned issues are not executed by the heartbeat scheduler. + +- Paperclip can track the ownership and status +- Paperclip cannot rely on heartbeat/run semantics to keep them moving +- stranded-work reconciliation does not apply to them + +This is why `in_progress` can be strict for agents without forcing the same runtime rules onto human-held work. + +## 5. Checkout and Active Execution + +Checkout is the bridge from issue ownership to active agent execution. + +- checkout is required to move an issue into agent-owned `in_progress` +- `checkoutRunId` represents issue-ownership lock for the current agent run +- `executionRunId` represents the currently active execution path for the issue + +These are related but not identical: + +- `checkoutRunId` answers who currently owns execution rights for the issue +- `executionRunId` answers which run is actually live right now + +Paperclip already clears stale execution locks and can adopt some stale checkout locks when the original run is gone. + +## 6. Parent/Sub-Issue vs Blockers + +Paperclip uses two different relationships for different jobs. + +### Parent/Sub-Issue (`parentId`) + +This is structural. + +Use it for: + +- work breakdown +- rollup context +- explaining why a child issue exists +- waking the parent assignee when all direct children become terminal + +Do not treat `parentId` as execution dependency by itself. + +### Blockers (`blockedByIssueIds`) + +This is dependency semantics. + +Use it for: + +- \"this issue cannot continue until that issue changes state\" +- explicit waiting relationships +- automatic wakeups when all blockers resolve + +If a parent is truly waiting on a child, model that with blockers. Do not rely on the parent/child relationship alone. + +## 7. Consistent Execution Path Rules + +For agent-assigned, non-terminal, actionable issues, Paperclip should not leave work in a state where nobody is working it and nothing will wake it. + +The relevant execution path depends on status. + +### Agent-assigned `todo` + +This is dispatch state: ready to start, not yet actively claimed. + +A healthy dispatch state means at least one of these is true: + +- the issue already has a queued/running wake path +- the issue is intentionally resting in `todo` after a successful agent heartbeat, not after an interrupted dispatch +- the issue has been explicitly surfaced as stranded + +### Agent-assigned `in_progress` + +This is active-work state. + +A healthy active-work state means at least one of these is true: + +- there is an active run for the issue +- there is already a queued continuation wake +- the issue has been explicitly surfaced as stranded + +## 8. Crash and Restart Recovery + +Paperclip now treats crash/restart recovery as a stranded-assigned-work problem, not just a stranded-run problem. + +There are two distinct failure modes. + +### 8.1 Stranded assigned `todo` + +Example: + +- issue is assigned to an agent +- status is `todo` +- the original wake/run died during or after dispatch +- after restart there is no queued wake and nothing picks the issue back up + +Recovery rule: + +- if the latest issue-linked run failed/timed out/cancelled and no live execution path remains, Paperclip queues one automatic assignment recovery wake +- if that recovery wake also finishes and the issue is still stranded, Paperclip moves the issue to `blocked` and posts a visible comment + +This is a dispatch recovery, not a continuation recovery. + +### 8.2 Stranded assigned `in_progress` + +Example: + +- issue is assigned to an agent +- status is `in_progress` +- the live run disappeared +- after restart there is no active run and no queued continuation + +Recovery rule: + +- Paperclip queues one automatic continuation wake +- if that continuation wake also finishes and the issue is still stranded, Paperclip moves the issue to `blocked` and posts a visible comment + +This is an active-work continuity recovery. + +## 9. Startup and Periodic Reconciliation + +Startup recovery and periodic recovery are different from normal wakeup delivery. + +On startup and on the periodic recovery loop, Paperclip now does three things in sequence: + +1. reap orphaned `running` runs +2. resume persisted `queued` runs +3. reconcile stranded assigned work + +That last step is what closes the gap where issue state survives a crash but the wake/run path does not. + +## 10. What This Does Not Mean + +These semantics do not change V1 into an auto-reassignment system. + +Paperclip still does not: + +- automatically reassign work to a different agent +- infer dependency semantics from `parentId` alone +- treat human-held work as heartbeat-managed execution + +The recovery model is intentionally conservative: + +- preserve ownership +- retry once when the control plane lost execution continuity +- escalate visibly when the system cannot safely keep going + +## 11. Practical Interpretation + +For a board operator, the intended meaning is: + +- agent-owned `in_progress` should mean \"this is live work or clearly surfaced as a problem\" +- agent-owned `todo` should not stay assigned forever after a crash with no remaining wake path +- parent/sub-issue explains structure +- blockers explain waiting + +That is the execution contract Paperclip should present to operators. diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 616483231c6..769c77a4ae0 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -3,12 +3,16 @@ import { spawn, type ChildProcess } from "node:child_process"; import { eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { + activityLog, agents, + agentRuntimeState, agentWakeupRequests, + companySkills, companies, createDb, heartbeatRunEvents, heartbeatRuns, + issueComments, issues, } from "@paperclipai/db"; import { @@ -33,6 +37,24 @@ vi.mock("@paperclipai/shared/telemetry", async () => { }; }); +vi.mock("../adapters/index.ts", async () => { + const actual = await vi.importActual("../adapters/index.ts"); + return { + ...actual, + getServerAdapter: vi.fn(() => ({ + supportsLocalAgentJwt: false, + execute: vi.fn(async () => ({ + exitCode: 0, + signal: null, + timedOut: false, + errorMessage: null, + provider: "test", + model: "test-model", + })), + })), + }; +}); + import { heartbeatService } from "../services/heartbeat.ts"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -68,6 +90,20 @@ async function waitForPidExit(pid: number, timeoutMs = 2_000) { return !isPidAlive(pid); } +async function waitForRunToSettle( + heartbeat: ReturnType, + runId: string, + timeoutMs = 3_000, +) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const run = await heartbeat.getRun(runId); + if (!run || (run.status !== "queued" && run.status !== "running")) return run; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + return heartbeat.getRun(runId); +} + async function spawnOrphanedProcessGroup() { const leader = spawn( process.execPath, @@ -134,6 +170,10 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { } } cleanupPids.clear(); + await db.delete(activityLog); + await db.delete(agentRuntimeState); + await db.delete(companySkills); + await db.delete(issueComments); await db.delete(issues); await db.delete(heartbeatRunEvents); await db.delete(heartbeatRuns); @@ -246,6 +286,95 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { return { companyId, agentId, runId, wakeupRequestId, issueId }; } + async function seedStrandedIssueFixture(input: { + status: "todo" | "in_progress"; + runStatus: "failed" | "timed_out" | "cancelled" | "succeeded"; + retryReason?: "assignment_recovery" | "issue_continuation_needed" | null; + assignToUser?: boolean; + }) { + const companyId = randomUUID(); + const agentId = randomUUID(); + const runId = randomUUID(); + const wakeupRequestId = randomUUID(); + const issueId = randomUUID(); + const now = new Date("2026-03-19T00:00:00.000Z"); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + + await db.insert(agentWakeupRequests).values({ + id: wakeupRequestId, + companyId, + agentId, + source: "assignment", + triggerDetail: "system", + reason: input.retryReason === "assignment_recovery" ? "issue_assignment_recovery" : "issue_assigned", + payload: { issueId }, + status: input.runStatus === "cancelled" ? "cancelled" : "failed", + runId, + claimedAt: now, + finishedAt: new Date("2026-03-19T00:05:00.000Z"), + error: input.runStatus === "succeeded" ? null : "run failed before issue advanced", + }); + + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + invocationSource: "assignment", + triggerDetail: "system", + status: input.runStatus, + wakeupRequestId, + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: input.retryReason === "assignment_recovery" + ? "issue_assignment_recovery" + : input.retryReason ?? "issue_assigned", + ...(input.retryReason ? { retryReason: input.retryReason } : {}), + }, + startedAt: now, + finishedAt: new Date("2026-03-19T00:05:00.000Z"), + updatedAt: new Date("2026-03-19T00:05:00.000Z"), + errorCode: input.runStatus === "succeeded" ? null : "process_lost", + error: input.runStatus === "succeeded" ? null : "run failed before issue advanced", + }); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Recover stranded assigned work", + status: input.status, + priority: "medium", + assigneeAgentId: input.assignToUser ? null : agentId, + assigneeUserId: input.assignToUser ? "user-1" : null, + checkoutRunId: input.status === "in_progress" ? runId : null, + executionRunId: null, + issueNumber: 1, + identifier: `${issuePrefix}-1`, + startedAt: input.status === "in_progress" ? now : null, + }); + + return { companyId, agentId, runId, wakeupRequestId, issueId }; + } + it("keeps a local run active when the recorded pid is still alive", async () => { const child = spawnAliveProcess(); childProcesses.add(child); @@ -402,4 +531,120 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { agentRole: "engineer", }); }); + + it("re-enqueues assigned todo work when the last issue run died and no wake remains", async () => { + const { agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "todo", + runStatus: "failed", + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.dispatchRequeued).toBe(1); + expect(result.continuationRequeued).toBe(0); + expect(result.escalated).toBe(0); + expect(result.issueIds).toEqual([issueId]); + + const runs = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(2); + + const retryRun = runs.find((row) => row.id !== runId); + expect(retryRun?.id).toBeTruthy(); + expect((retryRun?.contextSnapshot as Record)?.retryReason).toBe("assignment_recovery"); + if (retryRun) { + await waitForRunToSettle(heartbeat, retryRun.id); + } + }); + + it("blocks assigned todo work after the one automatic dispatch recovery was already used", async () => { + const { issueId } = await seedStrandedIssueFixture({ + status: "todo", + runStatus: "failed", + retryReason: "assignment_recovery", + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.dispatchRequeued).toBe(0); + expect(result.escalated).toBe(1); + expect(result.issueIds).toEqual([issueId]); + + const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); + expect(issue?.status).toBe("blocked"); + + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); + expect(comments).toHaveLength(1); + expect(comments[0]?.body).toContain("retried dispatch"); + }); + + it("re-enqueues continuation for stranded in-progress work with no active run", async () => { + const { agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "failed", + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.dispatchRequeued).toBe(0); + expect(result.continuationRequeued).toBe(1); + expect(result.escalated).toBe(0); + expect(result.issueIds).toEqual([issueId]); + + const runs = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(2); + + const retryRun = runs.find((row) => row.id !== runId); + expect(retryRun?.id).toBeTruthy(); + expect((retryRun?.contextSnapshot as Record)?.retryReason).toBe("issue_continuation_needed"); + if (retryRun) { + await waitForRunToSettle(heartbeat, retryRun.id); + } + }); + + it("blocks stranded in-progress work after the continuation retry was already used", async () => { + const { issueId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "failed", + retryReason: "issue_continuation_needed", + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.continuationRequeued).toBe(0); + expect(result.escalated).toBe(1); + expect(result.issueIds).toEqual([issueId]); + + const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); + expect(issue?.status).toBe("blocked"); + + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); + expect(comments).toHaveLength(1); + expect(comments[0]?.body).toContain("retried continuation"); + }); + + it("does not reconcile user-assigned work through the agent stranded-work recovery path", async () => { + const { issueId, runId } = await seedStrandedIssueFixture({ + status: "todo", + runStatus: "failed", + assignToUser: true, + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.dispatchRequeued).toBe(0); + expect(result.continuationRequeued).toBe(0); + expect(result.escalated).toBe(0); + + const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); + expect(issue?.status).toBe("todo"); + + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(runs).toHaveLength(1); + }); }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 056e812af3b..f8589544e53 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -119,6 +119,13 @@ vi.mock("../services/index.js", () => ({ heartbeatService: vi.fn(() => ({ reapOrphanedRuns: vi.fn(async () => undefined), resumeQueuedRuns: vi.fn(async () => undefined), + reconcileStrandedAssignedIssues: vi.fn(async () => ({ + dispatchRequeued: 0, + continuationRequeued: 0, + escalated: 0, + skipped: 0, + issueIds: [], + })), tickTimers: vi.fn(async () => ({ enqueued: 0 })), })), reconcilePersistedRuntimeServicesOnStartup: vi.fn(async () => ({ reconciled: 0 })), diff --git a/server/src/index.ts b/server/src/index.ts index 0e8db5a6291..74199e36cc4 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -583,6 +583,16 @@ export async function startServer(): Promise { void heartbeat .reapOrphanedRuns() .then(() => heartbeat.resumeQueuedRuns()) + .then(async () => { + const reconciled = await heartbeat.reconcileStrandedAssignedIssues(); + if ( + reconciled.dispatchRequeued > 0 || + reconciled.continuationRequeued > 0 || + reconciled.escalated > 0 + ) { + logger.warn({ ...reconciled }, "startup stranded-issue reconciliation changed assigned issue state"); + } + }) .catch((err) => { logger.error({ err }, "startup heartbeat recovery failed"); }); @@ -614,6 +624,16 @@ export async function startServer(): Promise { void heartbeat .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000 }) .then(() => heartbeat.resumeQueuedRuns()) + .then(async () => { + const reconciled = await heartbeat.reconcileStrandedAssignedIssues(); + if ( + reconciled.dispatchRequeued > 0 || + reconciled.continuationRequeued > 0 || + reconciled.escalated > 0 + ) { + logger.warn({ ...reconciled }, "periodic stranded-issue reconciliation changed assigned issue state"); + } + }) .catch((err) => { logger.error({ err }, "periodic heartbeat recovery failed"); }); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index f2ce72b7c0d..af0798b8534 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -89,6 +89,7 @@ const MAX_INLINE_WAKE_COMMENTS = 8; const MAX_INLINE_WAKE_COMMENT_BODY_CHARS = 4_000; const MAX_INLINE_WAKE_COMMENT_BODY_TOTAL_CHARS = 12_000; const execFile = promisify(execFileCallback); +const ACTIVE_HEARTBEAT_RUN_STATUSES = ["queued", "running"] as const; const SESSIONED_LOCAL_ADAPTERS = new Set([ "claude_local", "codex_local", @@ -2658,6 +2659,256 @@ export function heartbeatService(db: Db) { } } + async function getLatestIssueRun(companyId: string, issueId: string) { + return db + .select() + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + ), + ) + .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)) + .limit(1) + .then((rows) => rows[0] ?? null); + } + + async function hasActiveExecutionPath(companyId: string, issueId: string) { + const [run, deferredWake] = await Promise.all([ + db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + inArray(heartbeatRuns.status, [...ACTIVE_HEARTBEAT_RUN_STATUSES]), + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + db + .select({ id: agentWakeupRequests.id }) + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), + sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + ]); + + return Boolean(run || deferredWake); + } + + async function enqueueStrandedIssueRecovery(input: { + issueId: string; + agentId: string; + reason: "issue_assignment_recovery" | "issue_continuation_needed"; + retryReason: "assignment_recovery" | "issue_continuation_needed"; + source: string; + retryOfRunId?: string | null; + }) { + const queued = await enqueueWakeup(input.agentId, { + source: "automation", + triggerDetail: "system", + reason: input.reason, + payload: { + issueId: input.issueId, + ...(input.retryOfRunId ? { retryOfRunId: input.retryOfRunId } : {}), + }, + requestedByActorType: "system", + requestedByActorId: null, + contextSnapshot: { + issueId: input.issueId, + taskId: input.issueId, + wakeReason: input.reason, + retryReason: input.retryReason, + source: input.source, + ...(input.retryOfRunId ? { retryOfRunId: input.retryOfRunId } : {}), + }, + }); + + if (queued && input.retryOfRunId) { + return db + .update(heartbeatRuns) + .set({ + retryOfRunId: input.retryOfRunId, + updatedAt: new Date(), + }) + .where(eq(heartbeatRuns.id, queued.id)) + .returning() + .then((rows) => rows[0] ?? queued); + } + + return queued; + } + + async function escalateStrandedAssignedIssue(input: { + issue: typeof issues.$inferSelect; + previousStatus: "todo" | "in_progress"; + latestRun: typeof heartbeatRuns.$inferSelect | null; + comment: string; + }) { + const updated = await issuesSvc.update(input.issue.id, { + status: "blocked", + }); + if (!updated) return null; + + await issuesSvc.addComment(input.issue.id, input.comment, {}); + + await logActivity(db, { + companyId: input.issue.companyId, + actorType: "system", + actorId: "system", + agentId: null, + runId: null, + action: "issue.updated", + entityType: "issue", + entityId: input.issue.id, + details: { + identifier: input.issue.identifier, + status: "blocked", + previousStatus: input.previousStatus, + source: "heartbeat.reconcile_stranded_assigned_issue", + latestRunId: input.latestRun?.id ?? null, + latestRunStatus: input.latestRun?.status ?? null, + latestRunErrorCode: input.latestRun?.errorCode ?? null, + }, + }); + + return updated; + } + + async function reconcileStrandedAssignedIssues() { + const candidates = await db + .select() + .from(issues) + .where( + and( + isNull(issues.assigneeUserId), + inArray(issues.status, ["todo", "in_progress"]), + sql`${issues.assigneeAgentId} is not null`, + ), + ); + + const result = { + dispatchRequeued: 0, + continuationRequeued: 0, + escalated: 0, + skipped: 0, + issueIds: [] as string[], + }; + + for (const issue of candidates) { + const agentId = issue.assigneeAgentId; + if (!agentId) { + result.skipped += 1; + continue; + } + + const agent = await getAgent(agentId); + if (!agent || agent.companyId !== issue.companyId) { + result.skipped += 1; + continue; + } + if (agent.status === "paused" || agent.status === "terminated" || agent.status === "pending_approval") { + result.skipped += 1; + continue; + } + + if (await hasActiveExecutionPath(issue.companyId, issue.id)) { + result.skipped += 1; + continue; + } + + const latestRun = await getLatestIssueRun(issue.companyId, issue.id); + const latestContext = parseObject(latestRun?.contextSnapshot); + const latestRetryReason = readNonEmptyString(latestContext.retryReason); + + if (issue.status === "todo") { + if (!latestRun || latestRun.status === "succeeded") { + result.skipped += 1; + continue; + } + + if (latestRetryReason === "assignment_recovery") { + const updated = await escalateStrandedAssignedIssue({ + issue, + previousStatus: "todo", + latestRun, + comment: + "Paperclip automatically retried dispatch for this assigned `todo` issue after a lost wake/run, " + + "but it still has no live execution path. Moving it to `blocked` so it is visible for intervention.", + }); + if (updated) { + result.escalated += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + + const queued = await enqueueStrandedIssueRecovery({ + issueId: issue.id, + agentId, + reason: "issue_assignment_recovery", + retryReason: "assignment_recovery", + source: "issue.assignment_recovery", + retryOfRunId: latestRun.id, + }); + if (queued) { + result.dispatchRequeued += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + + if (latestRetryReason === "issue_continuation_needed") { + const updated = await escalateStrandedAssignedIssue({ + issue, + previousStatus: "in_progress", + latestRun, + comment: + "Paperclip automatically retried continuation for this assigned `in_progress` issue after its live " + + "execution disappeared, but it still has no live execution path. Moving it to `blocked` so it is " + + "visible for intervention.", + }); + if (updated) { + result.escalated += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + + const queued = await enqueueStrandedIssueRecovery({ + issueId: issue.id, + agentId, + reason: "issue_continuation_needed", + retryReason: "issue_continuation_needed", + source: "issue.continuation_recovery", + retryOfRunId: latestRun?.id ?? issue.checkoutRunId ?? null, + }); + if (queued) { + result.continuationRequeued += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + } + + return result; + } + async function updateRuntimeState( agent: typeof agents.$inferSelect, run: typeof heartbeatRuns.$inferSelect, @@ -4729,6 +4980,8 @@ export function heartbeatService(db: Db) { resumeQueuedRuns, + reconcileStrandedAssignedIssues, + tickTimers: async (now = new Date()) => { const allAgents = await db.select().from(agents); let checked = 0;