feat(worker): simulation mode (skip HTTP server, disable load limit, agent name override)

Add a --simulation CLI flag (on the start command, set by `lk simulate`) that
threads through to worker options. In simulation mode: skip worker HTTP server
creation/start/close so concurrent runs don't collide on a fixed port, disable
the worker load limit (loadThreshold = Infinity) so runs can saturate the
agent, and resolve agentName from LIVEKIT_AGENT_NAME_OVERRIDE first so lk
simulate dispatch matches.

Ported from livekit/agents#6055

Author: u9g

.changeset/simulation-http-server.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Simulation fixes: do not start the worker HTTP server, disable the worker load limit so runs can saturate the agent, and honor `LIVEKIT_AGENT_NAME_OVERRIDE` so the worker registers under the name `lk simulate` dispatches to.

agents/src/cli.ts

@@ -30,7 +30,7 @@ const runServer = async (args: CliArgs) => {
 
   // though `production` is defined in ServerOptions, it will always be overridden by CLI.
   const { production: _, ...opts } = args.opts; // eslint-disable-line @typescript-eslint/no-unused-vars
-  const server = new AgentServer(new ServerOptions({ production: args.production, ...opts }));
+  const server = new AgentServer(new ServerOptions({ ...opts, production: args.production }));
 
   if (args.room) {
     server.event.once('worker_registered', () => {
@@ -136,6 +136,13 @@ export const runApp = (opts: ServerOptions) => {
     .command('start')
     .description('Start the worker in production mode')
     .addOption(logLevelOption('info'))
+    .addOption(
+      new Option(
+        '--simulation',
+        'Run under an agent simulation: the worker load limit is disabled so runs ' +
+          'can saturate the agent, and the worker HTTP server is not started. Set by `lk simulate`.',
+      ).hideHelp(),
+    )
     .action((...[, command]) => {
       const globalOptions = program.optsWithGlobals();
       const commandOptions = command.opts();
@@ -144,6 +151,7 @@ export const runApp = (opts: ServerOptions) => {
       opts.apiSecret = globalOptions.apiSecret || opts.apiSecret;
       opts.logLevel = commandOptions.logLevel;
       opts.workerToken = globalOptions.workerToken || opts.workerToken;
+      opts.simulation = commandOptions.simulation || opts.simulation;
       runServer({
         opts,
         production: true,

agents/src/worker.ts

@@ -144,6 +144,7 @@ export class ServerOptions {
   port: number;
   logLevel: string;
   production: boolean;
+  simulation: boolean;
   jobMemoryWarnMB: number;
   jobMemoryLimitMB: number;
 
@@ -169,6 +170,7 @@ export class ServerOptions {
     port = undefined,
     logLevel = 'info',
     production = false,
+    simulation = false,
     jobMemoryWarnMB = 500,
     jobMemoryLimitMB = 0,
   }: {
@@ -210,6 +212,7 @@ export class ServerOptions {
     port?: number;
     logLevel?: string;
     production?: boolean;
+    simulation?: boolean;
     jobMemoryWarnMB?: number;
     jobMemoryLimitMB?: number;
   }) {
@@ -219,14 +222,19 @@ export class ServerOptions {
     }
     this.requestFunc = requestFunc;
     this.loadFunc = loadFunc;
-    this.loadThreshold = loadThreshold || Default.loadThreshold(production);
+    this.loadThreshold = simulation ? Infinity : loadThreshold || Default.loadThreshold(production);
     this.numIdleProcesses = numIdleProcesses || Default.numIdleProcesses(production);
     this.shutdownProcessTimeout = shutdownProcessTimeout;
     this.initializeProcessTimeout = initializeProcessTimeout;
     this.permissions = permissions;
     // agentNameIsEnv may be passed explicitly when ServerOptions is re-constructed (e.g.
     // cli.ts spreads an existing ServerOptions instance), so prefer it when defined.
-    if (agentName) {
+    if (process.env.LIVEKIT_AGENT_NAME_OVERRIDE) {
+      // Highest priority: `lk simulate` sets this to force the worker to register
+      // under the agent name it dispatches to, overriding any configured agentName.
+      this.agentName = process.env.LIVEKIT_AGENT_NAME_OVERRIDE;
+      this.agentNameIsEnv = agentNameIsEnv ?? true;
+    } else if (agentName) {
       this.agentName = agentName;
       this.agentNameIsEnv = agentNameIsEnv ?? false;
     } else if (process.env.LIVEKIT_AGENT_NAME) {
@@ -246,6 +254,7 @@ export class ServerOptions {
     this.port = port || Default.port(production);
     this.logLevel = logLevel;
     this.production = production;
+    this.simulation = simulation;
     this.jobMemoryWarnMB = jobMemoryWarnMB;
     this.jobMemoryLimitMB = jobMemoryLimitMB;
   }
@@ -283,7 +292,7 @@ export class AgentServer {
 
   event = new EventEmitter();
   #session: WebSocket | undefined = undefined;
-  #httpServer: HTTPServer;
+  #httpServer?: HTTPServer;
   #logger = log().child({ version });
   #inferenceExecutor?: InferenceProcExecutor;
 
@@ -336,35 +345,39 @@ export class AgentServer {
 
     this.#opts = opts;
 
-    const healthCheck = () => {
-      // Check if inference executor exists and is not alive
-      if (this.#inferenceExecutor && !this.#inferenceExecutor.isAlive) {
-        return { healthy: false, message: 'inference process not running' };
-      }
+    // Simulations run ephemeral workers side by side; a health endpoint on a fixed port would make
+    // concurrent runs collide.
+    if (!opts.simulation) {
+      const healthCheck = () => {
+        // Check if inference executor exists and is not alive
+        if (this.#inferenceExecutor && !this.#inferenceExecutor.isAlive) {
+          return { healthy: false, message: 'inference process not running' };
+        }
 
-      // Only healthy when fully connected with an active WebSocket
-      if (
-        this.#closed ||
-        this.#connecting ||
-        !this.#session ||
-        this.#session.readyState !== WebSocket.OPEN
-      ) {
-        return { healthy: false, message: 'not connected to livekit' };
-      }
+        // Only healthy when fully connected with an active WebSocket
+        if (
+          this.#closed ||
+          this.#connecting ||
+          !this.#session ||
+          this.#session.readyState !== WebSocket.OPEN
+        ) {
+          return { healthy: false, message: 'not connected to livekit' };
+        }
 
-      return { healthy: true, message: 'OK' };
-    };
+        return { healthy: true, message: 'OK' };
+      };
 
-    const getWorkerInfo = () => ({
-      agent_name: opts.agentName,
-      agent_name_is_env: opts.agentNameIsEnv,
-      worker_type: JobType[opts.serverType],
-      active_jobs: this.activeJobs.length,
-      sdk_version: version,
-      project_type: PROJECT_TYPE,
-    });
+      const getWorkerInfo = () => ({
+        agent_name: opts.agentName,
+        agent_name_is_env: opts.agentNameIsEnv,
+        worker_type: JobType[opts.serverType],
+        active_jobs: this.activeJobs.length,
+        sdk_version: version,
+        project_type: PROJECT_TYPE,
+      });
 
-    this.#httpServer = new HTTPServer(opts.host, opts.port, healthCheck, getWorkerInfo);
+      this.#httpServer = new HTTPServer(opts.host, opts.port, healthCheck, getWorkerInfo);
+    }
   }
 
   /** @throws {@link WorkerError} if worker failed to connect or already running */
@@ -431,7 +444,11 @@ export class AgentServer {
       }
     };
 
-    await ThrowsPromise.all([workerWS(), this.#httpServer.run()]);
+    const tasks = [workerWS()];
+    if (this.#httpServer) {
+      tasks.push(this.#httpServer.run());
+    }
+    await ThrowsPromise.all(tasks);
     this.#close.resolve();
   }
 
@@ -875,7 +892,7 @@ export class AgentServer {
 
     await this.#inferenceExecutor?.close();
     await this.#procPool.close();
-    await this.#httpServer.close();
+    await this.#httpServer?.close();
     await ThrowsPromise.allSettled(this.#tasks);
 
     this.#session?.close();

turbo.json

@@ -42,6 +42,7 @@
     "LLAMA_API_KEY",
     "LIVEKIT_AGENT_ID",
     "LIVEKIT_AGENT_NAME",
+    "LIVEKIT_AGENT_NAME_OVERRIDE",
     "LOG_LEVEL",
     "OCTOAI_TOKEN",
     "OPENAI_API_KEY",