mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-01 15:35:12 +01:00
fix: serialize native completion replies and bounded retries (#18502)
## Thinking Path > - Paperclip manages agents and their work. > - A delegated task reports its result to the source conversation through a durable completion outbox. > - Native execution can succeed before the presentation worker saves the accepted final reply. > - The outbox treated that short interval as a missing answer and cleared the target run. > - The presentation worker then rejected the valid answer because its delivery no longer belonged to that run. > - This change retains the existing target for one fixed presentation grace while accepted presentation is pending. > - The existing presentation owner can save the reply without another model run or more budget. ## Linked Issues or Issue Description **What happened?** A completion-delivery retry can run immediately after a native callback succeeds. It clears the callback's delivery target before the presentation worker saves the accepted answer. The reply then fails with `chat_completion_superseded`. A daily run limit correctly blocks the unnecessary replacement run, leaving the conversation active without its saved answer. **Expected behavior** The outbox must retain its current target and attempt while that exact native run has accepted presentation work. The existing presentation worker must save the answer once, acknowledge the delivery, and settle the conversation. Existing actor, session, result, digest, budget and retry checks must remain enforced. **Steps to reproduce** 1. Accept a native DONE result and current-turn final message for a completion callback. 2. Run its due completion-outbox handler after terminal projection but before reply presentation. 3. Run the existing presentation worker. Before this fix it rejects the reply because the outbox has cleared the target. **Paperclip version or commit** Reproduced on `b7f1c3769c00a107ffe4128b48b0ef8cc452cf3f` through the real PostgreSQL outbox and presentation services. The reported failure remains preserved. A new QA run is required after deployment; local tests do not qualify the deployed fix. Related: #17696 schedules accepted-response recovery. #18118 preserves canonical completion receipts. #18364 handles provider output and response-stage contracts. Those changes do not protect the completion-outbox target during this interval. ## What Changed - Defer a new execution attempt when the existing selector finds pending presentation for the same company, conversation, run and agent. - Retain the queued delivery target and attempt with an atomic compare-and-set. Await the existing scoped worker notification before that write in the same transaction. - Serialize retry with reply insertion under the existing child-task-first lock. Recheck locked task Done state and the delivery's stable completion-event version, status, target, attempt and exact lease. A delivered, retargeted or newer-claim winner prevents another wake. Admit the next attempt only after the winning transaction commits and releases its locks. - Limit presentation grace to five existing one-minute delivery leases, measured from the original run `finishedAt`. Clamp both the delivery lease and the native retry deadline to that immutable limit. Repeated sweeps cannot extend it. After expiry, resume the existing retry, exhaustion and daily-limit checks. - Keep publication authority in the existing materializer. The candidate query only defers a retry. Read the freshly reloaded delivery status, target and saved receipt after acquiring its child-task lock. - Fence both hold and retry writes with the exact claimed event version, attempt and lease. A stalled old hold cannot overwrite a newer lease. - Keep completion-event identity separate from mutable task ownership versions. Ordinary Done reassignment preserves its existing completion event; a real reopen retains normal supersession. - Add real PostgreSQL coverage for the observed ordering, outer commit and rollback, one saved reply, conversation settlement, idle SQL, retry boundaries and superseded deliveries. - Document the completion-outbox and presentation contract. ## Verification - Before the fix, the real PostgreSQL regression fails with `chat_completion_superseded` in `existingChatCompletionReply` when the outbox handles the terminal run first. After the fix, the same test passes without a new model wake. - Focused command: `vitest run --project @paperclipai/server server/src/services/native-runtime/agent-chat-response-recovery.test.ts server/src/__tests__/chat-completion-delivery.test.ts` — **95 pass** on `45f53a8c32fcca0bbdacf717286d37e30c1ba308`, using a disposable database on an owned local PostgreSQL 17 container. - Controls preserve retries for failed, cancelled, missing-final, missing-accepted-result, old-generation, foreign-company and foreign-agent targets. Reset or superseded task versions retain supersession. Digest and external-context declines never authorize a reply, never extend the fixed window, and leave it at exact expiry. An external native deadline cannot extend the limit. A valid reply at the last millisecond saves once without another model run. - The first reviewed head `b6f9c484b9567bc60af6d684c04e00058c1048ed` received Greptile 5/5 but an independent 3/5 review correctly found an unbounded hold. Three new real PostgreSQL controls failed on that source. This head adds the fixed bound and makes them pass. The earlier review, controls and CI remain retained; its review score does not qualify the new head. - The bounded predecessor `7300d2a978a5c4b3c13d85899bed3e98ae3c49fa` received an independent 4/5 review that correctly found an expiry-versus-in-flight-reply race. A real PostgreSQL advisory gate reproduced it after reply authority checked the delivery but before insertion. The retry lock and exact claim CAS now make that control pass. Both orderings, outer rollback, delivered/retargeted/superseded winners and competing lease claims pass. The next-wake test can acquire both parent and child locks, proving dispatch occurs after release. The valid 4/5 review and failing control remain retained. - At `39dd86e8f51d937cf382b2757736e0477201d21d`, both reviews found valid remaining defects: the reply helper still used its pre-lock delivery snapshot, and ordinary Done reassignment could stall a delivery by changing the task version. Four real PostgreSQL controls fail on that source and pass here: concurrent retry-first, a newly saved receipt after waiting, and reassignment within grace or at expiry. A real reopen through the issue service still supersedes the old event. - A later coordinator review found the hold CAS still lacked the attempt and lease fences already present on retry. A real PostgreSQL producer barrier reproduced an old hold shortening a newer claim from 120 seconds to 60 seconds. The same exact claim fences now prevent that overwrite. This failed control is retained; no old run was blindly retried. - The real worker saves one exact run-owned reply and acknowledges the delivery only after outer commit. An outer rollback publishes no reply. After settlement, advancing the scheduler clock by 60 seconds causes no further SQL. - Module boundaries, map structure, `git diff --check` and the canonical server build (including TypeScript and Runner vendor preparation) pass. The first local build stopped in the host package-manager setup; the installed pinned pnpm completed the build. Full repository typecheck, test and build gates pass in current-head CI; they have not all been run locally. - Final local checks also caught a test cleanup type error (TS2677) and one 94/95 readiness failure: the worker saves its comment before post-commit conversation settlement finishes. The test-only follow-up removes an invalid cleanup type predicate and waits on the existing bounded worker-quiet check before reading final state. All assertions and the same five-second bound remain. Source `45f53a8c32fcca0bbdacf717286d37e30c1ba308` passes all 95 cases and the canonical server build. The earlier failures are retained. - The final policy cleanup removes only 15 newly added narrative comment lines (seven service, eight test), as required by CONTRIBUTING.md. Both TypeScript AST signatures and `removeComments=true` emitted JavaScript are identical to tested source `45f53a8c32`. This was checked with the TypeScript 6.0.3 API and the installed native TypeScript 7.0.2 compiler. Assertions, timeouts and behavior are unchanged; the 95 tests were not repeated for identical compiled code. - The superseded `45f53a8c32` CI run recorded two 15-second timeouts in attention-retention cases. That failure remains retained. The test and its ten inspected owner or support files are identical on the tested master revisions. The canonical base run passes all 53 attention cases, and the comparison PR passes the same server lane, including those two cases in 9.3 and 9.8 seconds. This does not establish the timeout cause. Current-head CI is a new run after the required comment cleanup, not a retry of the failed source. - The optional Paperclip review bot ended neutral without a completed assessment because its execution paused. That result is not an approval. A separate [independent read-only audit](https://github.com/paperclipai/paperclip/pull/18502#issuecomment-6550216227) reports 5/5 confidence on exact head `9cfd3f74c69cc7a2da7b4ee5d1fca80f15d4af0c`, with no findings. Its manual assessment does not replace or relabel the incomplete bot result. - Current head `9cfd3f74c69cc7a2da7b4ee5d1fca80f15d4af0c` has a fresh Greptile 5/5 review with no open threads. [CI run 41249007430](https://github.com/paperclipai/paperclip/actions/runs/41249007430) passes all primary jobs and both required gates (`ci / verify` and `ci / e2e`). All 59 PR contexts are terminal, with no failed contexts. The actual CI checkout is `18db9e2750d6a799f26b5b8ba31bcccbea254227`, which combines master `681f7b77262a9df6c1b4aad55291bf960061e491` with this exact head. Fresh QA verification on the deployed application remains required before claiming the live failure is fixed. ## Risks A candidate can retain a delivery whose stricter presentation checks still prevent publication, but only within the fixed five-minute window. The existing worker keeps publication authority. If a valid reply remains unavailable at expiry, the normal callback retry may require another admitted model run; the original result and completed child work remain preserved. Existing retry, exhaustion, model-run and spending limits still apply. Failed or cancelled execution and genuinely missing accepted output retain the existing retry path. ## Model Used OpenAI GPT-6 through Codex, with repository inspection, tool use and code execution. The runtime does not expose the exact hosted revision or context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
681f7b7726
commit
b889ac39b7
3 files changed
+589
-20
No files matched your search
@@ -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.
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<ReturnType<typeof fixture>>) {
|
||||
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<ReturnType<typeof fixture>> | 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<void>(resolve => { release = resolve; });
|
||||
const locked = new Promise<void>(resolve => { acquired = resolve; });
|
||||
let blocker: Promise<unknown> | undefined, presentation: Promise<unknown> | undefined, expiry: Promise<unknown> | 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<unknown> => 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<Db["transaction"]>[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<void>(resolve => { release = resolve; });
|
||||
const locked = new Promise<void>(resolve => { acquired = resolve; });
|
||||
let first: Promise<unknown> | undefined, second: Promise<unknown> | 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<unknown> => 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<void>(resolve => { release = resolve; });
|
||||
const locked = new Promise<void>(resolve => { acquired = resolve; });
|
||||
let blocker: Promise<unknown> | undefined, presentation: Promise<unknown> | undefined, expiry: Promise<unknown> | 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<unknown> => 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<void>(resolve => { release = resolve; });
|
||||
const locked = new Promise<void>(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<Awaited<ReturnType<typeof existingChatCompletionReply>>> | 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<void>(resolve => { release = resolve; });
|
||||
const ready = new Promise<void>(resolve => { notified = resolve; });
|
||||
const spy = vi.spyOn(nativeResponseNotifications, "notifyNativeResponsePresentation").mockImplementationOnce(async (tx, scope) => {
|
||||
await notify(tx, scope); notified(); await hold;
|
||||
});
|
||||
let old: Promise<unknown> | 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([]);
|
||||
|
||||
Reference in new issue
Block a user