diff --git a/extensions/browser/plugin-registration.ts b/extensions/browser/plugin-registration.ts index 31e2b3edc84a..0dc5ce4b7093 100644 --- a/extensions/browser/plugin-registration.ts +++ b/extensions/browser/plugin-registration.ts @@ -205,8 +205,9 @@ function createLazyBrowserPluginService(): OpenClawPluginService { let service: OpenClawPluginService | null = null; const loadService = async () => { if (!service) { - const { createBrowserPluginService } = await loadBrowserRegistrationRuntimeModule(); - service = createBrowserPluginService(); + const { createBrowserPluginService, stopBrowserControlService } = + await loadBrowserRegistrationRuntimeModule(); + service = createBrowserPluginService({ stopOnDemand: stopBrowserControlService }); } return service; }; @@ -221,7 +222,11 @@ function createLazyBrowserPluginService(): OpenClawPluginService { }, stop: async (ctx) => { if (!service) { - const { stopBrowserControlService } = await import("./src/control-service.js"); + const loadedRuntime = loadBrowserRegistrationRuntimeModule.peek(); + if (!loadedRuntime) { + return; + } + const { stopBrowserControlService } = await loadedRuntime; await stopBrowserControlService(); return; } diff --git a/extensions/browser/register.runtime.ts b/extensions/browser/register.runtime.ts index e33d843dc77c..a65baac6a603 100644 --- a/extensions/browser/register.runtime.ts +++ b/extensions/browser/register.runtime.ts @@ -7,4 +7,5 @@ export { ensureBrowserProxyUploadCleanup } from "./src/browser-proxy-upload.js"; export { handleBrowserGatewayRequest } from "./src/gateway/browser-request.js"; export { runBrowserProxyCommand } from "./src/node-host/invoke-browser.js"; export { createBrowserPluginService } from "./src/plugin-service.js"; +export { stopBrowserControlService } from "./src/control-service.js"; export { collectBrowserSecurityAuditFindings } from "./src/security-audit.js"; diff --git a/extensions/browser/src/plugin-service.test.ts b/extensions/browser/src/plugin-service.test.ts index 063d91fda595..080022d3070a 100644 --- a/extensions/browser/src/plugin-service.test.ts +++ b/extensions/browser/src/plugin-service.test.ts @@ -52,8 +52,11 @@ describe("createBrowserPluginService", () => { return { validateOverrideSpecifier: params.validateOverrideSpecifier }; } + const createService = () => + createBrowserPluginService({ stopOnDemand: runtimeMocks.stopBrowserControlService }); + it("does not start the control server during gateway startup by default", async () => { - const service = createBrowserPluginService(); + const service = createService(); await service.start(SERVICE_CONTEXT); @@ -63,7 +66,7 @@ describe("createBrowserPluginService", () => { for (const value of ["0", "", "disabled"]) { it(`does not start the control server for eager env value ${JSON.stringify(value)}`, async () => { vi.stubEnv("OPENCLAW_EAGER_BROWSER_CONTROL_SERVER", value); - const service = createBrowserPluginService(); + const service = createService(); await service.start(SERVICE_CONTEXT); @@ -73,7 +76,7 @@ describe("createBrowserPluginService", () => { it("passes a browser override validator to the eager service loader", async () => { vi.stubEnv("OPENCLAW_EAGER_BROWSER_CONTROL_SERVER", "1"); - const service = createBrowserPluginService(); + const service = createService(); await service.start(SERVICE_CONTEXT); @@ -83,7 +86,7 @@ describe("createBrowserPluginService", () => { it("rejects unsafe browser override specifiers", async () => { vi.stubEnv("OPENCLAW_EAGER_BROWSER_CONTROL_SERVER", "1"); - const service = createBrowserPluginService(); + const service = createService(); await service.start(SERVICE_CONTEXT); @@ -100,7 +103,7 @@ describe("createBrowserPluginService", () => { }); it("stops an on-demand browser runtime even when startup stayed lazy", async () => { - const service = createBrowserPluginService(); + const service = createService(); await service.stop?.(SERVICE_CONTEXT); @@ -109,7 +112,7 @@ describe("createBrowserPluginService", () => { it("propagates on-demand cleanup failures", async () => { runtimeMocks.stopBrowserControlService.mockRejectedValueOnce(new Error("cleanup failed")); - const service = createBrowserPluginService(); + const service = createService(); await expect(service.stop?.(SERVICE_CONTEXT)).rejects.toThrow("cleanup failed"); }); @@ -121,7 +124,7 @@ describe("createBrowserPluginService", () => { .mockRejectedValueOnce(new Error("loaded cleanup failed")) .mockResolvedValue(undefined); runtimeMocks.startLazyPluginServiceModule.mockResolvedValue({ stop } as never); - const service = createBrowserPluginService(); + const service = createService(); await service.start(SERVICE_CONTEXT); await expect(service.stop?.(SERVICE_CONTEXT)).rejects.toThrow("loaded cleanup failed"); diff --git a/extensions/browser/src/plugin-service.ts b/extensions/browser/src/plugin-service.ts index 1df200c67048..5bfd5596b90a 100644 --- a/extensions/browser/src/plugin-service.ts +++ b/extensions/browser/src/plugin-service.ts @@ -21,7 +21,9 @@ function validateBrowserControlOverrideSpecifier(specifier: string): string { } /** Creates the Browser plugin service registered by the plugin entrypoint. */ -export function createBrowserPluginService(): OpenClawPluginService { +export function createBrowserPluginService(params: { + stopOnDemand: () => Promise; +}): OpenClawPluginService { let handle: BrowserControlHandle = null; return { @@ -55,8 +57,7 @@ export function createBrowserPluginService(): OpenClawPluginService { } return; } - const { stopBrowserControlService } = await import("./control-service.js"); - await stopBrowserControlService(); + await params.stopOnDemand(); }, }; } diff --git a/src/cli/gateway-cli/run-loop.test.ts b/src/cli/gateway-cli/run-loop.test.ts index 1659a72a0111..b18672b167ab 100644 --- a/src/cli/gateway-cli/run-loop.test.ts +++ b/src/cli/gateway-cli/run-loop.test.ts @@ -529,6 +529,27 @@ describe("runGatewayLoop", () => { }); }); + it("names and canonically formats a gateway close failure", async () => { + vi.clearAllMocks(); + + await withIsolatedSignals(async ({ captureSignal }) => { + const close = vi.fn(async () => { + throw new TypeError("close owner failed"); + }); + const { start, started } = createSignaledStart(close); + const { runtime, exited } = createRuntimeWithExitSignal(); + await runLoopWithStart({ start, runtime }); + await waitForStart(started); + + captureSignal("SIGTERM")(); + + await expect(exited).resolves.toBe(0); + expect(gatewayLog.error).toHaveBeenCalledWith( + "shutdown step failed (gateway server close): close owner failed", + ); + }); + }); + it.each(["SIGTERM", "SIGINT"] as const)( "drains admitted root work before closing on %s", async (signal) => { diff --git a/src/cli/gateway-cli/run-loop.ts b/src/cli/gateway-cli/run-loop.ts index 98a98126f34d..47aba84329d5 100644 --- a/src/cli/gateway-cli/run-loop.ts +++ b/src/cli/gateway-cli/run-loop.ts @@ -790,7 +790,7 @@ export async function runGatewayLoop(params: { ...(closeDrainTimeoutMs !== null ? { drainTimeoutMs: closeDrainTimeoutMs } : {}), }); } catch (err) { - gatewayLog.error(`shutdown error: ${String(err)}`); + gatewayLog.error(`shutdown step failed (gateway server close): ${formatErrorMessage(err)}`); } finally { server = null; if (isRestart) { diff --git a/src/gateway/active-sessions-shutdown-drain.ts b/src/gateway/active-sessions-shutdown-drain.ts new file mode 100644 index 000000000000..ffa1100e834a --- /dev/null +++ b/src/gateway/active-sessions-shutdown-drain.ts @@ -0,0 +1,73 @@ +import { buildSessionEndHookPayload } from "../auto-reply/reply/session-hooks.js"; +import { logVerbose } from "../globals.js"; +import { getGlobalHookRunner } from "../plugins/hook-runner-global.js"; +import { + forgetActiveSessionForShutdown, + listActiveSessionsForShutdown, +} from "./active-sessions-shutdown-tracker.js"; +import { resolveStableSessionEndTranscript } from "./session-transcript-files.fs.js"; + +export async function drainActiveSessionsForShutdown(params: { + reason: "shutdown" | "restart"; + totalTimeoutMs?: number; +}): Promise<{ emittedSessionIds: string[]; timedOut: boolean }> { + const tracked = listActiveSessionsForShutdown(); + if (tracked.length === 0) { + return { emittedSessionIds: [], timedOut: false }; + } + const totalTimeoutMs = Math.max(100, Math.floor(params.totalTimeoutMs ?? 2_000)); + const emittedSessionIds: string[] = []; + const hookRunner = getGlobalHookRunner(); + let settledEmissions = 0; + // Start all emissions before the bounded aggregate so one slow plugin cannot + // prevent later tracked sessions from receiving session_end. + const drain = Promise.allSettled( + tracked.map(async (entry) => { + try { + forgetActiveSessionForShutdown(entry.sessionId); + emittedSessionIds.push(entry.sessionId); + if (!hookRunner?.hasHooks("session_end")) { + return; + } + const transcript = resolveStableSessionEndTranscript({ + sessionId: entry.sessionId, + storePath: entry.storePath, + sessionFile: entry.sessionFile, + agentId: entry.agentId, + }); + const payload = buildSessionEndHookPayload({ + sessionId: entry.sessionId, + sessionKey: entry.sessionKey, + agentId: entry.agentId, + reason: params.reason, + sessionFile: transcript.sessionFile, + transcriptArchived: transcript.transcriptArchived, + }); + await hookRunner.runSessionEnd(payload.event, payload.context); + } catch (err) { + logVerbose(`session_end hook failed during shutdown drain: ${String(err)}`); + } finally { + settledEmissions++; + } + }), + ); + let timer: ReturnType | undefined; + const timeout = new Promise<"timeout">((resolve) => { + timer = setTimeout(() => resolve("timeout"), totalTimeoutMs); + timer.unref?.(); + }); + try { + const result = await Promise.race([drain.then(() => "ok" as const), timeout]); + if (result === "timeout") { + logVerbose( + `shutdown session-end drain timed out after ${totalTimeoutMs}ms with ${tracked.length - settledEmissions} session_end handler(s) still pending`, + ); + return { emittedSessionIds, timedOut: true }; + } + return { emittedSessionIds, timedOut: false }; + } finally { + if (timer) { + clearTimeout(timer); + } + } +} diff --git a/src/gateway/drain-active-sessions-for-shutdown.test.ts b/src/gateway/drain-active-sessions-for-shutdown.test.ts index c6c5fed14ec5..ea6f817dbb3f 100644 --- a/src/gateway/drain-active-sessions-for-shutdown.test.ts +++ b/src/gateway/drain-active-sessions-for-shutdown.test.ts @@ -48,11 +48,9 @@ vi.mock("../auto-reply/reply/session-hooks.js", () => ({ buildSessionStartHookPayload: vi.fn(() => ({ event: {}, context: {} })), })); -const { - drainActiveSessionsForShutdown, - emitGatewaySessionEndPluginHook, - emitGatewaySessionStartPluginHook, -} = await import("./session-reset-service.js"); +const { emitGatewaySessionEndPluginHook, emitGatewaySessionStartPluginHook } = + await import("./session-reset-service.js"); +const { drainActiveSessionsForShutdown } = await import("./active-sessions-shutdown-drain.js"); const { forgetActiveSessionForShutdown, listActiveSessionsForShutdown } = await import("./active-sessions-shutdown-tracker.js"); diff --git a/src/gateway/server-close.runtime.ts b/src/gateway/server-close.runtime.ts index aba96228b41d..6a44af9a006c 100644 --- a/src/gateway/server-close.runtime.ts +++ b/src/gateway/server-close.runtime.ts @@ -1,3 +1,3 @@ // Runtime close barrel keeps shutdown imports narrow for lazy server paths. export * from "./server-close.js"; -export { drainActiveSessionsForShutdown } from "./session-reset-service.js"; +export { drainActiveSessionsForShutdown } from "./active-sessions-shutdown-drain.js"; diff --git a/src/gateway/server-close.test.ts b/src/gateway/server-close.test.ts index 371e8c138805..5d7daab75d43 100644 --- a/src/gateway/server-close.test.ts +++ b/src/gateway/server-close.test.ts @@ -27,6 +27,7 @@ const mocks = vi.hoisted(() => ({ triggerInternalHook: vi.fn(async (_eventValue) => undefined), disposeAllBundleLspRuntimes: vi.fn(async () => undefined), drainRetainedEmbeddingProviders: vi.fn(async () => undefined), + stopGmailWatcher: vi.fn(async () => undefined), disposeAcpSessionManagerInstance: vi.fn(async () => undefined), getAcpSessionManager: vi.fn(() => ({})), fenceSessionSuspensionWritesForGatewayShutdown: vi.fn(), @@ -47,7 +48,7 @@ vi.mock("../channels/plugins/index.js", async () => ({ })); vi.mock("../hooks/gmail-watcher.js", () => ({ - stopGmailWatcher: vi.fn(async () => undefined), + stopGmailWatcher: mocks.stopGmailWatcher, })); vi.mock("../hooks/internal-hooks.js", async () => { @@ -151,6 +152,11 @@ function createGatewayCloseTestDeps( tailscaleCleanup: null, stopChannel: vi.fn(async () => undefined), pluginServices: null, + disposeAllBundleLspRuntimes: mocks.disposeAllBundleLspRuntimes, + drainRetainedOpenAiEmbeddingProviders: mocks.drainRetainedEmbeddingProviders, + stopGmailWatcher: mocks.stopGmailWatcher, + disposeAllCodeModeRuns: mocks.disposeAllCodeModeRuns, + closeProviderTransportDispatcherPool: mocks.closeProviderTransportDispatcherPool, cron: { stop: vi.fn() }, heartbeatRunner: { stop: vi.fn() } as never, updateCheckStop: null, @@ -209,6 +215,10 @@ describe("createGatewayCloseHandler", () => { mocks.disposeAllBundleLspRuntimes.mockResolvedValue(undefined); mocks.drainRetainedEmbeddingProviders.mockClear(); mocks.drainRetainedEmbeddingProviders.mockResolvedValue(undefined); + mocks.stopGmailWatcher.mockClear(); + mocks.stopGmailWatcher.mockResolvedValue(undefined); + mocks.closeProviderTransportDispatcherPool.mockClear(); + mocks.closeProviderTransportDispatcherPool.mockResolvedValue(undefined); mocks.disposeAcpSessionManagerInstance.mockReset(); mocks.disposeAcpSessionManagerInstance.mockResolvedValue(undefined); mocks.getAcpSessionManager.mockClear(); diff --git a/src/gateway/server-close.ts b/src/gateway/server-close.ts index a932602f0b7c..3f51c15dec3f 100644 --- a/src/gateway/server-close.ts +++ b/src/gateway/server-close.ts @@ -57,7 +57,7 @@ const RESTART_REPLY_POST_ABORT_DRAIN_POLL_MS = 50; const RESTART_TERMINAL_PERSISTENCE_WAIT_TIMEOUT_MS = 1_000; const RESTART_MARKER_SLOW_WARNING_MS = 1_000; -export type ShutdownResult = { +type ShutdownResult = { durationMs: number; warnings: string[]; }; @@ -603,21 +603,6 @@ async function disposeRuntimeWithShutdownGrace(params: { disposeTimeout.clear(); } -async function disposeAllBundleLspRuntimesOnDemand(): Promise { - const { disposeAllBundleLspRuntimes } = await import("../agents/agent-bundle-lsp-runtime.js"); - await disposeAllBundleLspRuntimes(); -} - -async function drainRetainedEmbeddingProvidersOnDemand(): Promise { - const { drainRetainedOpenAiEmbeddingProviders } = await import("./embeddings-http.js"); - await drainRetainedOpenAiEmbeddingProviders(); -} - -async function stopGmailWatcherOnDemand(): Promise { - const { stopGmailWatcher } = await import("../hooks/gmail-watcher.js"); - await stopGmailWatcher(); -} - export async function runGatewayClosePrelude(params: { stopDiagnostics?: () => void; clearSkillsRefreshTimer?: () => void; @@ -685,6 +670,11 @@ export function createGatewayCloseHandler( postReadySidecars?: readonly GatewayPostReadySidecarHandle[]; disposeSessionMcpRuntimes?: () => Promise; disposeBundleLspRuntimes?: () => Promise; + disposeAllBundleLspRuntimes: () => Promise; + drainRetainedOpenAiEmbeddingProviders: () => Promise; + stopGmailWatcher: () => Promise; + disposeAllCodeModeRuns: () => Promise | void; + closeProviderTransportDispatcherPool: () => Promise; cron: { stop: () => void; stopAndDrain?: () => Promise }; heartbeatRunner: HeartbeatRunner; updateCheckStop?: (() => void) | null; @@ -882,25 +872,12 @@ export function createGatewayCloseHandler( await shutdownStep(`channel/${channelId}`, () => params.stopChannel(channelId), warnings); } }); - // Load the bridge only at shutdown; eager imports boot the subagent registry at startup. - // Cancel parked calls before their agent harnesses and MCP transports disappear. - await shutdownStep( - "code-mode-runs", - async () => { - const { disposeAllCodeModeRuns } = await import("../agents/code-mode-state.js"); - return disposeAllCodeModeRuns(); - }, - warnings, - ); + await shutdownStep("code-mode-runs", () => params.disposeAllCodeModeRuns(), warnings); await shutdownStep("agent-harnesses", () => disposeRegisteredAgentHarnesses(), warnings); await shutdownStep("ai-session-resources", () => cleanupSessionResources(), warnings); await shutdownStep( "provider-transport-dispatchers", - async () => { - const { closeProviderTransportDispatcherPool } = - await import("../agents/provider-transport-dispatcher-pool.js"); - await closeProviderTransportDispatcherPool(); - }, + () => params.closeProviderTransportDispatcherPool(), warnings, ); await measureCloseStep("bundle-runtimes", async () => { @@ -913,7 +890,7 @@ export function createGatewayCloseHandler( }), disposeRuntimeWithShutdownGrace({ label: "bundle-lsp", - dispose: params.disposeBundleLspRuntimes ?? disposeAllBundleLspRuntimesOnDemand, + dispose: params.disposeBundleLspRuntimes ?? params.disposeAllBundleLspRuntimes, graceMs: LSP_RUNTIME_CLOSE_GRACE_MS, warnings, }), @@ -934,7 +911,7 @@ export function createGatewayCloseHandler( recordShutdownWarning(warnings, "media-cleanup"); } await measureCloseStep("gmail-watcher", () => - shutdownStep("gmail-watcher", () => stopGmailWatcherOnDemand(), warnings), + shutdownStep("gmail-watcher", () => params.stopGmailWatcher(), warnings), ); await shutdownStep( "cron", @@ -1093,7 +1070,7 @@ export function createGatewayCloseHandler( } await disposeRuntimeWithShutdownGrace({ label: "embedding-providers", - dispose: drainRetainedEmbeddingProvidersOnDemand, + dispose: params.drainRetainedOpenAiEmbeddingProviders, graceMs: EMBEDDING_PROVIDER_CLOSE_GRACE_MS, warnings, }); diff --git a/src/gateway/server-core-runtime.ts b/src/gateway/server-core-runtime.ts index e15949baae99..ce3aff001285 100644 --- a/src/gateway/server-core-runtime.ts +++ b/src/gateway/server-core-runtime.ts @@ -151,6 +151,7 @@ export async function startGatewayCoreRuntime(input: { broadcastPluginEvent, activateRuntimeSecrets, residentRegistry, + shutdownRuntime, } = runtime; if (desktopSessionRegistry) { kernel.addGatewayLifetimeSidecar({ stop: () => desktopSessionRegistry.stopAll() }); @@ -234,8 +235,7 @@ export async function startGatewayCoreRuntime(input: { stop: async () => { const earlyRuntime = await startEarlyRuntime(); earlyRuntime.skillsChangeUnsub(); - const { stopTaskRegistryMaintenance } = await import("../tasks/task-registry.maintenance.js"); - stopTaskRegistryMaintenance(); + shutdownRuntime.stopTaskRegistryMaintenance(); }, }); const earlyRuntime = await startupTrace.measure("runtime.early", () => diff --git a/src/gateway/server-import-boundary.test.ts b/src/gateway/server-import-boundary.test.ts index 9c27d457c940..7d627a05201e 100644 --- a/src/gateway/server-import-boundary.test.ts +++ b/src/gateway/server-import-boundary.test.ts @@ -88,6 +88,7 @@ function readServerImplementation(): string { return [ "src/gateway/server-start.ts", "src/gateway/server-kernel.ts", + "src/gateway/server-shutdown.runtime.ts", "src/gateway/server-startup-bootstrap.ts", "src/gateway/server-runtime-state-prepare.ts", "src/gateway/server-lifecycle.ts", @@ -99,6 +100,14 @@ function readServerImplementation(): string { } describe("gateway startup import boundaries", () => { + it("keeps ordinary session lifecycle code out of the prepared shutdown graph", () => { + const graph = collectStaticValueImportGraph("src/gateway/server-close.runtime.ts"); + + expect([...graph.keys()].map((filePath) => path.relative(repoRoot, filePath))).not.toContain( + "src/gateway/session-reset-service.ts", + ); + }); + it("keeps the kernel static import graph free of HTTP server and WebSocket construction", () => { const graph = collectStaticValueImportGraph("src/gateway/server-kernel.ts"); const violations: string[] = []; @@ -240,9 +249,9 @@ describe("gateway startup import boundaries", () => { it("fences config reload before gateway teardown and gateway_stop hooks", () => { const serverImpl = readServerImplementation(); const closeStart = /close:\s*async\s*\([^)]*\)\s*=>/u.exec(serverImpl)?.index ?? -1; - const hookStart = serverImpl.indexOf("runGlobalGatewayStopSafely", closeStart); - const reloadStopStart = serverImpl.indexOf("await beginClosePrelude();", closeStart); - const terminalStopStart = serverImpl.indexOf("terminalSessions.disposeAll();", closeStart); + const hookStart = serverImpl.indexOf('name: "gateway_stop plugin hooks"', closeStart); + const reloadStopStart = serverImpl.indexOf('name: "close prelude fence"', closeStart); + const terminalStopStart = serverImpl.indexOf('name: "terminal sessions"', closeStart); const markHelperStart = serverImpl.indexOf("const markClosePreludeStarted = () => {"); const markHelperEnd = serverImpl.indexOf("};", markHelperStart); const beginHelperStart = serverImpl.indexOf("const beginClosePrelude = async () => {"); @@ -255,6 +264,7 @@ describe("gateway startup import boundaries", () => { expect(reloadStopStart).toBeGreaterThan(closeStart); expect(reloadStopStart).toBeLessThan(terminalStopStart); expect(reloadStopStart).toBeLessThan(hookStart); + expect(serverImpl.slice(closeStart, hookStart)).not.toContain("await import("); expect(markHelperStart).toBeGreaterThan(-1); expect(serverImpl.slice(markHelperStart, markHelperEnd)).toContain( "clearPostReadyMaintenanceTimer();", diff --git a/src/gateway/server-kernel-request-runtime.ts b/src/gateway/server-kernel-request-runtime.ts index 4645f06f9ef9..ce177b32d4ea 100644 --- a/src/gateway/server-kernel-request-runtime.ts +++ b/src/gateway/server-kernel-request-runtime.ts @@ -103,6 +103,7 @@ export async function prepareGatewayKernelRequestRuntime(params: { startupState, clearFallbackGatewayContextForServer, kernel, + shutdownRuntime, } = runtime; const chatMetadataLifecycle = await createGatewayChatMetadataLifecycle({ getConfig: getRuntimeConfig, @@ -221,6 +222,7 @@ export async function prepareGatewayKernelRequestRuntime(params: { await attachInitialGatewayLifetimeSidecars({ chatMetadataLifecycle, gatewayRequestContext, + flushPendingSessionsChangedEvents: shutdownRuntime.flushPendingSessionsChangedEvents, sidecars: runtimeState.gatewayLifetimeSidecars, }); pluginGatewayContext.current = gatewayRequestContext; diff --git a/src/gateway/server-kernel.test.ts b/src/gateway/server-kernel.test.ts index adace9ba16e4..66efbbda68b3 100644 --- a/src/gateway/server-kernel.test.ts +++ b/src/gateway/server-kernel.test.ts @@ -256,6 +256,7 @@ describe("createGatewayKernel", () => { "control-ui.root", "tls.runtime", "runtime.state", + "gateway.shutdown-runtime-import", "runtime.early", "runtime.early.discovery", "runtime.post-early-imports", diff --git a/src/gateway/server-kernel.ts b/src/gateway/server-kernel.ts index d789c2938747..788d15742c21 100644 --- a/src/gateway/server-kernel.ts +++ b/src/gateway/server-kernel.ts @@ -38,7 +38,9 @@ const loadGatewayStartupEarlyModule = createLazyRuntimeModule( const loadGatewayPluginBootstrapModule = createLazyRuntimeModule( () => import("./server-plugin-bootstrap.js"), ); -const loadGatewayCloseModule = createLazyRuntimeModule(() => import("./server-close.runtime.js")); +const loadGatewayShutdownModule = createLazyRuntimeModule( + () => import("./server-shutdown.runtime.js"), +); const log = createSubsystemLogger("gateway"); const logDiscovery = log.child("discovery"); @@ -112,16 +114,6 @@ function formatRuntimeGatewayAuthTokenWarning(): string { ].join(" "); } -async function closeMcpLoopbackServerOnDemand(): Promise { - const { closeMcpLoopbackServer } = await import("./mcp-http.js"); - await closeMcpLoopbackServer(); -} - -async function stopTaskRegistryMaintenanceOnDemand(): Promise { - const { stopTaskRegistryMaintenance } = await import("../tasks/task-registry.maintenance.js"); - stopTaskRegistryMaintenance(); -} - export async function resetPreparedModelCatalogForTestCore(): Promise { const { resetPreparedModelCatalogStateForTest } = await loadGatewayModelCatalogModule(); await resetPreparedModelCatalogStateForTest(); @@ -153,15 +145,19 @@ export async function createGatewayKernel(port = 18789, opts: GatewayServerOptio loadWorkerEnvironmentStartupModule, loadWorkerPlacementStartupModule, }); + // An in-place update may replace every hashed chunk before SIGTERM arrives. + // Resolve and retain the complete shutdown graph while the install is healthy. + const shutdownRuntime = await runtime.startupTrace.measure( + "gateway.shutdown-runtime-import", + async () => (await loadGatewayShutdownModule()).prepareGatewayShutdownRuntime(), + ); lifecycleRuntime = await prepareGatewayLifecycle({ runtime, port, log, logCron, diagnosticsEnabled: bootstrap.diagnosticsEnabled, - loadGatewayCloseModule, - closeMcpLoopbackServerOnDemand, - stopTaskRegistryMaintenanceOnDemand, + shutdownRuntime, }); if (bootstrap.cfgAtStart.gateway?.tls?.enabled && !runtime.gatewayTls.enabled) { throw new Error(runtime.gatewayTls.error ?? "gateway tls: failed to enable"); diff --git a/src/gateway/server-lifecycle.ts b/src/gateway/server-lifecycle.ts index f19165713552..66172a49b37b 100644 --- a/src/gateway/server-lifecycle.ts +++ b/src/gateway/server-lifecycle.ts @@ -34,6 +34,7 @@ import { shouldRetainControlUiDeviceAuthMigrationSession, } from "./server-public.js"; import type { prepareGatewayKernelState } from "./server-runtime-state-prepare.js"; +import type { GatewayShutdownRuntime } from "./server-shutdown.runtime.js"; import { getHealthVersion, incrementPresenceVersion, @@ -51,20 +52,9 @@ export async function prepareGatewayLifecycle(params: { log: GatewayLogger; logCron: GatewayLogger; diagnosticsEnabled: boolean; - loadGatewayCloseModule: () => Promise; - closeMcpLoopbackServerOnDemand: () => Promise; - stopTaskRegistryMaintenanceOnDemand: () => Promise; + shutdownRuntime: GatewayShutdownRuntime; }) { - const { - runtime, - port, - log, - logCron, - diagnosticsEnabled, - loadGatewayCloseModule, - closeMcpLoopbackServerOnDemand, - stopTaskRegistryMaintenanceOnDemand, - } = params; + const { runtime, port, log, logCron, diagnosticsEnabled, shutdownRuntime } = params; const { minimalTestGateway, controlUiDeviceAuthMigration, @@ -485,8 +475,7 @@ export async function prepareGatewayLifecycle(params: { disposeNodeConnectionNotifications(nodeRegistry); watchNodeHttpRuntime.close(); clearPluginMetadataLifecycleCaches(); - const { runGatewayClosePrelude } = await loadGatewayCloseModule(); - await runGatewayClosePrelude({ + await shutdownRuntime.runGatewayClosePrelude({ ...(diagnosticsEnabled ? { stopDiagnostics: stopDiagnosticHeartbeat } : {}), clearSkillsRefreshTimer: () => { if (!runtimeState?.skillsRefreshTimer) { @@ -507,7 +496,7 @@ export async function prepareGatewayLifecycle(params: { await monitor?.waitForIdle(); }, stopReadinessEventLoopHealth: readinessEventLoopHealth.stop, - closeMcpServer: closeMcpLoopbackServerOnDemand, + closeMcpServer: shutdownRuntime.closeMcpLoopbackServer, }); }; const { getRuntimeSnapshot, startChannels, startChannel, stopChannel, markChannelLoggedOut } = @@ -537,11 +526,9 @@ export async function prepareGatewayLifecycle(params: { }; const createCloseHandler = () => async (optsValue?: GatewayCloseOptions) => { const channelIds = listLoadedChannelPlugins().map((plugin) => plugin.id as ChannelId); - const { createGatewayCloseHandler, drainActiveSessionsForShutdown } = - await loadGatewayCloseModule(); const transport = transportBridge.current(); await transport?.portalService.closeAll(); - await createGatewayCloseHandler({ + await shutdownRuntime.createGatewayCloseHandler({ bonjourStop: runtimeState.bonjourStop, tailscaleCleanup: runtimeState.tailscaleCleanup, clearSecretsRuntimeSnapshot: clearSecretsRuntimeSnapshotState, @@ -552,7 +539,7 @@ export async function prepareGatewayLifecycle(params: { cron: runtimeState.cronState.cron, heartbeatRunner: runtimeState.heartbeatRunner, updateCheckStop: runtimeState.stopGatewayUpdateCheck, - stopTaskRegistryMaintenance: stopTaskRegistryMaintenanceOnDemand, + stopTaskRegistryMaintenance: shutdownRuntime.stopTaskRegistryMaintenance, nodePresenceTimers, broadcast, tickInterval: runtimeState.tickInterval, @@ -584,9 +571,7 @@ export async function prepareGatewayLifecycle(params: { if (sessionKeys.size === 0 && sessionIds.size === 0) { return; } - const { markRestartAbortedMainSessions } = - await import("../agents/main-session-recovery/main-session-restart-recovery.js"); - await markRestartAbortedMainSessions({ + await shutdownRuntime.markRestartAbortedMainSessions({ cfg: getRuntimeConfig(), sessionKeys, sessionIds, @@ -605,7 +590,12 @@ export async function prepareGatewayLifecycle(params: { httpServers: transport.httpServers, } : {}), - drainActiveSessionsForShutdown, + drainActiveSessionsForShutdown: shutdownRuntime.drainActiveSessionsForShutdown, + disposeAllBundleLspRuntimes: shutdownRuntime.disposeAllBundleLspRuntimes, + drainRetainedOpenAiEmbeddingProviders: shutdownRuntime.drainRetainedOpenAiEmbeddingProviders, + stopGmailWatcher: shutdownRuntime.stopGmailWatcher, + disposeAllCodeModeRuns: shutdownRuntime.disposeAllCodeModeRuns, + closeProviderTransportDispatcherPool: shutdownRuntime.closeProviderTransportDispatcherPool, })(optsValue); }; let clearFallbackGatewayContextForServer = () => {}; @@ -676,6 +666,7 @@ export async function prepareGatewayLifecycle(params: { unavailableGatewayMethods, kernel, pluginHostServices, + shutdownRuntime, lifecycle, postReadyState, cronReconciliation, diff --git a/src/gateway/server-lifetime-sidecars.ts b/src/gateway/server-lifetime-sidecars.ts index 57d17a0cced5..262e65d24808 100644 --- a/src/gateway/server-lifetime-sidecars.ts +++ b/src/gateway/server-lifetime-sidecars.ts @@ -7,14 +7,13 @@ type GatewayChatMetadataLifecycle = Awaited void; sidecars: GatewayPostReadySidecarHandle[]; }): Promise { await params.chatMetadataLifecycle.attachContext(params.gatewayRequestContext, params.sidecars); params.sidecars.push({ - stop: async () => { - const { flushPendingSessionsChangedEvents } = - await import("./server-methods/session-change-event.js"); - flushPendingSessionsChangedEvents(params.gatewayRequestContext); + stop: () => { + params.flushPendingSessionsChangedEvents(params.gatewayRequestContext); }, }); } diff --git a/src/gateway/server-shutdown.runtime.test.ts b/src/gateway/server-shutdown.runtime.test.ts new file mode 100644 index 000000000000..2ed37a144393 --- /dev/null +++ b/src/gateway/server-shutdown.runtime.test.ts @@ -0,0 +1,96 @@ +import { describe, expect, it, vi } from "vitest"; + +const state = vi.hoisted(() => ({ + loaded: [] as string[], + close: vi.fn(), + flushSessionChanges: vi.fn(), + stopPlugins: vi.fn(), + clearPluginRegistry: vi.fn(), + preparePluginRegistryShutdown: vi.fn(async () => undefined), +})); + +vi.mock("./server-close.runtime.js", () => { + state.loaded.push("server-close"); + return { + createGatewayCloseHandler: state.close, + drainActiveSessionsForShutdown: vi.fn(), + runGatewayClosePrelude: vi.fn(), + }; +}); +vi.mock("../plugins/hook-runner-global.js", () => { + state.loaded.push("plugin-hooks"); + return { runGlobalGatewayStopSafely: state.stopPlugins }; +}); +vi.mock("./server-methods/session-change-event.js", () => { + state.loaded.push("session-change-events"); + return { flushPendingSessionsChangedEvents: state.flushSessionChanges }; +}); +vi.mock("./mcp-http.js", () => { + state.loaded.push("mcp-http"); + return { closeMcpLoopbackServer: vi.fn() }; +}); +vi.mock("../tasks/task-registry.maintenance.js", () => { + state.loaded.push("task-maintenance"); + return { stopTaskRegistryMaintenance: vi.fn() }; +}); +vi.mock("../agents/main-session-recovery/main-session-restart-recovery.js", () => { + state.loaded.push("restart-recovery"); + return { markRestartAbortedMainSessions: vi.fn() }; +}); +vi.mock("../agents/agent-bundle-lsp-runtime.js", () => { + state.loaded.push("bundle-lsp"); + return { disposeAllBundleLspRuntimes: vi.fn() }; +}); +vi.mock("./embeddings-http.js", () => { + state.loaded.push("embeddings"); + return { drainRetainedOpenAiEmbeddingProviders: vi.fn() }; +}); +vi.mock("../hooks/gmail-watcher.js", () => { + state.loaded.push("gmail-watcher"); + return { stopGmailWatcher: vi.fn() }; +}); +vi.mock("../agents/code-mode-state.js", () => { + state.loaded.push("code-mode"); + return { disposeAllCodeModeRuns: vi.fn() }; +}); +vi.mock("../agents/provider-transport-dispatcher-pool.js", () => { + state.loaded.push("provider-transports"); + return { closeProviderTransportDispatcherPool: vi.fn() }; +}); +vi.mock("../plugins/runtime.js", () => { + state.loaded.push("plugin-runtime"); + return { + clearActivePluginRegistry: state.clearPluginRegistry, + prepareActivePluginRegistryShutdown: state.preparePluginRegistryShutdown, + }; +}); + +const { prepareGatewayShutdownRuntime } = await import("./server-shutdown.runtime.js"); + +describe("gateway shutdown runtime", () => { + it("resolves every shutdown dependency during preparation", async () => { + const runtime = await prepareGatewayShutdownRuntime(); + + expect(state.loaded.toSorted()).toEqual( + [ + "server-close", + "plugin-hooks", + "session-change-events", + "mcp-http", + "task-maintenance", + "restart-recovery", + "bundle-lsp", + "embeddings", + "gmail-watcher", + "code-mode", + "provider-transports", + "plugin-runtime", + ].toSorted(), + ); + expect(runtime.createGatewayCloseHandler).toBe(state.close); + expect(runtime.flushPendingSessionsChangedEvents).toBe(state.flushSessionChanges); + expect(runtime.runGlobalGatewayStopSafely).toBe(state.stopPlugins); + expect(runtime.clearActivePluginRegistry).toBe(state.clearPluginRegistry); + expect(state.preparePluginRegistryShutdown).toHaveBeenCalledOnce(); + }); +}); diff --git a/src/gateway/server-shutdown.runtime.ts b/src/gateway/server-shutdown.runtime.ts new file mode 100644 index 000000000000..a9b6e6d52717 --- /dev/null +++ b/src/gateway/server-shutdown.runtime.ts @@ -0,0 +1,49 @@ +export async function prepareGatewayShutdownRuntime() { + const [ + { createGatewayCloseHandler, drainActiveSessionsForShutdown, runGatewayClosePrelude }, + { runGlobalGatewayStopSafely }, + { flushPendingSessionsChangedEvents }, + { closeMcpLoopbackServer }, + { stopTaskRegistryMaintenance }, + { markRestartAbortedMainSessions }, + { disposeAllBundleLspRuntimes }, + { drainRetainedOpenAiEmbeddingProviders }, + { stopGmailWatcher }, + { disposeAllCodeModeRuns }, + { closeProviderTransportDispatcherPool }, + { clearActivePluginRegistry, prepareActivePluginRegistryShutdown }, + ] = await Promise.all([ + import("./server-close.runtime.js"), + import("../plugins/hook-runner-global.js"), + import("./server-methods/session-change-event.js"), + import("./mcp-http.js"), + import("../tasks/task-registry.maintenance.js"), + import("../agents/main-session-recovery/main-session-restart-recovery.js"), + import("../agents/agent-bundle-lsp-runtime.js"), + import("./embeddings-http.js"), + import("../hooks/gmail-watcher.js"), + import("../agents/code-mode-state.js"), + import("../agents/provider-transport-dispatcher-pool.js"), + import("../plugins/runtime.js"), + ]); + await prepareActivePluginRegistryShutdown(); + + return { + createGatewayCloseHandler, + drainActiveSessionsForShutdown, + runGatewayClosePrelude, + runGlobalGatewayStopSafely, + flushPendingSessionsChangedEvents, + closeMcpLoopbackServer, + stopTaskRegistryMaintenance, + markRestartAbortedMainSessions, + disposeAllBundleLspRuntimes, + drainRetainedOpenAiEmbeddingProviders, + stopGmailWatcher, + disposeAllCodeModeRuns, + closeProviderTransportDispatcherPool, + clearActivePluginRegistry, + }; +} + +export type GatewayShutdownRuntime = Awaited>; diff --git a/src/gateway/server-shutdown.test.ts b/src/gateway/server-shutdown.test.ts new file mode 100644 index 000000000000..77da9ab0f7bf --- /dev/null +++ b/src/gateway/server-shutdown.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it, vi } from "vitest"; +import { runGatewayShutdownSteps } from "./server-shutdown.js"; + +describe("gateway shutdown steps", () => { + it("names an unavailable module step and continues the remaining shutdown", async () => { + const missingModule = Object.assign(new Error("Cannot find module 'rotated-chunk.js'"), { + code: "ERR_MODULE_NOT_FOUND", + }); + const loadStopModule = vi.fn(async () => { + throw missingModule; + }); + const closeGateway = vi.fn(async () => {}); + const messages: string[] = []; + + await runGatewayShutdownSteps({ + steps: [ + { name: "gateway lifetime sidecars", run: loadStopModule }, + { name: "gateway close", run: closeGateway }, + ], + onError: (message) => messages.push(message), + }); + + expect(closeGateway).toHaveBeenCalledOnce(); + expect(messages).toEqual([ + "shutdown step failed (gateway lifetime sidecars): Cannot find module 'rotated-chunk.js'", + ]); + expect(messages.join("\n")).not.toContain("shutdown error"); + }); +}); diff --git a/src/gateway/server-shutdown.ts b/src/gateway/server-shutdown.ts new file mode 100644 index 000000000000..87e3266d1681 --- /dev/null +++ b/src/gateway/server-shutdown.ts @@ -0,0 +1,20 @@ +import { formatErrorMessage } from "../infra/errors.js"; + +type GatewayShutdownStep = { + name: string; + run: () => Promise | void; +}; + +/** Run every shutdown step even when one owner fails, with the failed owner named. */ +export async function runGatewayShutdownSteps(params: { + steps: readonly GatewayShutdownStep[]; + onError: (message: string) => void; +}): Promise { + for (const step of params.steps) { + try { + await step.run(); + } catch (error) { + params.onError(`shutdown step failed (${step.name}): ${formatErrorMessage(error)}`); + } + } +} diff --git a/src/gateway/server-start.ts b/src/gateway/server-start.ts index 7950ab49f920..23ccddefdd23 100644 --- a/src/gateway/server-start.ts +++ b/src/gateway/server-start.ts @@ -1,3 +1,4 @@ +import { formatErrorMessage } from "../infra/errors.js"; import { createLazyRuntimeModule } from "../shared/lazy-runtime.js"; import { createGatewayKernel, @@ -6,6 +7,7 @@ import { } from "./server-kernel.js"; import type { GatewayServer, GatewayServerOptions } from "./server-public.js"; import { createGatewayHttpTransport } from "./server-runtime-state.js"; +import { runGatewayShutdownSteps } from "./server-shutdown.js"; import { finishGatewayStartup } from "./server-startup-finish.js"; const loadGatewayStartupPostAttachModule = createLazyRuntimeModule( @@ -36,6 +38,7 @@ export async function startGatewayServerCore( stopRegisteredGatewayLifetimeSidecars, stopRegisteredPostReadySidecars, terminalSessions, + shutdownRuntime, } = gatewayKernel; try { const transport = await createGatewayHttpTransport(gatewayKernel.createHttpTransportOptions()); @@ -68,24 +71,33 @@ export async function startGatewayServerCore( return { close: async (optsLocal) => { - try { - await beginClosePrelude(); - // Kill any live operator shells before the socket layer tears down. - terminalSessions.disposeAll(); - await stopRegisteredGatewayLifetimeSidecars(); - await stopRegisteredPostReadySidecars(); - // Run gateway_stop plugin hook before shutdown - const { runGlobalGatewayStopSafely } = await import("../plugins/hook-runner-global.js"); - await runGlobalGatewayStopSafely({ - event: { reason: optsLocal?.reason ?? "gateway stopping" }, - ctx: { port }, - onError: (err) => log.warn(`gateway_stop hook failed: ${String(err)}`), - }); - await runClosePrelude(); - await close(optsLocal); - } finally { - clearFallbackGatewayContextForServer.get()(); - } + await runGatewayShutdownSteps({ + steps: [ + { name: "close prelude fence", run: beginClosePrelude }, + // Kill any live operator shells before the socket layer tears down. + { name: "terminal sessions", run: () => terminalSessions.disposeAll() }, + { name: "gateway lifetime sidecars", run: stopRegisteredGatewayLifetimeSidecars }, + { name: "post-ready sidecars", run: stopRegisteredPostReadySidecars }, + { + name: "gateway_stop plugin hooks", + run: async () => { + await shutdownRuntime.runGlobalGatewayStopSafely({ + event: { reason: optsLocal?.reason ?? "gateway stopping" }, + ctx: { port }, + onError: (error) => + log.warn(`gateway_stop hook failed: ${formatErrorMessage(error)}`), + }); + }, + }, + { name: "gateway close prelude", run: runClosePrelude }, + { name: "gateway close", run: () => close(optsLocal) }, + { + name: "fallback gateway context", + run: () => clearFallbackGatewayContextForServer.get()(), + }, + ], + onError: (message) => log.error(message), + }); }, }; } diff --git a/src/gateway/server-startup-post-attach.test.ts b/src/gateway/server-startup-post-attach.test.ts index 4c3c347ae796..35e0de9e177f 100644 --- a/src/gateway/server-startup-post-attach.test.ts +++ b/src/gateway/server-startup-post-attach.test.ts @@ -2518,6 +2518,11 @@ describe("startGatewayPostAttachRuntime", () => { stopChannel, pluginServices, postReadySidecars, + disposeAllBundleLspRuntimes: vi.fn(async () => {}), + drainRetainedOpenAiEmbeddingProviders: vi.fn(async () => {}), + stopGmailWatcher: vi.fn(async () => {}), + disposeAllCodeModeRuns: vi.fn(async () => {}), + closeProviderTransportDispatcherPool: vi.fn(async () => {}), cron: { stop: vi.fn() }, heartbeatRunner: { stop: vi.fn(), updateConfig: vi.fn() }, nodePresenceTimers: new Map(), diff --git a/src/gateway/session-reset-service.ts b/src/gateway/session-reset-service.ts index 4623b682edc0..3c0abda47347 100644 --- a/src/gateway/session-reset-service.ts +++ b/src/gateway/session-reset-service.ts @@ -90,7 +90,6 @@ import { resolveGlobalSingleton } from "../shared/global-singleton.js"; import { getOrCreatePromise } from "../shared/lazy-promise.js"; import { forgetActiveSessionForShutdown, - listActiveSessionsForShutdown, noteActiveSessionForShutdown, } from "./active-sessions-shutdown-tracker.js"; import { findDirectChildSessionsForParent } from "./session-child-sessions.js"; @@ -295,94 +294,6 @@ export function emitGatewaySessionStartPluginHook(params: { }); } -const SHUTDOWN_DRAIN_DEFAULT_TOTAL_TIMEOUT_MS = 2_000; - -type DrainActiveSessionsForShutdownResult = { - emittedSessionIds: string[]; - timedOut: boolean; -}; - -/** - * Emit a typed `session_end` for every session that received `session_start` - * but did not yet receive a paired `session_end`. The bounded total timeout - * mirrors the gateway lifecycle hook timeout so a slow plugin cannot block - * SIGTERM/SIGINT past the runtime's overall shutdown grace window. - * - * Sessions that have already been finalized through replace / reset / delete / - * compaction are forgotten from the tracker by `emitGatewaySessionEndPluginHook` - * before this drain runs, so they will not be double-fired here. - */ -export async function drainActiveSessionsForShutdown(params: { - reason: "shutdown" | "restart"; - totalTimeoutMs?: number; -}): Promise { - const tracked = listActiveSessionsForShutdown(); - if (tracked.length === 0) { - return { emittedSessionIds: [], timedOut: false }; - } - const totalTimeoutMs = Math.max( - 100, - Math.floor(params.totalTimeoutMs ?? SHUTDOWN_DRAIN_DEFAULT_TOTAL_TIMEOUT_MS), - ); - const emittedSessionIds: string[] = []; - const hookRunner = getGlobalHookRunner(); - let settledEmissions = 0; - // Inline the session_end emission instead of calling - // `emitGatewaySessionEndPluginHook`, because that helper uses fire-and-forget - // (`void hookRunner.runSessionEnd(...)`). Start every tracked session's - // emission before awaiting the bounded aggregate so one slow plugin write - // cannot prevent later active sessions from receiving `session_end`. - const drain = Promise.allSettled( - tracked.map(async (entry) => { - try { - forgetActiveSessionForShutdown(entry.sessionId); - emittedSessionIds.push(entry.sessionId); - if (!hookRunner?.hasHooks("session_end")) { - return; - } - const transcript = resolveStableSessionEndTranscript({ - sessionId: entry.sessionId, - storePath: entry.storePath, - sessionFile: entry.sessionFile, - agentId: entry.agentId, - }); - const payload = buildSessionEndHookPayload({ - sessionId: entry.sessionId, - sessionKey: entry.sessionKey, - agentId: entry.agentId, - reason: params.reason, - sessionFile: transcript.sessionFile, - transcriptArchived: transcript.transcriptArchived, - }); - await hookRunner.runSessionEnd(payload.event, payload.context); - } catch (err) { - logVerbose(`session_end hook failed during shutdown drain: ${String(err)}`); - } finally { - settledEmissions++; - } - }), - ); - let timer: ReturnType | undefined; - const timeout = new Promise<"timeout">((resolve) => { - timer = setTimeout(() => resolve("timeout"), totalTimeoutMs); - timer.unref?.(); - }); - try { - const result = await Promise.race([drain.then(() => "ok" as const), timeout]); - if (result === "timeout") { - logVerbose( - `shutdown session-end drain timed out after ${totalTimeoutMs}ms with ${tracked.length - settledEmissions} session_end handler(s) still pending`, - ); - return { emittedSessionIds, timedOut: true }; - } - return { emittedSessionIds, timedOut: false }; - } finally { - if (timer) { - clearTimeout(timer); - } - } -} - export async function emitSessionUnboundLifecycleEvent(params: { targetSessionKey: string; reason: "session-reset" | "session-delete"; diff --git a/src/plugins/plugin-command-runtime.test.ts b/src/plugins/plugin-command-runtime.test.ts index dfa054956abf..a6beb3bbfbfd 100644 --- a/src/plugins/plugin-command-runtime.test.ts +++ b/src/plugins/plugin-command-runtime.test.ts @@ -1,5 +1,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; -const cleanupReplacedPluginHostRegistry = vi.hoisted(() => vi.fn(async () => {})); +const cleanupReplacedPluginHostRegistry = vi.hoisted(() => + vi.fn(async () => ({ cleanupCount: 0, failures: [] })), +); vi.mock("./host-hook-cleanup.js", () => ({ cleanupReplacedPluginHostRegistry })); @@ -16,6 +18,7 @@ import { createEmptyPluginRegistry } from "./registry-empty.js"; import { markPluginRegistryRetired } from "./registry-lifecycle.js"; import { clearActivePluginRegistry, + prepareActivePluginRegistryShutdown, resetPluginRuntimeStateForTest, setActivePluginRegistry, } from "./runtime.js"; @@ -71,6 +74,17 @@ afterEach(() => { }); describe("plugin command runtime", () => { + it("prepares plugin host cleanup before gateway shutdown", async () => { + await prepareActivePluginRegistryShutdown(); + const registry = createEmptyPluginRegistry(); + registry.plugins.push({ status: "loaded" } as never); + setActivePluginRegistry(registry); + + await clearActivePluginRegistry(); + + expect(cleanupReplacedPluginHostRegistry).toHaveBeenCalledOnce(); + }); + it("binds the request-scoped registry and scopes provider aliases", async () => { const ambient = createEmptyPluginRegistry(); const scoped = createEmptyPluginRegistry(); @@ -339,8 +353,8 @@ describe("plugin command runtime", () => { let releaseCleanup!: () => void; cleanupReplacedPluginHostRegistry.mockImplementationOnce( async () => - await new Promise((resolve) => { - releaseCleanup = resolve; + await new Promise<{ cleanupCount: number; failures: [] }>((resolve) => { + releaseCleanup = () => resolve({ cleanupCount: 0, failures: [] }); }), ); let detachedClear!: Promise; diff --git a/src/plugins/runtime.ts b/src/plugins/runtime.ts index c2a7025a8d7b..ee9ef700e2f8 100644 --- a/src/plugins/runtime.ts +++ b/src/plugins/runtime.ts @@ -2,6 +2,7 @@ import { onAgentEvent } from "../infra/agent-events.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; import { drainGlobalSingletonLifecycleState } from "../shared/global-singleton.js"; +import { createLazyRuntimeModule } from "../shared/lazy-runtime.js"; import { getPluginCommandExecutionCount, isPluginCommandExecutionActiveHere, @@ -63,13 +64,19 @@ function isRegistryLive(registry: PluginRegistry): boolean { return state.activeRegistry === registry; } -async function cleanupPreviousPluginHostRegistry(params: { - previousRegistry: PluginRegistry; -}): Promise { +const loadPluginHostCleanupRuntime = createLazyRuntimeModule(async () => { const [{ getRuntimeConfig }, { cleanupReplacedPluginHostRegistry }] = await Promise.all([ import("../config/config.js"), import("./host-hook-cleanup.js"), ]); + return { getRuntimeConfig, cleanupReplacedPluginHostRegistry }; +}); + +async function cleanupPreviousPluginHostRegistry(params: { + previousRegistry: PluginRegistry; +}): Promise { + const { getRuntimeConfig, cleanupReplacedPluginHostRegistry } = + await loadPluginHostCleanupRuntime(); const nextRegistry = asPluginRegistry(state.activeRegistry); if (nextRegistry === params.previousRegistry) { return; @@ -438,6 +445,10 @@ export async function clearActivePluginRegistry(): Promise { await completion; } +export async function prepareActivePluginRegistryShutdown(): Promise { + await loadPluginHostCleanupRuntime(); +} + export function resetPluginRuntimeStateForTest(): void { state.registrationContext = undefined; clearActivePluginRegistryState();