diff --git a/src/gateway/server-connection-state.ts b/src/gateway/server-connection-state.ts new file mode 100644 index 000000000000..fc3edde7d3f7 --- /dev/null +++ b/src/gateway/server-connection-state.ts @@ -0,0 +1,59 @@ +// Gateway connection and run registries. +// This state is transport-fed but can be constructed without HTTP or WebSocket servers. +import type { ChatAbortControllerEntry } from "./chat-abort.js"; +import { createGatewayBroadcaster } from "./server-broadcast.js"; +import { + createChatRunState, + createSessionEventSubscriberRegistry, + createSessionMessageSubscriberRegistry, +} from "./server-chat-state.js"; +import type { GatewayWsClient } from "./server/ws-types.js"; +import { canReceiveSessionEvent } from "./session-sharing.js"; + +/** Creates transport-independent connection, subscription, and run state. */ +export function createGatewayConnectionState(params: { + cfg: import("../config/config.js").OpenClawConfig; + getRuntimeConfig?: () => import("../config/config.js").OpenClawConfig; +}) { + const loadRuntimeConfig = params.getRuntimeConfig ?? (() => params.cfg); + const clients = new Set(); + const sessionEventSubscribers = createSessionEventSubscriberRegistry(); + const sessionMessageSubscribers = createSessionMessageSubscriberRegistry(); + const gatewayBroadcaster = createGatewayBroadcaster({ + clients, + sessionMessageSubscribers, + canReceiveSessionEvent: (client, sessionKeys, agentId, event, payload) => + canReceiveSessionEvent({ + cfg: loadRuntimeConfig(), + client, + sessionKeys, + agentId, + event, + payload, + }), + }); + const agentRunSeq = new Map(); + const dedupe = new Map(); + const chatRunState = createChatRunState(); + const chatRunRegistry = chatRunState.registry; + const addChatRun = chatRunRegistry.add; + const removeChatRun = chatRunRegistry.remove; + const chatAbortControllers = new Map(); + const chatQueuedTurns = new Map(); + const toolEventRecipients = chatRunState.toolEventRecipients; + + return { + clients, + ...gatewayBroadcaster, + agentRunSeq, + dedupe, + chatRunState, + addChatRun, + removeChatRun, + chatAbortControllers, + chatQueuedTurns, + toolEventRecipients, + sessionEventSubscribers, + sessionMessageSubscribers, + }; +} diff --git a/src/gateway/server-runtime-state-prepare.ts b/src/gateway/server-runtime-state-prepare.ts index ad7ae792087d..c0edb22f6f61 100644 --- a/src/gateway/server-runtime-state-prepare.ts +++ b/src/gateway/server-runtime-state-prepare.ts @@ -16,11 +16,12 @@ import { resolveGatewayAuth } from "./auth.js"; import { isLoopbackHost } from "./net.js"; import { createNodeReapprovalCoordinator } from "./node-reapproval-coordinator.js"; import { resolveGatewayPluginConfig } from "./runtime-plugin-config.js"; +import { createGatewayConnectionState } from "./server-connection-state.js"; import { createGatewayControlUiRootLifecycle } from "./server-control-ui-root.js"; import type { GatewayInstanceRuntime } from "./server-instance-runtime.types.js"; import type { GatewayServerLiveState } from "./server-live-state.js"; import type { GatewayRequestContext } from "./server-methods/types.js"; -import { createGatewayRuntimeState } from "./server-runtime-state.js"; +import { createGatewayHttpTransport } from "./server-runtime-state.js"; import type { SharedGatewaySessionGenerationState } from "./server-shared-auth-generation.js"; import type { prepareGatewayServerBootstrap } from "./server-startup-bootstrap.js"; import { createWizardSessionTracker } from "./server-wizard-sessions.js"; @@ -353,13 +354,53 @@ export async function prepareGatewayRuntimeState(params: { const watchNodeRequestHandler: { current?: (req: IncomingMessage, res: ServerResponse) => Promise; } = {}; + const { connectionState, httpTransport } = await startupTrace.measure( + "runtime.state", + async () => { + const createdConnectionState = createGatewayConnectionState({ + cfg: cfgAtStart, + getRuntimeConfig, + }); + const createdHttpTransport = await createGatewayHttpTransport({ + cfg: cfgAtStart, + getRuntimeConfig, + bindHost, + port, + controlUiEnabled, + controlUiBasePath, + controlUiRoot: controlUiRootLifecycle.state, + openAiChatCompletionsEnabled, + openAiChatCompletionsConfig, + openResponsesEnabled, + openResponsesConfig, + strictTransportSecurityHeader, + resolvedAuth, + rateLimiter: authRateLimiter, + isTerminalEnabled: terminalLaunchPolicy.isEnabled, + gatewayTls, + getResolvedAuth, + hooksConfig: () => runtimeStateRef.current?.hooksConfig ?? initialHooksConfig, + getHookClientIpConfig: () => + runtimeStateRef.current?.hookClientIpConfig ?? initialHookClientIpConfig, + pluginRegistry: pluginRuntime.registry, + getPluginRouteRegistry: () => pluginRuntime.registry, + isStartupPluginRuntimeReady: () => startupState.sidecarsReady, + getGatewayRequestContext: () => pluginGatewayContext.current, + deps, + log, + logHooks, + logPlugins, + getReadiness, + handleWatchNodeRequest: async (req, res) => + (await watchNodeRequestHandler.current?.(req, res)) ?? false, + workerIngressEnabled: Boolean(workerEnvironmentService), + workerDesktopTunnels: workerTunnelManager?.desktop, + clients: createdConnectionState.clients, + }); + return { connectionState: createdConnectionState, httpTransport: createdHttpTransport }; + }, + ); const { - httpServer, - httpServers, - httpBindHosts, - startListening, - wss, - preauthConnectionBudget, clients, broadcast, broadcastToConnIds, @@ -375,46 +416,18 @@ export async function prepareGatewayRuntimeState(params: { toolEventRecipients, sessionEventSubscribers, sessionMessageSubscribers, + } = connectionState; + const { + httpServer, + httpServers, + httpBindHosts, + startListening, + wss, + preauthConnectionBudget, getWorkerIngressEndpoint, getMcpAppSandboxPort, ensureSandboxHostPort, - } = await startupTrace.measure("runtime.state", () => - createGatewayRuntimeState({ - cfg: cfgAtStart, - getRuntimeConfig, - bindHost, - port, - controlUiEnabled, - controlUiBasePath, - controlUiRoot: controlUiRootLifecycle.state, - openAiChatCompletionsEnabled, - openAiChatCompletionsConfig, - openResponsesEnabled, - openResponsesConfig, - strictTransportSecurityHeader, - resolvedAuth, - rateLimiter: authRateLimiter, - isTerminalEnabled: terminalLaunchPolicy.isEnabled, - gatewayTls, - getResolvedAuth, - hooksConfig: () => runtimeStateRef.current?.hooksConfig ?? initialHooksConfig, - getHookClientIpConfig: () => - runtimeStateRef.current?.hookClientIpConfig ?? initialHookClientIpConfig, - pluginRegistry: pluginRuntime.registry, - getPluginRouteRegistry: () => pluginRuntime.registry, - isStartupPluginRuntimeReady: () => startupState.sidecarsReady, - getGatewayRequestContext: () => pluginGatewayContext.current, - deps, - log, - logHooks, - logPlugins, - getReadiness, - handleWatchNodeRequest: async (req, res) => - (await watchNodeRequestHandler.current?.(req, res)) ?? false, - workerIngressEnabled: Boolean(workerEnvironmentService), - workerDesktopTunnels: workerTunnelManager?.desktop, - }), - ); + } = httpTransport; return { ...bootstrap, diff --git a/src/gateway/server-runtime-state.ts b/src/gateway/server-runtime-state.ts index 4bae07655b5a..779d1c8dca7e 100644 --- a/src/gateway/server-runtime-state.ts +++ b/src/gateway/server-runtime-state.ts @@ -17,26 +17,11 @@ import type { createSubsystemLogger } from "../logging/subsystem.js"; import type { PluginRegistry } from "../plugins/registry.js"; import type { AuthRateLimiter } from "./auth-rate-limit.js"; import type { ResolvedGatewayAuth } from "./auth.js"; -import type { ChatAbortControllerEntry } from "./chat-abort.js"; import type { ControlUiRootState } from "./control-ui.js"; import type { HooksConfigResolved } from "./hooks.js"; import type { AuthorizedGatewayHttpRequest } from "./http-auth-utils.js"; import { createSandboxHostHttpServer } from "./mcp-app-sandbox-http.js"; import { isLoopbackHost, resolveGatewayListenHosts } from "./net.js"; -import type { - GatewayBroadcastFn, - GatewayBroadcastToConnIdsFn, - GatewayBufferedAmountFn, - GatewayPluginEventBroadcastFn, -} from "./server-broadcast-types.js"; -import { createGatewayBroadcaster } from "./server-broadcast.js"; -import { - type ChatRunEntry, - type ChatRunRegistration, - createChatRunState, - createSessionEventSubscriberRegistry, - createSessionMessageSubscriberRegistry, -} from "./server-chat-state.js"; import { MAX_PREAUTH_PAYLOAD_BYTES } from "./server-constants.js"; import { attachGatewayUpgradeHandler, @@ -44,7 +29,6 @@ import { createGatewayHttpServer, } from "./server-http.js"; import type { GatewayRequestContext } from "./server-methods/types.js"; -import type { DedupeEntry } from "./server-shared.js"; import type { HookClientIpConfig, HooksRequestHandler } from "./server/hooks-request-handler.js"; import { listenGatewayHttpServer } from "./server/http-listen.js"; import { runWithGatewayHttpWorkAdmission } from "./server/http-work-admission.js"; @@ -59,7 +43,6 @@ import { import type { ReadinessChecker } from "./server/readiness.js"; import type { GatewayTlsRuntime } from "./server/tls.js"; import type { GatewayWsClient } from "./server/ws-types.js"; -import { canReceiveSessionEvent } from "./session-sharing.js"; import type { WorkerDesktopTunnels } from "./worker-environments/desktop-tunnel.js"; type GatewayPluginRequestHandler = ( @@ -101,8 +84,8 @@ function hasMatchingGatewayPluginRoute( : matchingRoutes.length > 0; } -/** Creates the HTTP/WebSocket runtime state for one gateway start. */ -export async function createGatewayRuntimeState(params: { +/** Creates the HTTP/WebSocket transport for one gateway start. */ +export async function createGatewayHttpTransport(params: { cfg: import("../config/config.js").OpenClawConfig; getRuntimeConfig?: () => import("../config/config.js").OpenClawConfig; bindHost: string; @@ -135,6 +118,7 @@ export async function createGatewayRuntimeState(params: { handleWatchNodeRequest?: (req: IncomingMessage, res: ServerResponse) => Promise; workerIngressEnabled?: boolean; workerDesktopTunnels?: WorkerDesktopTunnels; + clients: Set; }): Promise<{ httpServer: HttpServer; httpServers: HttpServer[]; @@ -142,25 +126,6 @@ export async function createGatewayRuntimeState(params: { startListening: () => Promise; wss: WebSocketServer; preauthConnectionBudget: PreauthConnectionBudget; - clients: Set; - broadcast: GatewayBroadcastFn; - broadcastToConnIds: GatewayBroadcastToConnIdsFn; - getBufferedAmount: GatewayBufferedAmountFn; - broadcastPluginEvent: GatewayPluginEventBroadcastFn; - agentRunSeq: Map; - dedupe: Map; - chatRunState: ReturnType; - addChatRun: (sessionId: string, entry: ChatRunRegistration) => void; - removeChatRun: ( - sessionId: string, - clientRunId: string, - sessionKey?: string, - ) => ChatRunEntry | undefined; - chatAbortControllers: Map; - chatQueuedTurns: Map; - toolEventRecipients: ReturnType["toolEventRecipients"]; - sessionEventSubscribers: ReturnType; - sessionMessageSubscribers: ReturnType; getWorkerIngressEndpoint: () => { host: "127.0.0.1"; port: number } | undefined; getMcpAppSandboxPort: () => number | undefined; ensureSandboxHostPort: () => Promise; @@ -168,22 +133,6 @@ export async function createGatewayRuntimeState(params: { const loadRuntimeConfig = params.getRuntimeConfig ?? (() => params.cfg); const resolvePluginRouteRegistry = () => params.getPluginRouteRegistry?.() ?? params.pluginRegistry; - const clients = new Set(); - const sessionEventSubscribers = createSessionEventSubscriberRegistry(); - const sessionMessageSubscribers = createSessionMessageSubscriberRegistry(); - const gatewayBroadcaster = createGatewayBroadcaster({ - clients, - sessionMessageSubscribers, - canReceiveSessionEvent: (client, sessionKeys, agentId, event, payload) => - canReceiveSessionEvent({ - cfg: loadRuntimeConfig(), - client, - sessionKeys, - agentId, - event, - payload, - }), - }); let loadedHooksRequestHandler: HooksRequestHandler | null = null; const handleHooksRequest: HooksRequestHandler = async (req, res) => { @@ -306,7 +255,7 @@ export async function createGatewayRuntimeState(params: { const httpBindHosts: string[] = []; for (const _ of bindHosts) { const httpServer = createGatewayHttpServer({ - clients, + clients: params.clients, controlUiEnabled: params.controlUiEnabled, controlUiBasePath: params.controlUiBasePath, controlUiRoot: params.controlUiRoot, @@ -336,7 +285,7 @@ export async function createGatewayRuntimeState(params: { handlePluginUpgrade, shouldEnforcePluginGatewayAuth, resolvePluginNodeCapabilityRoute, - clients, + clients: params.clients, preauthConnectionBudget, resolvedAuth: params.resolvedAuth, getResolvedAuth: params.getResolvedAuth, @@ -519,16 +468,6 @@ export async function createGatewayRuntimeState(params: { })(); await startListeningPromise; }; - const agentRunSeq = new Map(); - const dedupe = new Map(); - const chatRunState = createChatRunState(); - const chatRunRegistry = chatRunState.registry; - const addChatRun = chatRunRegistry.add; - const removeChatRun = chatRunRegistry.remove; - const chatAbortControllers = new Map(); - const chatQueuedTurns = new Map(); - const toolEventRecipients = chatRunState.toolEventRecipients; - return { httpServer, httpServers, @@ -536,18 +475,6 @@ export async function createGatewayRuntimeState(params: { startListening, wss, preauthConnectionBudget, - clients, - ...gatewayBroadcaster, - agentRunSeq, - dedupe, - chatRunState, - addChatRun, - removeChatRun, - chatAbortControllers, - chatQueuedTurns, - toolEventRecipients, - sessionEventSubscribers, - sessionMessageSubscribers, getWorkerIngressEndpoint: () => workerIngressPort === undefined ? undefined diff --git a/src/gateway/test-helpers.server-runtime-state.ts b/src/gateway/test-helpers.server-runtime-state.ts index 0b1419fa6d89..0bbb40c1588c 100644 --- a/src/gateway/test-helpers.server-runtime-state.ts +++ b/src/gateway/test-helpers.server-runtime-state.ts @@ -1,19 +1,20 @@ // Server runtime-state test helper builds minimal gateway runtime state with a // configurable plugin registry. import { createEmptyPluginRegistry } from "../plugins/registry.js"; -import { createGatewayRuntimeState } from "./server-runtime-state.js"; +import { createGatewayConnectionState } from "./server-connection-state.js"; +import { createGatewayHttpTransport } from "./server-runtime-state.js"; /** * Runtime-state fixture factory for gateway server tests. */ -type GatewayRuntimeStateParams = Parameters[0]; +type GatewayRuntimeStateParams = Omit[0], "clients">; /** Creates a minimal gateway runtime state with optional plugin registry fixture. */ export async function createGatewayRuntimeStateForTest( pluginRegistry: GatewayRuntimeStateParams["pluginRegistry"] = createEmptyPluginRegistry(), overrides: Partial = {}, ) { - return await createGatewayRuntimeState({ + const params = { cfg: {}, bindHost: "127.0.0.1", port: 0, @@ -32,5 +33,11 @@ export async function createGatewayRuntimeStateForTest( logHooks: { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} } as never, logPlugins: { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} } as never, ...overrides, + }; + const connectionState = createGatewayConnectionState(params); + const httpTransport = await createGatewayHttpTransport({ + ...params, + clients: connectionState.clients, }); + return { ...httpTransport, ...connectionState }; }