refactor(gateway): split connection state from HTTP transport (#121931)

* refactor(gateway): split connection state from HTTP transport

* test(ci): drop stale unit-fast helper expectations

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<GatewayWsClient>();
+  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<string, number>();
+  const dedupe = new Map<string, import("./server-shared.js").DedupeEntry>();
+  const chatRunState = createChatRunState();
+  const chatRunRegistry = chatRunState.registry;
+  const addChatRun = chatRunRegistry.add;
+  const removeChatRun = chatRunRegistry.remove;
+  const chatAbortControllers = new Map<string, ChatAbortControllerEntry>();
+  const chatQueuedTurns = new Map<string, import("./chat-queued-turns.js").QueuedChatTurnEntry>();
+  const toolEventRecipients = chatRunState.toolEventRecipients;
+
+  return {
+    clients,
+    ...gatewayBroadcaster,
+    agentRunSeq,
+    dedupe,
+    chatRunState,
+    addChatRun,
+    removeChatRun,
+    chatAbortControllers,
+    chatQueuedTurns,
+    toolEventRecipients,
+    sessionEventSubscribers,
+    sessionMessageSubscribers,
+  };
+}

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<boolean>;
   } = {};
+  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,

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<boolean>;
   workerIngressEnabled?: boolean;
   workerDesktopTunnels?: WorkerDesktopTunnels;
+  clients: Set<GatewayWsClient>;
 }): Promise<{
   httpServer: HttpServer;
   httpServers: HttpServer[];
@@ -142,25 +126,6 @@ export async function createGatewayRuntimeState(params: {
   startListening: () => Promise<void>;
   wss: WebSocketServer;
   preauthConnectionBudget: PreauthConnectionBudget;
-  clients: Set<GatewayWsClient>;
-  broadcast: GatewayBroadcastFn;
-  broadcastToConnIds: GatewayBroadcastToConnIdsFn;
-  getBufferedAmount: GatewayBufferedAmountFn;
-  broadcastPluginEvent: GatewayPluginEventBroadcastFn;
-  agentRunSeq: Map<string, number>;
-  dedupe: Map<string, DedupeEntry>;
-  chatRunState: ReturnType<typeof createChatRunState>;
-  addChatRun: (sessionId: string, entry: ChatRunRegistration) => void;
-  removeChatRun: (
-    sessionId: string,
-    clientRunId: string,
-    sessionKey?: string,
-  ) => ChatRunEntry | undefined;
-  chatAbortControllers: Map<string, ChatAbortControllerEntry>;
-  chatQueuedTurns: Map<string, import("./chat-queued-turns.js").QueuedChatTurnEntry>;
-  toolEventRecipients: ReturnType<typeof createChatRunState>["toolEventRecipients"];
-  sessionEventSubscribers: ReturnType<typeof createSessionEventSubscriberRegistry>;
-  sessionMessageSubscribers: ReturnType<typeof createSessionMessageSubscriberRegistry>;
   getWorkerIngressEndpoint: () => { host: "127.0.0.1"; port: number } | undefined;
   getMcpAppSandboxPort: () => number | undefined;
   ensureSandboxHostPort: () => Promise<number>;
@@ -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<GatewayWsClient>();
-  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<string, number>();
-  const dedupe = new Map<string, DedupeEntry>();
-  const chatRunState = createChatRunState();
-  const chatRunRegistry = chatRunState.registry;
-  const addChatRun = chatRunRegistry.add;
-  const removeChatRun = chatRunRegistry.remove;
-  const chatAbortControllers = new Map<string, ChatAbortControllerEntry>();
-  const chatQueuedTurns = new Map<string, import("./chat-queued-turns.js").QueuedChatTurnEntry>();
-  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

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<typeof createGatewayRuntimeState>[0];
+type GatewayRuntimeStateParams = Omit<Parameters<typeof createGatewayHttpTransport>[0], "clients">;
 
 /** Creates a minimal gateway runtime state with optional plugin registry fixture. */
 export async function createGatewayRuntimeStateForTest(
   pluginRegistry: GatewayRuntimeStateParams["pluginRegistry"] = createEmptyPluginRegistry(),
   overrides: Partial<GatewayRuntimeStateParams> = {},
 ) {
-  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 };
 }