diff --git a/doc/native-workspace-finalization-recovery.md b/doc/native-workspace-finalization-recovery.md index 01e7a7031c4..c8353239234 100644 --- a/doc/native-workspace-finalization-recovery.md +++ b/doc/native-workspace-finalization-recovery.md @@ -505,3 +505,11 @@ acknowledgement receipt, and process identity. It does not load saved execution input, runner profiles, excerpts, or unrelated result fields. Lease cleanup, remote termination receipts, local process checks, task ownership, and admission checks keep their existing behavior. Stored runs retain all of their data. + +One periodic stranded-message pass coalesces successful native release checks +for the same company, task, source run, and current source row version. The release +path already drains that +task's queue through current admission checks. Each saved request still advances +its recovery cursor. Rejected attempts, changed source runs, and updated source evidence remain eligible, +and the next pass checks current state again. Legacy per-request handling stays +unchanged. Pending work is awaited before a source is marked checked. diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index 5a31b5ee131..ef889747669 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/server/src/services/heartbeat/queue.test.ts @@ -1,12 +1,14 @@ import { claimNativeSessionResumptions } from "../native-runtime/native-finalization-reconciler.js"; import type { Db } from "@paperclipai/db"; import { agents, agentWakeupRequests, caseHandoffs, cases, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issueComments, issues, nativeRunFinalizations } from "@paperclipai/db"; -import { eq, sql } from "drizzle-orm"; +import { eq, inArray, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { createPostgresWakeQueueAdapter } from "../../modules/wake-queue/adapters/postgres.js"; import { createReleaseIssueExecution } from "../../modules/wake-queue/application/use-cases.js"; import { createHeartbeatQueue, type HeartbeatQueueDependencies } from "./queue.js"; +import { createHeartbeatRunControl } from "./run-control.js"; +import { callbacks as runControlCallbacks } from "./__tests__/run-control-fixtures.js"; import { createHeartbeatQueueTestDependencies as callbacks } from "../../__tests__/helpers/heartbeat-queue.js"; @@ -113,6 +115,116 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { return { ...seeded, recipient, pending }; } + async function addStrandedMessages(seeded: Awaited>) { + const messages = [seeded.pending]; + for (let index = 0; index < 2; index++) { + const [comment] = await db.insert(issueComments).values({ companyId: seeded.company.id, issueId: seeded.issue.id, + body: `Additional saved feedback ${index}`, authorAgentId: seeded.agent.id }).returning(); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: seeded.company.id, agentId: seeded.recipient.id, + source: "automation", reason: "issue_execution_deferred", status: "deferred_issue_execution", + payload: { issueId: seeded.issue.id, _paperclipWakeContext: { issueId: seeded.issue.id, + wakeReason: "issue_comment_mentioned", wakeCommentIds: [comment.id] } } }).returning(); + messages.push(wake); + } + await db.update(agentWakeupRequests).set({ updatedAt: new Date(0) }).where(inArray(agentWakeupRequests.id, messages.map(row => row.id))); + return messages; + } + + async function expectSavedMessages(messages: Awaited>) { + const rows = await db.select().from(agentWakeupRequests).where(inArray(agentWakeupRequests.id, messages.map(row => row.id))); + expect(rows).toHaveLength(messages.length); + for (const before of messages) { + const row = rows.find(row => row.id === before.id)!; + expect(row).toMatchObject({ payload: before.payload, status: "deferred_issue_execution", runId: null, + finishedAt: null, error: null }); + expect(row.updatedAt.getTime()).toBeGreaterThan(0); + } + } + + it.each(["legacy", "native"])("keeps saved %s messages while avoiding repeated native source release", async runtimeMode => { + const seeded = await strandedMention(runtimeMode); + const messages = await addStrandedMessages(seeded); + const deps = callbacks(db), queue = createHeartbeatQueue(db, deps); + await queue.resumeQueuedRuns(); + expect(deps.releaseIssueExecutionAndPromote).toHaveBeenCalledTimes(runtimeMode === "native" ? 1 : 3); + await expectSavedMessages(messages); + await queue.resumeQueuedRuns(); + expect(deps.releaseIssueExecutionAndPromote).toHaveBeenCalledTimes(runtimeMode === "native" ? 2 : 6); + expect(deps.executeRun).not.toHaveBeenCalled(); + }); + + it("retries a rejected native release on the next saved message", async () => { + const seeded = await strandedMention("native"), messages = await addStrandedMessages(seeded); + const deps = callbacks(db); + deps.releaseIssueExecutionAndPromote.mockRejectedValueOnce(new Error("Fixture release remains pending")); + await createHeartbeatQueue(db, deps).resumeQueuedRuns(); + expect(deps.releaseIssueExecutionAndPromote).toHaveBeenCalledTimes(2); + await expectSavedMessages(messages); + expect(deps.executeRun).not.toHaveBeenCalled(); + }); + + it("joins the native release before it suppresses another message's duplicate check", async () => { + const seeded = await strandedMention("native"), messages = await addStrandedMessages(seeded); + const deps = callbacks(db); + let finish!: () => void; + deps.releaseIssueExecutionAndPromote.mockImplementationOnce(() => new Promise(resolve => { finish = resolve; })); + let settled = false; + const pending = createHeartbeatQueue(db, deps).resumeQueuedRuns().then(() => { settled = true; }); + try { + await vi.waitFor(() => expect(deps.releaseIssueExecutionAndPromote).toHaveBeenCalledTimes(1)); + expect(settled).toBe(false); + } finally { finish?.(); } + await pending; + expect(deps.releaseIssueExecutionAndPromote).toHaveBeenCalledTimes(1); + await expectSavedMessages(messages); + }); + + it("rechecks a different native source that completes during the pass", async () => { + const seeded = await strandedMention("native"), messages = await addStrandedMessages(seeded); + await db.update(heartbeatRuns).set({ createdAt: new Date(0) }).where(eq(heartbeatRuns.id, seeded.run.id)); + const deps = callbacks(db); + let nextId: string | undefined; + deps.releaseIssueExecutionAndPromote.mockImplementationOnce(async () => { + const [next] = await db.insert(heartbeatRuns).values({ companyId: seeded.company.id, agentId: seeded.agent.id, + runtimeMode: "native", nativeIssueId: seeded.issue.id, status: "succeeded", finishedAt: new Date(), + contextSnapshot: { issueId: seeded.issue.id } }).returning(); + nextId = next.id; + }); + await createHeartbeatQueue(db, deps).resumeQueuedRuns(); + expect(deps.releaseIssueExecutionAndPromote.mock.calls.map(([run]) => run.id)).toEqual([seeded.run.id, nextId]); + await expectSavedMessages(messages); + }); + + it("rechecks changed native evidence on the same source run", async () => { + const seeded = await strandedMention("native"), messages = await addStrandedMessages(seeded); + const deps = callbacks(db); + deps.releaseIssueExecutionAndPromote.mockImplementationOnce(async () => { + await db.update(heartbeatRuns).set({ resultJson: { observedReceipt: "new evidence" } }).where(eq(heartbeatRuns.id, seeded.run.id)); + }); + await createHeartbeatQueue(db, deps).resumeQueuedRuns(); + expect(deps.releaseIssueExecutionAndPromote).toHaveBeenCalledTimes(2); + expect(deps.releaseIssueExecutionAndPromote.mock.calls.map(([run]) => run.id)).toEqual([seeded.run.id, seeded.run.id]); + await expectSavedMessages(messages); + }); + + it("checks each unrelated native task in the same pass", async () => { + const seeded = await strandedMention("native"), firstMessages = await addStrandedMessages(seeded); + const [issue] = await db.insert(issues).values({ companyId: seeded.company.id, title: "Another saved task", status: "todo", assigneeAgentId: seeded.agent.id }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: seeded.company.id, agentId: seeded.agent.id, + runtimeMode: "native", nativeIssueId: issue.id, status: "succeeded", finishedAt: new Date(), contextSnapshot: { issueId: issue.id } }).returning(); + const [comment] = await db.insert(issueComments).values({ companyId: seeded.company.id, issueId: issue.id, + body: "Other task feedback", authorAgentId: seeded.agent.id }).returning(); + const [pending] = await db.insert(agentWakeupRequests).values({ companyId: seeded.company.id, agentId: seeded.recipient.id, + source: "automation", reason: "issue_execution_deferred", status: "deferred_issue_execution", + payload: { issueId: issue.id, _paperclipWakeContext: { issueId: issue.id, wakeReason: "issue_comment_mentioned", wakeCommentIds: [comment.id] } } }).returning(); + const secondMessages = await addStrandedMessages({ ...seeded, issue, run, pending }); + const deps = callbacks(db); + await createHeartbeatQueue(db, deps).resumeQueuedRuns(); + expect(deps.releaseIssueExecutionAndPromote).toHaveBeenCalledTimes(2); + expect(new Set(deps.releaseIssueExecutionAndPromote.mock.calls.map(([row]) => row.id))).toEqual(new Set([seeded.run.id, run.id])); + await expectSavedMessages([...firstMessages, ...secondMessages]); + }); + it("rechecks a handoff's assignment version under the issue claim lock", async () => { const { company, agent, issue, wake, run } = await fixture("queued"); const [record] = await db.insert(cases).values({ companyId: company.id, caseNumber: 1, identifier: "QM-CASE-1", @@ -137,7 +249,9 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { }); it.each(["legacy", "native"])("periodic recovery preserves a paused recipient through the real %s release adapter", async runtimeMode => { - const { company, issue, recipient, pending } = await strandedMention(runtimeMode); + const seeded = await strandedMention(runtimeMode); + const { company, issue, recipient } = seeded; + const messages = await addStrandedMessages(seeded); await db.update(agents).set({ status: "paused" }).where(eq(agents.id, recipient.id)); await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, issue.id)); const deps = callbacks(db); @@ -146,16 +260,16 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }), resolveSessionBeforeForWakeup: async () => null, }), recovery: { escalateStrandedAssignedIssue: async () => undefined, escalateStrandedRecoveryIssueInPlace: async () => undefined } }); - deps.releaseIssueExecutionAndPromote.mockImplementation(async (run, options) => { - await release({ companyId: run.companyId, runId: run.id, now: new Date(), ...options }); - }); + const controlDeps = runControlCallbacks(db); + vi.mocked(controlDeps.wakeQueue.releaseIssueExecution).mockImplementation(release); + const control = createHeartbeatRunControl(db, controlDeps); + deps.releaseIssueExecutionAndPromote.mockImplementation(control.releaseIssueExecutionAndPromote); for (let pass = 0; pass < 2; pass++) { await createHeartbeatQueue(db, deps).resumeQueuedRuns(); - expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, pending.id)))[0]) - .toMatchObject({ status: "deferred_issue_execution", runId: null, finishedAt: null, error: null, payload: pending.payload }); + await expectSavedMessages(messages); expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, company.id))).toHaveLength(1); } - expect(deps.releaseIssueExecutionAndPromote).toHaveBeenCalledTimes(2); + expect(deps.releaseIssueExecutionAndPromote).toHaveBeenCalledTimes(runtimeMode === "native" ? 2 : 6); expect(deps.executeRun).not.toHaveBeenCalled(); }); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 515bd683fb3..35ff18e6e88 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1834,6 +1834,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende AND live_issue_run.status IN ('queued', 'running', 'scheduled_retry'))`, cutoff ? gte(agentWakeupRequests.requestedAt, cutoff) : undefined)) .orderBy(asc(agentWakeupRequests.updatedAt)).limit(50); + const releasedNativeSources = new Set(); for (const { wake } of strandedQueues) { if (!queuedCommentIdsFromWakePayload(wake.payload).length && !await readQueuedInteractionResponse(db, wake.companyId, String(wake.payload?.issueId), wake.payload)) continue; @@ -1845,6 +1846,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende startedAt: heartbeatRuns.startedAt, runtimeMode: heartbeatRuns.runtimeMode, errorCode: heartbeatRuns.errorCode, + sourceVersion: sql`${heartbeatRuns}.xmin::text`, }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, wake.companyId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${String(wake.payload?.issueId)}`, @@ -1855,6 +1857,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende )); if (!latest || !latest.finishedAt || !isHeartbeatRunTerminalStatus(latest.status)) continue; if (latest.runtimeMode !== "legacy" && latest.runtimeMode !== "native") continue; + const releaseSource = `${latest.companyId}:${String(wake.payload?.issueId)}:${latest.id}:${latest.sourceVersion}`; + if (latest.runtimeMode === "native" && releasedNativeSources.has(releaseSource)) continue; if (latest.runtimeMode === "legacy") { const cancelledAdmission = latest.status === "cancelled" && !latest.startedAt && latest.errorCode === "execution_reconciliation_required"; @@ -1864,7 +1868,9 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende }); } } - await releaseIssueExecutionAndPromote(latest, { suppressImmediateRecovery: true, resumeStrandedQueue: true }).catch(err => { + await releaseIssueExecutionAndPromote(latest, { suppressImmediateRecovery: true, resumeStrandedQueue: true }).then(() => { + if (latest.runtimeMode === "native") releasedNativeSources.add(releaseSource); + }).catch(err => { logger.warn({ err, queueId: wake.id }, "failed to promote stranded comments"); }); }