diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 50a386a359d..5084351556e 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1468,7 +1468,11 @@ 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 +owned by `native_workspace_finalizer` whose actual callback is still executing +in this server process. A persisted running row alone is not activity evidence: a +crash may leave it behind. The in-process claim ends when the callback joins and +is empty after restart, with no expiry that can misclassify a slow export. This +read-only presentation check never grants takeover authority. It 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`. diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index c7baf1775f5..357c3cf38f3 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -1,4 +1,7 @@ import { randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; import express from "express"; import request from "supertest"; import { and, eq } from "drizzle-orm"; @@ -30,6 +33,7 @@ import { errorHandler } from "../middleware/index.js"; import { issueRoutes } from "../routes/issues.js"; import { buildPaperclipWakePayload, heartbeatService } from "../services/heartbeat.js"; import { deliverReconciledExecutions } from "../services/execution-recovery-resolution.js"; +import { workspaceOperationService, isNativeWorkspaceFinalizationOperationActive } from "../services/workspace-operations.js"; import { issueRecoveryActionService } from "../services/issue-recovery-actions.js"; import { recoveryService } from "../services/recovery/service.js"; import { noticeMetadataReferencesRecoveryAction } from "../services/recovery/successful-run-handoff.js"; @@ -133,8 +137,12 @@ if (!embeddedPostgresSupport.supported) { describeEmbeddedPostgres("issue recovery actions", () => { let tempDb: Awaited> | null = null; let db: ReturnType; + let operationLogRoot: string; + const priorOperationLogRoot = process.env.WORKSPACE_OPERATION_LOG_BASE_PATH; beforeAll(async () => { + operationLogRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-recovery-operation-logs-")); + process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = operationLogRoot; tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-recovery-actions-"); db = createDb(tempDb.connectionString); }, 30_000); @@ -159,6 +167,9 @@ describeEmbeddedPostgres("issue recovery actions", () => { afterAll(async () => { await tempDb?.cleanup(); + if (priorOperationLogRoot === undefined) delete process.env.WORKSPACE_OPERATION_LOG_BASE_PATH; + else process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = priorOperationLogRoot; + await fs.rm(operationLogRoot, { recursive: true, force: true }); }); async function seedCompany() { @@ -287,17 +298,63 @@ describeEmbeddedPostgres("issue recovery actions", () => { expect(response.body.activeRecoveryAction.nativeRunActivity).toEqual(expected); }); - it("projects running export work while its original heartbeat remains failed, then clears it", async () => { + it("projects an actual long-running export 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({ + let release!: () => void; + let started!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + const entered = new Promise((resolve) => { started = resolve; }); + const work = workspaceOperationService(db).createRecorder({ companyId, issueId: sourceIssueId, heartbeatRunId: runId }).recordOperation({ + phase: "workspace_finalize", metadata: { owningService: "native_workspace_finalizer" }, + run: async () => { started(); await held; return { status: "succeeded" }; }, + }); + await entered; + let operationId: string; + try { + const [operation] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.heartbeatRunId, runId)); + operationId = operation!.id; + // A slow callback remains live independently of elapsed wall-clock time. + await db.update(workspaceOperations).set({ startedAt: new Date("2000-01-01T00:00:00Z") }).where(eq(workspaceOperations.id, operationId)); + const scope = { companyId, issueId: sourceIssueId, runId, operationId }; + expect(isNativeWorkspaceFinalizationOperationActive(scope)).toBe(true); + expect(isNativeWorkspaceFinalizationOperationActive({ ...scope, companyId: randomUUID() })).toBe(false); + expect((await svc.getActiveForIssue(companyId, sourceIssueId))?.nativeRunActivity).toEqual({ + runId, status: "running", workspaceOperationId: operationId, + }); + } finally { release(); await work; } + expect((await svc.getActiveForIssue(companyId, sourceIssueId))?.nativeRunActivity).toBeNull(); + // Even a stranded DB row cannot turn a joined callback back into activity. + await db.update(workspaceOperations).set({ status: "running", finishedAt: null }).where(eq(workspaceOperations.id, operationId!)); + expect((await svc.getActiveForIssue(companyId, sourceIssueId))?.nativeRunActivity).toBeNull(); + }); + + it("keeps the activity claim until a timed-out export callback actually joins", async () => { + const { companyId, sourceIssueId, runId } = await seedNativeFinalizationRecovery("failed"); + let release!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + const work = workspaceOperationService(db).createRecorder({ companyId, issueId: sourceIssueId, heartbeatRunId: runId }).recordOperation({ + phase: "workspace_finalize", metadata: { owningService: "native_workspace_finalizer" }, timeoutMs: 10, + run: async () => { await held; return { status: "succeeded" }; }, + }); + try { + await expect(work).rejects.toThrow("exceeded"); + const [operation] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.heartbeatRunId, runId)); + const scope = { companyId, issueId: sourceIssueId, runId, operationId: operation!.id }; + expect(isNativeWorkspaceFinalizationOperationActive(scope)).toBe(true); + release(); + await held; + await new Promise((resolve) => setImmediate(resolve)); + expect(isNativeWorkspaceFinalizationOperationActive(scope)).toBe(false); + } finally { release(); } + }); + + it("does not treat an orphaned running export record as live after controller loss", async () => { + const { companyId, sourceIssueId, runId, svc } = await seedNativeFinalizationRecovery("failed"); + 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(); }); diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts index 4dcb831ecae..aeb7384bee3 100644 --- a/server/src/services/issue-recovery-actions.ts +++ b/server/src/services/issue-recovery-actions.ts @@ -10,6 +10,8 @@ import type { IssueRecoveryActionStatus, } from "@paperclipai/shared"; +import { isNativeWorkspaceFinalizationOperationActive } from "./workspace-operations.js"; + const ACTIVE_RECOVERY_ACTION_STATUSES = ["active", "escalated"] as const satisfies readonly IssueRecoveryActionStatus[]; const MAX_UPSERT_RETRIES = 3; @@ -169,6 +171,7 @@ export function issueRecoveryActionService(db: Db) { runId: heartbeatRuns.id, issueId: heartbeatRuns.nativeIssueId, status: heartbeatRuns.status, + finishedAt: heartbeatRuns.finishedAt, workspaceOperationId: workspaceOperations.id, }).from(heartbeatRuns).leftJoin(workspaceOperations, and( eq(workspaceOperations.companyId, heartbeatRuns.companyId), @@ -189,7 +192,16 @@ export function issueRecoveryActionService(db: Db) { )).orderBy(desc(workspaceOperations.startedAt)); const activityByRun = new Map(); for (const row of activity) { - if (!activityByRun.has(row.runId)) activityByRun.set(row.runId, row); + const liveOperation = row.workspaceOperationId && row.issueId + && isNativeWorkspaceFinalizationOperationActive({ + operationId: row.workspaceOperationId, companyId, runId: row.runId, issueId: row.issueId, + }); + const liveHeartbeat = ["queued", "running"].includes(row.status) && row.finishedAt === null; + if (!liveHeartbeat && !liveOperation) continue; + const projected = { ...row, workspaceOperationId: liveOperation ? row.workspaceOperationId : null }; + if (!activityByRun.has(row.runId) || (!activityByRun.get(row.runId)?.workspaceOperationId && liveOperation)) { + activityByRun.set(row.runId, projected); + } } for (const action of nativeActions) { const row = activityByRun.get(String(action.wakePolicy?.runId).toLowerCase()); diff --git a/server/src/services/workspace-operations.ts b/server/src/services/workspace-operations.ts index 773744a40d7..233773bb110 100644 --- a/server/src/services/workspace-operations.ts +++ b/server/src/services/workspace-operations.ts @@ -96,6 +96,22 @@ function readRuntimeControlOwner(metadata: unknown): RuntimeControlOwnerStamp | */ const liveRuntimeControlOperationIds = new Set(); +type NativeFinalizationActivityScope = { + operationId: string; + companyId: string; + runId: string; + issueId: string; +}; +// Presentation-only positive proof. Persisted running rows can outlive a crash; +// this registry only exists while this process awaits the actual export callback. +const liveNativeFinalizationOperations = new Map(); + +export function isNativeWorkspaceFinalizationOperationActive(scope: NativeFinalizationActivityScope): boolean { + const active = liveNativeFinalizationOperations.get(scope.operationId); + return Boolean(active && active.companyId === scope.companyId + && active.runId === scope.runId && active.issueId === scope.issueId); +} + /** * Whether the process that stamped an operation is still alive on this host. A dead owner can * never heartbeat again, so its operation is recoverable immediately instead of after the @@ -579,8 +595,24 @@ export function workspaceOperationService(db: Db) { }; const timeoutMs = recordInput.timeoutMs ?? defaultRuntimeControlTimeoutMs(runtimeControlAction); + const runWithActivity = async () => { + const nativeFinalization = recordInput.phase === "workspace_finalize" + && recordInput.metadata?.owningService === "native_workspace_finalizer" + && input.heartbeatRunId && input.issueId; + if (nativeFinalization) liveNativeFinalizationOperations.set(id, { + operationId: id, companyId: input.companyId, + runId: input.heartbeatRunId!, issueId: input.issueId!, + }); + try { + return await recordInput.run(reportProgress); + } finally { + // Join the underlying callback, even if the recorder's timeout + // already fired. A deadline cannot prove physical export stopped. + liveNativeFinalizationOperations.delete(id); + } + }; const settle = async () => { - if (!timeoutMs || timeoutMs <= 0) return await recordInput.run(reportProgress); + if (!timeoutMs || timeoutMs <= 0) return await runWithActivity(); const timeout = new Promise((_resolve, reject) => { timeoutTimer = setTimeout( () => reject(new WorkspaceOperationTimeoutError(timeoutMs, runtimeControlAction)), @@ -588,7 +620,7 @@ export function workspaceOperationService(db: Db) { ); timeoutTimer.unref?.(); }); - return await Promise.race([recordInput.run(reportProgress), timeout]); + return await Promise.race([runWithActivity(), timeout]); }; try {