diff --git a/server/src/__integration__/worker-startup.test.ts b/server/src/__integration__/worker-startup.test.ts new file mode 100644 index 000000000..45e6cad6a --- /dev/null +++ b/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() + } +}) 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()