From 95c8491e2e33c6824cca2974ce039f06ca7b3e8d Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 26 Oct 2026 14:03:51 +0700 Subject: [PATCH] test(server): await finish cleanup and expose completion persistence errors (#7909) --- .../finish-subscriptions.test.ts | 8 +-- .../server/connected-hosts/hosts.e2e.test.ts | 57 ++++++++++++++++++- 2 files changed, 58 insertions(+), 7 deletions(-) diff --git a/packages/server/src/server/agent/finish-subscriptions/finish-subscriptions.test.ts b/packages/server/src/server/agent/finish-subscriptions/finish-subscriptions.test.ts index a4b4514add..323e4be48f 100644 --- a/packages/server/src/server/agent/finish-subscriptions/finish-subscriptions.test.ts +++ b/packages/server/src/server/agent/finish-subscriptions/finish-subscriptions.test.ts @@ -68,7 +68,7 @@ afterEach(async () => { for (const service of services.splice(0)) await service.close(); await Promise.all(homes.splice(0).map((home) => rm(home, { recursive: true, force: true }))); }); -async function scenario(directory?: string, removalFailures = 0) { +async function scenario(directory?: string, removalFailures = 0, retryDelayMs = 1) { const home = directory ?? (await mkdtemp(join(tmpdir(), "finish-recovery-"))); if (!directory) homes.push(home); const durableStore = createJsonFinishSubscriptionStore(home); @@ -92,7 +92,7 @@ async function scenario(directory?: string, removalFailures = 0) { callerIsActive: async () => active, serverId: "origin", logger: createTestLogger(), - retryDelayMs: 1, + retryDelayMs, }); services.push(service); await service.start(); @@ -388,13 +388,13 @@ test("removed-host notice retries parent dispatch without observing again or del }); test("failed removal after dispatch acknowledgment retries only storage, preserving the accepted outcome", async () => { - const state = await scenario(undefined, 1); + const state = await scenario(undefined, 2, 1000); await state.service.register(intent); await until(() => state.observation.waits.length === 1); state.observation.waits[0]!.result.resolve(outcome("idle")); await until(() => state.delivery.prompts.length === 1); state.delivery.acknowledgments[0]!.resolve("steered"); - await expect.poll(async () => (await state.store.list()).length).toBe(0); + await expect.poll(async () => (await state.store.list()).length, { timeout: 10000 }).toBe(0); expect(state.delivery.prompts).toHaveLength(1); expect(state.delivery.prompts[0]).toContain("finished."); expect(state.observation.waits).toHaveLength(1); diff --git a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts index 8a2d4f9c39..d8d5533cf6 100644 --- a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts +++ b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts @@ -1,4 +1,5 @@ import { z } from "zod"; +import pino from "pino"; import { createTestAgentClients } from "../test-utils/fake-agent-client.js"; import { Client } from "@modelcontextprotocol/sdk/client/index.js"; import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; @@ -9,6 +10,47 @@ import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo import type { HostDirectory } from "@getpaseo/protocol/hosts"; const cleanup: (() => Promise)[] = []; + +// These asynchronous failures otherwise disappear into the fixture's silent logger. +// Only completion errors and filesystem metadata belong in CI, never prompts or credentials. +function completionLogger() { + return pino( + { + level: "warn", + serializers: { + err: (error) => ({ + type: error.name, + code: error.code, + syscall: error.syscall, + path: typeof error.path === "string" ? basename(error.path) : undefined, + dest: typeof error.dest === "string" ? basename(error.dest) : undefined, + }), + }, + }, + { + write(line) { + const record = JSON.parse(line); + if ( + record.context !== "Failed to dispatch queued agent message" && + ![ + "Agent message batch dispatch failed", + "Retrying finish subscription", + "Finish subscription failed", + ].includes(record.msg) + ) + return; + console.error("Completion diagnostic", { + time: record.time, + message: record.msg, + agentId: record.agentId, + messageIds: record.messageIds, + subscription: record.subscription, + error: record.err, + }); + }, + }, + ); +} afterEach(async () => { for (const close of cleanup.splice(0).toReversed()) await close(); }); @@ -1512,6 +1554,7 @@ test( finishChild.resolve(); }); const a = await daemon({ + logger: completionLogger(), agentClients: createTestAgentClients({ onStartTurn: (prompt, config) => { if (config.title === "Batch parent") @@ -1635,7 +1678,12 @@ test( expect(parentPrompts[0]).toContain("LOCAL_CLI_REPORT"); expect(parentPrompts[0]).toContain('event="finished"'); expect(parentPrompts[0]).toContain(`source-agent-id="${child.id}"`); - await expect.poll(receiptStates, { timeout: 10000 }).toEqual(["completed", "completed"]); + await expect + .poll(receiptStates, { + timeout: 10000, + message: "Parent accepted the batch; waiting for durable receipt commits", + }) + .toEqual(["completed", "completed"]); const timeline = await origin.client.fetchAgentTimeline(parent.id); const messages = timeline.entries.flatMap((row) => row.item.type === "tool_call" && row.item.agentMessage ? [row.item.agentMessage] : [], @@ -1957,6 +2005,7 @@ test.each(["removed", "disabled"] as const)( cleanup.push(async () => finish.resolve()); const a = await daemon({ mcpEnabled: true, + logger: completionLogger(), agentClients: createTestAgentClients({ onStartTurn: (prompt) => { if (String(prompt).includes('kind="agent-notification"')) @@ -2015,8 +2064,10 @@ test.each(["removed", "disabled"] as const)( expect(parentPrompts[0]).toContain("is unknown."); expect(parentPrompts[0]).not.toContain("finished."); await expect - .poll(async () => - JSON.parse(await readFile(join(a.paseoHome, "finish-subscriptions.json"), "utf8")), + .poll( + async () => + JSON.parse(await readFile(join(a.paseoHome, "finish-subscriptions.json"), "utf8")), + { timeout: 10000 }, ) .toEqual([]); if (availability === "removed")