Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d35447005a | ||
|
|
2109458b3f | ||
|
|
1616d1dd6f | ||
|
|
7a97cfc4f3 |
No files matched your search
@@ -6343,3 +6343,5 @@ R5 quote fallback room (#797 D3, 2026-11-06, MVP3 lead): r4 review P2 found that
|
||||
2026-11-09 UTC | Organizations friction round, landings #1876-#1881 (#1873) | Before: refactor 00ef0bd30. A Console launched by a relative path showed Request status unavailable, the overview printed the north star as raw markdown, packages/business and the business command spellings were the only names, Quick Setup ended with no route to a north star, an archived org could not be restored from a fresh page, and a dead bus after setup had no browser control. After, each through one Codex Sol xhigh review round plus a lead or builder fix round: #1876 re-execs a relative Console launch with its absolute script path (build 744753d7e, reproduced 121/0; review APPROVE 0/0/1). #1877 F1 renders the north star as sanitized markdown with a short preview (review APPROVE 0/0/1; live retest 4/0 twice). #1878 rename stage B moves packages/business to packages/organization and adds the org command spellings, keeping the business ones and their messages (builder 6828/6, see report; review CHANGES 0/1/0, fixed in e19db4009). #1879 F2 adds the onboarding interview after Quick Setup and on the project page, with human-only north-star approval (`mosaic project north-star --approve`) and no new agent grant (builder 411/0; review CHANGES 0/1/1: a refused start left its button disabled, fixed in 5937a977f along with the P3 stale approval capability). #1880 lands the U0 unified workspace samples, approved by Jason with the four-tile logo kept; the module names stay undecided at his request. #1881 E2E round 5 fixes restore from a fresh page, show an owner bus Start whenever the org list reads the bus down, say room changes are unavailable on org Consoles and recover the composer (builder 258/0; review CHANGES 0/1/0, 231/0: Settings kept Done after the bus died because only opening the org reread health). Lead fix fdd6658a6: every org list read is announced and the open setup redraws on a flip, and app.js reads the list before bus-status so a failed bus-status read cannot skip it; regression red on 8a8edc44b, 18 touched webui files 199/0 (Node 26.12.1, one file at a time). uiorg redeployed at fdd6658a6, 19:53Z. Open follow-ups: a down bus offers both the F3 banner Restart and the Settings Start, focus can move when setup redraws, and a Console stop also stops the org buses it started. Reports: /mnt/storage/scratch/mosaic-lead/reports/org-{reqget-2,f1,rename-b,f2,u0-design,e2e5}-*.md.
|
||||
|
||||
2026-11-09 UTC, organizations F3 (#1873, PR #1884, a8f9f0957). The WebUI restarts the Console (supervised Consoles exit and `mosaic console start` relaunches them), restarts an organization's bus from the bus-down banner with a done receipt, and sends the owner's routing, agent, instance, name, assistant, limits and model proxy writes to the selected organization's bus. Routing apply lists how a change reaches each saved agent and offers Restart NAME now. Console restarts, bus restarts and routing, agent and instance changes append to `<dataRoot>/console/actions.jsonl`. Builder plus two refactor merges: combined list 354/0 on 999d173fd. Sol review CHANGES 1/0/1. P1: on a `--dev-local-owner` Console a local agent can send the right Origin and Fetch Metadata headers and reach these controls. The pinned startup warning and the accepted no-auth dev flags already cover this, but the README claimed no agent reaches them. Lead decision: correct the README and file the cooperative local-agent peer refusal as #1883, without building it in this round. P3: the instances route wrote no trail line. Fixed in 0a73279d6; the regression test fails on 999d173fd and passes on the fix. After the refactor merge, 25 touched or README-reading files 195/0 (Node 26.12.1, one file at a time, env -i). uiorg redeployed at a8f9f0957, 22:45Z; live config offers consoleRestart and org settingsSend. Correction to this session's process: after the test run completed I replied "No response requested" and stopped until Jason asked. Open follow-ups: defect l (a bus restart after adding a credentialed agent can refuse to start), the bus restart refusal code, a stale-lock Recover control, one control for banner Restart and Settings Start, #1883. Report: /mnt/storage/scratch/mosaic-lead/reports/org-f3-review.md.
|
||||
|
||||
2026-11-09 UTC, PRD 1.20 and defect l (#1873, #1886; PRs #1887 03fe321b1 and #1888 7a97cfc4f). PRD 1.20: Jason approved draft 2 at 22:57Z ("The file looks fine other than that"; the Montage module name stays deferred). The PRD now has 118 requirements, with REQ-WS-1..8, REQ-AGENT-1..3, REQ-SYS-1..2, REQ-ACCT-1..2 and REQ-SETUP-1..2 added and REQ-ASSIST-1, REQ-USER-1, REQ-TASK-1 and REQ-WEB-3 amended. Builder Codex Terra xhigh declined to commit because stack-copy failed 20/1 (node-gyp missing from the test Node's npm); that failure is pre-existing on 483eccb3d, so the lead committed. Sol review APPROVE 0 (word-for-word diff against draft 2, catalog order, both requirements tests). Defect l: before, an agent that declared a store credential made the next bus start refuse, because Credentials refuses a missing store root and only issue() created it. After: start creates the root 0700 when any reference names the store; a root that exists with loose modes or as a symlink still refuses. Two new credential-verbs tests fail on the parent. Bus suite 171 files one at a time, 1713/0 (Node 26.12.1, env -i). Sol review APPROVE 0. uiorg redeployed at 7a97cfc4f, 23:15Z. Follow-up: the stack-copy node-gyp test environment. Reports: /mnt/storage/scratch/mosaic-lead/reports/{prd120-land,prd120-review,defectl-review}.md.
|
||||
@@ -2122,3 +2122,4 @@ are never rewritten or removed; corrections are new entries.
|
||||
2026-11-07 UTC | Claude implementation sub-agent | Organizations S6 WebUI create and guided setup on org/s6 | Settings > Organizations: template form with JSON preview, C2 create/update with if-revision, setup cards for checkout, repositories and bus; webui suite 974/975 (inherited dev-owner test failure). Report and screenshots: /mnt/storage/scratch/mosaic-lead/uidev/org-release/report-s6.md. No push, merge or deployment.
|
||||
2026-11-07 UTC | Codex implementation sub-agent | Organizations release fix round 1 on ui-org-r1 | Local candidate for independent review; verification and findings recorded in /mnt/storage/scratch/mosaic-lead/uidev/org-release/report-fix-r1.md. Queue refused linked worktree; no push, merge or deployment.
|
||||
2026-11-09 UTC | Mos (lead, Claude Opus 5.5) | Organizations F3 fix round and landing (#1873) | PR #1884 merged a8f9f0957; review P3 fixed, P1 README corrected with follow-up #1883; 195/0 focused suite; uiorg redeployed 22:45Z.
|
||||
2026-11-09 UTC | Mos (lead, Claude Opus 5.5) | PRD 1.20 landing and defect l (#1873, #1886) | PR #1887 merged 03fe321b1, PR #1888 merged 7a97cfc4f; both Sol reviews APPROVE 0; bus suite 1713/0; uiorg redeployed 23:15Z; U1, U2 and U4a builds dispatched.
|
||||
+4
-1
@@ -311,7 +311,8 @@ scripts/mosaic decide <decision> <option> [--note <text>] [--yes]
|
||||
scripts/mosaic trail <task|decision> [--json]
|
||||
scripts/mosaic bus start <business> [--pm | --no-launch] [--dev-local-owner] | stop | status [--json]
|
||||
scripts/mosaic console start [--dev-local-owner [--dev-insecure-bind ADDR --dev-allow-host NAME[,NAME...]]] [--detach] [--port auto|N]
|
||||
scripts/mosaic console stop | status
|
||||
scripts/mosaic console stop | status | restart
|
||||
scripts/mosaic org reload <org>
|
||||
scripts/mosaic talk <role> <text> [--wait SECONDS] [--business <id>]
|
||||
scripts/mosaic project create <name> [--repo <repo>] | list [--json]
|
||||
scripts/mosaic threads [--project <project>] [--json] [--tester]
|
||||
@@ -332,6 +333,8 @@ For local development on a named instance only, `mosaic bus start BUSINESS --dev
|
||||
|
||||
`mosaic console start --dev-local-owner --dev-insecure-bind ADDR --dev-allow-host NAME[,NAME...]` opens that dev owner Console to other machines, for UI testing from another device on a trusted LAN. It has no sign-in: anyone who can reach ADDR:PORT acts as the owner. Start, and `mosaic console status` while that Console runs, print `WARNING: dev owner Console listens on ADDR:PORT with no sign-in; anyone who can reach it acts as the owner.` on stderr. ADDR is `0.0.0.0` or an IPv4 address, and a specific address must also be one of the names. NAME is a host name or IPv4 address without a port; give several as one comma-separated value. The Console answers only `NAME:<bound port>` and the loopback names, and takes a write only when Origin is `http://` plus that same Host. A browser on such an http origin sends no Fetch Metadata, so for an allowed NAME a write with that exact Origin and no `Sec-Fetch-Site` counts as same-origin; a `Sec-Fetch-Site` the browser did send is still checked. Both flags are refused without `--dev-local-owner`, and `--dev-insecure-bind` is refused without a `--dev-allow-host`. SSO, preview and tester Consoles stay on loopback. The Console state file records `bind` and `allowHosts`. `mosaic console status` prints the state as JSON (`bind` is `127.0.0.1` for a loopback Console) and never moves a stale state file.
|
||||
|
||||
`mosaic console restart` and `mosaic org reload ORG` are the Console's Settings controls (F3) from a terminal (U1, #1873). Each sends one request to the running Console named in the instance's console state, as the page sends it (the Console's own Host and Origin, `Sec-Fetch-Site: same-origin`, a JSON `{}` body), so the Console applies the same owner, `--send` and origin checks and writes the same action trail, `<dataRoot>/console/actions.jsonl`. Both refuse inside an agent run (exit 3) before reading anything, and refuse with exit 3 when no Console runs. `console restart` works only for a Console started in the foreground by `mosaic console start` (otherwise the Console answers `not-supervised`); it waits up to 30 seconds for the relaunched Console to answer and prints `{"outcome":"restarted","from":OLD,"boot":NEW}`. `org reload ORG` restarts a running bus or starts a stopped one and prints the Console's done receipt. A request that left but got no answer is reported as outcome unknown (exit 1) and is never retried; check `mosaic console status`, `mosaic bus status ORG` and the trail. `mosaic bus restart ORG` remains the trail-less host verb.
|
||||
|
||||
The instance processing switch is separate from per-business launches off/on.
|
||||
`mosaic halt` durably stops managed processing across every business in the
|
||||
selected configuration's data root. The bus stays up for reads and human
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { createRequirementsHost } from '../../tasks/src/builtin.mjs';
|
||||
import { createRequirementCatalog } from './requirement-catalog.mjs';
|
||||
import { readPrdMaterial } from '../../project/src/prd.mjs';
|
||||
import { lstatSync, readdirSync, readFileSync } from 'node:fs';
|
||||
import { lstatSync, mkdirSync, readdirSync, readFileSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
import { Store } from './store.mjs';
|
||||
@@ -360,6 +360,11 @@ export async function startBroker({
|
||||
// Row 236 (#1742, #436b r5 section 5): each gitea grant record gives its agent a store reference to
|
||||
// gitea, and `held` names the files memory must not load until reconcile or a revoke.
|
||||
const { held } = storeRoot ? giteaStart(store, storeRoot, businesses, references) : { held: [] };
|
||||
// #1886 (F3 defect l): a store reference with no file yet reads as no key (design D3a), but Credentials
|
||||
// refuses a missing store root, and only issue() made it. Start makes it the same way issue() does when
|
||||
// any reference names the store. A root that exists with loose modes or as a symlink still refuses.
|
||||
if (storeRoot && Object.values(references).some(refs => Object.values(refs ?? {}).some(ref => ref?.store === true)))
|
||||
mkdirSync(storeRoot, { recursive: true, mode: 0o700 });
|
||||
credentials = new Credentials({ references, repoRoots: protectedRoots, dataRoot, storeRoot, held,
|
||||
modelAgents: proxy ? Object.entries(businesses).flatMap(([id, b]) => Object.keys(b.roles).map(role => id + '/' + role)) : [] });
|
||||
if (proxy) {
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
// scratch root only.
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync } from 'node:fs';
|
||||
import { chmodSync, existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { startBroker } from '../src/runtime.mjs';
|
||||
@@ -18,7 +18,7 @@ const OTHER = 'fake_other_token_5555555555';
|
||||
const DATE = '2099-01-01';
|
||||
const role = { authority: { withinRole: ['message.send'], crossRole: [] } };
|
||||
|
||||
async function setup(t) {
|
||||
async function setup(t, { storeRoot = true } = {}) {
|
||||
const root = mkdtempSync(join(tmpdir(), 'bus-cred-verbs-'));
|
||||
t.after(() => rmSync(root, { recursive: true, force: true }));
|
||||
const rolesDir = rolesCopy(root);
|
||||
@@ -26,7 +26,7 @@ async function setup(t) {
|
||||
doc.roles.coder.credentials.gitea = { store: true, rotateBy: DATE };
|
||||
writeJson(join(root, 'businesses/demo.json'), doc);
|
||||
mkdirSync(join(root, 'data'), { mode: 0o700 });
|
||||
mkdirSync(join(root, 'secrets', 'agents'), { mode: 0o700 });
|
||||
if (storeRoot) mkdirSync(join(root, 'secrets', 'agents'), { mode: 0o700 });
|
||||
const store = (agent, service) => join(root, 'secrets', 'agents', 'demo', agent, service);
|
||||
// The bus's resolved business carries the same references; a store one names its derived file.
|
||||
const demo = {
|
||||
@@ -161,3 +161,29 @@ test('T5: issue, rotate, revoke and delete are live, rewrite the business file a
|
||||
for (const path of walk(c.root))
|
||||
for (const token of [TOKEN, ROTATED, OTHER]) assert.equal(readFileSync(path).includes(token), false, path);
|
||||
});
|
||||
|
||||
test('#1886: a store reference with no store root yet starts the bus, makes the root private and issues', async (t) => {
|
||||
const c = await setup(t, { storeRoot: false });
|
||||
const dir = join(c.root, 'secrets', 'agents');
|
||||
assert.equal(statSync(dir).mode & 0o777, 0o700);
|
||||
assert.equal(statSync(join(c.root, 'secrets')).mode & 0o077, 0);
|
||||
assert.deepEqual(c.status('coder'), { gitea: 'not-issued', vikunja: 'valid' });
|
||||
const issued = c.ask(c.human, 'credentials.issue', { agent: 'coder', service: 'gitea', token: TOKEN + '\n', date: DATE });
|
||||
assert.deepEqual(issued, { agent: 'coder', verb: 'issue', service: 'gitea', date: DATE, state: 'valid' });
|
||||
assert.equal(statSync(c.store('coder', 'gitea')).mode & 0o777, 0o600);
|
||||
});
|
||||
|
||||
test('#1886: a store root that exists with loose modes still refuses start', async (t) => {
|
||||
const root = mkdtempSync(join(tmpdir(), 'bus-cred-loose-'));
|
||||
t.after(() => rmSync(root, { recursive: true, force: true }));
|
||||
const rolesDir = rolesCopy(root);
|
||||
writeJson(join(root, 'businesses/demo.json'), { ...businessDoc(root, 'demo'), plugins: { harness: ['pi'] } });
|
||||
mkdirSync(join(root, 'data'), { mode: 0o700 });
|
||||
mkdirSync(join(root, 'secrets', 'agents'), { recursive: true, mode: 0o750 });
|
||||
chmodSync(join(root, 'secrets', 'agents'), 0o750);
|
||||
const demo = {
|
||||
id: 'demo', human: 'jason', arbiters: { technical: 'coder', delivery: 'coder' }, vars: {},
|
||||
roles: { coder: { ...role, credentials: { gitea: { store: true, file: join(root, 'secrets', 'agents', 'demo', 'coder', 'gitea'), rotateBy: DATE } } } },
|
||||
};
|
||||
await assert.rejects(startBroker({ dataRoot: join(root, 'data'), businesses: { demo }, organization: { dir: root, rolesDir, system: systemFor(root) } }), /credential-file/);
|
||||
});
|
||||
@@ -43,7 +43,7 @@ import { SeatError } from "../../seat/src/seat.mjs";
|
||||
import { OrgError, resolveOrg, withOrgLifecycle } from "../../organization/src/org-registry.mjs";
|
||||
import { orgSelector } from "./org-selector.mjs";
|
||||
import { busStart, busStatus, busStop, busRestart, busRecover } from "./org-bus.mjs";
|
||||
import { consoleStart, consoleStatus, consoleStop } from "./org-console.mjs";
|
||||
import { consoleRestart, consoleStart, consoleStatus, consoleStop } from "./org-console.mjs";
|
||||
|
||||
import { usageCommand, USAGE_USAGE } from "./usage.mjs";
|
||||
import { orgCommand, ORG_USAGE } from "./org.mjs";
|
||||
@@ -91,6 +91,7 @@ export const USAGE = `usage:
|
||||
mosaic bus status <id> [--json]
|
||||
mosaic console start [--dev-local-owner [--dev-insecure-bind ADDR --dev-allow-host NAME[,NAME...]]] [--detach] [--port auto|N]
|
||||
mosaic console stop|status
|
||||
mosaic console restart
|
||||
${DEV_USAGE}
|
||||
${RUN_USAGE}`;
|
||||
|
||||
@@ -571,6 +572,14 @@ async function consoleCommand(rest, io, deps) {
|
||||
io.stdout.write(`${JSON.stringify(consoleStatus(organizationContext(io, deps), { warn: s => io.stderr.write(s) }))}\n`);
|
||||
return;
|
||||
}
|
||||
// U1 (#1873): the F3 Settings control from a terminal. The running Console applies its own owner
|
||||
// and trail checks (org-console.mjs consoleRestart); an agent run is refused before any read.
|
||||
if (sub === "restart") {
|
||||
if (args.length) throw usage("console restart takes no arguments");
|
||||
refuseInsideAgent(io.env);
|
||||
io.stdout.write(`${JSON.stringify(await consoleRestart(organizationContext(io, deps), deps.consoleRequest ?? {}))}\n`);
|
||||
return;
|
||||
}
|
||||
if (sub !== "start") throw usage();
|
||||
const parsed = parse(args, { flags: ["--dev-local-owner", "--detach"], values: ["--port", "--dev-insecure-bind", "--dev-allow-host"] });
|
||||
if (parsed.positional.length) throw usage("console start takes no positional arguments");
|
||||
|
||||
@@ -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 };
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
import { randomBytes } from 'node:crypto';
|
||||
import { processState } from '../../seat/src/process-state.mjs';
|
||||
import { spawn } from 'node:child_process';
|
||||
import { request as httpRequest } from 'node:http';
|
||||
import { constants, closeSync, fstatSync, lstatSync, mkdirSync, openSync, readFileSync, rmSync, renameSync, writeFileSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
@@ -219,3 +220,79 @@ export function consoleStatus(ctx, { warn = s => process.stderr.write(s) } = {})
|
||||
if (running && s.bind) warn(lanWarning(bind, port));
|
||||
return { running, ...(running ? {} : { stale: true }), pid: s.pid, url: s.url, bind, ...(s.allowHosts ? { allowHosts: s.allowHosts } : {}) };
|
||||
}
|
||||
|
||||
// U1 (#1873): `mosaic console restart` and `mosaic org reload ID` are the F3 page controls from a
|
||||
// terminal. Each one is a request to the running Console named by console.json, sent as the page
|
||||
// sends it: the Console's own Host and Origin, Sec-Fetch-Site: same-origin and a JSON `{}` body.
|
||||
// The Console's route applies its owner, --send and origin checks and writes the action trail;
|
||||
// nothing here writes the trail, signals a process or touches a bus. The CLI verbs refuse inside
|
||||
// an agent run before they get here (transport.mjs refuseInsideAgent).
|
||||
const ORG_ID = /^[a-z0-9][a-z0-9._-]{0,63}$/;
|
||||
function consoleTarget(ctx) {
|
||||
const s = state(ctx);
|
||||
if (!s) fail('no Console runs for this instance; start one with mosaic console start');
|
||||
if (!liveConsole(s)) fail('the Console in console state is not running; see mosaic console status');
|
||||
const url = /^http:\/\/([0-9.]+):(\d+)\/$/.exec(s.url);
|
||||
if (!url) fail('console state is invalid');
|
||||
const port = Number(url[2]);
|
||||
if (forbidden.has(port)) fail('console state names a forbidden port');
|
||||
// A LAN Console on 0.0.0.0 answers on loopback too; one bound to an address answers only there.
|
||||
const address = s.bind && s.bind !== '0.0.0.0' ? s.bind : '127.0.0.1';
|
||||
return { address, port, host: `${address}:${port}` };
|
||||
}
|
||||
function consoleCall(target, { method = 'GET', path, timeoutMs }) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const headers = { host: target.host, ...(method === 'POST' ? { origin: `http://${target.host}`, 'sec-fetch-site': 'same-origin', 'content-type': 'application/json', 'content-length': 2 } : {}) };
|
||||
let sent = false;
|
||||
const req = httpRequest({ host: target.address, port: target.port, method, path, headers, timeout: timeoutMs }, res => {
|
||||
let text = '';
|
||||
res.setEncoding('utf8');
|
||||
res.on('data', chunk => { text += chunk; if (text.length > 65536) req.destroy(new Error('the Console answer is too large')); });
|
||||
res.on('end', () => { let body = null; try { body = JSON.parse(text); } catch { /* not JSON */ } resolve({ status: res.statusCode, body }); });
|
||||
res.on('error', e => reject(Object.assign(e, { sent: true })));
|
||||
});
|
||||
req.on('timeout', () => req.destroy(Object.assign(new Error('timed out'), { code: 'timeout' })));
|
||||
req.on('finish', () => { sent = true; });
|
||||
req.on('error', e => reject(Object.assign(e, { sent })));
|
||||
req.end(method === 'POST' ? '{}' : undefined);
|
||||
});
|
||||
}
|
||||
// A refused connection never reached the Console. Anything else after the request left may have
|
||||
// changed something, so it is outcome-unknown and nothing is retried.
|
||||
async function consolePost(target, path, { timeoutMs, unknown }) {
|
||||
let answer;
|
||||
try { answer = await consoleCall(target, { method: 'POST', path, timeoutMs }); }
|
||||
catch (e) {
|
||||
if (e.code === 'ECONNREFUSED' && !e.sent) fail('the Console refused the connection; see mosaic console status');
|
||||
throw new CliError(`outcome unknown: ${unknown}`, 1);
|
||||
}
|
||||
return answer;
|
||||
}
|
||||
function refusal(answer, what) {
|
||||
const code = typeof answer.body?.error === 'string' && /^[a-z-]{1,64}$/.test(answer.body.error) ? answer.body.error : `http ${answer.status}`;
|
||||
const message = typeof answer.body?.message === 'string' ? `: ${answer.body.message.slice(0, 500)}` : '';
|
||||
return new CliError(`${what} refused (${code})${message}`, answer.status >= 500 ? 1 : 3);
|
||||
}
|
||||
// POST /api/console/restart, then wait for the relaunched Console (a new boot id on /api/config).
|
||||
export async function consoleRestart(ctx, { timeoutMs = 10000, waitMs = 30000 } = {}) {
|
||||
const target = consoleTarget(ctx);
|
||||
const answer = await consolePost(target, '/api/console/restart', { timeoutMs, unknown: 'the Console may be restarting; check mosaic console status and the action trail' });
|
||||
if (answer.status !== 202 || answer.body?.outcome !== 'restarting' || typeof answer.body.boot !== 'string') throw refusal(answer, 'console restart');
|
||||
const from = answer.body.boot, deadline = Date.now() + waitMs;
|
||||
while (Date.now() < deadline) {
|
||||
await sleep(250);
|
||||
try {
|
||||
const config = await consoleCall(target, { path: '/api/config', timeoutMs: 2000 });
|
||||
if (config.status === 200 && typeof config.body?.boot === 'string' && config.body.boot !== from) return { outcome: 'restarted', from, boot: config.body.boot };
|
||||
} catch { /* closed while it restarts */ }
|
||||
}
|
||||
throw new CliError('the Console accepted the restart but has not answered since; check mosaic console status', 1);
|
||||
}
|
||||
// POST /api/orgs/ID/restart/bus: a running bus restarts, a stopped one starts (org-admin.mjs restartBus).
|
||||
export async function consoleBusRestart(ctx, id, { timeoutMs = 120000 } = {}) {
|
||||
if (typeof id !== 'string' || !ORG_ID.test(id)) throw new CliError('invalid organization id', 4);
|
||||
const target = consoleTarget(ctx);
|
||||
const answer = await consolePost(target, `/api/orgs/${id}/restart/bus`, { timeoutMs, unknown: `the bus may be restarting; check mosaic bus status ${id} and the action trail` });
|
||||
if (answer.status !== 200 || answer.body?.outcome !== 'done' || answer.body.step !== 'bus-restart') throw refusal(answer, 'org reload');
|
||||
return answer.body;
|
||||
}
|
||||
@@ -6,7 +6,9 @@ import { DEV_OWNER, devOwnerInstance } from './dev-local-owner.mjs';
|
||||
import { loadSystem } from './config.mjs';
|
||||
import { CliError } from './errors.mjs';
|
||||
import { busStopForDelete } from './org-bus.mjs';
|
||||
export const ORG_USAGE = 'mosaic org lock show|clear ID | archive ID --if-revision N [--json] | restore ID --if-revision N [--json] | delete ID --confirm ID [--if-revision N] [--json] | template render ID [--fixture local] --output FILE | validate FILE | create ID --from FILE | show ID [--json] | list [--json] | update ID --from FILE --if-revision N | queue init ID (--local-only --accept-local-loss-risk | --remote URL) | queue show ID [--json] | project repo init ID PROJECT (--local-only --accept-local-loss-risk | --remote URL) | project repo show ID PROJECT [--json] | config validate ID | config resolve ID INSTANCE [--project ID]';
|
||||
import { consoleBusRestart } from './org-console.mjs';
|
||||
import { refuseInsideAgent } from './transport.mjs';
|
||||
export const ORG_USAGE = 'mosaic org lock show|clear ID | reload ID | archive ID --if-revision N [--json] | restore ID --if-revision N [--json] | delete ID --confirm ID [--if-revision N] [--json] | template render ID [--fixture local] --output FILE | validate FILE | create ID --from FILE | show ID [--json] | list [--json] | update ID --from FILE --if-revision N | queue init ID (--local-only --accept-local-loss-risk | --remote URL) | queue show ID [--json] | project repo init ID PROJECT (--local-only --accept-local-loss-risk | --remote URL) | project repo show ID PROJECT [--json] | config validate ID | config resolve ID INSTANCE [--project ID]';
|
||||
function parse(args) { const pos = [], values = {}, flags = new Set(); for (let i = 0; i < args.length; i++) { const a = args[i]; if (['--json', '--local-only', '--accept-local-loss-risk'].includes(a)) { if (flags.has(a)) throw new CliError(`duplicate ${a}`, 4); flags.add(a); } else if (['--fixture', '--output', '--from', '--if-revision', '--remote', '--confirm'].includes(a)) { if (values[a] !== undefined || !args[i+1] || args[i+1].startsWith('--')) throw new CliError(`${a} needs one value`, 4); values[a] = args[++i]; } else if (a.startsWith('--')) throw new CliError(`unknown option ${a}`, 4); else pos.push(a); } return { pos, values, flags }; }
|
||||
async function document(file) { try { return JSON.parse(await readFile(file, 'utf8')); } catch (e) { throw new CliError(`cannot read document ${file}: ${e.message}`, 2); } }
|
||||
export async function orgCommand(args, io, deps = {}) {
|
||||
@@ -16,12 +18,17 @@ export async function orgCommand(args, io, deps = {}) {
|
||||
const need = key => { if (!p.values[key]) throw new CliError(`${key} is required`, 4); return p.values[key]; };
|
||||
if (words[0] === 'template' && words[1] === 'render' && words.length === 3) { allowed([], ['--fixture', '--output']); result = await templateRender(words[2], { fixture: p.values['--fixture'] }); await writeFile(need('--output'), JSON.stringify(result, null, 2) + '\n', { flag: 'wx', mode: 0o600 }); io.stdout.write(`rendered ${words[2]} to ${p.values['--output']}\n`); return; }
|
||||
if (words[0] === 'validate' && words.length === 2) { allowed([], []); result = await validateOrg(await document(words[1])); if (!result.ok) throw new CliError(result.errors.map(e => `${e.path}: ${e.message}`).join('\n'), 2); io.stdout.write('valid\n'); return; }
|
||||
// U1 (#1873): `org reload` is a human action; an agent run is refused before any config is read.
|
||||
if (words[0] === 'reload') refuseInsideAgent(io.env);
|
||||
const instance = devOwnerInstance(io.env); const system = deps.system ?? loadSystem({ env: io.env, instance }); const ctx = { configDir: configDir(io.env), dataRoot: system.dataRoot, instance, principal: DEV_OWNER };
|
||||
if (words[0] === 'lock' && ['show', 'clear'].includes(words[1]) && words.length === 3) { allowed([], []); result = await orgLock(ctx, words[2], { clear: words[1] === 'clear' }); }
|
||||
else if (words[0] === 'create' && words.length === 2) { allowed([], ['--from']); result = await createOrg(ctx, words[1], await document(need('--from'))); }
|
||||
else if (words[0] === 'update' && words.length === 2) { allowed([], ['--from', '--if-revision']); const r = need('--if-revision'); if (!/^[1-9][0-9]*$/.test(r)) throw new CliError('invalid revision', 4); result = await updateOrg(ctx, words[1], await document(need('--from')), { ifRevision: Number(r) }); }
|
||||
else if (['archive', 'restore'].includes(words[0]) && words.length === 2) { allowed(['--json'], ['--if-revision']); const r = need('--if-revision'); if (!/^[1-9][0-9]*$/.test(r)) throw new CliError('invalid revision', 4); const opts = { ifRevision: Number(r) }; result = words[0] === 'archive' ? await archiveOrg(ctx, words[1], { ...opts, busStop: busStopForDelete }) : await restoreOrg(ctx, words[1], opts); }
|
||||
else if (words[0] === 'delete' && words.length === 2) { allowed(['--json'], ['--confirm', '--if-revision']); const r = p.values['--if-revision']; if (r !== undefined && !/^[1-9][0-9]*$/.test(r)) throw new CliError('invalid revision', 4); result = await deleteOrg(ctx, words[1], { busStop: busStopForDelete, confirm: need('--confirm'), ifRevision: r === undefined ? undefined : Number(r) }); }
|
||||
// U1 (#1873): the F3 bus control from a terminal, through the running Console's own route, which
|
||||
// checks the owner and writes the action trail (org-console.mjs consoleBusRestart).
|
||||
else if (words[0] === 'reload' && words.length === 2) { allowed([], []); result = await consoleBusRestart(ctx, words[1], deps.consoleRequest ?? {}); }
|
||||
else if (words[0] === 'show' && words.length === 2) { allowed(['--json'], []); result = await showOrg(ctx, words[1]); }
|
||||
else if (words[0] === 'list' && words.length === 1) { allowed(['--json'], []); for (const entry of await listOrgs(ctx)) if (entry.state === 'active' || entry.state === 'archived') accepted.push(await showOrg(ctx, entry.id, { includeArchived: true })); result = accepted; }
|
||||
else if ((words[0] === 'queue' && words.length === 3) || (words[0] === 'project' && words[1] === 'repo' && words.length === 5)) {
|
||||
|
||||
@@ -0,0 +1,187 @@
|
||||
// U1 (#1873): `mosaic console restart` and `mosaic org reload ID` are the F3 page controls from a
|
||||
// terminal. Each test runs the real scripts/mosaic on a scratch instance made by bootstrap.sh, and
|
||||
// checks the refusal before any read, the request to the Console's own route and the action trail
|
||||
// that route writes. The supervised Console is the real one; the org reload success path runs
|
||||
// against an in-process Console whose bus service is a fake, so no bus host starts.
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { spawn, spawnSync } from 'node:child_process';
|
||||
import { chmodSync, existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, writeFileSync } from 'node:fs';
|
||||
import { request } from 'node:http';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { reserveOrg, activateOrg } from '../../organization/src/org-registry.mjs';
|
||||
import { processState } from '../../seat/src/process-state.mjs';
|
||||
import { startServer } from '../../webui/src/serve.mjs';
|
||||
import { createDevOwner } from '../../webui/src/dev-owner.mjs';
|
||||
import { createOrgAdmin } from '../../webui/src/org-admin.mjs';
|
||||
import { actionTrail } from '../../webui/src/action-trail.mjs';
|
||||
import { close } from '../../webui/tests/fixture.mjs';
|
||||
import { humanEnv } from './helpers.mjs';
|
||||
import { consoleStart, consoleStop } from '../src/org-console.mjs';
|
||||
|
||||
const REPO = new URL('../../../', import.meta.url).pathname;
|
||||
const MOSAIC = join(REPO, 'scripts/mosaic');
|
||||
const FORBIDDEN = [7332, 7340, 7341, 33223, 17450, 17443, 44307, 7350];
|
||||
const DEV = { provider: 'local-dev', iss: 'local-dev', sub: 'local-dev-owner', name: 'Local operator' };
|
||||
|
||||
// A scratch instance, as bootstrap.sh makes one, with org alpha active.
|
||||
async function instance(t) {
|
||||
const home = mkdtempSync(join(tmpdir(), 'u1-'));
|
||||
const env = humanEnv({ HOME: home, MOSAIC_CONFIG: join(home, '.config/mosaic-u1test/config.json') });
|
||||
const setup = spawnSync(join(REPO, 'scripts/bootstrap.sh'), ['--instance', 'u1test'], { env, encoding: 'utf8' });
|
||||
assert.equal(setup.status, 0, setup.stderr);
|
||||
const ctx = { configDir: join(home, '.config/mosaic-u1test'), dataRoot: join(home, '.mosaic-u1test'), instance: 'u1test', env };
|
||||
await reserveOrg(ctx, 'alpha'); await activateOrg(ctx, 'alpha');
|
||||
t.after(async () => { try { await consoleStop(ctx); } catch {} rmSync(home, { recursive: true, force: true }); });
|
||||
return { ctx, env, trailFile: join(ctx.dataRoot, 'console/actions.jsonl') };
|
||||
}
|
||||
// The CLI as a child, without blocking this process: the in-process Console must keep answering.
|
||||
function mosaic(env, args, timeout = 60000) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const child = spawn(MOSAIC, args, { env, stdio: ['ignore', 'pipe', 'pipe'] });
|
||||
let stdout = '', stderr = '';
|
||||
child.stdout.setEncoding('utf8').on('data', c => { stdout += c; });
|
||||
child.stderr.setEncoding('utf8').on('data', c => { stderr += c; });
|
||||
const timer = setTimeout(() => child.kill('SIGKILL'), timeout);
|
||||
child.on('error', reject);
|
||||
child.on('close', status => { clearTimeout(timer); resolve({ status, stdout, stderr }); });
|
||||
});
|
||||
}
|
||||
const trailLines = file => existsSync(file) ? readFileSync(file, 'utf8').split('\n').filter(Boolean).map(l => { const { at, ...e } = JSON.parse(l); assert.ok(Number.isFinite(Date.parse(at))); return e; }) : [];
|
||||
function get(port, path) {
|
||||
return new Promise((resolve, reject) => {
|
||||
request({ host: '127.0.0.1', port, path, headers: { host: `127.0.0.1:${port}` } }, res => {
|
||||
let text = ''; res.setEncoding('utf8'); res.on('data', c => { text += c; });
|
||||
res.on('end', () => resolve({ status: res.statusCode, body: JSON.parse(text) }));
|
||||
}).on('error', reject).end();
|
||||
});
|
||||
}
|
||||
// console.json naming a Console that runs in this test process (same checks as a real one).
|
||||
function recordConsole(ctx, url, { pid = process.pid, startTime = processState(process.pid).startTime } = {}) {
|
||||
const dir = join(ctx.dataRoot, 'console');
|
||||
mkdirSync(dir, { recursive: true, mode: 0o700 }); chmodSync(dir, 0o700);
|
||||
const file = join(dir, 'console.json');
|
||||
writeFileSync(file, JSON.stringify({ pid, startTime, url }), { mode: 0o600 }); chmodSync(file, 0o600);
|
||||
}
|
||||
|
||||
test('both verbs refuse inside an agent run before any config is read', async () => {
|
||||
// The config path is not an instance config, so a read would refuse with another message.
|
||||
// humanEnv strips agent markers, so the marker goes in after it.
|
||||
const env = { ...humanEnv({ MOSAIC_CONFIG: '/nonexistent/u1/config.json' }), CLAUDECODE: '1' };
|
||||
for (const args of [['console', 'restart'], ['org', 'reload', 'alpha']]) {
|
||||
const r = await mosaic(env, args);
|
||||
assert.equal(r.status, 3, `${args.join(' ')}: ${r.stderr}`);
|
||||
assert.match(r.stderr, /outside any agent run \(found CLAUDECODE/);
|
||||
assert.equal(r.stdout, '');
|
||||
}
|
||||
});
|
||||
|
||||
test('with no live Console both verbs refuse and nothing is written', { timeout: 90000 }, async t => {
|
||||
const { ctx, env, trailFile } = await instance(t);
|
||||
for (const args of [['console', 'restart'], ['org', 'reload', 'alpha']]) {
|
||||
const r = await mosaic(env, args);
|
||||
assert.equal(r.status, 3, `${args.join(' ')}: ${r.stderr}`);
|
||||
assert.match(r.stderr, /no Console runs for this instance/);
|
||||
}
|
||||
recordConsole(ctx, 'http://127.0.0.1:1/', { pid: 2147483647, startTime: '1' });
|
||||
const stale = await mosaic(env, ['console', 'restart']);
|
||||
assert.equal(stale.status, 3, stale.stderr);
|
||||
assert.match(stale.stderr, /is not running/);
|
||||
const bad = await mosaic(env, ['org', 'reload', 'Bad/Id']);
|
||||
assert.equal(bad.status, 4, bad.stderr);
|
||||
assert.match(bad.stderr, /invalid organization id/);
|
||||
assert.equal((await mosaic(env, ['console', 'restart', 'extra'])).status, 4);
|
||||
assert.equal(existsSync(trailFile), false);
|
||||
});
|
||||
|
||||
test('mosaic console restart relaunches a supervised Console and the trail records it', { timeout: 120000 }, async t => {
|
||||
const { ctx, env, trailFile } = await instance(t);
|
||||
const warnings = [];
|
||||
const result = await consoleStart(ctx, { devLocalOwner: true, port: 'auto', warn: s => warnings.push(s) });
|
||||
t.after(async () => { result.stop(); await result.done; });
|
||||
const port = Number(new URL(result.url).port);
|
||||
assert.ok(!FORBIDDEN.includes(port));
|
||||
const before = (await get(port, '/api/config')).body;
|
||||
assert.equal(before.consoleRestart, true);
|
||||
const r = await mosaic(env, ['console', 'restart']);
|
||||
assert.equal(r.status, 0, r.stderr);
|
||||
const answer = JSON.parse(r.stdout);
|
||||
assert.equal(answer.outcome, 'restarted'); assert.equal(answer.from, before.boot);
|
||||
assert.equal(typeof answer.boot, 'string'); assert.notEqual(answer.boot, before.boot);
|
||||
assert.equal((await get(port, '/api/config')).body.boot, answer.boot);
|
||||
assert.deepEqual(trailLines(trailFile), [{ by: DEV, action: 'console.restart', outcome: 'restarting', boot: before.boot }]);
|
||||
assert.equal(statSync(trailFile).mode & 0o777, 0o600);
|
||||
assert.deepEqual(warnings, ['mosaic: console asked for a restart; relaunching on the same arguments\n']);
|
||||
});
|
||||
|
||||
test('a detached Console refuses mosaic console restart, and org reload of an unknown org, with no trail', { timeout: 90000 }, async t => {
|
||||
const { ctx, env, trailFile } = await instance(t);
|
||||
const result = await consoleStart(ctx, { devLocalOwner: true, detach: true, port: 'auto', warn: () => {} });
|
||||
const port = Number(new URL(result.url).port);
|
||||
assert.ok(!FORBIDDEN.includes(port));
|
||||
const r = await mosaic(env, ['console', 'restart']);
|
||||
assert.equal(r.status, 3, r.stderr);
|
||||
assert.match(r.stderr, /console restart refused \(not-supervised\)/);
|
||||
// The real organization service refuses an org that does not exist before any trail line.
|
||||
const unknown = await mosaic(env, ['org', 'reload', 'nosuch']);
|
||||
assert.equal(unknown.status, 3, unknown.stderr);
|
||||
assert.match(unknown.stderr, /org reload refused \(org-[a-z-]+\)/);
|
||||
assert.equal((await get(port, '/api/config')).status, 200);
|
||||
assert.equal(existsSync(trailFile), false);
|
||||
assert.equal((await consoleStop(ctx)).stopped, true);
|
||||
});
|
||||
|
||||
// The F3 bus control through the Console's own route: a stopped bus starts and a running one restarts.
|
||||
function fakeAdmin(ctx) {
|
||||
const calls = [];
|
||||
let running = false;
|
||||
const known = id => { if (id !== 'acme') throw Object.assign(new Error('unknown organization'), { code: 'org-unknown' }); };
|
||||
const service = { showOrg: async (c, id) => { known(id); return { id, root: join(ctx.dataRoot, 'org-runtime', id), revision: 1, doc: {}, principal: c.principal }; } };
|
||||
const bus = {
|
||||
busStatus: async (c, id) => { known(id); return { org: id, instance: ctx.instance, root: join(ctx.dataRoot, 'org-runtime', id), socket: '/run/u1.sock', healthy: running, pid: running ? 4242 : null }; },
|
||||
busStart: async (c, id, opts) => { calls.push(['busStart', id, opts]); running = true; return { org: id, pid: 4242, socket: '/run/u1.sock', root: join(ctx.dataRoot, 'org-runtime', id) }; },
|
||||
busRestart: async (c, id) => { calls.push(['busRestart', id]); return { org: id, pid: 4343, socket: '/run/u1.sock', root: join(ctx.dataRoot, 'org-runtime', id) }; },
|
||||
};
|
||||
const registry = { listOrgs: async () => [{ id: 'acme', state: 'active' }] };
|
||||
return { calls, admin: createOrgAdmin({ service, registry, bus, context: async () => ({ configDir: ctx.configDir, dataRoot: ctx.dataRoot, instance: ctx.instance }) }) };
|
||||
}
|
||||
|
||||
test('mosaic org reload starts or restarts the bus through the Console route and the trail records each', { timeout: 90000 }, async t => {
|
||||
const { ctx, env, trailFile } = await instance(t);
|
||||
const f = fakeAdmin(ctx);
|
||||
const server = await startServer({ port: 0, board: 'http://127.0.0.1:9', sso: createDevOwner(), send: async () => ({}), orgAdmin: f.admin, log: () => {}, trail: actionTrail(join(ctx.dataRoot, 'console')) });
|
||||
t.after(() => close(server));
|
||||
const port = server.address().port;
|
||||
assert.ok(!FORBIDDEN.includes(port));
|
||||
recordConsole(ctx, `http://127.0.0.1:${port}/`);
|
||||
let r = await mosaic(env, ['org', 'reload', 'acme']);
|
||||
assert.equal(r.status, 0, r.stderr);
|
||||
assert.deepEqual(JSON.parse(r.stdout), { outcome: 'done', step: 'bus-restart', restarted: false, healthy: true, receipt: { org: 'acme', pid: 4242, socket: '<host path>', root: '<data root>/org-runtime/acme' }, verb: 'mosaic bus start acme --dev-local-owner --detach' });
|
||||
r = await mosaic(env, ['org', 'reload', 'acme']);
|
||||
assert.equal(r.status, 0, r.stderr);
|
||||
const second = JSON.parse(r.stdout);
|
||||
assert.equal(second.restarted, true); assert.equal(second.verb, 'mosaic bus restart acme'); assert.equal(second.receipt.pid, 4343);
|
||||
assert.deepEqual(f.calls, [['busStart', 'acme', { detach: true, devLocalOwner: true, pm: false }], ['busRestart', 'acme']]);
|
||||
// An unknown org is refused by the route before the trail (as org-unavailable, which does not say
|
||||
// whether it exists); extra arguments never reach the Console.
|
||||
const unknown = await mosaic(env, ['org', 'reload', 'other']);
|
||||
assert.equal(unknown.status, 3, unknown.stderr);
|
||||
assert.match(unknown.stderr, /org reload refused \(org-unavailable\)/);
|
||||
assert.equal((await mosaic(env, ['org', 'reload', 'acme', 'extra'])).status, 4);
|
||||
assert.deepEqual(trailLines(trailFile), [
|
||||
{ by: DEV, action: 'bus.restart', org: 'acme', outcome: 'requested', running: false },
|
||||
{ by: DEV, action: 'bus.restart', org: 'acme', outcome: 'done' },
|
||||
{ by: DEV, action: 'bus.restart', org: 'acme', outcome: 'requested', running: true },
|
||||
{ by: DEV, action: 'bus.restart', org: 'acme', outcome: 'done' },
|
||||
]);
|
||||
// A read-only Console (no --send) refuses, and the CLI reports the Console's own reason.
|
||||
await close(server);
|
||||
const readonly = await startServer({ port: 0, board: 'http://127.0.0.1:9', sso: createDevOwner(), orgAdmin: f.admin, log: () => {}, trail: actionTrail(join(ctx.dataRoot, 'console')) });
|
||||
t.after(() => close(readonly));
|
||||
recordConsole(ctx, `http://127.0.0.1:${readonly.address().port}/`);
|
||||
const denied = await mosaic(env, ['org', 'reload', 'acme']);
|
||||
assert.equal(denied.status, 3, denied.stderr);
|
||||
assert.match(denied.stderr, /org reload refused \(read-only\): start the Console with --send/);
|
||||
assert.equal(trailLines(trailFile).length, 4);
|
||||
});
|
||||
+21
-38
@@ -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
|
||||
|
||||
@@ -156,9 +156,12 @@ try {
|
||||
const organizations = devOwner ? organizationConsole({
|
||||
ctx: { configDir: configDir(), dataRoot: system.dataRoot, instance },
|
||||
listOrgs, showOrg, resolveOrg, orgPaths, busStatus: orgBusStatus,
|
||||
options: ({ id, root, paths, session }) => {
|
||||
const owner = busCaller({ dataRoot: root, socket: paths.socket, business: id });
|
||||
const call = session.user === undefined ? owner : userCaller({ dataRoot: root, socket: paths.socket, business: id, user: session.user });
|
||||
// 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, 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 {
|
||||
...consoleBus(call, { send: sendOn, ...(session.user === undefined ? {} : { owner }) }),
|
||||
decisionWrite: session.user === undefined && sendOn ? consoleCall({ dataRoot: root, socket: paths.socket, business: id }) : null,
|
||||
|
||||
@@ -35,8 +35,40 @@ export function organizationConsole({ ctx, listOrgs, showOrg, resolveOrg, orgPat
|
||||
}
|
||||
return result;
|
||||
}
|
||||
// U1 (#1873): the orgs a session belongs to, for the cross-organization read (workspace-read.mjs).
|
||||
// The same admission as `list` and `resolve`: an org the session does not own is skipped, and a
|
||||
// broken org shows only to its verified owner, as invalid. Bus status and options are lazy, so a
|
||||
// stopped bus never gets a transport; `timeoutMs` bounds each bus read the options make.
|
||||
async function members(session, { timeoutMs } = {}) {
|
||||
const entries = await listOrgs(ctx), result = [];
|
||||
for (const entry of entries) {
|
||||
if (entry.state !== 'active') continue;
|
||||
let verified;
|
||||
try { verified = await admitted(entry.id, session); }
|
||||
catch (error) {
|
||||
if (error?.ownerVerified !== true) continue;
|
||||
result.push({ id: entry.id, invalid: true, reason: publicReason(ctx, error) });
|
||||
continue;
|
||||
}
|
||||
const { org, shown, context } = verified, id = entry.id;
|
||||
result.push({
|
||||
id, doc: shown.doc,
|
||||
// Host paths in a refusal or boot reason become relative to the instance before they leave.
|
||||
async status(bound) {
|
||||
let s;
|
||||
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: bound => options({ id, root: org.root, paths: orgPaths(org.root), session, context, doc: shown.doc, ...(timeoutMs === undefined ? {} : { timeoutMs }), ...bound }),
|
||||
});
|
||||
}
|
||||
return result;
|
||||
}
|
||||
return Object.freeze({
|
||||
list,
|
||||
members,
|
||||
async resolve(id, session) {
|
||||
const { org, shown, context } = await admitted(id, session);
|
||||
let busHealthy = false;
|
||||
|
||||
@@ -20,6 +20,7 @@ import { REQUEST_ROUTE, requestHTTP, requestRefuse } from './request-http.mjs';
|
||||
import { PROJECT_REMOTE_ROUTE, projectRemoteHTTP } from './project-remote-http.mjs';
|
||||
import { orgAdminHTTP, defaultOrgAdmin } from './org-admin.mjs';
|
||||
import { consoleRestartHTTP } from './console-restart-http.mjs';
|
||||
import { workspaceHTTP } from './workspace-read.mjs';
|
||||
import { actor } from './action-trail.mjs';
|
||||
import { devHostName } from '../../cli/src/dev-local-owner.mjs';
|
||||
|
||||
@@ -770,7 +771,7 @@ async function libraryStatusRead(res, read, search) {
|
||||
// `log` takes one line per send attempt that got past sign-in; never the body. `fieldsFile` is for
|
||||
// tests; the field table is otherwise the committed packages/tasks/fields.json.
|
||||
// `agentWrite` (row 194, D7) is `(verb, args) => Promise` for the Agents section's verbs and instance.request.
|
||||
export async function startServer({ host = '127.0.0.1', port = 7330, board = DEFAULT_BOARD, timeout = 20000, bus = null, reads = null, preview = null, sso = null, tester = false, testerSocket = null, testerAuthHeader = null, proof = null, testerProofHeader = null, send = null, settingsWrite = null, projectFiles = null, projectFile = null, projectSave = null, libraryStatus = null, libraryRevert = null, chat = null, nameWrite = null, assistantStop = null, instancesWrite = null, routesWrite = null, agentWrite = null, proxyCall = null, processing = null, decisionWrite = null, requestRead = null, requestWrite = null, userOptions = null, federation = null, organizations = null, preselect = null, routes = [], orgAdmin = null, devAllowHosts = [], log = line => console.log(line), fieldsFile = undefined, restart = null, trail = null, boot = randomUUID() } = {}) {
|
||||
export async function startServer({ host = '127.0.0.1', port = 7330, board = DEFAULT_BOARD, timeout = 20000, bus = null, reads = null, preview = null, sso = null, tester = false, testerSocket = null, testerAuthHeader = null, proof = null, testerProofHeader = null, send = null, settingsWrite = null, projectFiles = null, projectFile = null, projectSave = null, libraryStatus = null, libraryRevert = null, chat = null, nameWrite = null, assistantStop = null, instancesWrite = null, routesWrite = null, agentWrite = null, proxyCall = null, processing = null, decisionWrite = null, requestRead = null, requestWrite = null, userOptions = null, federation = null, organizations = null, preselect = null, routes = [], orgAdmin = null, devAllowHosts = [], log = line => console.log(line), fieldsFile = undefined, restart = null, trail = null, boot = randomUUID(), workspaceTimeoutMs = undefined } = {}) {
|
||||
// org-lan-bind: the dev owner on the LAN is an explicit opt-in. `devAllowHosts` names the Host
|
||||
// names (without a port) a remote browser may use; only the dev owner takes them, and only with
|
||||
// them may it bind a non-loopback IPv4 address. SSO, preview and tester listeners stay loopback.
|
||||
@@ -805,6 +806,8 @@ export async function startServer({ host = '127.0.0.1', port = 7330, board = DEF
|
||||
// F3: the owner restarts the Console (console-restart-http.mjs). `restart` is set only under the
|
||||
// supervisor; `boot` names this process so an open page can tell when the new one answers.
|
||||
registered.add('POST', '/api/console/restart', (req, res, context) => consoleRestartHTTP(req, res, { ...context, restart, trail, boot, log }));
|
||||
// U1 (#1873): one read across every organization the owner belongs to (workspace-read.mjs).
|
||||
registered.add('GET', '/api/workspace', (req, res, context) => workspaceHTTP(req, res, { ...context, organizations, log, ...(workspaceTimeoutMs === undefined ? {} : { timeoutMs: workspaceTimeoutMs }) }));
|
||||
const upstream = boardURL(board);
|
||||
// `userOptions` (FED R6, #1800) is `session => options`, the bus-backed options over that session
|
||||
// user's own transport (console-bus.mjs). Only a signed-in listener has user sessions.
|
||||
|
||||
@@ -0,0 +1,344 @@
|
||||
// U1 (#1873, PRD 1.20 REQ-WS-1..8): the cross-organization read model behind GET /api/workspace.
|
||||
//
|
||||
// One snapshot across the bus of every organization the viewer belongs to. Membership is the
|
||||
// per-org check the org list and the org pages already use (organizations.mjs `admitted`), so an
|
||||
// org the viewer does not belong to is never read and never named. Each org keeps its own bus;
|
||||
// this module only fans out over the same per-org options the org pages get (cli.mjs) and merges.
|
||||
// There is no second path to a bus and no write.
|
||||
//
|
||||
// Every org has one deadline. A bus that is down, still starting or slow is marked with a state
|
||||
// and a time, and the other orgs still answer. Every merged item carries its org id.
|
||||
import { plainMessage } from './org-admin.mjs';
|
||||
|
||||
export const WORKSPACE_SECTIONS = Object.freeze(['projects', 'sessions', 'inbox', 'tasks', 'metrics']);
|
||||
export const WORKSPACE_POLL_MS = 10000;
|
||||
export const WORKSPACE_TIMEOUT_MS = 4000;
|
||||
const PROJECT_STATUSES = ['active', 'archived'];
|
||||
const PROJECT_STATES = ['working', 'waiting', 'idle', 'none'];
|
||||
const SORTS = ['activity', 'name', 'org', 'status'];
|
||||
const ORG_ID = /^[a-z0-9][a-z0-9._-]{0,63}$/;
|
||||
// A project selector is `<org>:<project uuid or org-file project key>`.
|
||||
const PROJECT_SELECTOR = /^([a-z0-9][a-z0-9._-]{0,63}):([A-Za-z0-9._-]{1,128})$/;
|
||||
// Only these reach a copied shell command; anything else gets no command.
|
||||
const SHELL_SAFE = /^[A-Za-z0-9._:-]{1,128}$/;
|
||||
const WEEK_MS = 7 * 24 * 3600 * 1000;
|
||||
const PER_ORG_CONCURRENCY = 3;
|
||||
|
||||
// The broker reads each section needs. Each name is one human-side read the org pages already make.
|
||||
const NEEDS = {
|
||||
projects: ['projects', 'agents', 'claims', 'threads', 'tasks', 'inbox'],
|
||||
sessions: ['agents', 'claims', 'threads', 'projects', 'tasks'],
|
||||
inbox: ['inbox', 'agents', 'tasks', 'projects'],
|
||||
tasks: ['tasks', 'projects', 'inbox'],
|
||||
metrics: ['inbox', 'agents', 'claims', 'tasks', 'projects', 'usage', 'limits', 'decisions'],
|
||||
};
|
||||
const READS = {
|
||||
inbox: o => o.bus.inbox(),
|
||||
agents: o => o.bus.agents(),
|
||||
claims: o => o.bus['thread.claims'](),
|
||||
threads: o => o.chat.threads({}),
|
||||
tasks: o => o.bus.tasks(),
|
||||
projects: o => o.chat.projects({ includeArchived: true }),
|
||||
usage: o => o.bus.usage(),
|
||||
limits: o => o.bus.limits(),
|
||||
decisions: o => o.bus.decisions('recent'),
|
||||
};
|
||||
|
||||
export class WorkspaceQueryError extends Error {}
|
||||
const bad = message => { throw new WorkspaceQueryError(message); };
|
||||
const list = value => value.split(',').map(v => v.trim()).filter(Boolean);
|
||||
function oneOf(name, value, allowed) {
|
||||
const values = list(value);
|
||||
if (!values.length) bad(`${name} is empty`);
|
||||
for (const v of values) if (!allowed.includes(v)) bad(`${name} takes ${allowed.join(', ')}`);
|
||||
return new Set(values);
|
||||
}
|
||||
|
||||
// Strict: an unknown or repeated key is refused, so a misspelled filter never reads as "all".
|
||||
export function parseWorkspaceQuery(search = '') {
|
||||
const params = new URLSearchParams(search), seen = new Set();
|
||||
const q = { sections: new Set(WORKSPACE_SECTIONS), orgs: null, status: new Set(['active']), state: null, name: null, sort: 'activity', order: null, project: null };
|
||||
for (const [key, value] of params) {
|
||||
if (seen.has(key)) bad(`${key} is given twice`);
|
||||
seen.add(key);
|
||||
if (key === 'sections') q.sections = oneOf('sections', value, WORKSPACE_SECTIONS);
|
||||
else if (key === 'org') {
|
||||
const ids = list(value);
|
||||
if (!ids.length || ids.some(id => !ORG_ID.test(id))) bad('org takes organization ids, comma separated');
|
||||
q.orgs = new Set(ids);
|
||||
} else if (key === 'status') q.status = oneOf('status', value, PROJECT_STATUSES);
|
||||
else if (key === 'state') q.state = oneOf('state', value, PROJECT_STATES);
|
||||
else if (key === 'name') {
|
||||
if (!value.trim() || value.length > 200) bad('name takes 1 to 200 characters');
|
||||
q.name = value.trim().toLowerCase();
|
||||
} else if (key === 'sort') { if (!SORTS.includes(value)) bad(`sort takes ${SORTS.join(', ')}`); q.sort = value; }
|
||||
else if (key === 'order') { if (!['asc', 'desc'].includes(value)) bad('order takes asc or desc'); q.order = value; }
|
||||
else if (key === 'project') {
|
||||
const m = PROJECT_SELECTOR.exec(value);
|
||||
if (!m) bad('project takes <org>:<project id or key>');
|
||||
q.project = { org: m[1], key: m[2] };
|
||||
} else bad(`unknown query key ${key}`);
|
||||
}
|
||||
q.order ??= q.sort === 'activity' ? 'desc' : 'asc';
|
||||
return q;
|
||||
}
|
||||
|
||||
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(() => { 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');
|
||||
|
||||
// Runs `names` reads `limit` at a time and stops starting new ones once `stopped()` says so.
|
||||
async function pooled(names, limit, run, stopped) {
|
||||
const out = {}, queue = [...names];
|
||||
const worker = async () => {
|
||||
while (queue.length && !stopped()) {
|
||||
const name = queue.shift();
|
||||
try { out[name] = { ok: true, value: await run(name) }; } catch (e) { out[name] = { ok: false, error: code(e) }; }
|
||||
}
|
||||
};
|
||||
await Promise.all(Array.from({ length: Math.min(limit, names.length) }, worker));
|
||||
return out;
|
||||
}
|
||||
|
||||
const iso = ms => new Date(ms).toISOString();
|
||||
const latest = values => values.filter(v => typeof v === 'string' && !Number.isNaN(Date.parse(v))).sort().at(-1) ?? null;
|
||||
function median(values) {
|
||||
if (!values.length) return null;
|
||||
const s = [...values].sort((a, b) => a - b), mid = s.length >> 1;
|
||||
return s.length % 2 ? s[mid] : Math.round((s[mid - 1] + s[mid]) / 2);
|
||||
}
|
||||
|
||||
// One org's bus state and reads. Never throws: every failure becomes the org's state.
|
||||
async function readOrg(member, needs, { timeoutMs, now, ctx }) {
|
||||
const started = now();
|
||||
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 } };
|
||||
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(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 || 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 }) };
|
||||
const reason = boot && !boot.ok && typeof status.boot.reason === 'string'
|
||||
? `the bus refused to start: ${plainMessage(ctx, status.boot.reason).slice(0, 300)}`
|
||||
: 'the bus is not running';
|
||||
return { org: finish('down', { reason, boot }) };
|
||||
}
|
||||
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 || 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]));
|
||||
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 : []);
|
||||
|
||||
// Merges one org's reads into rows. Each row names its org; ids stay the bus's own.
|
||||
function orgRows(org, data, doc, nowMs) {
|
||||
const id = org.org;
|
||||
const projects = arr(data.projects).filter(p => p && typeof p.id === 'string');
|
||||
const byKey = new Map(projects.filter(p => typeof p.repo === 'string').map(p => [p.repo, p]));
|
||||
const byId = new Map(projects.map(p => [p.id, p]));
|
||||
const ref = p => (p ? { id: p.id, key: p.repo ?? null, name: p.name ?? null } : null);
|
||||
const threadProject = new Map(arr(data.threads).map(t => [t.id, byId.get(t.project) ?? null]));
|
||||
const taskRows = arr(data.tasks).filter(t => t && typeof t.task_ref === 'string' && !t.fields?.gone);
|
||||
const bots = new Map(Object.entries(doc?.roles ?? {}).filter(([, r]) => r?.tracker?.botId !== undefined).map(([role, r]) => [r.tracker.botId, role]));
|
||||
const tasks = taskRows.map(t => {
|
||||
const f = t.fields ?? {};
|
||||
return {
|
||||
org: id, ref: t.task_ref, tracker: org.tracker, title: f.title ?? null, done: Boolean(f.done), bucket: f.bucket ?? null,
|
||||
assignees: arr(f.assignees).filter(u => bots.has(u)).map(u => bots.get(u)).sort(), labels: arr(f.labels), priority: t.priority ?? null,
|
||||
requirement: t.requirement ?? null, project: ref(typeof t.project_key === 'string' ? byKey.get(t.project_key) : null), changedAt: t.at ?? null,
|
||||
};
|
||||
});
|
||||
const taskByRef = new Map(tasks.map(t => [t.ref, t]));
|
||||
// A role session's project and task: the task its open decision names, else its first open
|
||||
// assigned task. A role claim itself names no project, so this is inferred, and says so.
|
||||
const roleWork = role => {
|
||||
const open = tasks.filter(t => !t.done && t.assignees.includes(role));
|
||||
return open[0] ?? null;
|
||||
};
|
||||
const sessions = [];
|
||||
for (const c of arr(data.agents)) {
|
||||
if (!c || typeof c.role !== 'string') continue;
|
||||
const asked = arr(c.inputNeeded).map(d => taskByRef.get(d.task_ref)).find(Boolean);
|
||||
const task = asked ?? roleWork(c.role);
|
||||
sessions.push({
|
||||
org: id, kind: 'role', role: c.role, run: c.holder_run ?? null, harness: c.harness ?? null, state: c.state ?? null,
|
||||
since: c.working?.since ?? arr(c.inputNeeded)[0]?.at ?? c.at ?? null, claimedAt: c.at ?? null,
|
||||
messages: c.working?.messages ?? 0, inputNeeded: arr(c.inputNeeded).length, stranded: arr(c.stranded).length,
|
||||
task: task ? { ref: task.ref, title: task.title } : null, project: task?.project ?? null, projectFrom: task ? 'task' : null,
|
||||
});
|
||||
}
|
||||
for (const c of arr(data.claims)) {
|
||||
if (!c || typeof c.thread !== 'string') continue;
|
||||
const p = threadProject.get(c.thread) ?? null;
|
||||
sessions.push({ org: id, kind: 'thread', thread: c.thread, instance: c.instance ?? null, run: c.holder_run ?? null, harness: c.harness ?? null, state: 'live', since: c.at ?? null, claimedAt: c.at ?? null, project: ref(p), projectFrom: p ? 'thread' : null });
|
||||
}
|
||||
const sessionByRole = new Map(sessions.filter(s => s.kind === 'role').map(s => [s.role, s]));
|
||||
const inbox = arr(data.inbox).filter(d => d && typeof d.id === 'string').map(d => {
|
||||
const task = taskByRef.get(d.task_ref) ?? null, s = sessionByRole.get(d.raised_by_role) ?? null;
|
||||
const options = arr(d.options).map(o => ({ key: o.key, text: o.text }));
|
||||
const safe = SHELL_SAFE.test(d.id) && ORG_ID.test(id);
|
||||
return {
|
||||
org: id, id: d.id, question: d.question ?? null, class: d.class ?? null, blocking: Boolean(d.blocking), raisedBy: d.raised_by_role ?? null,
|
||||
taskRef: d.task_ref ?? null, task: task ? { ref: task.ref, title: task.title } : null, project: task?.project ?? null,
|
||||
session: s ? { role: s.role, run: s.run } : null, at: d.at ?? null, options, console: Boolean(d.console),
|
||||
// A shared view mixes organizations, so a copied command always names the org's bus.
|
||||
decide: options.map(o => ({ key: o.key, command: safe && SHELL_SAFE.test(String(o.key)) ? `mosaic decide ${d.id} ${o.key} --org ${id}` : null })),
|
||||
};
|
||||
});
|
||||
const projectRows = projects.map(p => {
|
||||
const mine = s => s.project?.id === p.id;
|
||||
const ps = sessions.filter(mine), needs = inbox.filter(mine), open = tasks.filter(t => !t.done && t.project?.id === p.id);
|
||||
const count = state => ps.filter(s => s.state === state).length;
|
||||
const counts = { working: count('working') + count('live'), waiting: count('waiting'), idle: count('idle') };
|
||||
const entry = typeof p.repo === 'string' ? doc?.projects?.[p.repo] : undefined;
|
||||
return {
|
||||
org: id, id: p.id, key: p.repo ?? null, name: p.name ?? null, status: p.archived ? 'archived' : 'active',
|
||||
state: counts.working ? 'working' : counts.waiting || needs.length ? 'waiting' : counts.idle ? 'idle' : 'none',
|
||||
git: entry ? (entry.git?.mode ?? (entry.remote ? 'remote' : null)) : null,
|
||||
sessions: counts, openTasks: open.length, needsYou: needs.length,
|
||||
lastActivity: latest([p.at, ...tasks.filter(t => t.project?.id === p.id).map(t => t.changedAt), ...needs.map(d => d.at), ...ps.map(s => s.claimedAt)]),
|
||||
};
|
||||
});
|
||||
const decided = arr(data.decisions).filter(d => d?.closed?.op === 'resolved' && typeof d.at === 'string' && typeof d.closed.at === 'string')
|
||||
.map(d => Date.parse(d.closed.at) - Date.parse(d.at)).filter(ms => Number.isFinite(ms) && ms >= 0);
|
||||
const roleCount = state => sessions.filter(s => s.kind === 'role' && s.state === state).length;
|
||||
const metrics = {
|
||||
org: id, needsYou: inbox.length, oldestNeedsYouAt: inbox.map(d => d.at).filter(Boolean).sort()[0] ?? null,
|
||||
sessions: { working: roleCount('working'), waiting: roleCount('waiting'), idle: roleCount('idle'), live: sessions.filter(s => s.kind === 'thread').length },
|
||||
openTasks: tasks.filter(t => !t.done).length,
|
||||
doneThisWeek: tasks.filter(t => t.done && t.changedAt && nowMs - Date.parse(t.changedAt) <= WEEK_MS).length,
|
||||
projects: { active: projectRows.filter(p => p.status === 'active').length, archived: projectRows.filter(p => p.status === 'archived').length },
|
||||
usage: data.usage?.total ?? null, limits: arr(data.limits?.tokens), decisionWaitMedianMs: median(decided),
|
||||
};
|
||||
return { projects: projectRows, sessions, inbox, tasks, metrics };
|
||||
}
|
||||
|
||||
function sortProjects(rows, { sort, order }) {
|
||||
const activity = (a, b) => (a.lastActivity ?? '').localeCompare(b.lastActivity ?? '');
|
||||
const name = (a, b) => (a.name ?? '').localeCompare(b.name ?? '') || a.id.localeCompare(b.id);
|
||||
const by = {
|
||||
activity: (a, b) => activity(a, b) || name(a, b),
|
||||
name,
|
||||
org: (a, b) => a.org.localeCompare(b.org) || name(a, b),
|
||||
status: (a, b) => PROJECT_STATUSES.indexOf(a.status) - PROJECT_STATUSES.indexOf(b.status) || PROJECT_STATES.indexOf(a.state) - PROJECT_STATES.indexOf(b.state) || name(a, b),
|
||||
}[sort];
|
||||
const sign = order === 'desc' ? -1 : 1;
|
||||
return [...rows].sort((a, b) => sign * by(a, b));
|
||||
}
|
||||
|
||||
// `organizations.members(session, { timeoutMs })` gives the orgs this session belongs to, each
|
||||
// `{ id, doc, status(), options() }` or `{ id, invalid, reason }`. A registry that cannot be read
|
||||
// throws, and the HTTP handler answers a fixed message.
|
||||
export async function readWorkspace({ organizations, session, query = parseWorkspaceQuery(''), timeoutMs = WORKSPACE_TIMEOUT_MS, now = Date.now, ctx = {} }) {
|
||||
const members = (await organizations.members(session, { timeoutMs })).filter(m => !query.orgs || query.orgs.has(m.id));
|
||||
const needs = [...new Set([...query.sections].flatMap(s => NEEDS[s]))];
|
||||
const read = await Promise.all(members.map(m => readOrg(m, needs, { timeoutMs, now, ctx })));
|
||||
const nowMs = now();
|
||||
const merged = { projects: [], sessions: [], inbox: [], tasks: [], metrics: [] };
|
||||
for (const r of read) {
|
||||
if (r.org.state !== 'ok') continue;
|
||||
const rows = r.rows;
|
||||
for (const key of Object.keys(merged)) merged[key].push(...(key === 'metrics' ? [rows.metrics] : rows[key]));
|
||||
}
|
||||
const body = {
|
||||
readAt: iso(nowMs), pollMs: WORKSPACE_POLL_MS, timeoutMs,
|
||||
summary: { read: read.filter(r => r.org.state === 'ok').length, of: read.length },
|
||||
orgs: read.map(r => r.org),
|
||||
};
|
||||
if (query.sections.has('projects')) {
|
||||
const rows = merged.projects.filter(p => query.status.has(p.status) && (!query.state || query.state.has(p.state)) && (!query.name || (p.name ?? '').toLowerCase().includes(query.name)));
|
||||
body.projects = sortProjects(rows, query);
|
||||
}
|
||||
if (query.sections.has('sessions')) body.sessions = merged.sessions.sort((a, b) => a.org.localeCompare(b.org) || (b.since ?? '').localeCompare(a.since ?? ''));
|
||||
if (query.sections.has('inbox')) body.inbox = merged.inbox.sort((a, b) => (a.class === 'gated' ? 0 : 1) - (b.class === 'gated' ? 0 : 1) || (a.at ?? '').localeCompare(b.at ?? ''));
|
||||
if (query.sections.has('tasks')) {
|
||||
const p = query.project;
|
||||
body.tasks = merged.tasks.filter(t => !p || (t.org === p.org && (t.project?.id === p.key || t.project?.key === p.key)))
|
||||
.sort((a, b) => a.org.localeCompare(b.org) || a.ref.localeCompare(b.ref));
|
||||
}
|
||||
if (query.sections.has('metrics')) {
|
||||
const m = merged.metrics;
|
||||
const sum = f => m.reduce((n, x) => n + f(x), 0);
|
||||
body.metrics = {
|
||||
orgs: m,
|
||||
totals: {
|
||||
needsYou: sum(x => x.needsYou), oldestNeedsYouAt: m.map(x => x.oldestNeedsYouAt).filter(Boolean).sort()[0] ?? null,
|
||||
sessions: { working: sum(x => x.sessions.working), waiting: sum(x => x.sessions.waiting), idle: sum(x => x.sessions.idle), live: sum(x => x.sessions.live) },
|
||||
openTasks: sum(x => x.openTasks), doneThisWeek: sum(x => x.doneThisWeek),
|
||||
projects: { active: sum(x => x.projects.active), archived: sum(x => x.projects.archived) },
|
||||
},
|
||||
};
|
||||
}
|
||||
return body;
|
||||
}
|
||||
|
||||
const json = (res, status, body) => {
|
||||
res.writeHead(status, { 'content-type': 'application/json; charset=utf-8', 'cache-control': 'no-store' });
|
||||
res.end(JSON.stringify(body));
|
||||
};
|
||||
|
||||
// GET /api/workspace. The hook fence has required a signed-in session on the page's own name.
|
||||
// Only the owner reads it: a granted user or the tester sees no organizations today, so they get
|
||||
// owner-only here too, and nothing about any org.
|
||||
export async function workspaceHTTP(req, res, { session = null, owner = false, organizations = null, search = '', timeoutMs = WORKSPACE_TIMEOUT_MS, now = Date.now, ctx = {}, log = () => {} }) {
|
||||
if (req.method !== 'GET') return json(res, 405, { error: 'method not allowed' });
|
||||
if (!organizations || typeof organizations.members !== 'function') return json(res, 404, { error: 'org-unavailable' });
|
||||
if (!owner || !session) return json(res, 403, { error: 'owner-only', message: 'the cross-organization view is the owner\'s' });
|
||||
let query;
|
||||
try { query = parseWorkspaceQuery(search); }
|
||||
catch (e) { if (e instanceof WorkspaceQueryError) return json(res, 400, { error: 'invalid-request', message: e.message }); throw e; }
|
||||
let body;
|
||||
try { body = await readWorkspace({ organizations, session, query, timeoutMs, now, ctx }); }
|
||||
catch { log('workspace: organization registry could not be read'); return json(res, 503, { error: 'org-unavailable', message: 'Organization registry could not be read' }); }
|
||||
return json(res, 200, body);
|
||||
}
|
||||
@@ -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'); })]);
|
||||
});
|
||||
@@ -0,0 +1,264 @@
|
||||
// U1 (#1873, PRD 1.20 REQ-WS-1..8): the cross-organization read behind GET /api/workspace.
|
||||
// Merge, membership, partial failure, timeout, sort and filter, the copied decide command, the
|
||||
// route through startServer, and the assistant-read stop. Every bus here is a fake options
|
||||
// object or a real org registry in this test's own scratch; no test reaches a live bus.
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import * as fs from 'node:fs/promises';
|
||||
import { readFileSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { createOrg, showOrg, templateRender } from '../../organization/src/org-service.mjs';
|
||||
import { listOrgs, resolveOrg } from '../../organization/src/org-registry.mjs';
|
||||
import { organizationConsole } from '../src/organizations.mjs';
|
||||
import { startServer } from '../src/serve.mjs';
|
||||
import { createDevOwner } from '../src/dev-owner.mjs';
|
||||
import { parseWorkspaceQuery, readWorkspace, workspaceHTTP, WorkspaceQueryError, WORKSPACE_POLL_MS } from '../src/workspace-read.mjs';
|
||||
import * as broker from '../../bus/src/broker.mjs';
|
||||
|
||||
const principal = { provider: 'local-dev', iss: 'local-dev', sub: 'local-dev-owner' };
|
||||
const NOW = Date.parse('2026-11-09T13:00:00.000Z');
|
||||
const now = () => NOW;
|
||||
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] ?? (['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') },
|
||||
chat: { threads: () => value('threads'), projects: () => value('projects') },
|
||||
};
|
||||
}
|
||||
const healthy = async () => ({ healthy: true, pid: 4242, boot: null });
|
||||
function member(id, data, { status = healthy, over = {}, doc = {} } = {}) {
|
||||
return { id, doc: { appearance: { theme: `${id}-theme` }, ...doc }, status, options: () => options(data, over) };
|
||||
}
|
||||
const organizationsOf = (...members) => ({ members: async () => members, list: async () => [], resolve: async () => null });
|
||||
|
||||
const ACME = {
|
||||
projects: [
|
||||
{ id: 'p-site', repo: 'site', name: 'Website', archived: false, at: '2026-11-09T10:00:00.000Z' },
|
||||
{ id: 'p-old', repo: 'old', name: 'Old brochure', archived: true, at: '2026-10-01T00:00:00.000Z' },
|
||||
],
|
||||
tasks: [
|
||||
{ task_ref: 'vk:1', project_key: 'site', priority: 2, at: '2026-11-09T11:00:00.000Z', fields: { title: 'Hero copy', done: false, assignees: [7, 99], bucket: 'Doing', labels: ['web'] } },
|
||||
{ task_ref: 'vk:2', project_key: 'site', at: '2026-11-08T00:00:00.000Z', fields: { title: 'Logo', done: true } },
|
||||
{ task_ref: 'vk:3', project_key: 'site', at: '2026-11-08T00:00:00.000Z', fields: { title: 'Removed', gone: true } },
|
||||
],
|
||||
agents: [{ role: 'coder', holder_run: 'run-1', harness: 'claude', state: 'waiting', at: '2026-11-09T09:00:00.000Z', inputNeeded: [{ task_ref: 'vk:1', at: '2026-11-09T12:00:00.000Z' }], stranded: [] }],
|
||||
claims: [{ thread: 't-1', instance: 'assistant', holder_run: 'run-2', harness: 'claude', at: '2026-11-09T08:00:00.000Z' }],
|
||||
threads: [{ id: 't-1', project: 'p-site' }],
|
||||
inbox: [{ id: 'd1234567', question: 'Ship the hero?', class: 'gated', blocking: true, raised_by_role: 'coder', task_ref: 'vk:1', at: '2026-11-09T12:00:00.000Z', options: [{ key: 'yes', text: 'Yes' }, { key: 'no', text: 'No' }] }],
|
||||
usage: { total: { tokens: 1200 } },
|
||||
limits: { tokens: [{ scope: 'org', used: 1200, cap: 5000 }] },
|
||||
decisions: [{ at: '2026-11-09T00:00:00.000Z', closed: { op: 'resolved', at: '2026-11-09T00:10:00.000Z' } }],
|
||||
};
|
||||
const ACME_DOC = { tracker: { kind: 'vikunja' }, roles: { coder: { tracker: { botId: 7 } }, assistant: {} } };
|
||||
const BETA = {
|
||||
projects: [{ id: 'p-shop', repo: 'shop', name: 'Shop', archived: false, at: '2026-11-07T00:00:00.000Z' }],
|
||||
tasks: [{ task_ref: 'local:1', project_key: 'shop', at: '2026-11-07T00:00:00.000Z', fields: { title: 'Price list', done: false } }],
|
||||
};
|
||||
|
||||
test('merge: every item from two buses carries its org, and the snapshot names both orgs', async () => {
|
||||
const body = await readWorkspace({ organizations: organizationsOf(member('acme', ACME, { doc: ACME_DOC }), member('beta', BETA)), session: principal, now });
|
||||
assert.equal(body.pollMs, WORKSPACE_POLL_MS);
|
||||
assert.deepEqual(body.summary, { read: 2, of: 2 });
|
||||
assert.deepEqual(body.orgs.map(o => [o.org, o.state, o.tracker, o.theme]), [['acme', 'ok', 'vikunja', 'acme-theme'], ['beta', 'ok', 'built-in', 'beta-theme']]);
|
||||
assert.deepEqual(body.orgs[0].roles, ['assistant', 'coder']);
|
||||
for (const key of ['projects', 'sessions', 'inbox', 'tasks']) for (const row of body[key]) assert.ok(['acme', 'beta'].includes(row.org), `${key} row without org`);
|
||||
assert.deepEqual(body.metrics.orgs.map(m => m.org), ['acme', 'beta']);
|
||||
// Archived projects are hidden by default; the default sort is newest activity first.
|
||||
assert.deepEqual(body.projects.map(p => `${p.org}/${p.key}`), ['acme/site', 'beta/shop']);
|
||||
const site = body.projects[0];
|
||||
assert.deepEqual({ ...site }, { org: 'acme', id: 'p-site', key: 'site', name: 'Website', status: 'active', state: 'working', git: null,
|
||||
sessions: { working: 1, waiting: 1, idle: 0 }, openTasks: 1, needsYou: 1, lastActivity: '2026-11-09T12:00:00.000Z' });
|
||||
const role = body.sessions.find(s => s.kind === 'role'), thread = body.sessions.find(s => s.kind === 'thread');
|
||||
assert.equal(role.org, 'acme'); assert.equal(role.role, 'coder'); assert.equal(role.state, 'waiting'); assert.equal(role.inputNeeded, 1);
|
||||
assert.deepEqual(role.task, { ref: 'vk:1', title: 'Hero copy' }); assert.equal(role.project.key, 'site'); assert.equal(role.projectFrom, 'task');
|
||||
assert.equal(thread.thread, 't-1'); assert.equal(thread.project.key, 'site'); assert.equal(thread.projectFrom, 'thread');
|
||||
const ask = body.inbox[0];
|
||||
assert.equal(ask.org, 'acme'); assert.equal(ask.raisedBy, 'coder'); assert.deepEqual(ask.session, { role: 'coder', run: 'run-1' }); assert.equal(ask.project.key, 'site');
|
||||
// The gone task is dropped; only bot ids from the org file become role names.
|
||||
assert.deepEqual(body.tasks.map(t => `${t.org}/${t.ref}`), ['acme/vk:1', 'acme/vk:2', 'beta/local:1']);
|
||||
assert.deepEqual(body.tasks[0].assignees, ['coder']); assert.equal(body.tasks[0].bucket, 'Doing'); assert.equal(body.tasks[2].tracker, 'built-in');
|
||||
assert.deepEqual(body.metrics.orgs[0], { org: 'acme', needsYou: 1, oldestNeedsYouAt: '2026-11-09T12:00:00.000Z', sessions: { working: 0, waiting: 1, idle: 0, live: 1 },
|
||||
openTasks: 1, doneThisWeek: 1, projects: { active: 1, archived: 1 }, usage: { tokens: 1200 }, limits: [{ scope: 'org', used: 1200, cap: 5000 }], decisionWaitMedianMs: 600000 });
|
||||
assert.deepEqual(body.metrics.totals, { needsYou: 1, oldestNeedsYouAt: '2026-11-09T12:00:00.000Z', sessions: { working: 0, waiting: 1, idle: 0, live: 1 }, openTasks: 2, doneThisWeek: 1, projects: { active: 2, archived: 1 } });
|
||||
});
|
||||
|
||||
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] ?? (['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']);
|
||||
});
|
||||
|
||||
test('membership: a foreign org is never read or named; a broken own org shows as invalid; options get the timeout', async t => {
|
||||
const root = await fs.mkdtemp(join(tmpdir(), 'wsr-'));
|
||||
t.after(() => fs.rm(root, { recursive: true, force: true }));
|
||||
const ctx = { configDir: join(root, 'config'), dataRoot: join(root, 'data'), instance: 'test', principal };
|
||||
await fs.mkdir(ctx.configDir, { mode: 0o700 }); await fs.mkdir(ctx.dataRoot, { mode: 0o700 });
|
||||
for (const id of ['alpha', 'beta', 'private-tenant']) await createOrg(ctx, id, await templateRender(id));
|
||||
await fs.writeFile(join(ctx.dataRoot, 'org-runtime', 'private-tenant', 'owner.json'), JSON.stringify({ ...principal, sub: 'different-owner' }));
|
||||
const statusAsked = [], optionsAsked = [];
|
||||
const orgs = organizationConsole({ ctx, listOrgs, showOrg, resolveOrg, orgPaths: () => ({}),
|
||||
busStatus: async (_ctx, id) => { statusAsked.push(id); return id === 'beta' ? { healthy: false, pid: null, boot: { ok: false, at: '2026-11-09T12:59:00.000Z', reason: `config missing at ${ctx.dataRoot}/org-runtime/beta/x.json` } } : { healthy: true, pid: 7 }; },
|
||||
options: o => { optionsAsked.push({ id: o.id, timeoutMs: o.timeoutMs }); return options(o.id === 'alpha' ? BETA : {}); } });
|
||||
const body = await readWorkspace({ organizations: orgs, session: principal, timeoutMs: 1500, now, ctx });
|
||||
assert.deepEqual(body.orgs.map(o => [o.org, o.state]), [['alpha', 'ok'], ['beta', 'down']]);
|
||||
assert.equal(JSON.stringify(body).includes('private-tenant'), false);
|
||||
assert.deepEqual(statusAsked.sort(), ['alpha', 'beta']);
|
||||
assert.deepEqual(optionsAsked, [{ id: 'alpha', timeoutMs: 1500 }]);
|
||||
// The boot refusal is shown with the data root made relative.
|
||||
assert.equal(body.orgs[1].reason, 'the bus refused to start: config missing at <data root>/org-runtime/beta/x.json');
|
||||
assert.equal(JSON.stringify(body).includes(root), false);
|
||||
// A different principal sees none of these orgs.
|
||||
const outsider = await readWorkspace({ organizations: orgs, session: { ...principal, sub: 'outsider' }, now, ctx });
|
||||
assert.deepEqual(outsider.orgs, []);
|
||||
// The owner of a broken org sees it as invalid, with no host path.
|
||||
await fs.unlink(join(ctx.configDir, 'businesses', 'alpha.json'));
|
||||
const broken = await readWorkspace({ organizations: orgs, session: principal, now, ctx });
|
||||
assert.equal(broken.orgs[0].org, 'alpha'); assert.equal(broken.orgs[0].state, 'invalid');
|
||||
assert.equal(broken.orgs[0].reason.includes(ctx.configDir), false);
|
||||
assert.deepEqual(broken.summary, { read: 0, of: 2 });
|
||||
});
|
||||
|
||||
test('partial failure: down, starting and a status that throws each get a state and a time; the others return', async () => {
|
||||
const ctx = { dataRoot: '/srv/fake-data', configDir: '/srv/fake-config' };
|
||||
const body = await readWorkspace({ now, ctx, session: principal, organizations: organizationsOf(
|
||||
member('acme', ACME, { doc: ACME_DOC }),
|
||||
member('down', BETA, { status: async () => ({ healthy: false, pid: null, boot: { ok: true, at: '2026-11-09T12:00:00.000Z' } }) }),
|
||||
member('starting', BETA, { status: async () => ({ healthy: false, pid: 991, boot: null }) }),
|
||||
member('stale', BETA, { status: async () => { throw new Error('host state names /srv/fake-data/org-runtime/stale/bus-host and /etc/secret-place'); } }),
|
||||
member('reads-fail', BETA, { over: { tasks: () => Promise.reject(Object.assign(new Error('nope'), { code: 'bus-unavailable' })), projects: () => Promise.reject(new Error('raw text /etc/x')) } }),
|
||||
) });
|
||||
const byOrg = Object.fromEntries(body.orgs.map(o => [o.org, o]));
|
||||
assert.equal(byOrg.acme.state, 'ok');
|
||||
assert.equal(byOrg.down.state, 'down'); assert.equal(byOrg.down.reason, 'the bus is not running');
|
||||
assert.equal(byOrg.starting.state, 'starting'); assert.equal(byOrg.starting.reason, 'the bus host is running but does not answer yet');
|
||||
assert.equal(byOrg.stale.state, 'down');
|
||||
assert.equal(byOrg.stale.reason, 'host state names <data root>/org-runtime/stale/bus-host and <host path>');
|
||||
// A read that fails inside an answering bus is that read's error code, never its message.
|
||||
assert.equal(byOrg['reads-fail'].state, 'ok');
|
||||
assert.deepEqual(byOrg['reads-fail'].errors, { tasks: 'bus-unavailable', projects: 'read-failed' });
|
||||
for (const o of body.orgs) { assert.equal(o.at, new Date(NOW).toISOString()); assert.equal(typeof o.readMs, 'number'); }
|
||||
assert.deepEqual(body.summary, { read: 2, of: 5 });
|
||||
assert.ok(body.projects.every(p => p.org === 'acme'));
|
||||
});
|
||||
|
||||
test('timeout: a bus that never answers, or whose state is never read, is marked timeout within the bound', async () => {
|
||||
const started = Date.now();
|
||||
const body = await readWorkspace({ timeoutMs: 150, session: principal, organizations: organizationsOf(
|
||||
member('acme', ACME, { doc: ACME_DOC }),
|
||||
member('slow', BETA, { over: { tasks: never } }),
|
||||
member('stuck', BETA, { status: never }),
|
||||
) });
|
||||
const elapsed = Date.now() - started;
|
||||
assert.ok(elapsed < 1500, `took ${elapsed} ms`);
|
||||
const byOrg = Object.fromEntries(body.orgs.map(o => [o.org, o]));
|
||||
assert.equal(byOrg.acme.state, 'ok');
|
||||
assert.equal(byOrg.slow.state, 'timeout'); assert.equal(byOrg.slow.reason, 'the bus did not answer within 150 ms');
|
||||
assert.equal(byOrg.stuck.state, 'timeout'); assert.equal(byOrg.stuck.reason, 'the bus state was not read within 150 ms');
|
||||
assert.equal(body.timeoutMs, 150);
|
||||
assert.ok(body.tasks.every(t => t.org === 'acme'));
|
||||
});
|
||||
|
||||
test('query: strict keys and values', () => {
|
||||
const q = parseWorkspaceQuery('?sections=projects,tasks&org=acme,beta&status=active,archived&state=working&name=Web&sort=name&project=acme:site');
|
||||
assert.deepEqual([...q.sections], ['projects', 'tasks']); assert.deepEqual([...q.orgs], ['acme', 'beta']);
|
||||
assert.deepEqual([...q.status], ['active', 'archived']); assert.deepEqual([...q.state], ['working']);
|
||||
assert.equal(q.name, 'web'); assert.equal(q.sort, 'name'); assert.equal(q.order, 'asc'); assert.deepEqual(q.project, { org: 'acme', key: 'site' });
|
||||
assert.equal(parseWorkspaceQuery('').order, 'desc');
|
||||
for (const bad of ['secton=projects', 'org=acme&org=beta', 'sections=', 'sections=chat', 'org=Bad Org', 'status=paused', 'state=busy', `name=${'x'.repeat(201)}`, 'name=%20', 'sort=size', 'order=up', 'project=acme', 'project=acme:a/b'])
|
||||
assert.throws(() => parseWorkspaceQuery(bad), WorkspaceQueryError, bad);
|
||||
});
|
||||
|
||||
test('sort and filter: projects by org, status, state, name and activity; tasks by org and project', async () => {
|
||||
const organizations = organizationsOf(member('acme', ACME, { doc: ACME_DOC }), member('beta', BETA));
|
||||
const read = search => readWorkspace({ organizations, session: principal, now, query: parseWorkspaceQuery(search) });
|
||||
const keys = rows => rows.map(p => `${p.org}/${p.key}`);
|
||||
assert.deepEqual(keys((await read('status=active,archived')).projects), ['acme/site', 'beta/shop', 'acme/old']);
|
||||
assert.deepEqual(keys((await read('status=active,archived&sort=activity&order=asc')).projects), ['acme/old', 'beta/shop', 'acme/site']);
|
||||
assert.deepEqual(keys((await read('status=active,archived&sort=name')).projects), ['acme/old', 'beta/shop', 'acme/site']);
|
||||
assert.deepEqual(keys((await read('status=active,archived&sort=org&order=desc')).projects), ['beta/shop', 'acme/site', 'acme/old']);
|
||||
assert.deepEqual(keys((await read('status=active,archived&sort=status')).projects), ['acme/site', 'beta/shop', 'acme/old']);
|
||||
assert.deepEqual(keys((await read('status=archived')).projects), ['acme/old']);
|
||||
assert.deepEqual(keys((await read('state=none')).projects), ['beta/shop']);
|
||||
assert.deepEqual(keys((await read('name=SHO')).projects), ['beta/shop']);
|
||||
const onlyBeta = await read('org=beta');
|
||||
assert.deepEqual(onlyBeta.orgs.map(o => o.org), ['beta']); assert.deepEqual(keys(onlyBeta.projects), ['beta/shop']);
|
||||
// An org filter naming an org the viewer is not in returns nothing, not an error.
|
||||
assert.deepEqual((await read('org=nobody')).orgs, []);
|
||||
assert.deepEqual((await read('project=acme:site')).tasks.map(t => t.ref), ['vk:1', 'vk:2']);
|
||||
assert.deepEqual((await read('project=acme:p-site')).tasks.map(t => t.ref), ['vk:1', 'vk:2']);
|
||||
assert.deepEqual((await read('project=beta:site')).tasks, []);
|
||||
});
|
||||
|
||||
test('decide: a copied command names the org; an unsafe id or key gets none', async () => {
|
||||
const data = { ...ACME, inbox: [...ACME.inbox, { id: 'd99 ; rm -rf ~', question: 'x', options: [{ key: 'ok', text: 'ok' }] }, { id: 'e7654321', question: 'y', options: [{ key: 'a b', text: 'a b' }] }] };
|
||||
const body = await readWorkspace({ organizations: organizationsOf(member('acme', data, { doc: ACME_DOC })), session: principal, now, query: parseWorkspaceQuery('sections=inbox') });
|
||||
const byId = Object.fromEntries(body.inbox.map(d => [d.id, d]));
|
||||
assert.deepEqual(byId.d1234567.decide, [{ key: 'yes', command: 'mosaic decide d1234567 yes --org acme' }, { key: 'no', command: 'mosaic decide d1234567 no --org acme' }]);
|
||||
assert.deepEqual(byId['d99 ; rm -rf ~'].decide, [{ key: 'ok', command: null }]);
|
||||
assert.deepEqual(byId.e7654321.decide, [{ key: 'a b', command: null }]);
|
||||
// Gated first, then oldest first.
|
||||
assert.equal(body.inbox[0].id, 'd1234567');
|
||||
});
|
||||
|
||||
function fakeRes() {
|
||||
return { status: null, body: null, writeHead(status) { this.status = status; }, end(text) { this.body = JSON.parse(text); } };
|
||||
}
|
||||
test('handler: non-GET, no registry, non-owner, bad query and an unreadable registry', async () => {
|
||||
const organizations = organizationsOf(member('acme', ACME));
|
||||
const run = async (opts, method = 'GET') => { const res = fakeRes(); await workspaceHTTP({ method }, res, { session: principal, owner: true, organizations, ...opts }); return res; };
|
||||
assert.equal((await run({}, 'POST')).status, 405);
|
||||
assert.deepEqual((await run({ organizations: null })).body, { error: 'org-unavailable' });
|
||||
// A granted user or the tester reaches the hook with owner false and no owner session.
|
||||
for (const who of [{ owner: false }, { session: null }]) {
|
||||
const res = await run(who);
|
||||
assert.equal(res.status, 403); assert.deepEqual(res.body, { error: 'owner-only', message: 'the cross-organization view is the owner\'s' });
|
||||
}
|
||||
const bad = await run({ search: '?org=acme&org=acme' });
|
||||
assert.equal(bad.status, 400); assert.equal(bad.body.error, 'invalid-request');
|
||||
const logged = [];
|
||||
const down = await run({ organizations: { members: async () => { throw new Error('/srv/x/org-runtime.json: Unexpected token'); } }, log: line => logged.push(line) });
|
||||
assert.equal(down.status, 503); assert.deepEqual(down.body, { error: 'org-unavailable', message: 'Organization registry could not be read' });
|
||||
assert.deepEqual(logged, ['workspace: organization registry could not be read']);
|
||||
});
|
||||
|
||||
async function serve(t, organizations, sso = createDevOwner()) {
|
||||
const s = await startServer({ port: 0, board: 'http://127.0.0.1:9', sso, organizations, log: () => {}, workspaceTimeoutMs: 200 });
|
||||
t.after(() => new Promise(r => { s.close(r); s.closeIdleConnections(); }));
|
||||
return `http://127.0.0.1:${s.address().port}`;
|
||||
}
|
||||
test('route: the dev owner reads GET /api/workspace; a granted user gets owner-only; the timeout option applies', async t => {
|
||||
const seen = [];
|
||||
const organizations = { list: async () => [], resolve: async () => null, members: async (session, opts) => { seen.push({ sub: session.sub, timeoutMs: opts.timeoutMs }); return [member('acme', ACME, { doc: ACME_DOC }), member('slow', BETA, { status: never })]; } };
|
||||
const base = await serve(t, organizations);
|
||||
const response = await fetch(`${base}/api/workspace?sections=projects,metrics`), body = await response.json();
|
||||
assert.equal(response.status, 200); assert.equal(response.headers.get('cache-control'), 'no-store');
|
||||
assert.deepEqual(body.orgs.map(o => [o.org, o.state]), [['acme', 'ok'], ['slow', 'timeout']]);
|
||||
assert.equal(body.timeoutMs, 200); assert.deepEqual(seen, [{ sub: 'local-dev-owner', timeoutMs: 200 }]);
|
||||
assert.equal((await fetch(`${base}/api/workspace?bogus=1`)).status, 400);
|
||||
// POST has no route here: with no Origin the dev fence refuses it, and with the page's Origin the
|
||||
// organizations fence answers as for any unknown /api path.
|
||||
assert.equal((await fetch(`${base}/api/workspace`, { method: 'POST' })).status, 403);
|
||||
const post = await fetch(`${base}/api/workspace`, { method: 'POST', headers: { origin: base, 'content-type': 'application/json' }, body: '{}' });
|
||||
assert.equal(post.status, 404); assert.deepEqual(await post.json(), { error: 'org-unavailable' });
|
||||
// A granted user's session (session.user set) is fenced before any org is read.
|
||||
const granted = { ...createDevOwner(), session: () => ({ ...principal, sub: 'alice-sub', user: 'alice', name: 'Alice' }) };
|
||||
const userBase = await serve(t, organizations, granted);
|
||||
const refused = await fetch(`${userBase}/api/workspace`);
|
||||
assert.equal(refused.status, 403); assert.equal((await refused.json()).error, 'owner-only');
|
||||
assert.equal(seen.length, 1);
|
||||
});
|
||||
|
||||
test('assistant read is designed, not built: no workspace broker verb, assistant authority unchanged', () => {
|
||||
const role = JSON.parse(readFileSync(new URL('../../../roles/assistant.json', import.meta.url), 'utf8'));
|
||||
assert.deepEqual(role.authority, { withinRole: ['message.send', 'record.write'], crossRole: [] });
|
||||
const verbs = new Set(Object.values(broker).filter(Array.isArray).flat().filter(v => typeof v === 'string'));
|
||||
assert.ok(verbs.has('inbox'), 'the verb lists were read');
|
||||
assert.deepEqual([...verbs].filter(v => /^(workspace|orgs?)\b/.test(v)), []);
|
||||
});
|
||||
Reference in new issue
Block a user