From d35447005a75c63550d10d1a11973bf57e9194e0 Mon Sep 17 00:00:00 2001 From: Jason Woltje Date: Mon, 9 Nov 2026 18:07:40 -0600 Subject: [PATCH] Isolate invalid workspace reads and cancel timed-out org work --- packages/bus/tests/channel-proof.test.mjs | 4 +- packages/cli/src/org-bus.mjs | 17 ++- packages/webui/src/bus.mjs | 59 ++++----- packages/webui/src/cli.mjs | 4 +- packages/webui/src/organizations.mjs | 6 +- packages/webui/src/workspace-read.mjs | 52 ++++++-- .../webui/tests/workspace-failures.test.mjs | 114 ++++++++++++++++++ packages/webui/tests/workspace-read.test.mjs | 4 +- 8 files changed, 197 insertions(+), 63 deletions(-) create mode 100644 packages/webui/tests/workspace-failures.test.mjs diff --git a/packages/bus/tests/channel-proof.test.mjs b/packages/bus/tests/channel-proof.test.mjs index 02098242e..ee8e11e25 100644 --- a/packages/bus/tests/channel-proof.test.mjs +++ b/packages/bus/tests/channel-proof.test.mjs @@ -35,7 +35,9 @@ const baselines = { // Row 239 (#1745): outcome-unknown-revoked joins OPEN (duckworth-0165 C1), and STATUS gets // revision-conflict-revoked 409 and outcome-unknown-revoked 503; no proof or transport change. // #1839: artifact Console replies alone may use 2 MiB; the proof and other limits are unchanged. - "packages/webui/src/bus.mjs": "34fa64ffa17886d86243d7712fdb6f632c22b96412b642d785ddbc05fbd79692", + // #1873 U1: humanCall, busCaller, userCall and userCaller take an optional signal and absolute + // deadline; userCall shares transport(). Nonces, argv and limits are unchanged. + "packages/webui/src/bus.mjs": "3ef8064dd7911f9eee4475b52c74a1d551583a065474b1bcee968cd45f722747", }; const repoFile = path => new URL("../../../" + path, import.meta.url); diff --git a/packages/cli/src/org-bus.mjs b/packages/cli/src/org-bus.mjs index 6377f7ef2..1f866ba5c 100644 --- a/packages/cli/src/org-bus.mjs +++ b/packages/cli/src/org-bus.mjs @@ -43,12 +43,17 @@ function checkedState(org) { if (s.host && !lock) fail('bus host lock is missing'); return s; } -function answer(path) { +export function probeBusSocket(path, { signal, deadline } = {}) { return new Promise(resolve => { + const remaining = deadline === undefined ? 1000 : Math.min(1000, deadline - Date.now()); + if (signal?.aborted || remaining <= 0) return resolve(false); let done = false, data = ''; const socket = connect(path); - const finish = value => { if (done) return; done = true; socket.destroy(); resolve(value); }; - socket.setTimeout(1000, () => finish(false)); + const abort = () => finish(false); + const finish = value => { if (done) return; done = true; signal?.removeEventListener('abort', abort); socket.destroy(); resolve(value); }; + signal?.addEventListener('abort', abort, { once: true }); + if (signal?.aborted) return abort(); + socket.setTimeout(remaining, () => finish(false)); socket.on('error', () => finish(false)); socket.on('connect', () => socket.write('{}\n')); socket.on('data', chunk => { @@ -62,9 +67,9 @@ function answer(path) { socket.on('end', () => finish(false)); }); } -export async function busStatus(ctx, id) { +export async function busStatus(ctx, id, bound) { const org = await target(ctx, id), state = checkedState(org); - const healthy = Boolean(state.host?.live && state.socket && await answer(org.paths.socket)); + const healthy = Boolean(state.host?.live && state.socket && await probeBusSocket(org.paths.socket, bound)); return { org: id, instance: ctx.instance, root: org.root, socket: org.paths.socket, healthy, pid: state.host?.pid ?? null, boot: readBootStatus(org.root) }; } export async function busStop(ctx, id) { @@ -123,7 +128,7 @@ export async function busStart(ctx, id, { detach = false, devLocalOwner = false, while (Date.now() < deadline && !exited) { // During boot the host holds its lock before its state is published. const state = readHostState(org.root); - if (ready && state?.live && state.pid === child.pid && state.business === id && await answer(org.paths.socket)) { + if (ready && state?.live && state.pid === child.pid && state.business === id && await probeBusSocket(org.paths.socket)) { child.disconnect(); if (detach) child.unref(); const receipt = { org: id, pid: child.pid, socket: org.paths.socket, root: org.root }; diff --git a/packages/webui/src/bus.mjs b/packages/webui/src/bus.mjs index 593d12b15..cb87d3bac 100644 --- a/packages/webui/src/bus.mjs +++ b/packages/webui/src/bus.mjs @@ -27,19 +27,24 @@ export const subjectOk = s => typeof s === 'string' && s.length > 0 && s.length // Same protocol as humanTransport in packages/cli/src/transport.mjs: the // request on stdin, JSON on stdout, or a bus error code as the last // `^[a-z-]{1,64}$` line of stderr. -export function humanCall({ socket, business, env = process.env, cli = HUMAN_CLI, timeoutMs = 30000, maxBytes = 8 * 1024 * 1024 }) { - return (verb, args = {}) => transport({ cli, socket, env: cli === HUMAN_CLI ? { ...env, MOSAIC_BUS_CLI_NONCE: randomBytes(32).toString('hex') } : env, input: { business, verb, args }, timeoutMs, maxBytes }); +export function humanCall({ socket, business, env = process.env, cli = HUMAN_CLI, timeoutMs = 30000, maxBytes = 8 * 1024 * 1024, signal, deadline }) { + return (verb, args = {}) => transport({ cli, socket, signal, deadline, env: cli === HUMAN_CLI ? { ...env, MOSAIC_BUS_CLI_NONCE: randomBytes(32).toString('hex') } : env, input: { business, verb, args }, timeoutMs, maxBytes }); } -function transport({ cli, socket, env, input, timeoutMs, maxBytes }) { +function transport({ cli, socket, env, input, timeoutMs, maxBytes, signal, deadline }) { return new Promise((resolve, reject) => { + const remaining = deadline === undefined ? timeoutMs : Math.min(timeoutMs, deadline - Date.now()); + if (signal?.aborted || (deadline !== undefined && remaining <= 0)) return reject(new BusReadError('outcome-unknown', 'the bus transport deadline expired')); let child; try { child = spawn(process.execPath, [cli, socket], { env, stdio: ['pipe', 'pipe', 'pipe'] }); } catch { return reject(new BusReadError('outcome-unknown', 'the bus transport did not start')); } const out = [], err = []; let size = 0, settled = false; - const done = (fn, value) => { if (!settled) { settled = true; clearTimeout(timer); fn(value); } }; - const timer = setTimeout(() => { child.kill('SIGKILL'); done(reject, new BusReadError('outcome-unknown', 'the bus transport did not finish in time')); }, timeoutMs); + const done = (fn, value) => { if (!settled) { settled = true; clearTimeout(timer); signal?.removeEventListener('abort', abort); fn(value); } }; + const abort = () => { child.kill('SIGKILL'); done(reject, new BusReadError('outcome-unknown', 'the bus transport deadline expired')); }; + const timer = setTimeout(() => { child.kill('SIGKILL'); done(reject, new BusReadError('outcome-unknown', 'the bus transport did not finish in time')); }, remaining); + signal?.addEventListener('abort', abort, { once: true }); + if (signal?.aborted) abort(); child.stdout.on('data', chunk => { size += chunk.length; if (size > maxBytes) { child.kill('SIGKILL'); return done(reject, new BusReadError('response-too-large')); } @@ -82,9 +87,10 @@ export function consoleCall({ dataRoot, socket, business = null, env = process.e // The business is the --business flag, or else the live bus host's, read // on every request so a host started after the Console is picked up. -export function busCaller({ dataRoot, socket, business = null, env = process.env, cli = HUMAN_CLI, timeoutMs = 30000 }) { +export function busCaller({ dataRoot, socket, business = null, env = process.env, cli = HUMAN_CLI, timeoutMs = 30000, signal, deadline }) { return async (verb, args) => { - return humanCall({ socket, business: business || hostBusiness(dataRoot), env, cli, timeoutMs })(verb, args); + if (signal?.aborted || (deadline !== undefined && Date.now() >= deadline)) throw new BusReadError('outcome-unknown', 'the bus transport deadline expired'); + return humanCall({ socket, business: business || hostBusiness(dataRoot), env, cli, timeoutMs, signal, deadline })(verb, args); }; } function hostBusiness(dataRoot) { @@ -103,36 +109,10 @@ export const USER_CLI = fileURLToPath(new URL('../../bus/src/user-cli.mjs', impo // The Console sets the user nonce itself, so user-cli.mjs doesn't re-exec and the broker sees the // Console (`cli.mjs serve`) as its parent (principal.mjs). The human nonce is dropped, so this child // can never pass as human-cli, and it only ever runs USER_CLI. -export function userCall({ socket, business, user, env = process.env, timeoutMs = 30000, maxBytes = 8 * 1024 * 1024 }) { +export function userCall({ socket, business, user, env = process.env, timeoutMs = 30000, maxBytes = 8 * 1024 * 1024, signal, deadline }) { const { MOSAIC_BUS_CLI_NONCE, ...rest } = env; - return (verb, args = {}) => new Promise((resolve, reject) => { - let child; - try { - child = spawn(process.execPath, [USER_CLI, socket], { env: { ...rest, MOSAIC_BUS_USER_NONCE: randomBytes(32).toString('hex') }, stdio: ['pipe', 'pipe', 'pipe'] }); - } catch { return reject(new BusReadError('outcome-unknown', 'the bus transport did not start')); } - const out = [], err = []; - let size = 0, settled = false; - const done = (fn, value) => { if (!settled) { settled = true; clearTimeout(timer); fn(value); } }; - const timer = setTimeout(() => { child.kill('SIGKILL'); done(reject, new BusReadError('outcome-unknown', 'the bus transport did not finish in time')); }, timeoutMs); - child.stdout.on('data', chunk => { - size += chunk.length; - if (size > maxBytes) { child.kill('SIGKILL'); return done(reject, new BusReadError('response-too-large')); } - out.push(chunk); - }); - child.stderr.on('data', chunk => { if (err.length < 64) err.push(chunk); }); - child.on('error', () => done(reject, new BusReadError('outcome-unknown', 'the bus transport did not start'))); - child.on('close', status => { - if (status === null) return done(reject, new BusReadError('outcome-unknown', 'the bus transport did not finish')); - if (status !== 0) { - const code = Buffer.concat(err).toString('utf8').trim().split('\n').reverse().find(l => /^[a-z-]{1,64}$/.test(l)) ?? 'invalid-response'; - return done(reject, new BusReadError(code)); - } - try { done(resolve, JSON.parse(Buffer.concat(out).toString('utf8'))); } - catch { done(reject, new BusReadError('invalid-response')); } - }); - child.stdin.on('error', () => {}); // A child that exits early reports through 'close'. - child.stdin.end(`${JSON.stringify({ business, user, verb, args })}\n`); - }); + return (verb, args = {}) => transport({ cli: USER_CLI, socket, env: { ...rest, MOSAIC_BUS_USER_NONCE: randomBytes(32).toString('hex') }, + input: { business, user, verb, args }, timeoutMs, maxBytes, signal, deadline }); } // The business is the --business flag, or else the live bus host's, read on every call, as busCaller does. @@ -147,9 +127,12 @@ export function busBusiness({ dataRoot, business = null }) { } // One granted user's caller, for that user's session only. -export function userCaller({ dataRoot, socket, business = null, user, env = process.env, timeoutMs = 30000 }) { +export function userCaller({ dataRoot, socket, business = null, user, env = process.env, timeoutMs = 30000, signal, deadline }) { const which = busBusiness({ dataRoot, business }); - return async (verb, args) => userCall({ socket, business: which(), user, env, timeoutMs })(verb, args); + return async (verb, args) => { + if (signal?.aborted || (deadline !== undefined && Date.now() >= deadline)) throw new BusReadError('outcome-unknown', 'the bus transport deadline expired'); + return userCall({ socket, business: which(), user, env, timeoutMs, signal, deadline })(verb, args); + }; } // A send's error, sorted. The broker writes in one transaction, so any code it diff --git a/packages/webui/src/cli.mjs b/packages/webui/src/cli.mjs index 8853f9476..d0127c6ce 100644 --- a/packages/webui/src/cli.mjs +++ b/packages/webui/src/cli.mjs @@ -158,8 +158,8 @@ try { listOrgs, showOrg, resolveOrg, orgPaths, busStatus: orgBusStatus, // U1 (#1873): the cross-organization read passes `timeoutMs` so a slow bus's children are // killed at its deadline. The org pages pass none and keep the transport default. - options: ({ id, root, paths, session, timeoutMs }) => { - const bound = timeoutMs === undefined ? {} : { timeoutMs }; + options: ({ id, root, paths, session, timeoutMs, signal, deadline }) => { + const bound = timeoutMs === undefined ? {} : { timeoutMs, signal, deadline }; const owner = busCaller({ dataRoot: root, socket: paths.socket, business: id, ...bound }); const call = session.user === undefined ? owner : userCaller({ dataRoot: root, socket: paths.socket, business: id, user: session.user, ...bound }); return { diff --git a/packages/webui/src/organizations.mjs b/packages/webui/src/organizations.mjs index d8aa77db3..5a88e2d97 100644 --- a/packages/webui/src/organizations.mjs +++ b/packages/webui/src/organizations.mjs @@ -54,14 +54,14 @@ export function organizationConsole({ ctx, listOrgs, showOrg, resolveOrg, orgPat result.push({ id, doc: shown.doc, // Host paths in a refusal or boot reason become relative to the instance before they leave. - async status() { + async status(bound) { let s; - try { s = await busStatus(ctx, id); } + try { s = await busStatus(ctx, id, bound); } catch (error) { throw new Error(plainMessage(ctx, typeof error?.message === 'string' ? error.message : 'bus state could not be read')); } const boot = s?.boot && typeof s.boot === 'object' ? { ok: s.boot.ok, at: s.boot.at, lastGood: s.boot.lastGood ?? null, reason: s.boot.reason == null ? null : plainMessage(ctx, s.boot.reason) } : null; return { healthy: s?.healthy === true, pid: Number.isSafeInteger(s?.pid) ? s.pid : null, boot }; }, - options: () => options({ id, root: org.root, paths: orgPaths(org.root), session, context, doc: shown.doc, ...(timeoutMs === undefined ? {} : { timeoutMs }) }), + options: bound => options({ id, root: org.root, paths: orgPaths(org.root), session, context, doc: shown.doc, ...(timeoutMs === undefined ? {} : { timeoutMs }), ...bound }), }); } return result; diff --git a/packages/webui/src/workspace-read.mjs b/packages/webui/src/workspace-read.mjs index 3beecddb2..eba331237 100644 --- a/packages/webui/src/workspace-read.mjs +++ b/packages/webui/src/workspace-read.mjs @@ -85,9 +85,10 @@ export function parseWorkspaceQuery(search = '') { const TIMED_OUT = Symbol('timed-out'); function deadline(ms) { + const controller = new AbortController(), until = Date.now() + ms; let timer; - const promise = new Promise(resolve => { timer = setTimeout(() => resolve(TIMED_OUT), ms); timer.unref?.(); }); - return { promise, clear: () => clearTimeout(timer) }; + const promise = new Promise(resolve => { timer = setTimeout(() => { controller.abort(); resolve(TIMED_OUT); }, ms); timer.unref?.(); }); + return { promise, signal: controller.signal, deadline: until, abort: () => controller.abort(), clear: () => clearTimeout(timer) }; } const code = e => (typeof e?.code === 'string' && /^[a-z-]{1,64}$/.test(e.code) ? e.code : 'read-failed'); @@ -118,18 +119,20 @@ async function readOrg(member, needs, { timeoutMs, now, ctx }) { const head = { org: member.id, label: member.id, theme: member.doc?.appearance?.theme ?? null, tracker: member.doc?.tracker ? 'vikunja' : 'built-in', roles: Object.keys(member.doc?.roles ?? {}).sort() }; const finish = (state, extra = {}) => ({ ...head, state, at: iso(now()), readMs: now() - started, ...extra }); if (member.invalid) return { org: { ...head, roles: [], state: 'invalid', at: iso(now()), readMs: 0, reason: member.reason } }; - let expired = false; const limit = deadline(timeoutMs); + const bound = { signal: limit.signal, deadline: limit.deadline }; + const invalid = () => ({ org: finish('invalid', { error: 'invalid-response' }) }); try { let status; try { - status = await Promise.race([member.status(), limit.promise]); + status = await Promise.race([member.status(bound), limit.promise]); } catch (e) { + if (limit.signal.aborted || Date.now() >= limit.deadline) return { org: finish('timeout', { reason: `the bus state was not read within ${timeoutMs} ms` }) }; // A stale or mismatched host state (org-bus.mjs checkedState) is a stopped bus to the reader. // members() has already made host paths relative; any absolute path left is withheld. return { org: finish('down', { reason: plainMessage(ctx, typeof e?.message === 'string' ? e.message : 'bus state could not be read').slice(0, 300) }) }; } - if (status === TIMED_OUT) return { org: finish('timeout', { reason: `the bus state was not read within ${timeoutMs} ms` }) }; + if (status === TIMED_OUT || limit.signal.aborted || Date.now() >= limit.deadline) return { org: finish('timeout', { reason: `the bus state was not read within ${timeoutMs} ms` }) }; if (status?.healthy !== true) { const boot = status?.boot && typeof status.boot === 'object' ? { ok: status.boot.ok === true, at: status.boot.at ?? null, lastGood: status.boot.lastGood ?? null } : null; if (Number.isSafeInteger(status?.pid)) return { org: finish('starting', { reason: 'the bus host is running but does not answer yet', boot }) }; @@ -138,14 +141,41 @@ async function readOrg(member, needs, { timeoutMs, now, ctx }) { : 'the bus is not running'; return { org: finish('down', { reason, boot }) }; } - const options = member.options(); - const reads = pooled(needs, PER_ORG_CONCURRENCY, name => READS[name](options), () => expired); + const options = member.options(bound); + const reads = pooled(needs, PER_ORG_CONCURRENCY, name => READS[name](options), () => limit.signal.aborted || Date.now() >= limit.deadline); const results = await Promise.race([reads, limit.promise]); - if (results === TIMED_OUT) { expired = true; return { org: finish('timeout', { reason: `the bus did not answer within ${timeoutMs} ms` }) }; } + if (results === TIMED_OUT || limit.signal.aborted || Date.now() >= limit.deadline) return { org: finish('timeout', { reason: `the bus did not answer within ${timeoutMs} ms` }) }; const errors = Object.fromEntries(Object.entries(results).filter(([, r]) => !r.ok).map(([name, r]) => [name, r.error])); const data = Object.fromEntries(Object.entries(results).filter(([, r]) => r.ok).map(([name, r]) => [name, r.value])); - return { org: finish('ok', { errors }), data, doc: member.doc }; - } finally { limit.clear(); } + if (Object.values(errors).includes('invalid-response') || !Object.entries(data).every(([name, value]) => validRead(name, value))) return invalid(); + const org = finish('ok', { errors }); + return { org, rows: orgRows(org, data, member.doc, now()) }; + } catch { return invalid(); } finally { if (Date.now() >= limit.deadline) limit.abort(); limit.clear(); } +} + +// Validate the fields consumed by projection and by the cross-org filters and sorts. Unknown +// fields are retained by the bus contract; optional fields can be absent or null. +const record = v => v !== null && typeof v === 'object' && !Array.isArray(v); +const optional = (v, check) => v == null || check(v); +const string = v => typeof v === 'string'; +const strings = (v, keys) => keys.every(k => optional(v[k], string)); +const records = v => Array.isArray(v) && v.every(record); +function validRead(name, value) { + if (name === 'usage' || name === 'limits') { + if (!record(value)) return false; + return name === 'limits' ? optional(value.tokens, records) : optional(value.total, record); + } + if (!records(value)) return false; + return value.every(v => { + if (name === 'projects') return string(v.id) && strings(v, ['repo', 'name', 'at']); + if (name === 'threads') return strings(v, ['id', 'project']); + if (name === 'claims') return strings(v, ['thread', 'instance', 'holder_run', 'harness', 'at']); + if (name === 'inbox') return string(v.id) && strings(v, ['question', 'class', 'raised_by_role', 'task_ref', 'at']) && optional(v.options, a => records(a) && a.every(o => strings(o, ['key', 'text']))); + if (name === 'agents') return strings(v, ['role', 'holder_run', 'harness', 'state', 'at']) && optional(v.working, w => record(w) && strings(w, ['since'])) && ['inputNeeded', 'stranded'].every(k => optional(v[k], records)) && (v.inputNeeded ?? []).every(d => strings(d, ['task_ref', 'at'])); + if (name === 'tasks') return strings(v, ['task_ref', 'project_key', 'at']) && optional(v.fields, f => record(f) && strings(f, ['title', 'bucket']) && ['assignees', 'labels'].every(k => optional(f[k], Array.isArray))); + if (name === 'decisions') return strings(v, ['at']) && optional(v.closed, c => record(c) && strings(c, ['op', 'at'])); + return true; + }); } const arr = value => (Array.isArray(value) ? value : []); @@ -257,7 +287,7 @@ export async function readWorkspace({ organizations, session, query = parseWorks const merged = { projects: [], sessions: [], inbox: [], tasks: [], metrics: [] }; for (const r of read) { if (r.org.state !== 'ok') continue; - const rows = orgRows(r.org, r.data, r.doc, nowMs); + const rows = r.rows; for (const key of Object.keys(merged)) merged[key].push(...(key === 'metrics' ? [rows.metrics] : rows[key])); } const body = { diff --git a/packages/webui/tests/workspace-failures.test.mjs b/packages/webui/tests/workspace-failures.test.mjs new file mode 100644 index 000000000..9bd50d5e4 --- /dev/null +++ b/packages/webui/tests/workspace-failures.test.mjs @@ -0,0 +1,114 @@ +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'); })]); +}); diff --git a/packages/webui/tests/workspace-read.test.mjs b/packages/webui/tests/workspace-read.test.mjs index 64c589b80..8df92a8c5 100644 --- a/packages/webui/tests/workspace-read.test.mjs +++ b/packages/webui/tests/workspace-read.test.mjs @@ -23,7 +23,7 @@ const never = () => new Promise(() => {}); // One org's bus as the org pages see it: options().bus and options().chat. `over` replaces reads. function options(data, over = {}) { - const value = name => (over[name] ? over[name]() : Promise.resolve(data[name] ?? [])); + const value = name => (over[name] ? over[name]() : Promise.resolve(data[name] ?? (['usage', 'limits'].includes(name) ? {} : []))); return { bus: { inbox: () => value('inbox'), agents: () => value('agents'), 'thread.claims': () => value('claims'), tasks: () => value('tasks'), usage: () => value('usage'), limits: () => value('limits'), decisions: () => value('decisions') }, @@ -89,7 +89,7 @@ test('merge: every item from two buses carries its org, and the snapshot names b test('sections: only the asked sections are read and returned', async () => { const asked = []; - const tracked = member('acme', ACME, { over: Object.fromEntries(['inbox', 'agents', 'claims', 'threads', 'tasks', 'projects', 'usage', 'limits', 'decisions'].map(n => [n, () => { asked.push(n); return Promise.resolve(ACME[n] ?? []); }])) }); + const tracked = member('acme', ACME, { over: Object.fromEntries(['inbox', 'agents', 'claims', 'threads', 'tasks', 'projects', 'usage', 'limits', 'decisions'].map(n => [n, () => { asked.push(n); return Promise.resolve(ACME[n] ?? (['usage', 'limits'].includes(n) ? {} : [])); }])) }); const body = await readWorkspace({ organizations: organizationsOf(tracked), session: principal, query: parseWorkspaceQuery('sections=tasks'), now }); assert.deepEqual(Object.keys(body).filter(k => ['projects', 'sessions', 'inbox', 'tasks', 'metrics'].includes(k)), ['tasks']); assert.deepEqual(asked.sort(), ['inbox', 'projects', 'tasks']);