diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 48cf7fc3461..19e215df864 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1464,3 +1464,16 @@ that an agent turn is running. Task and inbox surfaces show recovery in progress only while its recorded retry is still due or its matching retry run is verified live. An expired or missing retry, exhausted budget, or board-owned failure shows recovery needed rather than “Observing active run.” + +Native recovery reads include a read-only `nativeRunActivity` projection, bound +to the company, source issue, and exact `resume_native_run.runId`. It reports a +queued/running native heartbeat, or a running `workspace_finalize` operation +owned by `native_workspace_finalizer`. The latter remains authoritative while +the original heartbeat still records its previous failure. Completed operations, +unrelated runs, and other companies do not establish activity. These retries do +not require or synthesize a legacy `scheduledRetryReason`. + +The card describes recovery of the existing run. It does not describe a fresh +owner disposition turn. This also covers native bootstrap/session recovery, +which shares the same resume policy. Inbox rows, source cards, and blocker chips +use the same activity projection and keep board-owned repairs actionable. diff --git a/packages/shared/src/types/issue.ts b/packages/shared/src/types/issue.ts index fbccc1aa554..197c3c883ac 100644 --- a/packages/shared/src/types/issue.ts +++ b/packages/shared/src/types/issue.ts @@ -538,6 +538,12 @@ export interface IssueUnblockDescriptor { } export interface IssueRecoveryAction { + /** Read-only activity of the exact native run named by the wake policy. */ + nativeRunActivity?: { + runId: string; + status: "queued" | "running"; + workspaceOperationId: string | null; + } | null; id: string; companyId: string; sourceIssueId: string; diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index 1160932558b..c7baf1775f5 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -19,6 +19,7 @@ import { issueRecoveryActions, issueRelations, issueThreadInteractions, + workspaceOperations, issues, } from "@paperclipai/db"; import { @@ -139,6 +140,7 @@ describeEmbeddedPostgres("issue recovery actions", () => { }, 30_000); afterEach(async () => { + await db.delete(workspaceOperations); await db.delete(issueThreadInteractions); await db.delete(issueRecoveryActions); await db.delete(issueComments); @@ -243,6 +245,91 @@ describeEmbeddedPostgres("issue recovery actions", () => { return app; } + async function seedNativeFinalizationRecovery(status: string) { + const fixture = await seedCompany(); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId: fixture.companyId, + agentId: fixture.coderId, + status, + runtimeMode: "native", + nativeIssueId: fixture.sourceIssueId, + nativePhase: "finalizing", + // Native coordinator retries resume this original run, without creating + // the legacy scheduled_retry_reason projection. + scheduledRetryReason: null, + contextSnapshot: { issueId: fixture.sourceIssueId }, + }); + const svc = issueRecoveryActionService(db); + await svc.upsertSourceScoped({ + companyId: fixture.companyId, + sourceIssueId: fixture.sourceIssueId, + kind: "active_run_watchdog", + ownerType: "agent", + ownerAgentId: fixture.coderId, + cause: "native_finalization_invalid", + fingerprint: `native:${runId}`, + nextAction: "Finish the existing run.", + wakePolicy: { kind: "resume_native_run", runId, notBefore: "2020-01-01T00:00:00Z" }, + maxAttempts: 3, + }); + return { ...fixture, runId, svc }; + } + + it.each(["queued", "running"])("projects the exact %s native run without a legacy scheduled retry", async (status) => { + const { companyId, sourceIssueId, runId, svc } = await seedNativeFinalizationRecovery(status); + const expected = { runId, status, workspaceOperationId: null }; + expect((await svc.getActiveForIssue(companyId, sourceIssueId))?.nativeRunActivity).toEqual(expected); + expect((await svc.listActiveForIssues(companyId, [sourceIssueId])).get(sourceIssueId)?.nativeRunActivity).toEqual(expected); + const response = await request(createApp()).get(`/api/issues/${sourceIssueId}`).expect(200); + expect(response.body.scheduledRetry).toBeNull(); + expect(response.body.activeRecoveryAction.nativeRunActivity).toEqual(expected); + }); + + it("projects running export work while its original heartbeat remains failed, then clears it", async () => { + const { companyId, sourceIssueId, runId, svc } = await seedNativeFinalizationRecovery("failed"); + const [operation] = await db.insert(workspaceOperations).values({ + companyId, issueId: sourceIssueId, heartbeatRunId: runId, + phase: "workspace_finalize", status: "running", + metadata: { owningService: "native_workspace_finalizer" }, + }).returning(); + expect((await svc.getActiveForIssue(companyId, sourceIssueId))?.nativeRunActivity).toEqual({ + runId, status: "running", workspaceOperationId: operation!.id, + }); + await db.update(workspaceOperations).set({ status: "succeeded", finishedAt: new Date() }).where(eq(workspaceOperations.id, operation!.id)); + expect((await svc.getActiveForIssue(companyId, sourceIssueId))?.nativeRunActivity).toBeNull(); + }); + + it.each(["company", "issue", "legacy", "finished"])("rejects an apparently live run with the wrong %s authority", async (mismatch) => { + const { companyId, sourceIssueId, runId, svc } = await seedNativeFinalizationRecovery("running"); + const other = await seedCompany(); + await db.update(heartbeatRuns).set({ + ...(mismatch === "company" ? { companyId: other.companyId, agentId: other.coderId } : {}), + ...(mismatch === "issue" ? { nativeIssueId: other.sourceIssueId } : {}), + ...(mismatch === "legacy" ? { runtimeMode: "legacy" } : {}), + ...(mismatch === "finished" ? { finishedAt: new Date() } : {}), + }).where(eq(heartbeatRuns.id, runId)); + expect((await svc.getActiveForIssue(companyId, sourceIssueId))?.nativeRunActivity).toBeNull(); + }); + + it.each(["company", "issue", "run", "phase", "owner", "finished"])( + "does not borrow native finalization activity with a different %s binding", async (mismatch) => { + const { companyId, sourceIssueId, runId, svc } = await seedNativeFinalizationRecovery("failed"); + const other = await seedNativeFinalizationRecovery("running"); + await db.insert(workspaceOperations).values({ + companyId: mismatch === "company" ? other.companyId : companyId, + issueId: mismatch === "issue" ? other.sourceIssueId : sourceIssueId, + heartbeatRunId: mismatch === "run" ? other.runId : runId, + phase: mismatch === "phase" ? "workspace_prepare" : "workspace_finalize", + status: "running", finishedAt: mismatch === "finished" ? new Date() : null, + metadata: { owningService: mismatch === "owner" ? "different_service" : "native_workspace_finalizer" }, + }); + expect((await svc.getActiveForIssue(companyId, sourceIssueId))?.nativeRunActivity).toBeNull(); + expect((await svc.listActiveForIssues(companyId, [sourceIssueId, other.sourceIssueId])).get(sourceIssueId)?.nativeRunActivity).toBeNull(); + }, + ); + it("upserts one active source-scoped action per issue and keeps company scoping explicit", async () => { const { companyId, managerId, sourceIssueId } = await seedCompany(); const svc = issueRecoveryActionService(db); diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts index 51d5df49adf..4dcb831ecae 100644 --- a/server/src/services/issue-recovery-actions.ts +++ b/server/src/services/issue-recovery-actions.ts @@ -1,6 +1,7 @@ -import { and, desc, eq, inArray } from "drizzle-orm"; +import { and, desc, eq, inArray, isNull, isNotNull, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; -import { issueRecoveryActions } from "@paperclipai/db"; +import { heartbeatRuns, issueRecoveryActions, workspaceOperations } from "@paperclipai/db"; +import { isUuidLike } from "@paperclipai/shared"; import type { IssueRecoveryAction, IssueRecoveryActionKind, @@ -149,6 +150,58 @@ export function issueRecoveryActionService(db: Db) { } } + // Coordinator retries reuse the original native run. They are not legacy + // scheduled retries, and workspace export can run while the heartbeat still + // records its earlier failure. Project their actual activity for every reader. + async function projectNativeRunActivity( + companyId: string, + actions: IssueRecoveryAction[], + dbOrTx: DbOrTransaction, + ) { + const nativeActions = actions.filter((action) => action.wakePolicy?.kind === "resume_native_run"); + for (const action of nativeActions) action.nativeRunActivity = null; + const runIds = [...new Set(nativeActions.flatMap((action) => { + const runId = action.wakePolicy?.runId; + return typeof runId === "string" && isUuidLike(runId) ? [runId] : []; + }))]; + if (runIds.length === 0) return; + const activity = await dbOrTx.select({ + runId: heartbeatRuns.id, + issueId: heartbeatRuns.nativeIssueId, + status: heartbeatRuns.status, + workspaceOperationId: workspaceOperations.id, + }).from(heartbeatRuns).leftJoin(workspaceOperations, and( + eq(workspaceOperations.companyId, heartbeatRuns.companyId), + eq(workspaceOperations.heartbeatRunId, heartbeatRuns.id), + eq(workspaceOperations.issueId, heartbeatRuns.nativeIssueId), + eq(workspaceOperations.phase, "workspace_finalize"), + eq(workspaceOperations.status, "running"), + isNull(workspaceOperations.finishedAt), + sql`${workspaceOperations.metadata}->>'owningService' = 'native_workspace_finalizer'`, + )).where(and( + eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.runtimeMode, "native"), + inArray(heartbeatRuns.id, runIds), + or( + and(inArray(heartbeatRuns.status, ["queued", "running"]), isNull(heartbeatRuns.finishedAt)), + isNotNull(workspaceOperations.id), + ), + )).orderBy(desc(workspaceOperations.startedAt)); + const activityByRun = new Map(); + for (const row of activity) { + if (!activityByRun.has(row.runId)) activityByRun.set(row.runId, row); + } + for (const action of nativeActions) { + const row = activityByRun.get(String(action.wakePolicy?.runId).toLowerCase()); + if (!row || row.issueId !== action.sourceIssueId) continue; + action.nativeRunActivity = { + runId: row.runId, + status: row.workspaceOperationId || row.status === "running" ? "running" : "queued", + workspaceOperationId: row.workspaceOperationId, + }; + } + } + async function getActiveForIssue( companyId: string, sourceIssueId: string, @@ -167,7 +220,10 @@ export function issueRecoveryActionService(db: Db) { .orderBy(desc(issueRecoveryActions.updatedAt)) .limit(1) .then((rows) => rows[0] ?? null); - return row ? toReadModel(row) : null; + if (!row) return null; + const action = toReadModel(row); + await projectNativeRunActivity(companyId, [action], dbOrTx); + return action; } async function listActiveForIssues(companyId: string, sourceIssueIds: string[]) { @@ -187,6 +243,7 @@ export function issueRecoveryActionService(db: Db) { for (const row of rows) { if (!result.has(row.sourceIssueId)) result.set(row.sourceIssueId, toReadModel(row)); } + await projectNativeRunActivity(companyId, [...result.values()], db); return result; } diff --git a/ui/src/components/IssueBlockedNotice.test.tsx b/ui/src/components/IssueBlockedNotice.test.tsx index 5bfa04246ae..8651ff7ac05 100644 --- a/ui/src/components/IssueBlockedNotice.test.tsx +++ b/ui/src/components/IssueBlockedNotice.test.tsx @@ -844,6 +844,17 @@ describe("IssueBlockedNotice", () => { ).querySelector('[data-testid="issue-blocked-notice-recovery-indicator"]'); } + it("uses native export activity on a blocker without a scheduled retry", () => { + const action = buildDispositionRepairAction({ + kind: "resume_native_run", runId: "native-run", notBefore: "2020-01-01T00:00:00Z", + }, { + kind: "active_run_watchdog", cause: "native_finalization_invalid", + nativeRunActivity: { runId: "native-run", status: "running", workspaceOperationId: "export-operation" }, + }); + expect(renderBlockerChip(action)?.getAttribute("data-recovery-state")).toBe("in_progress"); + expect(deriveRecoveryCardState(action)).toBe("in_progress"); + }); + it("reports the same liveness state as the source task's recovery card", () => { const liveAction = buildDispositionRepairAction({ type: "bounded_owner_disposition_repair", diff --git a/ui/src/components/IssueRecoveryActionCard.test.tsx b/ui/src/components/IssueRecoveryActionCard.test.tsx index d8480fb51c5..077b1707b8e 100644 --- a/ui/src/components/IssueRecoveryActionCard.test.tsx +++ b/ui/src/components/IssueRecoveryActionCard.test.tsx @@ -134,6 +134,22 @@ describe("deriveRecoveryCardState", () => { }); describe("IssueRecoveryActionCard", () => { + it.each(["running", "scheduled", "missed"] as const)("describes native %s finalization without claiming another agent turn", (state) => { + const node = render(); + expect(node.querySelector("section")?.getAttribute("data-recovery-state")).toBe(state === "missed" ? "needed" : "in_progress"); + expect(node.textContent).toContain("existing run"); + expect(node.textContent).not.toContain("retrying the original owner"); + expect(node.textContent).not.toContain("retrying itself"); + expect(node.textContent).not.toContain("repairs the next step only"); + expect(node.textContent).not.toContain("stopped to wait"); + expect(node.querySelector('[data-testid="recovery-retry-progress"]')?.getAttribute("data-recovery-lane")).toBe("native_run"); + }); + it("renders state and kind attributes with owner names and the recorded next action", () => { const node = render( - {recoveryOwnerIsSourceOwner ? ( + {lineage.lane === "native_run" && action.ownerType !== "board" ? ( + <> + Paperclip + recovers the existing run + + ) : recoveryOwnerIsSourceOwner ? ( Original owner — retrying itself ) : action.ownerType === "agent" && action.ownerAgentId ? ( <> diff --git a/ui/src/components/IssueRow.test.tsx b/ui/src/components/IssueRow.test.tsx index be507931c5d..188ece5f703 100644 --- a/ui/src/components/IssueRow.test.tsx +++ b/ui/src/components/IssueRow.test.tsx @@ -771,6 +771,18 @@ describe("IssueRow", () => { return snapshot; } + it("shows native export activity from the action even without a scheduled retry", () => { + const issue = recoveryIssue(at(-5 * 60_000)); + Object.assign(issue.activeRecoveryAction!, { + kind: "active_run_watchdog", cause: "native_finalization_invalid", + wakePolicy: { kind: "resume_native_run", runId: "native-run", notBefore: at(-5 * 60_000) }, + nativeRunActivity: { runId: "native-run", status: "running", workspaceOperationId: "export-operation" }, + }); + const chip = renderChip(issue); + expect(chip?.getAttribute("data-recovery-state")).toBe("in_progress"); + expect(chip?.getAttribute("aria-label")).not.toContain("missed"); + }); + it("stays calm while the next attempt is still ahead", () => { const chip = renderChip(recoveryIssue(at(3 * 60_000))); expect(chip?.getAttribute("data-recovery-state")).toBe("in_progress"); diff --git a/ui/src/lib/recovery-display.test.ts b/ui/src/lib/recovery-display.test.ts index 83890000067..5c71fed94e2 100644 --- a/ui/src/lib/recovery-display.test.ts +++ b/ui/src/lib/recovery-display.test.ts @@ -147,12 +147,20 @@ describe("deriveRecoveryDisplayState", () => { }, { now })).toBe("in_progress"); }); - it("recognizes only the matching verified run for an overdue retry", () => { + it.each(["queued", "running"] as const)("uses the API's %s finalization projection without a scheduled retry", (status) => { + expect(deriveRecoveryDisplayState({ + ...action, + nativeRunActivity: { runId: "finalizing-run", status, workspaceOperationId: null }, + }, { now })).toBe("in_progress"); + expect(deriveRecoveryDisplayState({ + ...action, + nativeRunActivity: { runId: "different-run", status, workspaceOperationId: null }, + }, { now })).toBe("needed"); + }); + + it("does not borrow an unrelated legacy retry to claim native finalization activity", () => { expect(deriveRecoveryDisplayState(action, { now, scheduledRetry: { runId: "finalizing-run", status: "running" }, - })).toBe("in_progress"); - expect(deriveRecoveryDisplayState(action, { - now, scheduledRetry: { runId: "different-run", status: "running" }, })).toBe("needed"); }); diff --git a/ui/src/lib/recovery-display.ts b/ui/src/lib/recovery-display.ts index d66951029fc..dff2fc5d67b 100644 --- a/ui/src/lib/recovery-display.ts +++ b/ui/src/lib/recovery-display.ts @@ -62,6 +62,7 @@ export type RecoveryDisplayInput = Pick< | "attemptCount" | "maxAttempts" | "timeoutAt" + | "nativeRunActivity" > >; @@ -83,6 +84,7 @@ export function deriveRecoveryDisplayState( attemptCount: action.attemptCount, maxAttempts: action.maxAttempts, timeoutAt: action.timeoutAt, + nativeRunActivity: action.nativeRunActivity, }, context); if (action.kind === "active_run_watchdog") { // Native finalization shares the watchdog kind, but resumes a failed diff --git a/ui/src/lib/recovery-lineage.ts b/ui/src/lib/recovery-lineage.ts index 8cffd1fa895..c172ef4db08 100644 --- a/ui/src/lib/recovery-lineage.ts +++ b/ui/src/lib/recovery-lineage.ts @@ -10,7 +10,7 @@ import { formatMonitorOffset } from "./issue-monitor"; * therefore the canonical lane signal — every surface reads the same stored value rather * than re-deriving liveness on its own. */ -export type RecoveryRetryLane = "source_owner" | "recovery_owner" | "board"; +export type RecoveryRetryLane = "source_owner" | "recovery_owner" | "native_run" | "board"; export interface RecoveryRetryLineage { lane: RecoveryRetryLane; @@ -55,7 +55,7 @@ export interface RecoveryRetryLineage { const SOURCE_LANE_POLICY = "bounded_owner_disposition_repair"; const RECOVERY_LANE_POLICY = "bounded_recovery_owner"; const BOARD_LANE_POLICY = "board_escalation"; -const NATIVE_FINALIZATION_POLICY = "resume_native_run"; +const NATIVE_RESUME_POLICY = "resume_native_run"; /** Scheduled-run states that mean the parked attempt is actually in flight right now. */ const LIVE_SCHEDULED_RUN_STATUSES = new Set(["queued", "running"]); @@ -71,7 +71,7 @@ const LIVE_SCHEDULED_RUN_STATUSES = new Set(["queued", "running"]); const RETRY_EXPIRY_GRACE_MS = 30_000; export type RecoveryLineageInput = Pick & - Partial>; + Partial>; /** * The source issue's scheduled-retry record, which is what lets a surface tell a run that is @@ -144,14 +144,15 @@ export function readRecoveryRetryLineage( const policy = asRecord(action.wakePolicy); if (!policy) return null; const type = asNonEmptyString(policy.type); - const nativeFinalization = policy.kind === NATIVE_FINALIZATION_POLICY; + const nativeResume = policy.kind === NATIVE_RESUME_POLICY; const preservesSourceAssignee = policy.preservesSourceAssignee === true; const evidence = asRecord(action.evidence) ?? {}; const evidenceSourceAttempt = asCount(evidence.sourceAttemptCount); const evidenceSourceMaxAttempts = asCount(evidence.sourceMaxAttempts); let lane: RecoveryRetryLane; - if (type === SOURCE_LANE_POLICY || nativeFinalization) lane = "source_owner"; + if (nativeResume) lane = "native_run"; + else if (type === SOURCE_LANE_POLICY) lane = "source_owner"; else if (type === RECOVERY_LANE_POLICY) lane = "recovery_owner"; else if ( type === BOARD_LANE_POLICY && @@ -162,16 +163,18 @@ export function readRecoveryRetryLineage( const attempt = asCount(policy.attempt) ?? asCount(action.attemptCount) ?? 0; const maxAttempts = asCount(policy.maxAttempts) ?? asCount(action.maxAttempts); - const scheduledRunId = asNonEmptyString(nativeFinalization ? policy.runId : policy.scheduledRunId); - // Native finalization resumes an existing coordinator; its scheduling fields + const scheduledRunId = asNonEmptyString(nativeResume ? policy.runId : policy.scheduledRunId); + // Native recovery resumes an existing run; its scheduling fields // differ from an agent wake. A retry without that run identity is not a path. - if (nativeFinalization && scheduledRunId === null) return null; - const storedRetryAt = nativeFinalization + if (nativeResume && scheduledRunId === null) return null; + const storedRetryAt = nativeResume ? asIsoDate(policy.notBefore) : asIsoDate(policy.retryAt) ?? asIsoDate(action.timeoutAt); const exhausted = lane === "board" || (maxAttempts !== null && attempt >= maxAttempts); const nextRetryAt = exhausted ? null : storedRetryAt; - const liveRunId = resolveLiveRunId(scheduledRunId, context?.scheduledRetry); + const liveRunId = nativeResume + ? resolveLiveRunId(scheduledRunId, action.nativeRunActivity) + : resolveLiveRunId(scheduledRunId, context?.scheduledRetry); // A run that is already executing is the reason its due time is behind us, so a confirmed // live run is never "missed" — it is the attempt happening. const retryExpired = @@ -191,9 +194,9 @@ export function readRecoveryRetryLineage( scheduledRunId, liveRunId, retryAgentId: asNonEmptyString(policy.retryAgentId) ?? asNonEmptyString(policy.ownerAgentId), - preservesSourceAssignee: preservesSourceAssignee || lane === "source_owner", - sourceAttempt: lane === "source_owner" ? attempt : evidenceSourceAttempt, - sourceMaxAttempts: lane === "source_owner" ? maxAttempts : evidenceSourceMaxAttempts, + preservesSourceAssignee: preservesSourceAssignee || lane === "source_owner" || nativeResume, + sourceAttempt: nativeResume ? null : lane === "source_owner" ? attempt : evidenceSourceAttempt, + sourceMaxAttempts: nativeResume ? null : lane === "source_owner" ? maxAttempts : evidenceSourceMaxAttempts, hasDurablePath: !exhausted && (liveRunId !== null || (nextRetryAt !== null && !retryExpired)), };