diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 970d2c28670..6229c02109c 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1342,6 +1342,33 @@ apply. Stream closure without a turn terminal is not proof of success. Event replay uses the existing source receipts and never repeats provider work merely to recover recorded output. +A delegated-task completion callback can finish native execution before its +accepted final reply is saved. The completion outbox must retain that callback's +queued target and attempt while the existing presentation selector finds pending +work for the same company, conversation, run and agent. The hold lasts at most +five existing one-minute delivery leases, measured from that run's original +`finishedAt`. Repeated sweeps cannot restart or extend this five-minute window. +It wakes the presentation owner and waits until the later of its current delivery +lease or the recorded native retry deadline, capped by that fixed window. +The selector only defers a new execution attempt; the presentation owner still +enforces all reply, session, actor and digest checks. A saved reply acknowledges +the outbox once and permits normal conversation settlement. If publication keeps +declining or cannot finish within the window, the existing retry, exhaustion and +daily-limit checks resume. Retry takes the same child-task-first lock held by +reply insertion and acknowledgment. Reply authority reloads the delivery after +acquiring that lock. Retry checks that the locked task is still Done. It then +matches the delivery's recorded completion-event version, pending state, target, +attempt and exact lease. Deferral uses that same exact claim fence, so a stalled +old hold cannot shorten or replace a newer claim's lease. +Reassigning a Done task can change its current task version without invalidating +that completion event; a real status change keeps its existing supersession. A delivered reply, +newer target or newer claim prevents that retry. Wake admission runs only after +the winning retry transaction commits and releases its locks. The original +accepted result and child work remain unchanged. Failed or cancelled execution, +missing accepted output and superseded +handoffs retain their existing retry or supersession paths. Daily limits do not +need to admit another model run to save an already accepted answer. + Native execution success can precede its durable presentation decision. GitHub response maintenance without a confirmed current-run reply holds that request until the native phase is committed and a valid presentation decision exists. diff --git a/server/src/services/chat-completion-delivery.ts b/server/src/services/chat-completion-delivery.ts index 8188b615262..e559044b894 100644 --- a/server/src/services/chat-completion-delivery.ts +++ b/server/src/services/chat-completion-delivery.ts @@ -1,5 +1,7 @@ import { notifyDeliveryWork, DELIVERY_QUEUES } from "./delivery-work-notifications.js"; -import { and, asc, desc, eq, inArray, lte, sql } from "drizzle-orm"; +import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; +import { nativeResponsePresentationCandidates } from "./native-runtime/native-response-presentation-query.js"; +import { and, asc, desc, eq, inArray, isNull, lte, sql } from "drizzle-orm"; import { agents, agentWakeupRequests, chatCompletionDeliveries as deliveries, chatTaskHandoffs as handoffs, heartbeatRuns, issueComments, issueDocuments, issueThreadInteractions, issues, type Db } from "@paperclipai/db"; import { instanceSettingsService } from "./instance-settings.js"; @@ -195,11 +197,11 @@ export async function existingChatCompletionReply(tx: Connection, runId: string, await tx.select({ id: issues.id }).from(issues).where(and(eq(issues.id, delivery.taskId), eq(issues.companyId, run.companyId))).for("update"); const row = await loadAudience(tx, delivery.id); if (!row || row.source?.id !== issueId || row.handoff.agentId !== run.agentId || !current(row)) throw new Error("chat_completion_superseded"); - if (delivery.responseCommentId) { - const [comment] = await tx.select().from(issueComments).where(and(eq(issueComments.id, delivery.responseCommentId), eq(issueComments.issueId, issueId))); + if (row.delivery.responseCommentId) { + const [comment] = await tx.select().from(issueComments).where(and(eq(issueComments.id, row.delivery.responseCommentId), eq(issueComments.issueId, issueId))); if (comment) return comment; } - if (delivery.status !== "queued" || delivery.targetRunId !== runId) throw new Error("chat_completion_superseded"); + if (row.delivery.status !== "queued" || row.delivery.targetRunId !== runId) throw new Error("chat_completion_superseded"); } if (rows.length !== ids(run).length) throw new Error("chat_completion_superseded"); return null; @@ -233,17 +235,51 @@ export function chatCompletionDeliveryService(db: Db, heartbeat: { wakeup(agentI if (runId) { const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); if (run && activeRuns.includes(run.status)) return; - // A published reply is acknowledged in its own transaction, independent of run outcome. - const [latest] = await db.select().from(deliveries).where(eq(deliveries.id, id)); - if (!latest || !pending.includes(latest.status as typeof pending[number])) return; - await db.update(deliveries).set({ status: claimed.attempts + 1 >= MAX_ATTEMPTS ? "exhausted" : "pending", - attempts: claimed.attempts + 1, targetRunId: null, nextAttemptAt: new Date(), - error: run?.error ?? "Completion turn ended without a reply" }) - .where(and(eq(deliveries.id, id), inArray(deliveries.status, [...pending]))); - // The failed run is already terminal. Reclaim the advanced attempt now; - // a second lease interval would only delay restart recovery. The atomic - // claim and fresh wake key still serialize competing sweepers. - if (claimed.attempts + 1 < MAX_ATTEMPTS) await deliver(id); + + const presentationDeadline = run?.runtimeMode === "native" && run.status === "succeeded" && run.finishedAt + ? run.finishedAt.getTime() + MAX_ATTEMPTS * LEASE_MS : null; + if (run && presentationDeadline !== null && Date.now() < presentationDeadline) { + const scope = { companyId: claimed.companyId, issueId: row.handoff.conversationId, + runId: run.id, agentId: row.handoff.agentId }; + const deferred = await db.transaction(async tx => { + + + const candidates = nativeResponsePresentationCandidates(scope).as("pending_native_completion_reply"); + const [response] = await tx.select().from(candidates); + if (!response || Date.now() >= presentationDeadline) return false; + await notifyNativeResponsePresentation(tx, scope); + const updated = await tx.update(deliveries).set({ nextAttemptAt: new Date(Math.min(presentationDeadline, Math.max( + claimed.nextAttemptAt.getTime(), response.nextAttemptAt?.getTime() ?? 0, + ))) }).where(and(eq(deliveries.id, id), eq(deliveries.companyId, claimed.companyId), + eq(deliveries.status, "queued"), eq(deliveries.targetRunId, run.id), + eq(deliveries.statusVersion, claimed.statusVersion), eq(deliveries.attempts, claimed.attempts), + eq(deliveries.nextAttemptAt, claimed.nextAttemptAt))).returning({ id: deliveries.id }); + return updated.length > 0; + }); + if (deferred) return; + } + const advanced = await db.transaction(async tx => { + + + const [task] = await tx.select({ status: issues.status }).from(issues) + .where(and(eq(issues.id, claimed.taskId), eq(issues.companyId, claimed.companyId))).for("update"); + if (task?.status !== "done") return false; + const sameClaim = and(eq(deliveries.id, id), eq(deliveries.companyId, claimed.companyId), + inArray(deliveries.status, [...pending]), eq(deliveries.statusVersion, claimed.statusVersion), + eq(deliveries.attempts, claimed.attempts), + eq(deliveries.nextAttemptAt, claimed.nextAttemptAt), + claimed.targetRunId ? eq(deliveries.targetRunId, claimed.targetRunId) : isNull(deliveries.targetRunId)); + const [latest] = await tx.select({ id: deliveries.id }).from(deliveries).where(sameClaim).for("update"); + if (!latest) return false; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatCompletion); + const updated = await tx.update(deliveries).set({ status: claimed.attempts + 1 >= MAX_ATTEMPTS ? "exhausted" : "pending", + attempts: claimed.attempts + 1, targetRunId: null, nextAttemptAt: new Date(), + error: run?.error ?? "Completion turn ended without a reply" }).where(sameClaim).returning({ id: deliveries.id }); + return updated.length > 0; + }); + + + if (advanced && claimed.attempts + 1 < MAX_ATTEMPTS) await deliver(id); return; } if (wakes.some(w => ["queued", "claimed", "deferred_issue_execution", "coalesced"].includes(w.status))) return; 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 1c2cc4399a3..aca26894574 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 @@ -1,13 +1,19 @@ import { randomUUID } from "node:crypto"; -import { and, eq } from "drizzle-orm"; -import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { agents, authUsers, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issues, - nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; + nativeRunFinalizations, nativeRunResults, chatTaskHandoffs, chatCompletionDeliveries, agentWakeupRequests, type Db } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js"; import { finalizeNativeRun, recoverCommittedNativeChatResponse } from "./native-run-finalizer.js"; import { nativeResponsePresentationWork } from "./native-response-presentation-work.js"; +import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../task-admission.js"; +import { chatCompletionDeliveryService, existingChatCompletionReply } from "../chat-completion-delivery.js"; +import { instanceSettingsService } from "../instance-settings.js"; +import { issueService } from "../issues.js"; +import * as nativeResponseNotifications from "../native-response-presentation-notifications.js"; const external = resolveExternalTestDatabaseUrl("PAPERCLIP_AGENT_CHAT_RESPONSE_TEST_DATABASE_URL"); const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); @@ -20,7 +26,7 @@ describe.skipIf(!support.supported)("committed Agent Chat reply recovery", () => }, 60_000); afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); - async function fixture(conversation = true) { + async function fixture(conversation = true, options: { succeed?: boolean } = {}) { const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); const sessionId = randomUUID(), runnerId = randomUUID(), userId = `fixture-${randomUUID()}`, wakeId = randomUUID(); await db.insert(authUsers).values({ id: userId, name: "Conversation owner", email: `${userId}@example.test`, @@ -52,14 +58,514 @@ describe.skipIf(!support.supported)("committed Agent Chat reply recovery", () => objectiveSatisfied: true, criteria: [], remainingWork: [] }, evidence: [], verification: [], attentionRequests: [], artifacts: [] }, terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "succeeded", turnTerminalState: "completed", reportedWorkDisposition: "done" }, turnId: "hire-turn" }); await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: false }); - await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date("2020-01-01T00:02:00Z") }).where(eq(heartbeatRuns.id, runId)); expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0].phase).toBe("committed"); + if (options.succeed !== false) await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date("2020-01-01T00:02:00Z") }).where(eq(heartbeatRuns.id, runId)); const input = { companyId, issueId, runId }; return { ...input, input, agentId, userId, wakeId, prpEvent, reply }; } 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]; + async function completion(f: Awaited>) { + const taskId = randomUUID(), deliveryId = randomUUID(); + await instanceSettingsService(db).updateExperimental({ enableAgentChat: true }); + await db.insert(issues).values({ id: taskId, companyId: f.companyId, title: "Completed review", status: "done", statusVersion: 2 }); + await db.insert(chatTaskHandoffs).values({ companyId: f.companyId, taskId, conversationId: f.issueId, + agentId: f.agentId, sessionGeneration: 0 }); + await db.insert(chatCompletionDeliveries).values({ id: deliveryId, companyId: f.companyId, taskId, + statusVersion: 2, status: "queued", targetRunId: f.runId, nextAttemptAt: new Date(0) }); + await db.update(heartbeatRuns).set({ finishedAt: new Date() }).where(and(eq(heartbeatRuns.id, f.runId), eq(heartbeatRuns.status, "succeeded"))); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, source: "automation", wakeReason: "chat_task_completed", + conversationMode: true, conversationSessionGeneration: 0, chatCompletionDeliveryIds: [deliveryId], chatCompletionUpdates: [{ taskId }] } }) + .where(eq(heartbeatRuns.id, f.runId)); + const wakeup = vi.fn(async (_agentId: string, options: { idempotencyKey: string }) => { + await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "automation", + status: "skipped", reason: "heartbeat.daily_run_limit", idempotencyKey: options.idempotencyKey }); + return null; + }); + await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "automation", + status: "completed", reason: "chat_task_completed", idempotencyKey: `chat-completion:${deliveryId}:0`, runId: f.runId }); + const delivery = async () => (await db.select().from(chatCompletionDeliveries).where(eq(chatCompletionDeliveries.id, deliveryId)))[0]; + return { taskId, deliveryId, wakeup, delivery, service: chatCompletionDeliveryService(db, { wakeup }) }; + } + + it("preserves accepted native callback presentation when outbox retry runs first at the daily cap", async () => { + const f = await fixture(), outbox = await completion(f); + const work = nativeResponsePresentationWork(db); + const [entry] = await work.load(f.input); + expect(entry).toMatchObject({ id: f.runId }); + + + await outbox.service.deliver(outbox.deliveryId); + await expect(work.check(entry)).resolves.toBeNull(); + const [comment] = await comments(f.runId); + expect(comment).toMatchObject({ body: f.reply, authorAgentId: f.agentId, createdByRunId: f.runId }); + expect(await comments(f.runId)).toHaveLength(1); + expect(await issue(f.issueId)).toMatchObject({ status: "in_review", conversationState: "waiting" }); + expect(await outbox.delivery()).toMatchObject({ status: "delivered", targetRunId: f.runId, attempts: 0, responseCommentId: comment.id }); + expect(outbox.wakeup).not.toHaveBeenCalled(); + }); + + it.each([false, true])("wakes the real reply worker only after outer commit, becomes idle and persists once (rollback=%s)", async rollback => { + const f = await fixture(true, { succeed: false }), outbox = await completion(f); + let queries = 0; + const debug = db.$client.options.debug; + db.$client.options.debug = () => { queries++; }; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeResponsePresentationWork(db).register(coordinator, () => true); + const quiet = () => vi.waitFor(() => { + expect(coordinator.nextWakeAt()).toBeNull(); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + const projectAndDefer = async (abort: boolean) => db.transaction(async tx => { + await tx.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); + await chatCompletionDeliveryService(tx as unknown as Db, { wakeup: outbox.wakeup }).deliver(outbox.deliveryId); + expect(await comments(f.runId)).toEqual([]); + if (abort) throw new Error("fixture-outbox-rollback"); + }); + try { + await worker.ready; await quiet(); + if (rollback) { + await expect(projectAndDefer(true)).rejects.toThrow("fixture-outbox-rollback"); + await quiet(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].status).toBe("running"); + expect(await outbox.delivery()).toMatchObject({ status: "queued", targetRunId: f.runId, attempts: 0, nextAttemptAt: new Date(0) }); + expect(await comments(f.runId)).toEqual([]); + } + await projectAndDefer(false); + await vi.waitFor(async () => expect(await comments(f.runId)).toHaveLength(1), { timeout: 5000 }); + await quiet(); + const [comment] = await comments(f.runId); + expect(comment).toMatchObject({ body: f.reply, authorAgentId: f.agentId, createdByRunId: f.runId }); + expect(await issue(f.issueId)).toMatchObject({ status: "in_review", conversationState: "waiting" }); + expect(await outbox.delivery()).toMatchObject({ status: "delivered", targetRunId: f.runId, attempts: 0, responseCommentId: comment.id }); + const settledQueries = queries; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + await vi.advanceTimersByTimeAsync(60_000); + expect(queries).toBe(settledQueries); + expect(coordinator.nextWakeAt()).toBeNull(); + vi.useRealTimers(); + await outbox.service.deliver(outbox.deliveryId); + expect(await comments(f.runId)).toHaveLength(1); + expect(outbox.wakeup).not.toHaveBeenCalled(); + expect(errors).toEqual([]); + } finally { + vi.useRealTimers(); worker.stop(); await coordinator.stop(); db.$client.options.debug = debug; + } + }); + + it.each(["failed", "cancelled", "missing-final", "no-accepted", "old-generation", "foreign-company", "foreign-agent"])( + "keeps the existing retry path for %s targets", async fault => { + const f = await fixture(), outbox = await completion(f); + if (fault === "failed" || fault === "cancelled") await db.update(heartbeatRuns).set({ status: fault }).where(eq(heartbeatRuns.id, f.runId)); + if (fault === "missing-final") await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.runId)); + if (fault === "no-accepted") await db.update(nativeRunResults).set({ schemaStatus: "rejected" }).where(eq(nativeRunResults.runId, f.runId)); + if (fault === "old-generation") { + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + await db.update(heartbeatRuns).set({ contextSnapshot: { ...run.contextSnapshot, conversationSessionGeneration: 1 } }).where(eq(heartbeatRuns.id, f.runId)); + } + let foreign: Awaited> | undefined; + if (fault === "foreign-company") { + foreign = await fixture(); + await db.update(chatCompletionDeliveries).set({ targetRunId: foreign.runId }).where(eq(chatCompletionDeliveries.id, outbox.deliveryId)); + } + if (fault === "foreign-agent") { + const otherAgent = randomUUID(); + await db.insert(agents).values({ id: otherAgent, companyId: f.companyId, name: "Other agent", adapterType: "paperclip_runner" }); + await db.update(heartbeatRuns).set({ agentId: otherAgent }).where(eq(heartbeatRuns.id, f.runId)); + } + await outbox.service.deliver(outbox.deliveryId); + expect(outbox.wakeup).toHaveBeenCalledTimes(1); + expect(await outbox.delivery()).toMatchObject({ status: "pending", targetRunId: null, attempts: 1, responseCommentId: null }); + expect(await comments(f.runId)).toEqual([]); + if (foreign) expect(await comments(foreign.runId)).toEqual([]); + }, + ); + + it.each(["reset-source", "reopened-task"])("keeps %s delivery supersession", async fault => { + const f = await fixture(), outbox = await completion(f); + if (fault === "reset-source") await db.update(issues).set({ conversationSessionGeneration: 1 }).where(eq(issues.id, f.issueId)); + else { + await issueService(db).update(outbox.taskId, { status: "todo" }); + await issueService(db).update(outbox.taskId, { status: "done" }); + expect(await issue(outbox.taskId)).toMatchObject({ status: "done", statusVersion: 4 }); + } + await outbox.service.deliver(outbox.deliveryId); + expect(await outbox.delivery()).toMatchObject({ status: "superseded", attempts: 0, responseCommentId: null }); + expect(outbox.wakeup).not.toHaveBeenCalled(); + expect(await comments(f.runId)).toEqual([]); + }); + + it("lets an in-flight presentation acknowledge before expiry can retarget the delivery", async () => { + const started = Date.now(); + vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(started); + const f = await fixture(), outbox = await completion(f); + const suffix = randomUUID().replaceAll("-", ""), lockId = Number.parseInt(suffix.slice(0, 6), 16) + 1; + const trigger = `completion_reply_hold_${suffix}`; + let release!: () => void, acquired!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const locked = new Promise(resolve => { acquired = resolve; }); + let blocker: Promise | undefined, presentation: Promise | undefined, expiry: Promise | undefined; + try { + await db.execute(sql.raw(`create function ${trigger}() returns trigger language plpgsql as $$ + begin + if NEW.created_by_run_id = '${f.runId}'::uuid then perform pg_advisory_xact_lock(${lockId}); end if; + return NEW; + end $$`)); + await db.execute(sql.raw(`create trigger ${trigger} before insert on issue_comments for each row execute function ${trigger}()`)); + blocker = db.transaction(async tx => { await tx.execute(sql`select pg_advisory_xact_lock(${lockId}::bigint)`); acquired(); await hold; }); + await Promise.race([locked, blocker]); + const work = nativeResponsePresentationWork(db), [entry] = await work.load(f.input); + presentation = work.check(entry); void presentation.catch(() => undefined); + + + await vi.waitFor(async () => { + const [row] = await db.execute(sql`select exists (select 1 from pg_locks + where locktype = 'advisory' and classid = 0 and objid = ${lockId}::oid and not granted + and database = (select oid from pg_database where datname = current_database())) as waiting`); + expect(row.waiting).toBe(true); + }, { timeout: 5000 }); + vi.setSystemTime(started + 300_000); + let expirySettled = false; + expiry = outbox.service.deliver(outbox.deliveryId).then(() => { expirySettled = true; }); + void expiry.catch(() => undefined); + await vi.waitFor(async () => { + const [row] = await db.execute(sql`select exists (select 1 from pg_stat_activity + where datname = current_database() and pid <> pg_backend_pid() and wait_event_type = 'Lock' + and query like '%from "issues"%' and query like '%for update%') as waiting`); + expect(expirySettled || row.waiting).toBe(true); + }, { timeout: 5000 }); + expect(expirySettled).toBe(false); + expect(await outbox.delivery()).toMatchObject({ status: "queued", targetRunId: f.runId, attempts: 0, responseCommentId: null }); + release(); await blocker; await presentation; await expiry; + const [comment] = await comments(f.runId); + expect(comment).toMatchObject({ body: f.reply, authorAgentId: f.agentId, createdByRunId: f.runId }); + expect(await comments(f.runId)).toHaveLength(1); + expect(await outbox.delivery()).toMatchObject({ status: "delivered", targetRunId: f.runId, attempts: 0, responseCommentId: comment.id }); + expect(outbox.wakeup).not.toHaveBeenCalled(); + expect(await issue(f.issueId)).toMatchObject({ status: "in_review", conversationState: "waiting" }); + } finally { + release?.(); await Promise.allSettled([blocker, presentation, expiry].filter((value): value is Promise => Boolean(value))); + await db.execute(sql.raw(`drop trigger if exists ${trigger} on issue_comments`)); + await db.execute(sql.raw(`drop function if exists ${trigger}()`)); + vi.useRealTimers(); + } + }); + + it("lets expiry win before presentation and releases the child lock before wake admission", async () => { + const started = Date.now(); + vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(started); + try { + const f = await fixture(), outbox = await completion(f), lockErrors: unknown[] = []; + const work = nativeResponsePresentationWork(db), [entry] = await work.load(f.input); + vi.setSystemTime(started + 300_000); + const wakeup = vi.fn(async (agentId: string, options: { idempotencyKey: string }) => { + try { + await db.transaction(async tx => { + await tx.execute(sql`set local lock_timeout = '500ms'`); + await tx.select({ id: issues.id }).from(issues).where(eq(issues.id, f.issueId)).for("update"); + await tx.select({ id: issues.id }).from(issues).where(eq(issues.id, outbox.taskId)).for("update"); + }); + } catch (error) { lockErrors.push(error); throw error; } + return outbox.wakeup(agentId, options); + }); + await chatCompletionDeliveryService(db, { wakeup }).deliver(outbox.deliveryId); + expect(lockErrors).toEqual([]); + expect(wakeup).toHaveBeenCalledTimes(1); + expect(await outbox.delivery()).toMatchObject({ status: "pending", targetRunId: null, attempts: 1, responseCommentId: null }); + await expect(work.check(entry)).rejects.toThrow("chat_completion_superseded"); + expect(await comments(f.runId)).toEqual([]); + expect(await issue(outbox.taskId)).toMatchObject({ status: "done", statusVersion: 2 }); + } finally { vi.useRealTimers(); } + }); + + it("rolls back an expiry transition without waking or losing the original target", async () => { + const started = Date.now(); + vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(started); + try { + const f = await fixture(), outbox = await completion(f); + vi.setSystemTime(started + 300_000); + const rollback = new Error("fixture-expiry-rollback"); + const connection = new Proxy(db, { get(target, key) { + return key === "transaction" ? async (callback: Parameters[0]) => target.transaction(async tx => { + await callback(tx); throw rollback; + }) : Reflect.get(target, key); + } }); + await chatCompletionDeliveryService(connection, { wakeup: outbox.wakeup }).deliver(outbox.deliveryId); + expect(await outbox.delivery()).toMatchObject({ status: "queued", targetRunId: f.runId, attempts: 0, responseCommentId: null, + error: rollback.message }); + expect(outbox.wakeup).not.toHaveBeenCalled(); + expect(await comments(f.runId)).toEqual([]); + } finally { vi.useRealTimers(); } + }); + + it.each(["new-claim", "delivered", "retargeted", "superseded"])("respects a %s winner while expiry waits for the child lock", async winner => { + const started = Date.now(); + vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(started); + const f = await fixture(), outbox = await completion(f); + const replacementId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: replacementId, companyId: f.companyId, agentId: f.agentId, status: "queued" }); + let release!: () => void, acquired!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const locked = new Promise(resolve => { acquired = resolve; }); + let first: Promise | undefined, second: Promise | undefined; + const blocker = db.transaction(async tx => { + await tx.select({ id: issues.id }).from(issues).where(eq(issues.id, outbox.taskId)).for("update"); acquired(); await hold; + if (winner === "delivered") await tx.update(chatCompletionDeliveries).set({ status: "delivered" }).where(eq(chatCompletionDeliveries.id, outbox.deliveryId)); + if (winner === "retargeted") await tx.update(chatCompletionDeliveries).set({ targetRunId: replacementId, attempts: 1 }) + .where(eq(chatCompletionDeliveries.id, outbox.deliveryId)); + if (winner === "superseded") { + await tx.update(issues).set({ status: "todo", statusVersion: 3 }).where(eq(issues.id, outbox.taskId)); + await tx.update(chatCompletionDeliveries).set({ status: "superseded" }).where(eq(chatCompletionDeliveries.id, outbox.deliveryId)); + } + }); + const waiting = async (count: number) => vi.waitFor(async () => { + const [row] = await db.execute(sql`select count(*)::int as waiting from pg_stat_activity + where datname = current_database() and pid <> pg_backend_pid() and wait_event_type = 'Lock' + and query like '%from "issues"%' and query like '%for update%'`); + expect(row.waiting).toBeGreaterThanOrEqual(count); + }, { timeout: 5000 }); + try { + await Promise.race([locked, blocker]); + vi.setSystemTime(started + 300_000); + first = outbox.service.deliver(outbox.deliveryId); void first.catch(() => undefined); + await waiting(1); + if (winner === "new-claim") { + vi.setSystemTime(started + 360_000); + second = outbox.service.deliver(outbox.deliveryId); void second.catch(() => undefined); + await waiting(2); + } + release(); await blocker; await first; await second; + if (winner === "new-claim") { + expect(outbox.wakeup).toHaveBeenCalledTimes(1); + expect(await outbox.delivery()).toMatchObject({ status: "pending", targetRunId: null, attempts: 1 }); + } else { + expect(outbox.wakeup).not.toHaveBeenCalled(); + if (winner === "delivered") expect((await outbox.delivery()).status).toBe("delivered"); + if (winner === "retargeted") expect(await outbox.delivery()).toMatchObject({ status: "queued", targetRunId: replacementId, attempts: 1 }); + if (winner === "superseded") expect((await outbox.delivery()).status).toBe("superseded"); + } + expect(await comments(f.runId)).toEqual([]); + } finally { + release?.(); await Promise.allSettled([blocker, first, second].filter((value): value is Promise => Boolean(value))); + vi.useRealTimers(); + } + }); + + it("rechecks the fresh delivery when retry holds the child lock before presentation reads", async () => { + const started = Date.now(); + vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(started); + const f = await fixture(), outbox = await completion(f); + const suffix = randomUUID().replaceAll("-", ""), lockId = Number.parseInt(suffix.slice(0, 6), 16) + 1; + const trigger = `completion_retry_hold_${suffix}`; + let release!: () => void, acquired!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const locked = new Promise(resolve => { acquired = resolve; }); + let blocker: Promise | undefined, presentation: Promise | undefined, expiry: Promise | undefined; + try { + await db.execute(sql.raw(`create function ${trigger}() returns trigger language plpgsql as $$ + begin + if OLD.id = '${outbox.deliveryId}'::uuid and NEW.target_run_id is null then perform pg_advisory_xact_lock(${lockId}); end if; + return NEW; + end $$`)); + await db.execute(sql.raw(`create trigger ${trigger} before update of target_run_id on chat_completion_deliveries for each row execute function ${trigger}()`)); + blocker = db.transaction(async tx => { await tx.execute(sql`select pg_advisory_xact_lock(${lockId}::bigint)`); acquired(); await hold; }); + await Promise.race([locked, blocker]); + const work = nativeResponsePresentationWork(db), [entry] = await work.load(f.input); + vi.setSystemTime(started + 300_000); + expiry = outbox.service.deliver(outbox.deliveryId); void expiry.catch(() => undefined); + + + await vi.waitFor(async () => { + const [row] = await db.execute(sql`select exists (select 1 from pg_locks + where locktype = 'advisory' and classid = 0 and objid = ${lockId}::oid and not granted + and database = (select oid from pg_database where datname = current_database())) as waiting`); + expect(row.waiting).toBe(true); + }, { timeout: 5000 }); + presentation = work.check(entry); void presentation.catch(() => undefined); + await vi.waitFor(async () => { + const [row] = await db.execute(sql`select exists (select 1 from pg_stat_activity + where datname = current_database() and pid <> pg_backend_pid() and wait_event_type = 'Lock' + and query like '%from "issues"%' and query like '%for update%') as waiting`); + expect(row.waiting).toBe(true); + }, { timeout: 5000 }); + release(); await blocker; await expiry; + const [outcome] = await Promise.allSettled([presentation]); + expect(outcome.status).toBe("rejected"); + if (outcome.status === "rejected") expect(outcome.reason.message).toBe("chat_completion_superseded"); + expect(await comments(f.runId)).toEqual([]); + expect(await outbox.delivery()).toMatchObject({ status: "pending", targetRunId: null, attempts: 1, responseCommentId: null }); + expect(outbox.wakeup).toHaveBeenCalledTimes(1); + } finally { + release?.(); await Promise.allSettled([blocker, presentation, expiry].filter((value): value is Promise => Boolean(value))); + await db.execute(sql.raw(`drop trigger if exists ${trigger} on chat_completion_deliveries`)); + await db.execute(sql.raw(`drop function if exists ${trigger}()`)); + vi.useRealTimers(); + } + }); + + it.each(["within-grace", "at-expiry"])("preserves a Done reassignment's completion event %s and still retries at expiry", async when => { + const started = Date.now(); + vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(started); + try { + const f = await fixture(), outbox = await completion(f), reassignedAgent = randomUUID(); + await db.insert(agents).values({ id: reassignedAgent, companyId: f.companyId, name: "New task owner", adapterType: "paperclip_runner" }); + await db.update(nativeRunResults).set({ canonicalSha256: "invalid-digest" }).where(eq(nativeRunResults.runId, f.runId)); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(started + 120_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + const updated = await issueService(db).update(outbox.taskId, { assigneeAgentId: reassignedAgent }); + expect(updated).toMatchObject({ status: "done", statusVersion: 3, assigneeAgentId: reassignedAgent }); + expect(await outbox.delivery()).toMatchObject({ status: "queued", statusVersion: 2, targetRunId: f.runId, attempts: 0 }); + if (when === "within-grace") { + await outbox.service.deliver(outbox.deliveryId); + expect(await outbox.delivery()).toMatchObject({ status: "queued", targetRunId: f.runId, attempts: 0, + nextAttemptAt: new Date(started + 120_000) }); + expect(outbox.wakeup).not.toHaveBeenCalled(); + } + vi.setSystemTime(started + 300_000); + await outbox.service.deliver(outbox.deliveryId); + expect(await outbox.delivery()).toMatchObject({ status: "pending", targetRunId: null, attempts: 1, statusVersion: 2 }); + expect(await comments(f.runId)).toEqual([]); + expect(outbox.wakeup).toHaveBeenCalledTimes(1); + expect(await issue(outbox.taskId)).toMatchObject({ status: "done", statusVersion: 3, assigneeAgentId: reassignedAgent }); + } finally { vi.useRealTimers(); } + }); + + it("returns the fresh saved reply receipt after waiting for the child lock", async () => { + const f = await fixture(), outbox = await completion(f); + const [saved] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorAgentId: f.agentId, createdByRunId: f.runId, body: f.reply }).returning(); + let release!: () => void, acquired!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const locked = new Promise(resolve => { acquired = resolve; }); + const blocker = db.transaction(async tx => { + await tx.select({ id: issues.id }).from(issues).where(eq(issues.id, outbox.taskId)).for("update"); acquired(); await hold; + await tx.update(chatCompletionDeliveries).set({ status: "delivered", responseCommentId: saved.id }) + .where(eq(chatCompletionDeliveries.id, outbox.deliveryId)); + }); + let reader: Promise>> | undefined; + try { + await Promise.race([locked, blocker]); + reader = db.transaction(tx => existingChatCompletionReply(tx, f.runId, f.issueId)); void reader.catch(() => undefined); + await vi.waitFor(async () => { + const [row] = await db.execute(sql`select exists (select 1 from pg_stat_activity + where datname = current_database() and pid <> pg_backend_pid() and wait_event_type = 'Lock' + and query like '%from "issues"%' and query like '%for update%') as waiting`); + expect(row.waiting).toBe(true); + }, { timeout: 5000 }); + release(); await blocker; + expect(await reader).toMatchObject({ id: saved.id, body: f.reply, createdByRunId: f.runId }); + expect(await comments(f.runId)).toHaveLength(1); + } finally { release?.(); await Promise.allSettled([blocker, reader]); } + }); + + it("does not let an old hold shorten a newer claim's lease", async () => { + const started = Date.now(); + vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(started); + const f = await fixture(), outbox = await completion(f); + const notify = nativeResponseNotifications.notifyNativeResponsePresentation; + let release!: () => void, notified!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { notified = resolve; }); + const spy = vi.spyOn(nativeResponseNotifications, "notifyNativeResponsePresentation").mockImplementationOnce(async (tx, scope) => { + await notify(tx, scope); notified(); await hold; + }); + let old: Promise | undefined; + try { + + + old = outbox.service.deliver(outbox.deliveryId); void old.catch(() => undefined); + await vi.waitFor(() => expect(spy).toHaveBeenCalledTimes(1), { timeout: 5000 }); await ready; + vi.setSystemTime(started + 60_000); + await outbox.service.deliver(outbox.deliveryId); + expect(await outbox.delivery()).toMatchObject({ status: "queued", targetRunId: f.runId, attempts: 0, + nextAttemptAt: new Date(started + 120_000) }); + release(); await old; + expect(await outbox.delivery()).toMatchObject({ status: "queued", targetRunId: f.runId, attempts: 0, + nextAttemptAt: new Date(started + 120_000) }); + expect(outbox.wakeup).not.toHaveBeenCalled(); + expect(await comments(f.runId)).toEqual([]); + } finally { release?.(); await Promise.allSettled(old ? [old] : []); spy.mockRestore(); vi.useRealTimers(); } + }); + + it("saves a legitimate delayed reply before the fixed limit without another model run", async () => { + const started = Date.now(); + vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(started); + try { + const f = await fixture(), outbox = await completion(f); + for (const elapsed of [0, 60_000, 120_000, 240_000]) { + vi.setSystemTime(started + elapsed); + await outbox.service.deliver(outbox.deliveryId); + expect(await outbox.delivery()).toMatchObject({ status: "queued", targetRunId: f.runId, attempts: 0, responseCommentId: null }); + expect(outbox.wakeup).not.toHaveBeenCalled(); + } + vi.setSystemTime(started + 299_999); + const work = nativeResponsePresentationWork(db), [entry] = await work.load(f.input); + await expect(work.check(entry)).resolves.toBeNull(); + const [comment] = await comments(f.runId); + expect(comment).toMatchObject({ body: f.reply, authorAgentId: f.agentId, createdByRunId: f.runId }); + expect(await issue(f.issueId)).toMatchObject({ status: "in_review", conversationState: "waiting" }); + expect(await outbox.delivery()).toMatchObject({ status: "delivered", targetRunId: f.runId, attempts: 0, responseCommentId: comment.id }); + vi.setSystemTime(started + 300_000); + await outbox.service.deliver(outbox.deliveryId); + expect(await comments(f.runId)).toHaveLength(1); + expect(outbox.wakeup).not.toHaveBeenCalled(); + } finally { vi.useRealTimers(); } + }); + + it("clamps an external native retry deadline to the immutable five-lease presentation limit", async () => { + const started = Date.now(); + vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(started); + try { + const f = await fixture(), outbox = await completion(f); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(started + 86_400_000) }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + await outbox.service.deliver(outbox.deliveryId); + expect(await outbox.delivery()).toMatchObject({ status: "queued", targetRunId: f.runId, attempts: 0, + nextAttemptAt: new Date(started + 300_000) }); + expect(outbox.wakeup).not.toHaveBeenCalled(); + expect(await comments(f.runId)).toEqual([]); + vi.setSystemTime(started + 300_000); + await outbox.service.deliver(outbox.deliveryId); + expect(await outbox.delivery()).toMatchObject({ status: "pending", targetRunId: null, attempts: 1 }); + expect(outbox.wakeup).toHaveBeenCalledTimes(1); + } finally { vi.useRealTimers(); } + }); + + it.each(["digest", "external-context"])("bounds permanent %s declines without sliding, publishing or changing accepted work", async fault => { + const started = Date.now(); + vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(started); + try { + const f = await fixture(), outbox = await completion(f); + if (fault === "digest") await db.update(nativeRunResults).set({ canonicalSha256: "invalid-digest" }).where(eq(nativeRunResults.runId, f.runId)); + else { + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + await db.update(heartbeatRuns).set({ contextSnapshot: { ...run.contextSnapshot, externalChatContinuation: true } }).where(eq(heartbeatRuns.id, f.runId)); + } + const [acceptedBefore] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId)); + const work = nativeResponsePresentationWork(db), [entry] = await work.load(f.input); + for (const elapsed of [0, 60_000, 120_000, 240_000]) { + vi.setSystemTime(started + elapsed); + await outbox.service.deliver(outbox.deliveryId); + await expect(work.check(entry)).resolves.toBeNull(); + expect(await comments(f.runId)).toEqual([]); + expect(await outbox.delivery()).toMatchObject({ status: "queued", targetRunId: f.runId, attempts: 0, + responseCommentId: null, nextAttemptAt: new Date(started + Math.min(elapsed + 60_000, 300_000)) }); + expect(outbox.wakeup).not.toHaveBeenCalled(); + } + vi.setSystemTime(started + 299_999); + await outbox.service.deliver(outbox.deliveryId); + expect((await outbox.delivery()).nextAttemptAt).toEqual(new Date(started + 300_000)); + vi.setSystemTime(started + 300_000); + await outbox.service.deliver(outbox.deliveryId); + expect(await outbox.delivery()).toMatchObject({ status: "pending", targetRunId: null, attempts: 1, responseCommentId: null }); + expect(outbox.wakeup).toHaveBeenCalledTimes(1); + expect(await comments(f.runId)).toEqual([]); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId)))[0]).toEqual(acceptedBefore); + expect(await issue(outbox.taskId)).toMatchObject({ status: "done", statusVersion: 2 }); + } finally { vi.useRealTimers(); } + }); + it("finds a committed final Agent Chat turn, persists the exact reply once and only then settles waiting", async () => { const f = await fixture(); expect(await comments(f.runId)).toEqual([]);