From 1734671a8688ab1d06064351f07c662fcf9b5237 Mon Sep 17 00:00:00 2001 From: Lyyzka <144704183+lyyzka@users.noreply.github.com> Date: Sat, 17 Oct 2026 18:57:54 +0800 Subject: [PATCH 1/4] test: reproduce production Worker bootstrap ownership [skip deploy] --- .../__integration__/worker-startup.test.ts | 55 +++++++++++++++++++ 1 file changed, 55 insertions(+) create mode 100644 server/src/__integration__/worker-startup.test.ts diff --git a/server/src/__integration__/worker-startup.test.ts b/server/src/__integration__/worker-startup.test.ts new file mode 100644 index 000000000..e1784eb2a --- /dev/null +++ b/server/src/__integration__/worker-startup.test.ts @@ -0,0 +1,55 @@ +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 + env.NODE_ENV = 'production' + 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 () => { + await assert.rejects(async () => { + const listener = await listenLingxiOSControl() + await listener.stop() + }, /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() + } +}) From 90f27451e341b8dd84a416507d77df3d34844cc7 Mon Sep 17 00:00:00 2001 From: Lyyzka <144704183+lyyzka@users.noreply.github.com> Date: Sat, 17 Oct 2026 19:01:01 +0800 Subject: [PATCH 2/4] test: isolate Worker ownership from production storage configuration [skip deploy] --- server/src/__integration__/worker-startup.test.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/server/src/__integration__/worker-startup.test.ts b/server/src/__integration__/worker-startup.test.ts index e1784eb2a..309013f01 100644 --- a/server/src/__integration__/worker-startup.test.ts +++ b/server/src/__integration__/worker-startup.test.ts @@ -21,7 +21,6 @@ test('production Worker reconciliation uses passive local context before runner const originalMode = env.NODE_ENV const originalPort = process.env.LINGXIOS_CONTROL_PORT const originalToken = process.env.LINGXIOS_SERVICE_TOKEN - env.NODE_ENV = 'production' process.env.LINGXIOS_CONTROL_PORT = '53971' process.env.LINGXIOS_SERVICE_TOKEN = 'isolated-worker-startup-service-token' let started = false @@ -29,6 +28,8 @@ test('production Worker reconciliation uses passive local context before runner 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(async () => { const listener = await listenLingxiOSControl() await listener.stop() From 1f145756d0c56efdb54388936b00ffc90ccab285 Mon Sep 17 00:00:00 2001 From: Lyyzka <144704183+lyyzka@users.noreply.github.com> Date: Sat, 17 Oct 2026 19:03:38 +0800 Subject: [PATCH 3/4] test: verify control ownership through the public listener [skip deploy] --- server/src/__integration__/worker-startup.test.ts | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/server/src/__integration__/worker-startup.test.ts b/server/src/__integration__/worker-startup.test.ts index 309013f01..45e6cad6a 100644 --- a/server/src/__integration__/worker-startup.test.ts +++ b/server/src/__integration__/worker-startup.test.ts @@ -30,10 +30,7 @@ test('production Worker reconciliation uses passive local context before runner startAgentRuntime: async () => { // Exercise production ownership after real reconciliation, using isolated test stores. env.NODE_ENV = 'production' - await assert.rejects(async () => { - const listener = await listenLingxiOSControl() - await listener.stop() - }, /Worker-local control cannot own the API control plane/) + await assert.rejects(listenLingxiOSControl, /Worker-local control cannot own the API control plane/) started = true return { stop: stopLingxiOSControl } }, From be28a68461542d849b47ecf3be2a8d50a76841fc Mon Sep 17 00:00:00 2001 From: Lyyzka <144704183+lyyzka@users.noreply.github.com> Date: Sat, 17 Oct 2026 19:06:31 +0800 Subject: [PATCH 4/4] fix: prepare passive Worker context before channel reconciliation [skip deploy] --- server/src/agent-runtime/runtime.ts | 12 ++++++++---- server/src/worker.ts | 3 ++- 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/server/src/agent-runtime/runtime.ts b/server/src/agent-runtime/runtime.ts index b9bb7807c..2aacb6979 100644 --- a/server/src/agent-runtime/runtime.ts +++ b/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, diff --git a/server/src/worker.ts b/server/src/worker.ts index bf6f4ba9c..b160b969e 100644 --- a/server/src/worker.ts +++ b/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()