diff --git a/src/gateway/server-startup-post-attach.test.ts b/src/gateway/server-startup-post-attach.test.ts index 81e1c105c1e2..a164887b99ef 100644 --- a/src/gateway/server-startup-post-attach.test.ts +++ b/src/gateway/server-startup-post-attach.test.ts @@ -254,15 +254,137 @@ vi.mock("./server-tailscale.js", () => ({ startGatewayTailscaleExposure: hoisted.startGatewayTailscaleExposure, })); -const { startGatewayPostAttachRuntime, startGatewaySidecars, testing } = - await import("./server-startup-post-attach.js"); +const { + startGatewayPostAttachRuntime: startGatewayPostAttachRuntimeImpl, + startGatewaySidecars: startGatewaySidecarsImpl, + testing, +} = await import("./server-startup-post-attach.js"); const { scheduleContextCachePrewarm } = await import("./server-startup-context-cache-prewarm.js"); const { STARTUP_UNAVAILABLE_GATEWAY_METHODS } = await import("./methods/core-descriptors.js"); const { createGatewayCloseHandler } = await import("./server-close.js"); const { createChatRunState } = await import("./server-chat-state.js"); -type PostAttachParams = Parameters[0]; -type PostAttachRuntimeDeps = NonNullable[1]>; +type PostAttachParams = Parameters[0]; +type PostAttachRuntimeDeps = NonNullable[1]>; +type SidecarPublisher = NonNullable; +type SidecarHandle = Parameters[0][number]; +type GatewaySidecarsResult = Awaited>; + +const publishedGatewayLifetimeSidecars = new Set(); +const publishedPostReadySidecars = new Set(); +const transferredSidecars = new Set(); + +function adoptSidecars(target: Set, sidecars: ReadonlyArray): void { + for (const sidecar of sidecars) { + if (!transferredSidecars.has(sidecar)) { + target.add(sidecar); + } + } +} + +function composeTrackedPublisher( + publishedSidecars: Set, + publisher: SidecarPublisher | undefined, +): SidecarPublisher { + return (sidecars) => { + adoptSidecars(publishedSidecars, sidecars); + publisher?.(sidecars); + }; +} + +function adoptPostReadyResult(result: GatewaySidecarsResult): GatewaySidecarsResult { + adoptSidecars(publishedPostReadySidecars, result.postReadySidecars); + return result; +} + +async function startGatewaySidecars( + ...args: Parameters +): Promise { + return adoptPostReadyResult(await startGatewaySidecarsImpl(...args)); +} + +function transferBeforeStop(sidecar: SidecarHandle): void { + publishedGatewayLifetimeSidecars.delete(sidecar); + publishedPostReadySidecars.delete(sidecar); + transferredSidecars.add(sidecar); +} + +async function stopTrackedSidecar(sidecar: SidecarHandle): Promise { + transferBeforeStop(sidecar); + await sidecar.stop(); +} + +function stopTrackedPostReadySidecarsAfterCloseStarted( + params: Parameters[0], +): void { + if (params.closeStarted) { + for (const sidecar of params.postReadySidecars) { + transferBeforeStop(sidecar); + } + } + testing.stopPostReadySidecarsAfterCloseStarted(params); +} + +async function cleanupGatewayTestState(): Promise { + let firstError: Error | undefined; + const cleanup = async (run: () => void | Promise) => { + try { + await run(); + } catch (error) { + firstError ??= error instanceof Error ? error : new Error(String(error)); + } + }; + + const lifetimeSidecars = [...publishedGatewayLifetimeSidecars]; + publishedGatewayLifetimeSidecars.clear(); + const postReadySidecars = [...publishedPostReadySidecars]; + publishedPostReadySidecars.clear(); + + for (const sidecar of lifetimeSidecars) { + transferredSidecars.add(sidecar); + await cleanup(() => sidecar.stop()); + } + for (const sidecar of postReadySidecars) { + transferredSidecars.add(sidecar); + await cleanup(() => sidecar.stop()); + } + + publishedGatewayLifetimeSidecars.clear(); + publishedPostReadySidecars.clear(); + transferredSidecars.clear(); + await cleanup(() => resetGatewayWorkAdmission()); + await cleanup(() => closeOpenClawStateDatabaseForTest()); + await cleanup(() => { + vi.useRealTimers(); + }); + await cleanup(() => { + vi.unstubAllEnvs(); + }); + + if (firstError !== undefined) { + throw firstError; + } +} + +function startGatewayPostAttachRuntime( + params: PostAttachParams, + runtimeDeps?: PostAttachRuntimeDeps, +) { + return startGatewayPostAttachRuntimeImpl( + { + ...params, + onGatewayLifetimeSidecars: composeTrackedPublisher( + publishedGatewayLifetimeSidecars, + params.onGatewayLifetimeSidecars, + ), + onPostReadySidecars: composeTrackedPublisher( + publishedPostReadySidecars, + params.onPostReadySidecars, + ), + }, + runtimeDeps, + ); +} async function waitForGatewayTestState( assertion: () => T | Promise, @@ -381,11 +503,46 @@ describe("startGatewayPostAttachRuntime", () => { hoisted.createTranscriptsAutoStartService.mockClear(); }); - afterEach(() => { - resetGatewayWorkAdmission(); - closeOpenClawStateDatabaseForTest(); - vi.useRealTimers(); - vi.unstubAllEnvs(); + afterEach(async () => { + await cleanupGatewayTestState(); + }); + + it("drains tracked sidecars and resets fixture state after the first cleanup failure", async () => { + const firstError = new Error("first cleanup failure"); + const stopOrder: string[] = []; + const firstLifetimeSidecar = { + stop: vi.fn(async () => { + stopOrder.push("lifetime:first"); + throw firstError; + }), + }; + const secondLifetimeSidecar = { + stop: vi.fn(async () => { + stopOrder.push("lifetime:second"); + }), + }; + const postReadySidecar = { + stop: vi.fn(async () => { + stopOrder.push("post-ready"); + }), + }; + const originalCleanupEnv = process.env.OPENCLAW_CLEANUP_TEST; + + adoptSidecars(publishedGatewayLifetimeSidecars, [firstLifetimeSidecar, secondLifetimeSidecar]); + adoptSidecars(publishedPostReadySidecars, [postReadySidecar]); + vi.useFakeTimers(); + vi.stubEnv("OPENCLAW_CLEANUP_TEST", "dirty"); + expect(tryBeginGatewayRootWorkAdmission()).not.toBeNull(); + + await expect(cleanupGatewayTestState()).rejects.toBe(firstError); + + expect(stopOrder).toEqual(["lifetime:first", "lifetime:second", "post-ready"]); + expect(publishedGatewayLifetimeSidecars.size).toBe(0); + expect(publishedPostReadySidecars.size).toBe(0); + expect(transferredSidecars.size).toBe(0); + expect(getActiveGatewayRootWorkCount()).toBe(0); + expect(vi.isFakeTimers()).toBe(false); + expect(process.env.OPENCLAW_CLEANUP_TEST).toBe(originalCleanupEnv); }); it("re-enables startup-gated methods after post-attach sidecars start", async () => { @@ -516,7 +673,7 @@ describe("startGatewayPostAttachRuntime", () => { expect(lifetimeSidecars).toContain(recoverySidecar); for (const sidecar of lifetimeSidecars ?? []) { - await sidecar.stop(); + await stopTrackedSidecar(sidecar); } expect(recoverySidecar.stop).toHaveBeenCalledOnce(); }); @@ -636,7 +793,7 @@ describe("startGatewayPostAttachRuntime", () => { await waitForGatewayTestState(() => { expect(getActiveGatewayRootWorkCount()).toBe(0); }); - await sidecar.stop(); + await stopTrackedSidecar(sidecar); }); it("cancels delayed restart sentinel recovery when the gateway closes", async () => { @@ -646,7 +803,7 @@ describe("startGatewayPostAttachRuntime", () => { log: { warn: vi.fn() }, }); - await sidecar.stop(); + await stopTrackedSidecar(sidecar); await vi.advanceTimersByTimeAsync(750); expect(hoisted.scheduleRestartSentinelWake).not.toHaveBeenCalled(); @@ -945,7 +1102,7 @@ describe("startGatewayPostAttachRuntime", () => { expect(startGatewaySidecarsPending).not.toHaveBeenCalled(); expect(buildSignal?.aborted).toBe(false); - await earlySidecar.stop(); + await stopTrackedSidecar(earlySidecar); expect(buildSignal?.aborted).toBe(true); expect(stopControlUiBuild).toHaveBeenCalledOnce(); @@ -955,6 +1112,10 @@ describe("startGatewayPostAttachRuntime", () => { expect(onGatewayLifetimeSidecars).toHaveBeenCalledTimes(2); expect(onGatewayLifetimeSidecars.mock.calls[1]?.[0]).toContain(earlySidecar); expect(startControlUiBuild).toHaveBeenCalledOnce(); + expect(publishedGatewayLifetimeSidecars).not.toContain(earlySidecar); + await cleanupGatewayTestState(); + expect(stopControlUiBuild).toHaveBeenCalledOnce(); + expect(publishedGatewayLifetimeSidecars).not.toContain(earlySidecar); }); it("loads startup plugins after bind and before channel sidecars", async () => { @@ -1289,7 +1450,7 @@ describe("startGatewayPostAttachRuntime", () => { }); } finally { admission.release(); - await sidecar.stop(); + await stopTrackedSidecar(sidecar); } }); @@ -1300,7 +1461,7 @@ describe("startGatewayPostAttachRuntime", () => { log: { warn: vi.fn() }, }); - await sidecar.stop(); + await stopTrackedSidecar(sidecar); await vi.runAllTimersAsync(); expect(hoisted.prewarmContextWindowCacheAfterReady).not.toHaveBeenCalled(); }); @@ -1351,7 +1512,7 @@ describe("startGatewayPostAttachRuntime", () => { expect(lifetimeSidecars).toHaveLength(4); for (const sidecar of gmailSidecars ?? []) { - sidecar.stop(); + await stopTrackedSidecar(sidecar); } await vi.dynamicImportSettled(); await waitForGatewayTestState(() => { @@ -1417,12 +1578,12 @@ describe("startGatewayPostAttachRuntime", () => { }); for (const sidecar of gmailSidecars ?? []) { - await sidecar.stop(); + await stopTrackedSidecar(sidecar); } expect(hoisted.transcriptsAutoStartService.stop).not.toHaveBeenCalled(); for (const sidecar of lifetimeSidecars ?? []) { - await sidecar.stop(); + await stopTrackedSidecar(sidecar); } expect(hoisted.transcriptsAutoStartService.stop).toHaveBeenCalledTimes(1); }); @@ -1443,7 +1604,7 @@ describe("startGatewayPostAttachRuntime", () => { expect(hoisted.setAuthProfileFailureHook).toHaveBeenCalledTimes(1); }); - await sidecar.stop(); + await stopTrackedSidecar(sidecar); await vi.advanceTimersByTimeAsync(1_000); expect(hoisted.warmCurrentProviderAuthStateOffMainThread).not.toHaveBeenCalled(); @@ -2060,24 +2221,28 @@ describe("startGatewayPostAttachRuntime", () => { it("stops post-ready sidecars registered after close started", () => { const postReadySidecar = { stop: vi.fn() }; + adoptSidecars(publishedPostReadySidecars, [postReadySidecar]); - testing.stopPostReadySidecarsAfterCloseStarted({ + stopTrackedPostReadySidecarsAfterCloseStarted({ postReadySidecars: [postReadySidecar], closeStarted: true, }); expect(postReadySidecar.stop).toHaveBeenCalledTimes(1); + expect(publishedPostReadySidecars).not.toContain(postReadySidecar); }); it("keeps post-ready sidecars running when close has not started", () => { const postReadySidecar = { stop: vi.fn() }; + adoptSidecars(publishedPostReadySidecars, [postReadySidecar]); - testing.stopPostReadySidecarsAfterCloseStarted({ + stopTrackedPostReadySidecarsAfterCloseStarted({ postReadySidecars: [postReadySidecar], closeStarted: false, }); expect(postReadySidecar.stop).not.toHaveBeenCalled(); + expect(publishedPostReadySidecars).toContain(postReadySidecar); }); it("runs Gmail watcher after sidecars are ready", async () => { @@ -2129,7 +2294,7 @@ describe("startGatewayPostAttachRuntime", () => { throw new Error("Expected gmail watcher resolver to be initialized"); } for (const sidecar of result.postReadySidecars) { - await sidecar.stop(); + await stopTrackedSidecar(sidecar); } expect(watcherSignal?.aborted).toBe(true); resolveWatcher(); @@ -2190,7 +2355,7 @@ describe("startGatewayPostAttachRuntime", () => { expect(result.postReadySidecars).toHaveLength(2); for (const sidecar of result.postReadySidecars) { - await sidecar.stop(); + await stopTrackedSidecar(sidecar); } await new Promise((resolve) => { setImmediate(resolve); @@ -2214,31 +2379,33 @@ describe("startGatewayPostAttachRuntime", () => { const { startGatewaySidecars: startGatewaySidecarsWithDelayedImport } = await import("./server-startup-post-attach.js"); - const result = await startGatewaySidecarsWithDelayedImport({ - cfg: { - hooks: { enabled: true, internal: { enabled: false }, gmail: { account: "me" } }, - } as never, - pluginRegistry: createPostAttachParams().pluginRegistry, - defaultWorkspaceDir: "/tmp/openclaw-workspace", - deps: {} as never, - startChannels: vi.fn(async () => {}), - log: { warn: vi.fn() }, - logHooks: { - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - }, - logChannels: { - info: vi.fn(), - error: vi.fn(), - }, - }); + const result = adoptPostReadyResult( + await startGatewaySidecarsWithDelayedImport({ + cfg: { + hooks: { enabled: true, internal: { enabled: false }, gmail: { account: "me" } }, + } as never, + pluginRegistry: createPostAttachParams().pluginRegistry, + defaultWorkspaceDir: "/tmp/openclaw-workspace", + deps: {} as never, + startChannels: vi.fn(async () => {}), + log: { warn: vi.fn() }, + logHooks: { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, + logChannels: { + info: vi.fn(), + error: vi.fn(), + }, + }), + ); await waitForGatewayTestState(() => { expect(releaseImport).toBeDefined(); }); for (const sidecar of result.postReadySidecars) { - await sidecar.stop(); + await stopTrackedSidecar(sidecar); } releaseImport?.(); await new Promise((resolve) => { @@ -2611,7 +2778,7 @@ describe("startGatewayPostAttachRuntime", () => { }); expect(result.postReadySidecars).toHaveLength(2); - testing.stopPostReadySidecarsAfterCloseStarted({ + stopTrackedPostReadySidecarsAfterCloseStarted({ postReadySidecars: result.postReadySidecars, closeStarted: true, });