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 }); }); }); diff --git a/server/src/__tests__/skill-version-retention.test.ts b/server/src/__tests__/skill-version-retention.test.ts index a84adff3042..d88985b23d3 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(); @@ -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(); diff --git a/tests/e2e/agent-chat-projects.spec.ts b/tests/e2e/agent-chat-projects.spec.ts index 994df38746c..89ec94ca54c 100644 --- a/tests/e2e/agent-chat-projects.spec.ts +++ b/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`), ); diff --git a/tests/e2e/fixtures/agent-chat.mjs b/tests/e2e/fixtures/agent-chat.mjs index acc5b25c789..2cbb19557a3 100644 --- a/tests/e2e/fixtures/agent-chat.mjs +++ b/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) ?? diff --git a/ui/src/hooks/useTaskPluginPanels.test.tsx b/ui/src/hooks/useTaskPluginPanels.test.tsx index c013137a057..bec7ca5b66c 100644 --- a/ui/src/hooks/useTaskPluginPanels.test.tsx +++ b/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([]);