fix(recovery): project exact native run activity across task views

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-09-27 21:38:40 -05:00
1 parent 34ebc558f5
commit 2321ddd573
11 files changed
+248 -21

No files matched your search

+13
View File
@@ -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.
+6
View File
@@ -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;
@@ -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);
+60 -3
View File
@@ -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<string, (typeof activity)[number]>();
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;
}
@@ -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",
@@ -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(<IssueRecoveryActionCard action={buildAction({
kind: "active_run_watchdog", cause: "native_finalization_invalid",
nextAction: "Finish the existing run.",
wakePolicy: { kind: "resume_native_run", runId: "native-run", notBefore: state === "scheduled" ? "2099-01-01T00:00:00Z" : "2020-01-01T00:00:00Z" },
nativeRunActivity: state === "running" ? { runId: "native-run", status: "running", workspaceOperationId: "export-operation" } : null,
})} />);
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(
<IssueRecoveryActionCard
+13 -1
View File
@@ -903,6 +903,13 @@ function formatTimeAbsolute(value: string | Date | null | undefined): string | n
* the deliverable.
*/
function lineageHeadline(lineage: RecoveryRetryLineage): string {
if (lineage.lane === "native_run") {
if (lineage.exhausted) return "Paperclip could not recover the existing run within its retry budget. Review the recorded failure before retrying. The task stays with its original owner.";
if (lineage.liveRunId) return "Paperclip is recovering the existing run. The task stays with its original owner.";
if (lineage.retryExpired) return "The retry for the existing run came due and did not start. Review the recorded failure and retry when ready. The task stays with its original owner.";
if (lineage.nextRetryAt) return "Paperclip has scheduled another attempt to resume the existing run. The task stays with its original owner.";
return "The existing run still needs recovery. Review the recorded failure before retrying. The task stays with its original owner.";
}
// An attempt that came due and never ran leaves nobody working on this task, even though
// attempts remain on paper. Say so before any lane wording that ends in "no action needed".
if (lineage.retryExpired) {
@@ -1177,7 +1184,12 @@ export function IssueRecoveryActionCard({
className="inline-flex flex-wrap items-center gap-1.5"
data-testid="recovery-recovery-owner"
>
{recoveryOwnerIsSourceOwner ? (
{lineage.lane === "native_run" && action.ownerType !== "board" ? (
<>
<span className="font-medium">Paperclip</span>
<span className="text-muted-foreground">recovers the existing run</span>
</>
) : recoveryOwnerIsSourceOwner ? (
<span className="font-medium">Original owner — retrying itself</span>
) : action.ownerType === "agent" && action.ownerAgentId ? (
<>
+12
View File
@@ -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");
+12 -4
View File
@@ -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");
});
+2
View File
@@ -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
+16 -13
View File
@@ -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<IssueRecoveryAction, "wakePolicy"> &
Partial<Pick<IssueRecoveryAction, "evidence" | "attemptCount" | "maxAttempts" | "timeoutAt">>;
Partial<Pick<IssueRecoveryAction, "evidence" | "attemptCount" | "maxAttempts" | "timeoutAt" | "nativeRunActivity">>;
/**
* 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)),
};