From 97dcec42262e666bd1db42c0a31e9028d4102409 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 22:42:23 -0500 Subject: [PATCH 1/2] test: await resolved interaction execution before outcome assertions Observe the target adapter-start event, then await its existing run-specific completion barrier. Preserve succeeded, execution-count and review-promotion assertions, including deliberately slow completion beyond the old terminal poll. Co-Authored-By: Paperclip --- ...heartbeat-stale-queue-invalidation.test.ts | 25 +++++++++++++++---- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts index 65191c24d2f..f1447508ed5 100644 --- a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts +++ b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts @@ -1796,8 +1796,22 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { const interactionId = randomUUID(); const wakeCommentId = randomUUID(); let claimedIssue: { status: string; executionRunId: string | null } | null = null; + let resumedRunId: string | undefined; + let signalAdapterStarted!: () => void; + const adapterStarted = new Promise(resolve => { signalAdapterStarted = resolve; }); + const execute = mockAdapterExecute.getMockImplementation()!; + mockAdapterExecute.mockImplementation(async (...args: unknown[]) => { + if ((args[0] as { runId: string }).runId === resumedRunId) { + signalAdapterStarted(); + // A claimed continuation can finish after the old terminal poll. + if (interactionStatus === "rejected" && interactionKind === "connection_intent") { + await new Promise(resolve => setTimeout(resolve, 3100)); + } + } + return execute(); + }); afterContinuationDispatchCheck = async ({ runId: checkedRunId, issueId: checkedIssueId }) => { - if (checkedIssueId !== issueId) return; + if (checkedIssueId !== issueId || checkedRunId !== resumedRunId) return; claimedIssue = await db .select({ status: issues.status, executionRunId: issues.executionRunId }) .from(issues) @@ -1831,13 +1845,14 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { contextExtras: { interactionId, interactionKind, interactionStatus, wakeCommentId, originCommentIds: [wakeCommentId], interactionResolvedAt: new Date().toISOString(), mutation: "interaction", source: `${interactionKind}.resolved`, forceFreshSession: true } }); + resumedRunId = runId; await heartbeat.resumeQueuedRuns(); - expect(await waitForCondition(async () => (await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]?.status === "succeeded")).toBe(true); - // Terminal status precedes completion bookkeeping. Drain those writes before - // afterEach truncates the fixture, otherwise PostgreSQL can deadlock. + // Adapter entry guarantees this exact execution is registered. Await its + // owning completion barrier before terminal assertions and fixture cleanup. + await adapterStarted; await heartbeat.waitForRunExecutionDrain(runId); + expect((await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]?.status).toBe("succeeded"); expect(countExecuteCallsForRun(runId)).toBe(1); - await waitForCondition(async () => claimedIssue !== null); expect(claimedIssue).toEqual({ status: "in_progress", executionRunId: runId }); }); }); From 21aed2b3e490d396554189766410a50816b921e4 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 22:53:14 -0500 Subject: [PATCH 2/2] test: measure retention page fairness on a controlled clock Keep virtual time fixed while actual filesystem and database work settles, then advance the coordinator deadline. Preserve the three-second page-gap limit and snapshot grace assertions; starvation mutation remains detected. Co-Authored-By: Paperclip --- .../__tests__/skill-version-retention.test.ts | 34 ++++++++++++++++--- 1 file changed, 30 insertions(+), 4 deletions(-) diff --git a/server/src/__tests__/skill-version-retention.test.ts b/server/src/__tests__/skill-version-retention.test.ts index bb757c7b6a6..ac56833140e 100644 --- a/server/src/__tests__/skill-version-retention.test.ts +++ b/server/src/__tests__/skill-version-retention.test.ts @@ -83,6 +83,7 @@ describeEmbeddedPostgres("skill version snapshot retention", () => { let coordinator: ReturnType | undefined; let worker: ReturnType | undefined; let admitted = true; + const passes = new Set>(); afterEach(async () => { worker?.unsubscribe(); await coordinator?.stop(); @@ -157,11 +158,17 @@ describeEmbeddedPostgres("skill version snapshot retention", () => { return dir; } - async function startRetention(graceMs = 20_000, service = skillVersionRetentionService(db, { graceMs })) { + async function startRetention(graceMs = 20_000, service = skillVersionRetentionService(db, { graceMs }), clock: { shouldAdvanceTime?: boolean; now?: number } = {}) { await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select 1`))); - vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true, ...clock }); coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => true, onError: error => { throw error; } }); + const register = coordinator.register.bind(coordinator); + coordinator.register = (queue, task) => register(queue, { ...task, run: signal => { + const pass = task.run(signal); passes.add(pass); + void pass.then(() => passes.delete(pass), () => passes.delete(pass)); + return pass; + } }); worker = registerSkillVersionRetentionWork(coordinator, service, () => admitted); await worker.ready; return service; @@ -577,7 +584,6 @@ describeEmbeddedPostgres("skill version snapshot retention", () => { it("keeps directory pages moving while another company's grace deadline is due", async () => { const dueCompany = await createCompany(), pagedCompany = await createCompany(); const dueSkill = await createSkill(dueCompany, 2), pagedSkill = await createSkill(pagedCompany, 1); - const expiring = await writeSnapshot(dueCompany, dueSkill.skillId, dueSkill.versionIds[0]!, 0); const parent = path.join(resolveCompanySkillVersionsRoot(pagedCompany), pagedSkill.skillId); await fs.mkdir(parent, { recursive: true }); for (let index = 0; index < 410; index++) await fs.mkdir(path.join(parent, `operator-${index}`)); @@ -588,7 +594,27 @@ describeEmbeddedPostgres("skill version snapshot retention", () => { if (id === pagedCompany) pageTimes.push(Date.now()); return prune(id, options); }; - await startRetention(5000, service); await settled(); + const expiring = await writeSnapshot(dueCompany, dueSkill.skillId, dueSkill.versionIds[0]!, 0); + const completeIo = async (work: Promise) => { + let done = false; + void work.then(() => { done = true; }, () => { done = true; }); + while (!done) { + if (vi.isFakeTimers()) await vi.advanceTimersByTimeAsync(0); + await realDelay(0); + } + await work; + }; + const accessedAt = (await lastUsedAt(skillVersionCachePaths(dueCompany, dueSkill.skillId, dueSkill.versionIds[0]!).access))!; + await completeIo(startRetention(5000, service, { shouldAdvanceTime: false, now: accessedAt })); + expect(await exists(expiring)).toBe(true); + // Keep virtual time fixed during real I/O; measure scheduler fairness, + // independent of filesystem latency and CI CPU scheduling. + while (coordinator!.nextWakeAt() !== null) { + await vi.advanceTimersByTimeAsync(Math.max(0, coordinator!.nextWakeAt()! - Date.now())); + await completeIo(Promise.all([...passes])); + await realDelay(0); + expect(idleWorkSnapshot().active).toBe(0); + } expect(pageTimes.length).toBeGreaterThan(5); expect(Math.max(...pageTimes.slice(1).map((at, index) => at - pageTimes[index]!))).toBeLessThan(3000); expect(await exists(expiring)).toBe(false); await quiet();