test: await async outcomes and drive saved deadlines deterministically (#18416)

Reviewed and verified at ab0b15d234.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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<void>(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 });
   });
 });

server/src/__tests__/skill-version-retention.test.ts

@@ -83,6 +83,7 @@ describeEmbeddedPostgres("skill version snapshot retention", () => {
   let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
   let worker: ReturnType<typeof registerSkillVersionRetentionWork> | undefined;
   let admitted = true;
+  const passes = new Set<Promise<unknown>>();
   afterEach(async () => {
     worker?.unsubscribe();
     await coordinator?.stop();
@@ -160,12 +161,20 @@ describeEmbeddedPostgres("skill version snapshot retention", () => {
   async function startRetention(
     graceMs = 20_000,
     service = skillVersionRetentionService(db, { graceMs }),
-    advanceAutomatically = true,
+    clock: boolean | { shouldAdvanceTime?: boolean; now?: number } = true,
   ) {
+    const clockOptions = typeof clock === "boolean" ? { shouldAdvanceTime: clock } : clock;
+    const advanceAutomatically = clockOptions.shouldAdvanceTime ?? true;
     await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select 1`)));
-    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: advanceAutomatically });
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: advanceAutomatically, ...clockOptions });
     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);
     if (advanceAutomatically) await worker.ready;
     else await waitForIo(worker.ready);
@@ -194,6 +203,7 @@ describeEmbeddedPostgres("skill version snapshot retention", () => {
     for (;;) {
       if (process.hrtime.bigint() >= deadline) throw new Error("Retention work did not settle within 10000ms");
       if (idleWorkSnapshot().active > 0) {
+        if (passes.size) await waitForIo(Promise.all([...passes]));
         await vi.advanceTimersByTimeAsync(0);
         await realDelay(0);
         continue;
@@ -631,7 +641,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}`));
@@ -642,7 +651,13 @@ describeEmbeddedPostgres("skill version snapshot retention", () => {
       if (id === pagedCompany) pageTimes.push(Date.now());
       return prune(id, options);
     };
-    await startRetention(5000, service, false); await settledAtDeadlines();
+    const expiring = await writeSnapshot(dueCompany, dueSkill.skillId, dueSkill.versionIds[0]!, 0);
+    const accessedAt = (await lastUsedAt(skillVersionCachePaths(dueCompany, dueSkill.skillId, dueSkill.versionIds[0]!).access))!;
+    await 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.
+    await settledAtDeadlines();
     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();

tests/e2e/agent-chat-projects.spec.ts

@@ -459,12 +459,15 @@ test("plan approval hands the preserved revision to an assigned project task", a
         { timeout: 60_000 },
       )
       .toBe(1);
+    // Task completion is delivered back to the chat as another real turn.
+    await idle(request, f.chatPath, 3);
     const task = (
       await json(await request.get(`/api/companies/${f.company.id}/issues`))
     )[0];
     expect(task.parentId).toBeNull();
     expect(task.projectId).toBeTruthy();
     expect(task.assigneeAgentId).toBe(f.agent.id);
+    await expect(page.getByText(`Delegated work completed: ${task.identifier}`, { exact: true })).toBeVisible();
     const plan = await json(
       await request.get(`/api/issues/${task.id}/documents/plan`),
     );

tests/e2e/fixtures/agent-chat.mjs

@@ -44,6 +44,10 @@ if (!task.conversationAgentId) {
   });
   process.exit(0);
 }
+if (ctx.wakeReason === "chat_task_completed") {
+  await comment(`Delegated work completed: ${ctx.chatCompletionUpdates.map((update) => update.identifier).join(", ")}`);
+  process.exit(0);
+}
 const comments = await api(`/issues/${task.id}/comments?order=asc`);
 const current =
   comments.find((c) => c.id === ctx.wakeCommentId) ??

ui/src/hooks/useTaskPluginPanels.test.tsx

@@ -143,9 +143,17 @@ it("discovers a panel started after the task is already open, then hides it on s
   get.mockRejectedValue(new Error("Access denied")); await refresh(); expect(current).toEqual([]);
 });
 it("clears automatic activation when the run ends even if its view is still ready", async () => {
-  await render(); expect(current).not.toEqual([]);
+  await render();
+  await vi.waitFor(async () => {
+    await act(async () => {});
+    expect(current).not.toEqual([]);
+  });
   await act(async () => { client.setQueryData(queryKeys.issues.liveRuns("task"), []); });
-  await render(); expect(current).toEqual([]);
+  await render();
+  await vi.waitFor(async () => {
+    await act(async () => {});
+    expect(current).toEqual([]);
+  });
 });
 it("does not carry an old task's ready panel to a new task or disabled plugin", async () => {
   await render(); expect(current).not.toEqual([]);