Isolate invalid workspace reads and cancel timed-out org work

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);
 

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 };

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

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 {

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;

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 = {

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'); })]);
+});

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']);