115 lines
7.6 KiB
JavaScript
115 lines
7.6 KiB
JavaScript
import test from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import { mkdtemp, writeFile, readFile, rm } from 'node:fs/promises';
|
|
import { join } from 'node:path';
|
|
import { tmpdir } from 'node:os';
|
|
import { setTimeout as delay } from 'node:timers/promises';
|
|
import { createServer } from 'node:net';
|
|
import * as orgBus from '../../cli/src/org-bus.mjs';
|
|
import { busCaller } from '../src/bus.mjs';
|
|
import { consoleBus } from '../src/console-bus.mjs';
|
|
import { organizationConsole } from '../src/organizations.mjs';
|
|
import { readWorkspace, workspaceHTTP, parseWorkspaceQuery } from '../src/workspace-read.mjs';
|
|
|
|
const session = { provider: 'local-dev', iss: 'local-dev', sub: 'local-dev-owner' };
|
|
const goodOptions = overrides => ({
|
|
bus: { tasks: async () => [], agents: async () => [], inbox: async () => [], 'thread.claims': async () => [], usage: async () => ({}), limits: async () => ({}), decisions: async () => [], ...overrides?.bus },
|
|
chat: { projects: async () => [{ id: 'p', name: 'Healthy' }], threads: async () => [], ...overrides?.chat },
|
|
});
|
|
const member = (id, overrides) => ({ id, doc: {}, status: async () => ({ healthy: true }), options: () => goodOptions(overrides) });
|
|
const orgs = (...members) => ({ members: async () => members });
|
|
const response = () => ({ writeHead(status) { this.status = status; }, end(body) { this.body = JSON.parse(body); } });
|
|
|
|
for (const [name, overrides] of [
|
|
['null thread', { chat: { threads: async () => [null] } }],
|
|
['null inbox option', { bus: { inbox: async () => [{ id: 'd', options: [null] }] } }],
|
|
['invalid project name', { chat: { projects: async () => [{ id: 'bad', name: 42 }] } }],
|
|
['non-array tasks', { bus: { tasks: async () => ({ tasks: [] }) } }],
|
|
['invalid usage total', { bus: { usage: async () => ({ total: 42 }) } }],
|
|
]) {
|
|
test(`malformed ${name} is isolated from a healthy organization`, async () => {
|
|
const res = response();
|
|
await workspaceHTTP({ method: 'GET' }, res, { organizations: orgs(member('good'), member('bad', overrides)), session, owner: true });
|
|
assert.equal(res.status, 200);
|
|
assert.deepEqual(res.body.projects.map(p => p.org), ['good']);
|
|
assert.deepEqual(res.body.summary, { read: 1, of: 2 });
|
|
const bad = res.body.orgs.find(o => o.org === 'bad');
|
|
assert.equal(bad.state, 'invalid');
|
|
assert.equal(bad.error, 'invalid-response');
|
|
assert.ok(Number.isFinite(Date.parse(bad.at)));
|
|
});
|
|
}
|
|
|
|
test('options construction failure is isolated from a healthy organization', async () => {
|
|
const bad = { ...member('bad'), options() { throw new Error('private detail'); } };
|
|
const res = response();
|
|
await workspaceHTTP({ method: 'GET' }, res, { organizations: orgs(member('good'), bad), session, owner: true });
|
|
assert.equal(res.status, 200);
|
|
assert.deepEqual(res.body.projects.map(p => p.org), ['good']);
|
|
assert.deepEqual(res.body.summary, { read: 1, of: 2 });
|
|
assert.equal(res.body.orgs[1].state, 'invalid');
|
|
assert.equal(res.body.orgs[1].error, 'invalid-response');
|
|
assert.ok(Number.isFinite(Date.parse(res.body.orgs[1].at)));
|
|
});
|
|
|
|
for (const [name, statusDelay, pluginDelay] of [['in-flight child', 0, 100], ['late channel continuation', 220, 300]]) {
|
|
test(`deadline cancels ${name} and prevents any later child`, async t => {
|
|
const dir = await mkdtemp(join(tmpdir(), 'ws-'));
|
|
t.after(() => rm(dir, { recursive: true, force: true }));
|
|
const cli = join(dir, 'cli.mjs'), log = join(dir, 'log');
|
|
await writeFile(cli, `import {appendFileSync} from 'node:fs';let input='';process.stdin.on('data',x=>input+=x);process.stdin.on('end',()=>{const q=JSON.parse(input);appendFileSync(${JSON.stringify(log)},JSON.stringify({pid:process.pid,verb:q.verb})+'\\n');if(q.verb==='plugins')setTimeout(()=>process.stdout.write(JSON.stringify({plugins:{channel:['webui']}})),${pluginDelay});else if(q.verb==='project.list')setInterval(()=>{},1000);else process.stdout.write('[]');});`);
|
|
const timeoutMs = 500;
|
|
const organizations = organizationConsole({ ctx: {}, listOrgs: async () => ['good', 'slow'].map(id => ({ id, state: 'active' })),
|
|
showOrg: async () => ({ doc: {} }), resolveOrg: async (_, id) => ({ id, state: 'active', root: dir }), orgPaths: () => ({ socket: join(dir, 'unused.sock') }),
|
|
busStatus: async (_, id) => { if (id === 'slow') await delay(statusDelay); return { healthy: true }; },
|
|
options: ({ id, ...bound }) => id === 'good' ? goodOptions() : consoleBus(busCaller({ dataRoot: dir, socket: bound.paths.socket, business: id, cli, timeoutMs, ...bound })) });
|
|
const body = await readWorkspace({ organizations, session, timeoutMs, query: parseWorkspaceQuery('sections=tasks') });
|
|
assert.equal(body.orgs.find(o => o.org === 'slow').state, 'timeout');
|
|
assert.deepEqual(body.summary, { read: 1, of: 2 });
|
|
const rows = async () => (await readFile(log, 'utf8')).trim().split('\n').map(JSON.parse);
|
|
const atReturn = await rows();
|
|
if (!statusDelay) {
|
|
await delay(30);
|
|
for (const { pid } of atReturn) assert.throws(() => process.kill(pid, 0), { code: 'ESRCH' }, `transport ${pid} still alive`);
|
|
}
|
|
await delay(350);
|
|
const later = await rows();
|
|
assert.deepEqual(later, atReturn, 'no transport starts after timeout');
|
|
assert.ok(atReturn.some(r => r.verb === (statusDelay ? 'plugins' : 'project.list')));
|
|
for (const { pid } of later) assert.throws(() => process.kill(pid, 0), { code: 'ESRCH' }, `transport ${pid} still alive`);
|
|
});
|
|
}
|
|
|
|
// The real status probe must release its socket as soon as its org deadline expires.
|
|
test('status probe closes its in-flight socket on cancellation', async t => {
|
|
const dir = await mkdtemp(join(tmpdir(), 'sock-')), path = join(dir, 's');
|
|
const server = createServer(socket => { socket.on('data', () => {}); });
|
|
await new Promise(resolve => server.listen(path, resolve));
|
|
t.after(async () => { await new Promise(resolve => server.close(resolve)); await rm(dir, { recursive: true, force: true }); });
|
|
const controller = new AbortController();
|
|
assert.equal(typeof orgBus.probeBusSocket, 'function');
|
|
const connected = new Promise(resolve => server.once('connection', socket => socket.once('close', resolve)));
|
|
const result = orgBus.probeBusSocket(path, { signal: controller.signal, deadline: Date.now() + 100 });
|
|
await delay(30);
|
|
controller.abort();
|
|
assert.equal(await result, false);
|
|
await Promise.race([connected, delay(250).then(() => { throw new Error('status socket still open'); })]);
|
|
});
|
|
|
|
test('workspace deadline reaches the organization status socket', async t => {
|
|
const dir = await mkdtemp(join(tmpdir(), 'sock-')), path = join(dir, 's');
|
|
let closed;
|
|
const socketClosed = new Promise(resolve => { closed = resolve; });
|
|
const server = createServer(socket => { socket.on('data', () => {}); socket.once('close', closed); });
|
|
await new Promise(resolve => server.listen(path, resolve));
|
|
t.after(async () => { await new Promise(resolve => server.close(resolve)); await rm(dir, { recursive: true, force: true }); });
|
|
const organizations = organizationConsole({ ctx: {}, listOrgs: async () => ['good', 'slow'].map(id => ({ id, state: 'active' })),
|
|
showOrg: async () => ({ doc: {} }), resolveOrg: async (_, id) => ({ id, state: 'active', root: dir }), orgPaths: () => ({ socket: path }),
|
|
busStatus: async (_, id, bound) => id === 'good' ? { healthy: true } : { healthy: await orgBus.probeBusSocket(path, bound) },
|
|
options: () => goodOptions() });
|
|
const body = await readWorkspace({ organizations, session, timeoutMs: 150 });
|
|
assert.equal(body.orgs.find(o => o.org === 'slow').state, 'timeout');
|
|
assert.deepEqual(body.summary, { read: 1, of: 2 });
|
|
await Promise.race([socketClosed, delay(250).then(() => { throw new Error('status socket still open'); })]);
|
|
});
|