diff --git a/server/src/__tests__/native-run-projection-work.test.ts b/server/src/__tests__/native-run-projection-work.test.ts index 26f0a9d239a..f9718354a1c 100644 --- a/server/src/__tests__/native-run-projection-work.test.ts +++ b/server/src/__tests__/native-run-projection-work.test.ts @@ -1,3 +1,4 @@ +import * as wakeSettlementNotifications from "../services/external-conversation-settlement-notifications.js"; import { heartbeatService } from "../services/heartbeat.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { createHeartbeatLifecycle, type HeartbeatLifecycleDependencies } from "../services/heartbeat/run-lifecycle.js"; @@ -8,14 +9,14 @@ import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { setTimeout as delay } from "node:timers/promises"; -import { eq } from "drizzle-orm"; +import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { agents, companies, completionContracts, createDb, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { agentWakeupRequests, agents, companies, completionContracts, createDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { nativeRunProjectionWork } from "../services/native-runtime/native-run-projection-work.js"; import { projectCommittedRun } from "../services/native-runtime/native-run-projection.js"; import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js"; -import { finalizeNativeRun } from "../services/native-runtime/native-run-finalizer.js"; +import { finalizeNativeRun, recordNativeFinalizationFailure } from "../services/native-runtime/native-run-finalizer.js"; import * as statusCommitter from "../services/native-runtime/status-decision-committer.js"; import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js"; import { NativeRunCoordinatorStore } from "../services/native-runtime/native-run-coordinator-store.js"; @@ -120,6 +121,154 @@ describe("committed native run projection lifecycle", () => { async function accepted(status: "succeeded" | "failed" | "cancelled" = "succeeded") { const binding = await seed(); await complete(binding, "control_plane", db, status); await barrier(binding); return binding; } + async function claimedWake(binding: Awaited>) { + const [wake] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "issue", + status: "claimed", runId: binding.runId, claimedAt: new Date(), payload: { issueId: binding.issueId } }).returning(); + await db.update(heartbeatRuns).set({ wakeupRequestId: wake.id }).where(eq(heartbeatRuns.id, binding.runId)); + return wake; + } + const readWake = async (id: string) => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, id)))[0]; + + it.each(["succeeded", "failed", "cancelled"] as const)("settles the exact wake atomically with a fresh %s projection", async status => { + const binding = await accepted(status), wake = await claimedWake(binding); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + const run = await read(binding.runId), settled = await readWake(wake.id); + expect(run.status).toBe(status); + expect(settled).toMatchObject({ status: status === "succeeded" ? "completed" : status, runId: binding.runId, + companyId, agentId, finishedAt: run.finishedAt, error: run.error }); + expect(settled.finishedAt).not.toBeNull(); + await projectCommittedRun({ db, run, coordinator: await state(binding.runId) }); + expect(await readWake(wake.id)).toEqual(settled); + }); + + it.each(["provider", "workspace"] as const)("settles the exact wake when %s finalization retries exhaust", async scope => { + const binding = await accepted(), wake = await claimedWake(binding); + if (scope === "workspace") await db.update(workspaceOperations).set({ status: "failed" }) + .where(eq(workspaceOperations.heartbeatRunId, binding.runId)); + if (scope === "provider") await db.update(nativeRunResults).set({ + resultJson: sql`jsonb_set(${nativeRunResults.resultJson}, '{terminal,runTerminalState}', '"unknown"'::jsonb)`, + }).where(eq(nativeRunResults.runId, binding.runId)); + for (let attempt = 0; attempt < 3; attempt++) { + await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(0) }) + .where(eq(nativeRunFinalizations.runId, binding.runId)); + if (scope === "provider") await reconcileNativeFinalizations(db, [binding.runId]); + else await recordNativeFinalizationFailure({ db, runId: binding.runId, projectRunStatus: true, + failureScope: "workspace", error: new Error("native_workspace_finalize_failed") }); + if (attempt < 2) { + expect((await read(binding.runId)).status).toBe("running"); + expect(await readWake(wake.id)).toMatchObject({ status: "claimed", finishedAt: null }); + } + } + const run = await read(binding.runId); + expect((await state(binding.runId)).phase).toBe("terminal_failure"); + expect(run).toMatchObject({ status: scope === "provider" ? "succeeded" : "failed", nativePhase: "terminal_failure" }); + expect(run.finishedAt).not.toBeNull(); + expect(await readWake(wake.id)).toMatchObject({ status: scope === "provider" ? "completed" : "failed", + runId: binding.runId, finishedAt: run.finishedAt }); + }); + + it("keeps the source wake claimed when exhausted failure does not own terminal projection", async () => { + const binding = await accepted(), wake = await claimedWake(binding); + await db.update(workspaceOperations).set({ status: "failed" }).where(eq(workspaceOperations.heartbeatRunId, binding.runId)); + for (let attempt = 0; attempt < 3; attempt++) await recordNativeFinalizationFailure({ db, runId: binding.runId, + projectRunStatus: false, failureScope: "workspace", error: new Error("native_workspace_finalize_failed") }); + expect((await state(binding.runId)).phase).toBe("terminal_failure"); + expect((await read(binding.runId)).status).toBe("running"); + expect(await readWake(wake.id)).toEqual(wake); + }); + + it("rolls back exhausted projection and its claim if wake notification registration fails", async () => { + const binding = await accepted(), wake = await claimedWake(binding); + await db.update(workspaceOperations).set({ status: "failed" }).where(eq(workspaceOperations.heartbeatRunId, binding.runId)); + const fail = () => recordNativeFinalizationFailure({ db, runId: binding.runId, + projectRunStatus: true, failureScope: "workspace" as const, error: new Error("native_workspace_finalize_failed") }); + await fail(); await fail(); + const beforeRun = await read(binding.runId), beforeCoordinator = await state(binding.runId); + const notify = wakeSettlementNotifications.notifyExternalConversationSettlement; + vi.spyOn(wakeSettlementNotifications, "notifyExternalConversationSettlement").mockImplementation(async (tx, scope) => { + const [current] = await tx.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + expect(current.status).toBe("claimed"); + await notify(tx, scope); + throw new Error("exhaustion wake notification failure"); + }); + await expect(fail()).rejects.toThrow("exhaustion wake notification failure"); + expect(await read(binding.runId)).toEqual(beforeRun); + expect(await state(binding.runId)).toEqual(beforeCoordinator); + expect(await readWake(wake.id)).toEqual(wake); + }); + + it.each(["succeeded", "failed", "cancelled"] as const)("discovers and repairs an old %s run's claimed wake once", async status => { + const binding = await accepted(status); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + const run = await read(binding.runId), wake = await claimedWake(binding); + const work = nativeRunProjectionWork(db); + const [entry] = await work.load({ companyId, runId: binding.runId }); + expect(entry).toMatchObject({ id: binding.runId }); + await work.check(entry); + expect(await readWake(wake.id)).toMatchObject({ status: status === "succeeded" ? "completed" : status, + finishedAt: run.finishedAt }); + expect((await read(binding.runId)).finishedAt).toEqual(run.finishedAt); + expect(await work.load({ companyId, runId: binding.runId })).toEqual([]); + }); + + it.each(["unbound", "newer-run", "queued", "cancelled", "foreign-company", "foreign-agent", "live-owner"] as const)( + "does not settle a %s wake claim", async fault => { + const binding = await accepted(), wake = await claimedWake(binding); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + if (fault === "unbound" || fault === "newer-run") await db.update(agentWakeupRequests) + .set({ runId: fault === "unbound" ? null : randomUUID() }).where(eq(agentWakeupRequests.id, wake.id)); + if (fault === "queued" || fault === "cancelled") await db.update(agentWakeupRequests) + .set({ status: fault }).where(eq(agentWakeupRequests.id, wake.id)); + if (fault === "foreign-company") { + const otherCompany = randomUUID(); + await db.insert(companies).values({ id: otherCompany, name: "Other wake owner", issuePrefix: `OW${randomUUID().slice(0, 6)}` }); + await db.update(agentWakeupRequests).set({ companyId: otherCompany }).where(eq(agentWakeupRequests.id, wake.id)); + } + if (fault === "foreign-agent") { + const otherAgent = randomUUID(); + await db.insert(agents).values({ id: otherAgent, companyId, name: "Other wake owner", adapterType: "paperclip_runner" }); + await db.update(agentWakeupRequests).set({ agentId: otherAgent }).where(eq(agentWakeupRequests.id, wake.id)); + } + if (fault === "live-owner") await db.update(heartbeatRuns).set({ status: "running", nativePhase: "terminal_failure", + errorCode: "native_execution_ownership_unverified" }).where(eq(heartbeatRuns.id, binding.runId)); + const before = await readWake(wake.id); + await projectCommittedRun({ db, run: await read(binding.runId), coordinator: await state(binding.runId) }); + expect(await readWake(wake.id)).toEqual(before); + if (fault === "live-owner") expect((await read(binding.runId)).status).toBe("running"); + }, + ); + + it("registers wake settlement intent before mutation and rolls back notification failure", async () => { + const binding = await accepted(), wake = await claimedWake(binding); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + const before = await read(binding.runId); + const notify = wakeSettlementNotifications.notifyExternalConversationSettlement; + const notification = vi.spyOn(wakeSettlementNotifications, "notifyExternalConversationSettlement") + .mockImplementation(async (tx, scope) => { + const [current] = await tx.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + expect(current.status).toBe("claimed"); + await notify(tx, scope); + throw new Error("wake intent registration failure"); + }); + await expect(projectCommittedRun({ db, run: before, coordinator: await state(binding.runId) })) + .rejects.toThrow("wake intent registration failure"); + expect(notification).toHaveBeenCalledOnce(); + expect(await read(binding.runId)).toEqual(before); + expect(await readWake(wake.id)).toEqual(wake); + }); + + it("rolls back terminal projection and exact wake settlement together", async () => { + const binding = await accepted(), wake = await claimedWake(binding); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + const before = await read(binding.runId); + await expect(db.transaction(async tx => { + await projectCommittedRun({ db: tx as unknown as Db, run: before, coordinator: await state(binding.runId) }); + throw new Error("projection rollback fixture"); + })).rejects.toThrow("projection rollback fixture"); + expect(await read(binding.runId)).toEqual(before); + expect(await readWake(wake.id)).toEqual(wake); + }); + it("reconstructs a committed decision with an unfinished run projection and stays quiet after repair", async () => { const binding = await accepted(); await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); diff --git a/server/src/services/native-runtime/agent-chat-response-recovery.test.ts b/server/src/services/native-runtime/agent-chat-response-recovery.test.ts index 918653a5fd1..b0d4ce5481a 100644 --- a/server/src/services/native-runtime/agent-chat-response-recovery.test.ts +++ b/server/src/services/native-runtime/agent-chat-response-recovery.test.ts @@ -65,6 +65,26 @@ describe.skipIf(!support.supported)("committed Agent Chat reply recovery", () => const input = { companyId, issueId, runId }; return { ...input, input, agentId, userId, wakeId, prpEvent, reply }; } + it("settles the exact claimed wake when recovery projects a committed native run", async () => { + // The provider and result are already committed. Recovery owns terminal + // projection because the live executor will not return to completeRun. + const f = await fixture(false, { succeed: false }); + const sourceWakeId = randomUUID(); + await db.insert(agentWakeupRequests).values({ id: sourceWakeId, companyId: f.companyId, + agentId: f.agentId, source: "issue", reason: "issue_assigned", status: "claimed", + runId: f.runId, claimedAt: new Date(), payload: { issueId: f.issueId } }); + await db.update(heartbeatRuns).set({ wakeupRequestId: sourceWakeId }).where(eq(heartbeatRuns.id, f.runId)); + + await finalizeNativeRun({ db, runId: f.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, sourceWakeId)); + expect(run).toMatchObject({ status: "succeeded", nativePhase: "committed" }); + expect(run.finishedAt).not.toBeNull(); + expect(wake).toMatchObject({ status: "completed", runId: f.runId }); + expect(wake.finishedAt).not.toBeNull(); + }); + const comments = (runId: string) => db.select().from(issueComments).where(eq(issueComments.createdByRunId, runId)); const issue = async (issueId: string) => (await db.select().from(issues).where(eq(issues.id, issueId)))[0]; diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 1cbba5703cd..3a68b7d8f3b 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -3,7 +3,7 @@ import { notifyExternalAgentSettlement } from "../../modules/external-agents/ind import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; import { resolveCommittedNativeRecoveryNotices } from "./native-recovery-notices.js"; import { notifyNativeRunProjection } from "../native-run-projection-notifications.js"; -import { projectCommittedRun, recoveredExecutionFailureMetadata } from "./native-run-projection.js"; +import { projectCommittedRun, recoveredExecutionFailureMetadata, settleNativeTerminalWake } from "./native-run-projection.js"; import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js"; import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; @@ -465,6 +465,7 @@ async function recordRetryableFailure(input: { }) .where(eq(heartbeatRuns.id, input.run.id)) .returning(); + if (projectsTerminalStatus && updatedRun) await settleNativeTerminalWake(tx as unknown as Db, updatedRun); if (projectsTerminalStatus) terminalRunToEmit = updatedRun ?? null; if (supersededByNewerRun) { await issueRecoveryActionService( @@ -1679,6 +1680,7 @@ export async function finalizeNativeRun(input: { await notifyExecutionRunWork(tx, { companyId: run.companyId }); await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: coordinator.issueId }); const [after] = await projectRun(tx); + if (after && input.projectRunStatus) await settleNativeTerminalWake(tx, after); return after ? { before, after } : null; }); const updatedRun = projection?.after; diff --git a/server/src/services/native-runtime/native-run-projection.ts b/server/src/services/native-runtime/native-run-projection.ts index 2542b8bb1e3..706f80e7366 100644 --- a/server/src/services/native-runtime/native-run-projection.ts +++ b/server/src/services/native-runtime/native-run-projection.ts @@ -1,3 +1,5 @@ +import { notifyExternalConversationSettlement } from "../external-conversation-settlement-notifications.js"; +import { notifyStrandedAgentStatus } from "../live-run-recovery-notifications.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; @@ -5,7 +7,7 @@ import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-wor import { notifyNativeRunProjection } from "../native-run-projection-notifications.js"; import { randomUUID } from "node:crypto"; import { and, eq, exists, inArray, isNotNull, isNull, not, or, sql, type SQL } from "drizzle-orm"; -import { heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; +import { agentWakeupRequests, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; @@ -26,11 +28,41 @@ export function projectNativeTerminalRunStatus( +function claimedNativeWakeNeedsSettlement() { + return sql`exists (select 1 from ${agentWakeupRequests} where + ${agentWakeupRequests.id} = ${heartbeatRuns.wakeupRequestId} + and ${agentWakeupRequests.companyId} = ${heartbeatRuns.companyId} + and ${agentWakeupRequests.agentId} = ${heartbeatRuns.agentId} + and ${agentWakeupRequests.runId} = ${heartbeatRuns.id} + and ${agentWakeupRequests.status} = 'claimed')`; +} + +/** Settle only the exact persisted source claim inside terminal projection. */ +export async function settleNativeTerminalWake(tx: Db, run: Pick) { + if (run.runtimeMode !== "native" || !run.wakeupRequestId || !run.finishedAt + || !["succeeded", "failed", "cancelled"].includes(run.status)) return; + const claim = and(eq(agentWakeupRequests.id, run.wakeupRequestId), + eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), + eq(agentWakeupRequests.runId, run.id), eq(agentWakeupRequests.status, "claimed")); + const [current] = await tx.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests) + .where(claim).for("update").limit(1); + if (!current) return; + await notifyExternalConversationSettlement(tx, { companyId: run.companyId }); + await notifyStrandedAgentStatus(tx, { companyId: run.companyId, agentId: run.agentId }); + await tx.update(agentWakeupRequests).set({ + status: run.status === "succeeded" ? "completed" : run.status, + finishedAt: run.finishedAt, error: run.error, updatedAt: new Date(), + }).where(claim); +} + export function nativeRunProjectionNeedsRepair(projectedStatus: "succeeded" | "failed" | "cancelled" | SQL) { return and( eq(heartbeatRuns.runtimeMode, "native"), or( inArray(heartbeatRuns.status, ["queued", "running", "failed"]), + and(eq(heartbeatRuns.status, projectedStatus), + inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled"]), claimedNativeWakeNeedsSettlement()), and( eq(heartbeatRuns.status, "succeeded"), or( @@ -51,6 +83,7 @@ export function nativeRunProjectionNeedsRepair(projectedStatus: "succeeded" | "f or( sql`${heartbeatRuns.status} is distinct from ${projectedStatus}`, + claimedNativeWakeNeedsSettlement(), isNull(heartbeatRuns.finishedAt), sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`, sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`, @@ -164,6 +197,7 @@ export async function projectCommittedRun(input: { }) .where(needsRepair) .returning(); + if (after) await settleNativeTerminalWake(tx, after); return after ? { before, after } : null; });