fix: prepare passive Worker context before reconciliation [skip deploy]

server/src/__integration__/worker-startup.test.ts

@@ -0,0 +1,53 @@
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import { listenLingxiOSControl, stopLingxiOSControl } from '../agent-runtime/runtime.js'
+import { pool } from '../db/pool.js'
+import { env } from '../env.js'
+import { startWorkerProcess } from '../worker.js'
+import { ensureSchemaOnce, installFakeWukong, resetAllTables, seedCompanyWithAgent, teardownAll } from './_helpers.js'
+
+// Failures: reconciliation creates an active control before Worker admission;
+// startup fails and closes the pool while active maintenance keeps the process alive.
+test('production Worker reconciliation uses passive local context before runner startup', { timeout: 30000 }, async () => {
+  await ensureSchemaOnce()
+  await resetAllTables()
+  installFakeWukong()
+  const { companyId, agentId } = await seedCompanyWithAgent()
+  await pool.query(`INSERT INTO conversations(id,company_id,kind,title,members)
+    VALUES('worker-startup-room',$1,'group','Startup',$2::jsonb)`, [companyId, JSON.stringify([agentId])])
+  await pool.query(`INSERT INTO im_channel_bindings(channel_id,company_id,profile)
+    VALUES('worker-startup-room',$1,$2::jsonb)`, [companyId, JSON.stringify({ channelType: 2, members: [agentId] })])
+  await stopLingxiOSControl()
+  const originalMode = env.NODE_ENV
+  const originalPort = process.env.LINGXIOS_CONTROL_PORT
+  const originalToken = process.env.LINGXIOS_SERVICE_TOKEN
+  process.env.LINGXIOS_CONTROL_PORT = '53971'
+  process.env.LINGXIOS_SERVICE_TOKEN = 'isolated-worker-startup-service-token'
+  let started = false
+  try {
+    const service = await startWorkerProcess({
+      initializeStorage: () => undefined,
+      startAgentRuntime: async () => {
+        // Exercise production ownership after real reconciliation, using isolated test stores.
+        env.NODE_ENV = 'production'
+        await assert.rejects(listenLingxiOSControl, /Worker-local control cannot own the API control plane/)
+        started = true
+        return { stop: stopLingxiOSControl }
+      },
+      startAgentIngress: () => null,
+      tasks: [],
+      closePostgres: () => undefined,
+      closeRedis: () => undefined,
+    })
+    assert.equal(started, true)
+    await service.stop('test')
+    assert.equal((await pool.query('SELECT 1')).rowCount, 1)
+  } finally {
+    env.NODE_ENV = originalMode
+    if (originalPort === undefined) delete process.env.LINGXIOS_CONTROL_PORT
+    else process.env.LINGXIOS_CONTROL_PORT = originalPort
+    if (originalToken === undefined) delete process.env.LINGXIOS_SERVICE_TOKEN
+    else process.env.LINGXIOS_SERVICE_TOKEN = originalToken
+    await teardownAll()
+  }
+})

server/src/agent-runtime/runtime.ts

@@ -174,13 +174,17 @@ export async function listenLingxiOSControl() {
     port: positiveInteger('LINGXIOS_CONTROL_PORT',5182), host: process.env.LINGXIOS_CONTROL_HOST?.trim() || '127.0.0.1' })
 }
 
-/** Worker ownership: the sole in-process agent runner; PostgreSQL leases fence replicas. */
-export async function startLingxiOSWorker() {
-  if (worker) throw new Error('LingxiOS worker already started in this process')
-  if (control && env.NODE_ENV !== 'test') throw new Error('Start the Worker in its own process before creating local context services')
+export function prepareLingxiOSWorkerContext(): void {
+  if (control && backgroundRole !== 'passive' && env.NODE_ENV !== 'test') throw new Error('Start the Worker in its own process before creating local context services')
   // Existing isolated model fixtures share their explicit API instance. A real
   // Worker process always constructs its local context services as passive.
   if (!control) backgroundRole = 'passive'
+}
+
+/** Worker ownership: the sole in-process agent runner; PostgreSQL leases fence replicas. */
+export async function startLingxiOSWorker() {
+  if (worker) throw new Error('LingxiOS worker already started in this process')
+  prepareLingxiOSWorkerContext()
   const concurrency = positiveInteger('AGENT_OS_MAX_CONCURRENT_RUNS', 2)
   const options = {
     modelBudget: common().modelBudget,

server/src/worker.ts

@@ -12,7 +12,7 @@ import { startPresentationStorageGc, startPresentationWorker } from './modules/p
 import { redis, sub } from './redis.js'
 import { Lifecycle, type ServiceHandle, startWorkerTasks, type WorkerTaskDefinition, type WorkerTaskHandle } from './runtime/lifecycle.js'
 import { initializeNativeStorage } from './storage.js'
-import { startLingxiOSWorker } from './agent-runtime/runtime.js'
+import { prepareLingxiOSWorkerContext, startLingxiOSWorker } from './agent-runtime/runtime.js'
 import { startAgentIngressRetry } from './agent-runtime/ingress.js'
 import { startNativeEventWorker } from './agents/native-event-worker.js'
 
@@ -61,6 +61,7 @@ export async function startWorkerProcess(options: WorkerProcessOptions = {}): Pr
   lifecycle.addDisposer('redis', options.closeRedis ?? (() => { sub.disconnect(); redis.disconnect() }))
 
   try {
+    prepareLingxiOSWorkerContext()
     const initializeStorage = options.initializeStorage ?? initializeNativeStorage
     initializeStorage()
     await prepare()