Compare commits

..
Author SHA1 Message Date
jason.woltje 900daf5213 fix(org): D2 security review: live identity refuses, clear only inspected entries, no revive from an agent run
- busRecover: a live pid with the recorded start time is the recorded process, so it refuses
  ("bus host identity is uncertain") whatever its argv. It used to read a mismatching argv as a dead
  host. The pinned recovery test now covers a reused pid (live, other start time) as the wrong
  process, and a new test covers the refusal.
- busRecover clears through setAside: each inspected entry is renamed aside and checked by device
  and inode; a swapped entry puts everything back and refuses before any receipt or deletion, and
  the deletion names only the aside entries.
- mosaic bus recover refuses inside an agent run (refuseInsideAgent), like mosaic org reload. The
  Console-start revive does nothing when the Console's own environment carries an agent marker.
- Tests: org-recovery (identity, agent refusal), org-revive (setAside swap), bus-revive (markers);
  the other reviveBuses tests pass an explicit env.
2026-11-11 14:03:18 -06:00
jason.woltje f9dae1ebad fix(org): D2 build review: pin pm:false, wait for the start-up revive, say when a clear was followed by a failed start
- The Console-start revive asks busRevive for pm:false; reviveStartOptions holds the rule (never
  wider than the dead host's record) and is tested on its own. A dropped --pm is logged with the
  way back.
- A Restart bus press waits for that organization's start-up revive (revivePending) before it reads
  the bus, so it no longer races it for the lifecycle lock; a press that still meets org-locked gets
  409 bus-busy, nothing changed.
- busRevive marks an error from busStart after a successful clear (error.recovered); the trail, the
  Console log and the press response then say "failed", not "refused". A non-CliError logs its code.
- Tests: bus-revive.test.mjs (new), org-admin and org-revive cases for the above.
2026-11-11 14:03:18 -06:00
jason.woltje f063158875 fix(org): Console start reattaches to org buses and revives a provably dead host (P0-3)
A Console unit stop (KillMode=mixed) SIGKILLs the bus hosts the Console
started, leaving host.json, host.lock and the socket behind, and the
dev Console's Restart bus refused 409 on that stale state.

- busRevive (org-bus): busRecover, then busStart detached on the
  options the dead host recorded. busRecover still refuses a live host
  pid, a lock that is not the dead host's, a live writer pid, or a
  socket that answers or is uncertain.
- reviveBuses (webui): when a dev-local-owner Console starts, a bus that
  answers is left (reattached); a bus with stale state goes to busRevive
  with pm:false, trailed as bus.revive before anything changes.
- Restart bus (org-admin): for the exact dev owner only, a stale-state
  refusal goes to busRevive; a signed-in owner keeps the refusal.
- The receipt reads "Dead bus host cleared; bus started".
2026-11-11 14:03:18 -06:00
jason.woltje ff20efb819 fix(autonomy): review round for #1873 account coverage
A replaced installed harness answers null for countsAccount until the
bus restarts and reads the new module. A runtime test pins the
startBroker wiring for claude-code (covered) and pi (refused). The
plugin README says countsAccount also decides authority coverage and
is read once at start.
2026-11-11 13:46:22 -06:00
jason.woltje cfe7535533 fix(autonomy): account quota coverage reads the harness module's countsAccount, not a name (#1873)
authorityBudgetCovered named claude-code to decide whether an account
quota covers every future launch, which fails the row 65 scan. The bus
host now passes each harness module's own countsAccount answer, read
once with its levels in harnessLevels, through autonomyHost to the
broker. Without the lookup no account quota counts (fail closed); the
provider/model branch is unchanged.
2026-11-11 13:39:44 -06:00
19 changed files with 690 additions and 37 deletions

No files matched your search

+4 -1
View File
@@ -5,8 +5,11 @@ import { randomBytes } from 'node:crypto';
import { prepareRevert, finishLand, dropLand } from '../../seat/src/working.mjs';
import { isMosaicStack, runGate, gateEnv } from '../../seat/src/landing.mjs';
const refuse=code=>{throw Object.assign(new Error(code),{code});};
export function autonomyHost({writer,dataRoot,business,hires,reload,processing,env=process.env,gate={classify:isMosaicStack,run:runGate}}) {
// countsAccount(name) answers what the named harness module declares (runtime harnessLevels);
// the broker's authority budget check reads it here instead of naming a harness.
export function autonomyHost({writer,dataRoot,business,hires,reload,processing,countsAccount=()=>null,env=process.env,gate={classify:isMosaicStack,run:runGate}}) {
return {
countsAccount,
get:id=>{const r=writer.read(id);return {revision:r.revision,autonomy:r.document.autonomy??{}};},
set:(id,args,actor)=>{
const old=writer.read(id);if(old.revision!==args.revision)refuse('revision-conflict');
+6 -4
View File
@@ -6415,7 +6415,7 @@ export class Broker {
if(!role?.authority?.withinRole.includes(right) || (cls==='authority' && name===b.launch?.by))fail('autonomy-precondition');
}
if(['merge','policy'].includes(cls) && !Object.values(b.roles).some(r=>r.authority?.withinRole.includes('review.verdict')))fail('autonomy-precondition');
if(cls==='authority' && !authorityBudgetCovered(b))fail('autonomy-precondition');
if(cls==='authority' && !this.#budgetCovered(b))fail('autonomy-precondition');
if(after==='out of the loop') {
if(before!=='near the loop')fail('autonomy-precondition');
const review=this.#store.get("SELECT seq,body FROM events WHERE business=? AND kind='action.allowed' AND json_extract(body,'$.operation')='autonomy.review' AND json_extract(body,'$.decisionClass')=? ORDER BY seq DESC LIMIT 1",s.business,cls);
@@ -6504,7 +6504,7 @@ export class Broker {
let route='human';
if(!immediate && level!=='in the loop' && p.level!=='root' && !p.flagged.length && !p.credentialLines.length) {
if(level==='out of the loop')fail('authority-limits-unavailable');
if(!authorityBudgetCovered(this.#business(s.business)))fail('autonomy-precondition');
if(!this.#budgetCovered(this.#business(s.business)))fail('autonomy-precondition');
route=b.autonomy?.authorityResolver;
if(!b.roles[route]?.authority?.withinRole.includes('decision.resolve.authority'))fail('autonomy-precondition');
if(route===s.role || resolverSubject(b,route,p.name))route='human';
@@ -6518,6 +6518,8 @@ export class Broker {
if(immediate) return this.#hireConsole(s,'hire.resolve.console',{id:did,choice:'yes',digest:p.digest});
return {...this.#decisionView(this.#decision(s,did)),hire:p};
}
// The harness fact comes from the host's loaded modules; no host, no account coverage.
#budgetCovered(b) { return authorityBudgetCovered(b,{countsAccount:this.#autonomyHost?.countsAccount}); }
// Trusted runtime inverse only; never a socket verb.
hireReverse(...args) {
if(!this.#hires?.reverse)fail('reversal-unavailable');
@@ -6534,7 +6536,7 @@ export class Broker {
if(this.#expireHire(s,d,p))fail('decision-expired');
if(this.#autonomyLevel(d)==='in the loop')fail('hire-resolution-required');
if(this.#autonomyLevel(d)==='out of the loop')fail('authority-limits-unavailable');
if(!authorityBudgetCovered(this.#business(s.business)))fail('autonomy-precondition');
if(!this.#budgetCovered(this.#business(s.business)))fail('autonomy-precondition');
if(!this.#business(s.business).roles[s.role]?.authority?.withinRole.includes('decision.resolve.authority'))fail('not-resolver');
};
validate();
@@ -6554,7 +6556,7 @@ export class Broker {
const recheck=()=>{
this.#agent(s);
if(this.#autonomyLevel(d)==='out of the loop')fail('authority-limits-unavailable');
if(!authorityBudgetCovered(this.#business(s.business)))fail('autonomy-precondition');
if(!this.#budgetCovered(this.#business(s.business)))fail('autonomy-precondition');
if(this.#autonomyLevel(d)==='in the loop' || !this.#business(s.business).roles[s.role]?.authority?.withinRole.includes('decision.resolve.authority') || resolverSubject(this.#business(s.business),s.role,p.name))fail('not-resolver');
};
recheck();
+21 -3
View File
@@ -99,6 +99,9 @@ export async function webServices(businesses, { log = (e) => process.stderr.writ
// checks a write runs, which are synchronous. A module that doesn't load, or answers anything but a
// list of names, answers null, which refuses the write. The tree is pinned, so a module changes only
// with a re-pin, and a re-pin restarts the bus.
// AU (#1873): the same pass records each module's countsAccount() answer, which decides whether an
// account quota covers authority hires. It is read once here; an installed harness whose lock pin
// has changed since answers null (not covered) until the bus restarts and reads the new module.
export async function harnessLevels({ log = (e) => process.stderr.write(`plugins: ${JSON.stringify(e)}\n`), dataRoot = null } = {}) {
const levels = new Map();
let names = [];
@@ -107,22 +110,37 @@ export async function harnessLevels({ log = (e) => process.stderr.write(`plugins
} catch (e) {
log({ event: 'harness-levels-unread', message: e.message });
}
let pins = {};
if (dataRoot !== null) {
try { names = [...new Set([...names,...Object.keys(readPluginLock(dataRoot).plugins).filter(name=>hasModule('harness',name,dataRoot))])]; }
try { pins = readPluginLock(dataRoot).plugins; names = [...new Set([...names,...Object.keys(pins).filter(name=>hasModule('harness',name,dataRoot))])]; }
catch { log({event:'harness-levels-unread',message:'installed harness discovery refused'}); }
}
const accounts = new Map();
for (const name of names) {
const loader = createLoader({ log, dataRoot });
if ((await loader.load({ harness: [name] })).refused.length) continue;
const [module] = loader.modules('harness');
try {
const list = loader.modules('harness')[0].levels();
const list = module.levels();
if (Array.isArray(list) && list.every((level) => typeof level === 'string')) levels.set(name, Object.freeze([...list]));
} catch (e) {
log({ event: 'harness-levels-failed', name, message: e.message });
}
// AU (#1873): whether this harness's launches spend launch.account, as its module declares.
try {
const counts = module.countsAccount();
if (typeof counts === 'boolean') accounts.set(name, counts);
} catch (e) {
log({ event: 'harness-account-failed', name, message: e.message });
}
}
const lookup = (name) => levels.get(name) ?? null;
lookup.names = [...levels.keys()].sort();
const pinned = (name) => {
if (!Object.hasOwn(pins, name)) return true;
try { return readPluginLock(dataRoot).plugins[name]?.sha256 === pins[name].sha256; } catch { return false; }
};
lookup.countsAccount = (name) => (accounts.has(name) && pinned(name) ? accounts.get(name) : null);
return lookup;
}
// Row 224 C2-05b: a host-only adapter. A failed fetch keeps serving immutable old
@@ -891,7 +909,7 @@ export async function startBroker({
return INSTANCE_VERBS.has(verb)?{...result,restart:!live}:result;
}):null,
// Row 194 (D2.3): the role files as they stand, for the hire checks.
autonomy: organization ? autonomyHost({writer:organizationWriter({dir:organization.dir,rolesDir:organization.rolesDir,dataRoot:instanceRoot??dataRoot}),dataRoot,business:id=>businesses[id],hires:{reverse:(...args)=>broker.hireReverse(...args)},reload,processing}) : null,
autonomy: organization ? autonomyHost({writer:organizationWriter({dir:organization.dir,rolesDir:organization.rolesDir,dataRoot:instanceRoot??dataRoot}),dataRoot,business:id=>businesses[id],countsAccount:levels.countsAccount,hires:{reverse:(...args)=>broker.hireReverse(...args)},reload,processing}) : null,
hires: organization ? (()=>{
const service=hireDecisions({writer:organizationWriter({dir:organization.dir,rolesDir:organization.rolesDir,dataRoot:instanceRoot??dataRoot}),rolesDir:organization.rolesDir,configDir:organization.dir,dataRoot,
system:organization.system,levels,catalogueFor:async(id,doc)=>doc.plugins?.['model-provider']?.length?catalogueFor(id,{fresh:true}):null,assertClean:value=>credentials.assertClean(value),scanDefinition:(name,entry)=>scanDefinitions({[name]:entry},value=>credentials.assertClean(value)),registryFor:doc=>organizationPlugins(instanceRoot??dataRoot,doc.plugins).registry});
+34 -2
View File
@@ -1,6 +1,6 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import {mkdtempSync,mkdirSync,rmSync} from 'node:fs';
import {mkdtempSync,mkdirSync,readFileSync,rmSync} from 'node:fs';
import {tmpdir} from 'node:os';
import {join} from 'node:path';
import {Broker} from '../src/broker.mjs';
@@ -17,6 +17,8 @@ function world(t,options={}) {
const root=mkdtempSync(join(tmpdir(),'au-authority-')),rolesDir=rolesCopy(root),dataRoot=join(root,'data');mkdirSync(dataRoot,{mode:0o700});
const doc=businessDoc(root,'demo');doc.plugins={harness:['pi'],'limit-check':['token-limit']};
doc.launch.limits.tokens={providers:{zai:cap}};
// #1873: an account quota alone, covered only through the host's declared harness fact.
if(options.account){doc.launch.account={name:'shared',credential:'FIXTURE_TOKEN'};doc.launch.limits.tokens={keys:{shared:cap}};}
doc.definitions={judge:{rev:1,role:{...inline,authority:{withinRole:['decision.resolve.authority'],crossRole:[]}},instructions:'Judge'},scout:{rev:3,role:structuredClone(inline),instructions:'Prior role'}};
// Definition instructions live beside the role composition.
for(const d of Object.values(doc.definitions))delete d.role.instructions;
@@ -31,7 +33,7 @@ function world(t,options={}) {
const reload=id=>{if(reversing && options.crossAfterCommit)Date.now=()=>now()+3600001;if(reversing && options.reloadRefused)return false;broker.reloadBusiness(id,loaded());return true;};
const service=hireDecisions({writer,rolesDir,configDir:root,dataRoot,assertClean:()=>{}});
const hires=authorityHost(service,{reload,modelKeys:{prepareRevoke:async(instance,guard)=>{guard();assert.equal(instance,'demo/scout');order.push('prepare');},completeRevoke:instance=>{assert.equal(instance,'demo/scout');assert.ok(store.get("SELECT 1 FROM events WHERE json_extract(body,'$.action')='fire.stopped'"));assert.equal(writer.read('demo').document.definitions.scout.instructions,'Changed role');order.push('complete');}}});
const host=autonomyHost({writer,dataRoot,business:()=>loaded(),hires,reload});
const host=autonomyHost({writer,dataRoot,business:()=>loaded(),hires,reload,...(options.countsAccount?{countsAccount:options.countsAccount}:{})});
const instances=instanceSettings({dir:root,rolesDir,writer,revoke:()=>{order.push('fire-revoke');}});
broker=new Broker({store,businesses:{demo:loaded()},hires,autonomy:host,organization:(id,verb,args,actor,context)=>{if(verb==='instance.fireCheck'){instances.checkFire(id,args.name);return {};}if(verb==='instance.fire')return instances.fire(id,args,{actor,context});throw Error('unexpected organization verb');}});
const caps={};const call=(who,verb,args={})=>broker.request(caps[who],{verb,args});
@@ -107,3 +109,33 @@ test('AU inverse receipt-write failure returns committed with trailPending and n
assert.equal(f.writer.read('demo').document.definitions.scout.instructions,'Prior role');
assert.equal(f.store.get("SELECT count(*) n FROM events WHERE json_extract(body,'$.operation')='autonomy.reverse.failed'").n,0);
});
test('AU an account-quota authority raise routes to the resolver only when the host says every harness spends the account (#1873)',async t=>{
const covered=world(t,{account:true,countsAccount:name=>name==='pi'}),d=await covered.raise();assert.equal(d.route_to,'judge');
for(const countsAccount of [undefined,()=>false,()=>null]){
const f=world(t,{account:true,countsAccount});
await assert.rejects(f.raise(),{code:'autonomy-precondition'});
assert.equal(f.store.get("SELECT count(*) n FROM decisions").n,0);
}
});
// #1873 review finding 2: the runtime itself carries each harness module's countsAccount into the
// broker's authority budget check. Only harness modules decide; the business names none here but
// through plugins.harness.
for(const [harness,code] of [['claude-code',null],['pi','autonomy-precondition']])test(`AU runtime: an account-quota authority setting with only ${harness} enabled ${code?'refuses':'saves'} (#1873)`,async t=>{
const {startBroker}=await import('../src/runtime.mjs');
const {systemFor}=await import('../../organization/tests/helpers.mjs');
const root=mkdtempSync(join(tmpdir(),'au-runtime-')),rolesDir=rolesCopy(root);t.after(()=>rmSync(root,{recursive:true,force:true}));
const doc=businessDoc(root,'demo');doc.plugins={harness:[harness],'limit-check':['token-limit']};
for(const r of Object.values(doc.roles))if(r.vars?.harness)r.vars.harness=harness;
doc.launch.account={name:'shared',credential:'FIXTURE_TOKEN'};doc.launch.limits.tokens={keys:{shared:cap}};
doc.definitions={judge:{rev:1,role:{...inline,authority:{withinRole:['decision.resolve.authority'],crossRole:[]}},instructions:'Judge'}};
delete doc.definitions.judge.role.instructions;
doc.roles.judge={definition:'org:judge',vars:{harness,model:'claude-opus-5-5'}};
writeJson(join(root,'businesses/demo.json'),doc);mkdirSync(join(root,'data'),{mode:0o700});
const loaded=()=>{const b=loadBusiness('demo',{dir:root,rolesDir});return {...b,roles:Object.fromEntries(Object.entries(b.roles).map(([n,e])=>[n,{...b.definitions[e.definition],...e}]))};};
const bus=await startBroker({dataRoot:join(root,'data'),businesses:{demo:loaded()},organization:{dir:root,rolesDir,system:systemFor(root)}});t.after(()=>bus.close());
const owner=bus.broker.bindConsole({business:'demo',human:'jason',subject:{provider:'test',iss:'test',sub:'owner'}});
const {revision}=await bus.broker.request(owner,{verb:'autonomy.get.console'});
const set=bus.broker.request(owner,{verb:'autonomy.set.console',args:{revision,autonomy:{authority:'near the loop',authorityResolver:'judge',notices:true}}});
if(code)await assert.rejects(set,{code});
else {await set;assert.equal(JSON.parse(readFileSync(join(root,'businesses/demo.json'),'utf8')).autonomy.authority,'near the loop');}
});
+3
View File
@@ -96,6 +96,9 @@ test('row 65 runtime: harnessLevels answers each module in the tree, and null fo
assert.deepEqual(levels('pi'), ['off', 'minimal', 'low', 'medium', 'high', 'xhigh', 'max']);
assert.deepEqual(levels('claude-code'), ['low', 'medium', 'high', 'xhigh', 'max']);
for (const name of ['cursor', '', '../pi']) assert.equal(levels(name), null, name);
// AU (#1873): each module's own countsAccount answer, read once with its levels.
assert.deepEqual([levels.countsAccount('pi'), levels.countsAccount('claude-code')], [false, true]);
for (const name of ['cursor', '', '../pi']) assert.equal(levels.countsAccount(name), null, name);
assert.deepEqual(logged, []);
});
+2 -1
View File
@@ -703,7 +703,8 @@ async function bus(rest, io, deps) {
if (["stop", "restart", "status", "recover"].includes(sub)) {
const parsed = parse(args, { flags: sub === "status" ? ["--json"] : [] });
if (parsed.positional.length !== 1) throw usage(`bus ${sub} takes one organization`);
if (sub === "recover") devOwnerInstance(io.env);
// D2 security review: recovery clears host state, so an agent run is refused before any read.
if (sub === "recover") { refuseInsideAgent(io.env); devOwnerInstance(io.env); }
const ctx = organizationContext(io, deps), id = parsed.positional[0];
const result = sub === "stop" ? await busStop(ctx, id) : sub === "restart" ? await busRestart(ctx, id) : sub === "recover" ? await busRecover(ctx, id) : await busStatus(ctx, id);
io.stdout.write(`${JSON.stringify(result, null, parsed.flags.has("--json") ? 2 : 0)}\n`);
+62 -10
View File
@@ -2,7 +2,7 @@
import { randomBytes } from 'node:crypto';
import { processState } from '../../seat/src/process-state.mjs';
import { spawn } from 'node:child_process';
import { constants, closeSync, lstatSync, mkdirSync, openSync, readFileSync, fstatSync, unlinkSync, writeFileSync } from 'node:fs';
import { constants, closeSync, linkSync, lstatSync, mkdirSync, openSync, readFileSync, fstatSync, renameSync, unlinkSync, writeFileSync } from 'node:fs';
import { connect } from 'node:net';
import { join } from 'node:path';
import { fileURLToPath } from 'node:url';
@@ -151,6 +151,27 @@ export async function busRestart(ctx, id) {
return busStart(ctx, id, { detach: true, ...options });
}
// D2 security review: clear only the entries busRecover inspected. Each is renamed aside (atomic)
// and the renamed entry is checked by device and inode; an entry swapped after the inspection is
// put back with everything moved so far, and the recovery refuses before any receipt or deletion.
// The deletion then names only the aside entries, never the live paths.
export function setAside(entries) {
const tag = randomBytes(4).toString('hex'), moved = [];
const restore = () => {
// link() refuses an existing path, so a put-back never replaces whatever now holds the name; an
// entry that cannot go back stays under its .recovering name for a person to inspect.
for (const m of moved.slice().reverse()) { try { linkSync(m.aside, m.path); unlinkSync(m.aside); } catch { /* left aside */ } }
};
for (const entry of entries) {
const aside = `${entry.path}.recovering-${tag}`;
let current;
try { renameSync(entry.path, aside); moved.push({ path: entry.path, aside }); current = lstatSync(aside); }
catch { restore(); fail('bus recovery state changed'); }
if (current.dev !== entry.st.dev || current.ino !== entry.st.ino) { restore(); fail('bus recovery state changed'); }
}
return { restore, remove: () => { for (const m of moved.slice().reverse()) unlinkSync(m.aside); } };
}
// Recover only while the org lifecycle fence excludes new starts and deletion.
export async function busRecover(ctx, id) {
return withOrgLifecycle(ctx, id, async () => {
@@ -177,10 +198,12 @@ export async function busRecover(ctx, id) {
const h = state.value;
if (!Number.isSafeInteger(h?.pid) || h.pid <= 0 || typeof h.startTime !== 'string' || !/^\d+$/.test(h.startTime) || typeof h.startedAt !== 'string' || !Number.isFinite(Date.parse(h.startedAt)) || h.business !== id) fail('invalid or mismatched bus host state');
let p; try { p = processState(h.pid); } catch { fail('bus host identity is uncertain'); }
// A live pid with the recorded start time is the recorded process: never proof of death. Its argv
// only picks the message (D2 security review: a mismatching argv used to read as a dead host).
if (p.alive && p.startTime === h.startTime) {
const argv = cmdlineOf(h.pid); if (!argv || !argv.length) fail('bus host identity is uncertain');
const i = argv.indexOf(CLI);
const argv = cmdlineOf(h.pid), i = argv?.indexOf(CLI) ?? -1;
if (i >= 0 && argv[i + 1] === 'bus' && argv[i + 2] === 'start' && argv[i + 3] === id) fail(`bus host pid ${h.pid} is alive`);
fail('bus host identity is uncertain');
}
if (lock && (lock.value?.pid !== h.pid || lock.value?.startTime !== h.startTime || lock.value?.business !== id || lock.value?.startedAt !== h.startedAt)) fail('bus host lock does not match host state');
if (writer) {
@@ -198,14 +221,43 @@ export async function busRecover(ctx, id) {
if (occupied) fail('bus socket answers or is uncertain');
}
const entries = [state, lock, writer, socket].filter(Boolean);
for (const entry of entries) {
const current = lstatSync(entry.path);
if (current.dev !== entry.st.dev || current.ino !== entry.st.ino) fail('bus recovery state changed');
}
const aside = setAside(entries);
const at = new Date().toISOString();
const receipt = join(org.paths.hostDir, `recovery-${at.replace(/[:.]/g, '-')}-${randomBytes(4).toString('hex')}.json`);
writeFileSync(receipt, JSON.stringify({ action: 'bus.recover', at, org: id, stopped: true, host: h, lock: lock?.value ?? null, writer: writer?.value ?? null, socket: Boolean(socket) }, null, 2) + '\n', { mode: 0o600, flag: 'wx' });
for (const entry of entries.slice().reverse()) unlinkSync(entry.path);
return { org: id, recovered: true, pid: h.pid, receipt };
try { writeFileSync(receipt, JSON.stringify({ action: 'bus.recover', at, org: id, stopped: true, host: h, lock: lock?.value ?? null, writer: writer?.value ?? null, socket: Boolean(socket) }, null, 2) + '\n', { mode: 0o600, flag: 'wx' }); }
catch (e) { aside.restore(); throw e; }
aside.remove();
const result = { org: id, recovered: true, pid: h.pid, receipt };
// The dead host's start options, for busRevive. Not part of the printed receipt.
const o = h.options && typeof h.options === 'object' ? h.options : {};
Object.defineProperty(result, 'options', { value: Object.freeze({ devLocalOwner: o.devLocalOwner === true, pm: o.pm === true, noLaunch: o.noLaunch === true }) });
return result;
});
}
// D2 (P0-3): bring back a bus whose host was killed, as a Console unit stop does to the hosts the
// Console started. busRecover is the proof: it refuses a live or unprovable host, a lock that does
// not match the dead host, a live writer and an answering socket. The host restarts on the options
// the dead host recorded, never wider; `pm: false` also keeps the chief from launching.
export function reviveStartOptions(recorded, { pm = true } = {}) {
const o = recorded && typeof recorded === 'object' ? recorded : {};
return { detach: true, devLocalOwner: o.devLocalOwner === true, pm: pm === true && o.pm === true, noLaunch: o.noLaunch === true };
}
export async function busRevive(ctx, id, { pm = true } = {}) {
await target(ctx, id);
const recovered = await busRecover(ctx, id);
const options = reviveStartOptions(recovered.options, { pm });
const cleared = { pid: recovered.pid, receipt: recovered.receipt };
let started;
try { started = await busStart(ctx, id, options); }
catch (e) {
// The dead host's state is gone by now: the caller must not report "nothing changed".
if (e && typeof e === 'object') { try { Object.defineProperty(e, 'recovered', { value: cleared }); } catch { /* frozen: the caller sees a plain failure */ } }
throw e;
}
const receipt = { ...started, recovered: cleared };
Object.defineProperty(receipt, 'done', { value: started.done });
// The chief was not relaunched although the dead host was started with --pm.
Object.defineProperty(receipt, 'pmDropped', { value: recovered.options.pm === true && !options.pm });
return receipt;
}
+25 -3
View File
@@ -21,8 +21,9 @@ async function fixture() {
const env = humanEnv({ HOME: root, MOSAIC_CONFIG: join(configDir, 'config.json') });
const moduleUrl = new URL('../src/cli.mjs', import.meta.url).href;
const code = `import { run } from ${JSON.stringify(moduleUrl)}; process.exitCode = await run(JSON.parse(process.argv[1]), undefined, {system: {dataRoot: process.argv[2]}});`;
const call = async (...args) => { const x = await exec(process.execPath, ['--input-type=module', '-e', code, JSON.stringify(args), dataRoot], { env }); return { out: { text: x.stdout }, err: { text: x.stderr } }; };
return { root, ctx, org, call };
const callWith = async (extra, ...args) => { const x = await exec(process.execPath, ['--input-type=module', '-e', code, JSON.stringify(args), dataRoot], { env: { ...env, ...extra } }); return { out: { text: x.stdout }, err: { text: x.stderr } }; };
const call = (...args) => callWith({}, ...args);
return { root, ctx, org, call, callWith };
}
const dead = 2147483647;
const json = (path, value) => writeFileSync(path, JSON.stringify(value), { mode: 0o600 });
@@ -121,8 +122,9 @@ test('org lock show and clear require a provably dead holder', async () => {
test('bus recover preserves evidence for dead and wrong-process hosts without signals', async () => {
const f = await fixture(), dir = join(f.org.root, 'bus-host'); mkdirSync(dir, { mode: 0o700 });
// A wrong process is a live pid whose start time is not the recorded one (the pid was reused).
for (const pid of [dead, process.pid]) {
const state = { pid, startTime: startTimeOf(pid) ?? '1', business: 'alpha', startedAt: '2026-01-01T00:00:00Z' };
const state = { pid, startTime: '1', business: 'alpha', startedAt: '2026-01-01T00:00:00Z' };
json(join(dir, 'host.json'), state); json(join(dir, 'host.lock'), state);
const x = await f.call('bus', 'recover', 'alpha');
const result = JSON.parse(x.out.text);
@@ -138,6 +140,26 @@ test('bus recover preserves evidence for dead and wrong-process hosts without si
assert.equal(process.pid > 0, true);
});
test('D2: bus recover refuses a live pid with the recorded start time, whatever its argv', async () => {
const f = await fixture(), dir = join(f.org.root, 'bus-host'); mkdirSync(dir, { mode: 0o700 });
const state = { pid: process.pid, startTime: startTimeOf(process.pid), business: 'alpha', startedAt: '2026-01-01T00:00:00Z' };
assert.match(state.startTime, /^\d+$/);
json(join(dir, 'host.json'), state); json(join(dir, 'host.lock'), state);
await assert.rejects(f.call('bus', 'recover', 'alpha'), /bus host identity is uncertain/);
assert.deepEqual(JSON.parse(readFileSync(join(dir, 'host.json'), 'utf8')), state);
assert.equal(existsSync(join(dir, 'host.lock')), true);
assert.ok(!readdirSync(dir).some(x => x.startsWith('recovery-')));
});
test('D2: bus recover refuses inside an agent run before it reads anything', async () => {
const f = await fixture(), dir = join(f.org.root, 'bus-host'); mkdirSync(dir, { mode: 0o700 });
const state = { pid: dead, startTime: '1', business: 'alpha', startedAt: '2026-01-01T00:00:00Z' };
json(join(dir, 'host.json'), state); json(join(dir, 'host.lock'), state);
await assert.rejects(f.callWith({ MOSAIC_AGENT_RUN: '1' }, 'bus', 'recover', 'alpha'), e => e.code === 3 && /outside any agent run/.test(e.stderr));
assert.equal(existsSync(join(dir, 'host.json')), true);
assert.ok(!readdirSync(dir).some(x => x.startsWith('recovery-')));
});
test('lock recovery refuses malformed or unreadable holders', async () => {
const f = await fixture(), path = join(f.ctx.configDir, 'org-lifecycle-alpha.lock');
for (const value of [{ pid: 0, at: '2026-01-01T00:00:00Z' }, { pid: dead }, null]) {
+162
View File
@@ -0,0 +1,162 @@
// D2 (P0-3): a Console unit stop (KillMode=mixed) SIGKILLs the bus hosts the Console started. These
// tests start real bus hosts, kill them the same way, and check that busRevive and the Console-start
// revive bring them back, and that a live host or a live writer lock is never taken over.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { spawn, spawnSync } from 'node:child_process';
import { existsSync, lstatSync, mkdtempSync, readFileSync, readdirSync, renameSync, rmSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { reserveOrg, activateOrg, listOrgs } from '../../organization/src/org-registry.mjs';
import { businessDoc, rolesCopy, writeJson } from '../../organization/tests/helpers.mjs';
import { humanEnv, notifyConfig } from './helpers.mjs';
import { busRevive, busStart, busStatus, busStop, setAside } from '../src/org-bus.mjs';
import { reviveBuses } from '../../webui/src/bus-revive.mjs';
const REPO = new URL('../../../', import.meta.url).pathname;
const IDS = ['alpha', 'beta'];
async function instance(t) {
const home = mkdtempSync(join(tmpdir(), 'rv-'));
const env = humanEnv({ HOME: home, MOSAIC_CONFIG: join(home, '.config/mosaic-rvtest/config.json') });
const setup = spawnSync(join(REPO, 'scripts/bootstrap.sh'), ['--instance', 'rvtest'], { env, encoding: 'utf8' });
assert.equal(setup.status, 0, setup.stderr);
const ctx = { configDir: join(home, '.config/mosaic-rvtest'), dataRoot: join(home, '.mosaic-rvtest'), instance: 'rvtest', env };
env.MOSAIC_ROLES_DIR = rolesCopy(home);
const roots = {};
for (const id of IDS) {
const org = await reserveOrg(ctx, id);
await activateOrg(ctx, id);
writeJson(join(ctx.configDir, 'businesses', `${id}.json`), businessDoc(home, id));
notifyConfig(org.root, id, null);
roots[id] = org.root;
}
const extra = [];
t.after(async () => {
for (const child of extra) { try { child.kill('SIGKILL'); } catch {} }
for (const id of IDS) { try { await busStop(ctx, id); } catch {} }
rmSync(home, { recursive: true, force: true });
});
return { ctx, roots, extra };
}
// What a Console unit stop does to a host it started: SIGKILL to the host and every process under it.
// The host is a session leader (detached), so its process group is the test's own; `done` resolves
// once the host is reaped, and the loop waits for the rest of the group to go.
async function kill(receipt) {
process.kill(-receipt.pid, 'SIGKILL');
await receipt.done;
for (let i = 0; i < 100; i++) {
try { process.kill(-receipt.pid, 0); } catch (e) { if (e.code === 'ESRCH') return; throw e; }
await new Promise(r => setTimeout(r, 50));
}
throw new Error(`process group ${receipt.pid} did not go`);
}
const host = root => JSON.parse(readFileSync(join(root, 'bus-host/host.json'), 'utf8'));
const recoveries = root => readdirSync(join(root, 'bus-host')).filter(name => /^recovery-.*\.json$/.test(name));
test('D2: a SIGKILLed bus host comes back from one revive, on the options it was started with', { timeout: 120000 }, async t => {
const { ctx, roots } = await instance(t);
const first = await busStart(ctx, 'alpha', { detach: true, devLocalOwner: true, noLaunch: true });
await kill(first);
// What a Console unit stop leaves behind: the dead host's state, lock and socket.
assert.equal(existsSync(join(roots.alpha, 'bus-host/host.json')), true);
await assert.rejects(busStatus(ctx, 'alpha'), e => e.name === 'CliError' && /stale/.test(e.message));
const revived = await busRevive(ctx, 'alpha');
assert.notEqual(revived.pid, first.pid);
assert.equal(revived.recovered.pid, first.pid);
assert.equal((await busStatus(ctx, 'alpha')).healthy, true);
assert.deepEqual(host(roots.alpha).options, { devLocalOwner: true, pm: false, noLaunch: true });
assert.equal(recoveries(roots.alpha).length, 1);
const receipt = JSON.parse(readFileSync(revived.recovered.receipt, 'utf8'));
assert.equal(receipt.host.pid, first.pid);
});
test('D2: a live bus host is never taken over, and its state is left as it was', { timeout: 120000 }, async t => {
const { ctx, roots } = await instance(t);
const live = await busStart(ctx, 'alpha', { detach: true, noLaunch: true });
const before = readFileSync(join(roots.alpha, 'bus-host/host.json'), 'utf8');
await assert.rejects(busRevive(ctx, 'alpha'), e => e.name === 'CliError' && e.message === `bus host pid ${live.pid} is alive`);
assert.equal(readFileSync(join(roots.alpha, 'bus-host/host.json'), 'utf8'), before);
assert.equal(recoveries(roots.alpha).length, 0);
const status = await busStatus(ctx, 'alpha');
assert.equal(status.healthy, true);
assert.equal(status.pid, live.pid);
});
test('D2: a writer lock held by a live process refuses the revive, even with the host dead', { timeout: 120000 }, async t => {
const { ctx, roots, extra } = await instance(t);
const first = await busStart(ctx, 'alpha', { detach: true, noLaunch: true });
await kill(first);
// A live process the test started holds the writer lock.
const holder = spawn('sleep', ['60'], { stdio: 'ignore' }); extra.push(holder);
writeFileSync(join(roots.alpha, 'bus/writer.lock'), JSON.stringify({ pid: holder.pid, at: new Date().toISOString() }), { mode: 0o600 });
const lockBefore = readFileSync(join(roots.alpha, 'bus/writer.lock'), 'utf8');
await assert.rejects(busRevive(ctx, 'alpha'), e => e.name === 'CliError' && e.message === `bus writer pid ${holder.pid} is alive`);
assert.equal(readFileSync(join(roots.alpha, 'bus/writer.lock'), 'utf8'), lockBefore);
assert.equal(existsSync(join(roots.alpha, 'bus-host/host.json')), true);
assert.equal(recoveries(roots.alpha).length, 0);
// The Console-start revive leaves it too, and logs the refusal.
const lines = [], trail = [];
const results = await reviveBuses({ ctx, listOrgs, busStatus, busRevive, env: {}, trail: e => trail.push(e), log: line => lines.push(line) });
assert.deepEqual(results.find(r => r.id === 'alpha'), { id: 'alpha', outcome: 'refused' });
assert.ok(lines.some(line => line.includes(`alpha: refused (bus writer pid ${holder.pid} is alive)`)), lines.join('\n'));
assert.deepEqual(trail.filter(e => e.org === 'alpha').map(e => e.outcome), ['requested', 'refused']);
assert.equal(existsSync(join(roots.alpha, 'bus-host/host.json')), true);
});
test('D2: Console start brings both killed buses back and reattaches to one that still runs', { timeout: 180000 }, async t => {
const { ctx, roots } = await instance(t);
const a = await busStart(ctx, 'alpha', { detach: true, devLocalOwner: true, noLaunch: true });
const b = await busStart(ctx, 'beta', { detach: true, devLocalOwner: true, noLaunch: true });
await kill(a); await kill(b);
const trail = [], lines = [];
const results = await reviveBuses({ ctx, listOrgs, busStatus, busRevive, env: {}, trail: e => trail.push(e), log: line => lines.push(line) });
assert.deepEqual(results.map(r => [r.id, r.outcome]), [['alpha', 'revived'], ['beta', 'revived']]);
for (const id of IDS) {
const status = await busStatus(ctx, id);
assert.equal(status.healthy, true, id);
assert.equal(host(roots[id]).options.devLocalOwner, true);
assert.deepEqual(trail.filter(e => e.org === id).map(e => [e.action, e.outcome, e.by.sub]), [['bus.revive', 'requested', 'local-dev-owner'], ['bus.revive', 'done', 'local-dev-owner']]);
}
// A second Console start finds them running and leaves them.
const pids = Object.fromEntries(await Promise.all(IDS.map(async id => [id, (await busStatus(ctx, id)).pid])));
const again = await reviveBuses({ ctx, listOrgs, busStatus, busRevive, env: {}, trail: e => trail.push(e), log: () => {} });
assert.deepEqual(again.map(r => [r.id, r.outcome]), [['alpha', 'running'], ['beta', 'running']]);
for (const id of IDS) assert.equal((await busStatus(ctx, id)).pid, pids[id]);
});
test('D2: a start that fails after the dead state was cleared says so (error.recovered)', { timeout: 120000 }, async t => {
const { ctx, roots } = await instance(t);
const first = await busStart(ctx, 'alpha', { detach: true, noLaunch: true });
await kill(first);
// A record busStart refuses (--pm with --no-launch), rewritten in place: the clear goes through and
// the start that follows fails, the way a host that cannot come up would.
const file = join(roots.alpha, 'bus-host/host.json');
writeFileSync(file, JSON.stringify({ ...host(roots.alpha), options: { devLocalOwner: false, pm: true, noLaunch: true } }));
await assert.rejects(busRevive(ctx, 'alpha'), e => e.name === 'CliError' && /--pm or --no-launch/.test(e.message) && e.recovered?.pid === first.pid);
assert.equal(existsSync(file), false);
assert.equal(recoveries(roots.alpha).length, 1);
});
test('D2: an entry swapped after inspection is put back and nothing is removed', t => {
const dir = mkdtempSync(join(tmpdir(), 'aside-'));
t.after(() => rmSync(dir, { recursive: true, force: true }));
const a = join(dir, 'host.json'), b = join(dir, 'writer.lock');
writeFileSync(a, 'dead host', { mode: 0o600 }); writeFileSync(b, 'dead writer', { mode: 0o600 });
const entries = [{ path: a, st: lstatSync(a) }, { path: b, st: lstatSync(b) }];
// The swap: the inspected writer.lock is moved off (its inode stays in use) and a new one takes the name.
renameSync(b, join(dir, 'kept')); writeFileSync(b, 'live writer', { mode: 0o600 });
assert.throws(() => setAside(entries), e => e.name === 'CliError' && e.message === 'bus recovery state changed');
assert.equal(readFileSync(a, 'utf8'), 'dead host');
assert.equal(readFileSync(b, 'utf8'), 'live writer');
assert.deepEqual(readdirSync(dir).sort(), ['host.json', 'kept', 'writer.lock']);
// Unswapped, the entries go aside and remove() deletes only those.
renameSync(join(dir, 'kept'), b);
const fresh = [{ path: a, st: lstatSync(a) }, { path: b, st: lstatSync(b) }];
const aside = setAside(fresh);
assert.equal(existsSync(a) || existsSync(b), false);
writeFileSync(a, 'a new host', { mode: 0o600 });
aside.remove();
assert.deepEqual(readdirSync(dir), ['host.json']);
assert.equal(readFileSync(a, 'utf8'), 'a new host');
});
+6 -3
View File
@@ -30,12 +30,15 @@ export function resolverSubject(business, resolver, target) {
}
// Aggregate provider/model limits cover names that do not exist yet. An account
// quota covers every future launch only when every enabled harness uses that account.
export function authorityBudgetCovered(business) {
// quota covers every future launch only when every enabled harness's module declares
// that its launches spend launch.account (its countsAccount export). The host passes
// that answer as countsAccount(name) -> true|false|null; core names no harness, and
// without the lookup no account quota counts.
export function authorityBudgetCovered(business, {countsAccount=()=>null}={}) {
const launch=business.launch, tokens=launch?.limits?.tokens, checks=business.plugins?.['limit-check']??[];
const quota=x=>Number.isSafeInteger(x?.limit) && x.limit>=0 && typeof x.since==='string' && !Number.isNaN(Date.parse(x.since));
const account=launch?.account?.name, harnesses=business.plugins?.harness??[];
if(account && harnesses.length && harnesses.every(h=>h==='claude-code') &&
if(account && harnesses.length && harnesses.every(h=>countsAccount(h)===true) &&
(checks.includes('daily-quota') && quota(tokens?.daily?.[account]) ||
checks.includes('token-limit') && quota(tokens?.keys?.[account]) && tokens.keys[account].reset===undefined))return true;
const models=Object.values(launch?.models??{});
+21 -4
View File
@@ -38,8 +38,25 @@ test('AU authority budgets cover all selectable models and only enforced account
b.launch.limits.tokens.models={two:{n:cap}};assert.equal(authorityBudgetCovered(b),true);
b.plugins['limit-check']=[];assert.equal(authorityBudgetCovered(b),false);
b.launch.account={name:'shared'};b.launch.limits.tokens={keys:{shared:cap}};b.plugins['limit-check']=['token-limit'];
assert.equal(authorityBudgetCovered(b),false);
b.plugins.harness=['claude-code'];assert.equal(authorityBudgetCovered(b),true);
b.plugins.harness.push('pi');assert.equal(authorityBudgetCovered(b),false);
b.plugins.harness=['claude-code'];b.plugins['limit-check']=['daily-quota'];b.launch.limits.tokens={daily:{shared:{...cap,reset:'00:00',timeZone:'UTC'}}};assert.equal(authorityBudgetCovered(b),true);
// What today's two harness modules declare (packages/plugins/harness/*/index.mjs countsAccount).
const declared={countsAccount:h=>({'claude-code':true,pi:false})[h]??null};
assert.equal(authorityBudgetCovered(b,declared),false);
b.plugins.harness=['claude-code'];assert.equal(authorityBudgetCovered(b,declared),true);
b.plugins.harness.push('pi');assert.equal(authorityBudgetCovered(b,declared),false);
b.plugins.harness=['claude-code'];b.plugins['limit-check']=['daily-quota'];b.launch.limits.tokens={daily:{shared:{...cap,reset:'00:00',timeZone:'UTC'}}};assert.equal(authorityBudgetCovered(b,declared),true);
});
test('AU an account quota covers authority only when every enabled harness declares it spends the launch account (#1873)',()=>{
const cap={limit:0,since:'2026-01-01T00:00:00.000Z'};
const b={plugins:{harness:['alpha','beta'],'limit-check':['token-limit']},launch:{account:{name:'shared'},models:{one:{provider:'one',model:'m'}},limits:{tokens:{keys:{shared:cap}}}}};
const declares=map=>({countsAccount:h=>map[h]??null});
// The declared fact decides, whatever the harness is called.
assert.equal(authorityBudgetCovered(b,declares({alpha:true,beta:true})),true);
assert.equal(authorityBudgetCovered(b,declares({alpha:true,beta:false})),false);
assert.equal(authorityBudgetCovered(b,declares({alpha:true})),false,'an undeclared harness is not covered');
assert.equal(authorityBudgetCovered(b,declares({alpha:true,beta:'yes'})),false,'only true counts');
b.plugins.harness=['claude-code'];
assert.equal(authorityBudgetCovered(b,declares({'claude-code':false})),false,'the name alone covers nothing');
assert.equal(authorityBudgetCovered(b),false,'no lookup: fail closed');
// The provider/model branch needs no harness fact.
b.launch.limits.tokens.providers={one:cap};assert.equal(authorityBudgetCovered(b),true);
});
+3 -1
View File
@@ -251,7 +251,9 @@ export returns:
session credential (`no-session-credential`) and `MOSAIC_CLAUDE_BIN`.
- `version({env})` is what the manifest records as `harnessVersion`.
- `countsAccount()` says whether its launches spend `launch.account` and get
the row 126 account checks.
the row 126 account checks. The same answer decides whether an account
quota covers authority hires (#1873); the bus reads it once at start, so
return a constant.
- `mounts`, `builtinWeb`, `builtins`, `typedName`, `reliesOn`,
`gateTimeout`, `bundle`, `env`, `turnEnd`, `outputPolicy` and `transcript` give its
sandbox binds, web, tool names, S0 lines, bundle files, adapter variables,
+7 -1
View File
@@ -181,10 +181,16 @@ test('PL1 runtime discovers installed harness levels from the instance root',asy
const {KINDS}=await import('../src/kinds.mjs');
const f=fixture(t);f.manifest.kinds=['harness'];f.manifest.interfaceVersions={harness:1};
mkdirSync(join(f.src,'harness'));
writeFileSync(join(f.src,'harness/index.mjs'),KINDS.harness.exports.map(name=>`export function ${name}(){return ${name==='levels'?"['high']":'null'};}`).join('\n')+'\nexport const interfaceVersion=1;');
writeFileSync(join(f.src,'harness/index.mjs'),KINDS.harness.exports.map(name=>`export function ${name}(){return ${name==='levels'?"['high']":name==='countsAccount'?'true':'null'};}`).join('\n')+'\nexport const interfaceVersion=1;');
const hash=f.seal();f.install(f.src,{expectedHash:hash});
const levels=await harnessLevels({dataRoot:f.dataRoot,log:()=>{}});
assert.deepEqual(levels('acmeplug'),['high']);assert.ok(levels.names.includes('pi'));
// #1873: the module's countsAccount answer, until the lock pins another package for that name.
assert.equal(levels.countsAccount('acmeplug'),true);
const lock=join(f.dataRoot,'plugins.lock.json'),doc=JSON.parse(readFileSync(lock,'utf8'));
doc.plugins.acmeplug.sha256='f'.repeat(64);writeFileSync(lock,JSON.stringify(doc));
assert.equal(levels.countsAccount('acmeplug'),null,'a replaced pin is not covered until restart');
assert.equal(levels.countsAccount('pi'),false,'a tree module keeps its answer');
});
test('PL1 invalid UTF-8 manifest bytes refuse instead of normalizing outside the pin',t=>{
+82
View File
@@ -0,0 +1,82 @@
// D2 (P0-3): when a dev-local-owner Console starts, it reattaches to the organization buses that
// still run and brings back the ones a Console unit stop killed. The unit's KillMode=mixed sends
// SIGKILL to the bus hosts the Console started; a host killed that way leaves its host.json behind,
// which a host stopped on purpose (mosaic bus stop, Stop in the page) removes as it closes.
//
// For each active organization:
// - a bus whose state checks out (running, or stopped cleanly) is left as it is;
// - a bus whose state does not check out goes to busRevive, whose busRecover clears the state only
// when the host is proven dead (no live pid with that start time, the lock is the dead host's, no
// live writer, no socket that answers). Anything else refuses and the state stays for a person.
// The chief is not relaunched (pm: false): no agent starts without a person's press.
// Every attempt is a trail line before anything changes; a trail that cannot be written skips it.
// A Console whose own environment carries an agent run's markers revives nothing (the human-only
// rule `mosaic bus recover` and Restart bus's `mosaic org reload` follow).
// While an organization's revive runs, revivePending(id) holds it, so a Restart bus press waits for
// it instead of racing it for the lifecycle lock.
import { DEV_OWNER } from '../../cli/src/dev-local-owner.mjs';
import { AGENT_MARKERS } from '../../cli/src/transport.mjs';
const isCliError = e => e?.name === 'CliError' && Number.isInteger(e.exitCode);
const BY = Object.freeze({ ...DEV_OWNER, name: 'Console start' });
const pending = new Map();
export const revivePending = id => pending.get(id) ?? null;
export async function reviveBuses({ ctx, listOrgs, busStatus, busRevive, trail, log = () => {}, env = process.env }) {
const results = [];
if (AGENT_MARKERS.some(k => env?.[k])) { log('bus revive: this Console runs inside an agent run; no bus was checked'); return results; }
let orgs;
try { orgs = await listOrgs(ctx); }
catch { log('bus revive: the organization list could not be read; no bus was checked'); return results; }
const active = (orgs ?? []).filter(org => org?.state === 'active').map(org => org.id);
const settle = new Map();
for (const id of active) {
let done;
const wait = new Promise(resolve => { done = resolve; });
pending.set(id, wait);
settle.set(id, () => { if (pending.get(id) === wait) pending.delete(id); done(); });
}
try {
for (const id of active) {
try { results.push(await reviveOne({ ctx, id, busStatus, busRevive, trail, log })); }
finally { settle.get(id)(); }
}
} finally { for (const release of settle.values()) release(); }
return results;
}
async function reviveOne({ ctx, id, busStatus, busRevive, trail, log }) {
let status = null;
try { status = await busStatus(ctx, id); }
catch (e) {
if (!isCliError(e)) { log(`bus revive: ${id}: bus state could not be read; left as it is`); return { id, outcome: 'uncertain' }; }
}
if (status) {
const outcome = status.healthy ? 'running' : Number.isSafeInteger(status.pid) ? 'not-answering' : 'stopped';
if (outcome === 'running') log(`bus revive: ${id}: bus answers; reattached`);
return { id, outcome };
}
try { trail({ by: BY, action: 'bus.revive', org: id, outcome: 'requested' }); }
catch { log(`bus revive: ${id}: the action trail could not be written; nothing was recovered`); return { id, outcome: 'trail-unavailable' }; }
const note = outcome => { try { trail({ by: BY, action: 'bus.revive', org: id, outcome, ...(outcome === 'failed' ? { recovered: true } : {}) }); } catch { log(`bus revive: ${id}: trail write failed after ${outcome}`); } };
try {
const receipt = await busRevive(ctx, id, { pm: false });
note('done');
log(`bus revive: ${id}: dead bus host cleared; bus started (pid ${receipt.pid})`);
if (receipt.pmDropped) log(`bus revive: ${id}: the dead host was started with --pm; the chief was not relaunched. mosaic bus stop ${id}; mosaic bus start ${id} --pm brings it back`);
return { id, outcome: 'revived', pid: receipt.pid };
} catch (e) {
// The reason names the organization id and pids only, never a host path (busRecover's and
// busStart's messages); an error that is not a CliError shows its code, not its text.
const reason = isCliError(e) ? e.message : `unexpected error${typeof e?.code === 'string' ? ` ${e.code}` : ''}`;
if (e?.recovered) {
note('failed');
log(`bus revive: ${id}: dead bus host cleared, but the bus did not start (${reason}); see mosaic bus status ${id}`);
return { id, outcome: 'failed' };
}
note('refused');
log(`bus revive: ${id}: refused (${reason}); left for mosaic bus status ${id}`);
return { id, outcome: 'refused' };
}
}
+5 -1
View File
@@ -14,7 +14,7 @@ import { startTesterSocket } from './tester-socket.mjs';
import { TESTER_CLI } from '../../cli/src/transport.mjs';
import { views } from '../../bus/src/views.mjs';
import { loadSystem, socketPath, orgPaths, REPO, rolesDir } from '../../cli/src/config.mjs';
import { busStatus as orgBusStatus } from '../../cli/src/org-bus.mjs';
import { busStatus as orgBusStatus, busRevive as orgBusRevive } from '../../cli/src/org-bus.mjs';
import { listOrgs, resolveOrg } from '../../organization/src/org-registry.mjs';
import { showOrg } from '../../organization/src/org-service.mjs';
import { configDir } from '../../organization/src/organization-config.mjs';
@@ -28,6 +28,7 @@ import { readFileSync, statSync } from 'node:fs';
import { fileURLToPath } from 'node:url';
import { isAbsolute, join } from 'node:path';
import { actionTrail } from './action-trail.mjs';
import { reviveBuses } from './bus-revive.mjs';
import { agentLevelsWriteRoutes } from './agent-levels-write.mjs';
import { RESTART_EXIT } from '../../cli/src/org-console.mjs';
import { ID_PATTERN } from '../../organization/src/vocabulary.mjs';
@@ -235,6 +236,9 @@ try {
console.log(`Listening: ${lan.bind}:${port}, allowed Host names ${lan.allowHosts.map(name => `${name}:${port}`).join(', ')}`);
}
console.log(`Mosaic Console: http://${urlHost}:${port}/\nBoard: ${options.board}\n${busLine}${queueLine}${previewLine}${ssoLines}${proofLine}${testerServer ? `\nTester Console: unix:${testerServer.address()} (binds the tester)` : ''}\nCtrl-C stops this server.`);
// D2 (P0-3): the dev Console reattaches to the buses that still run and brings back the ones its
// unit stop killed. It runs after the Console answers, so a slow bus never delays the page.
if (organizations && trail) reviveBuses({ ctx: { configDir: configDir(), dataRoot: system.dataRoot, instance }, listOrgs, busStatus: orgBusStatus, busRevive: orgBusRevive, trail, log: line => console.log(line) }).catch(() => console.log('bus revive: stopped on an unexpected error'));
for (const signal of ['SIGINT', 'SIGTERM']) process.once(signal, () => { sso?.close(); server.close(); testerServer?.close(); });
} catch (err) {
console.error(`refused: ${err.message}`);
+40 -2
View File
@@ -16,7 +16,8 @@
// POST /api/orgs/ID/setup/queue C2 queueInit mosaic org queue init
// POST /api/orgs/ID/setup/project/P C2 projectRepoInit mosaic org project repo init
// POST /api/orgs/ID/setup/bus C3 busStart mosaic bus start --detach
// POST /api/orgs/ID/restart/bus C3 busRestart mosaic bus restart (F3; busStart when stopped)
// POST /api/orgs/ID/restart/bus C3 busRestart mosaic bus restart (F3; busStart when stopped;
// busRevive for a host that died, dev-local owner only, D2)
//
// GET /api/orgs (the list) and other /api/orgs/ID/... routes are S5's; this
// handler returns false for them.
@@ -27,6 +28,8 @@ import { checkProject, initProject, projectRepo, recordContext } from '../../pro
import { refuseInsideAgent } from '../../cli/src/transport.mjs';
import { join } from 'node:path';
import { actor } from './action-trail.mjs';
import { DEV_OWNER } from '../../cli/src/dev-local-owner.mjs';
import { revivePending } from './bus-revive.mjs';
const ID = /^[a-z0-9][a-z0-9._-]{0,63}$/;
const KEY = /^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$/;
@@ -51,6 +54,7 @@ export function verbFor(step, id, opts = {}) {
if (step === 'records') return `mosaic project init ${id} ${opts.project}`;
if (step === 'bus') return `mosaic bus start ${id} --dev-local-owner --detach`;
if (step === 'bus-restart') return `mosaic bus restart ${id}`;
if (step === 'bus-revive') return `mosaic bus recover ${id}; mosaic bus start ${id} --dev-local-owner --detach`;
throw new Error(`unknown step ${step}`);
}
@@ -563,7 +567,8 @@ async function route(req, res, { path, search, org, admin, principal, write, log
// F3 (REQ-WEB-3): one control restarts an organization's bus and answers with a done receipt. A
// running host restarts on the options it was started with (C3 busRestart keeps --dev-local-owner);
// a stopped one starts as setup starts it. The trail line is written before anything changes, and
// a trail that cannot be written refuses the restart. Stale host state is never cleared here.
// a trail that cannot be written refuses the restart. Stale host state is cleared only for the
// dev-local owner, and only through busRevive (D2), whose busRecover refuses a host it cannot prove dead.
async function restartBus(req, res, { sub, write, id, ctx, service, admin, log, trail, by }) {
const step = 'bus-restart';
if (sub !== 'bus' || !write) return json(res, 404, { error: 'not found' });
@@ -571,11 +576,14 @@ async function restartBus(req, res, { sub, write, id, ctx, service, admin, log,
try { await service.showOrg(ctx, id); } catch (e) { return refusal(res, e, { step, id }); }
const bus = await need(admin, 'bus', res, step); if (!bus) return;
if (!trail) return json(res, 503, { error: 'trail-unavailable', step, message: 'this Console has no action trail, so it does not restart buses' });
// D2: the dev Console's start-up revive may be bringing this bus back; the press acts on the result.
await revivePending(id);
let status;
try { status = await bus.busStatus(ctx, id); }
catch (e) {
if (['org-unknown', 'org-mismatch', 'org-unavailable', 'org-archived'].includes(e?.code)) return refusal(res, e, { step, id });
if (isCliError(e)) {
if (isDevOwner(by) && typeof bus.busRevive === 'function') return reviveBus(res, { id, ctx, bus, log, trail, by, step });
// Refused before anything changed, but the owner asked: the trail records the attempt.
try { trail({ by, action: 'bus.restart', org: id, outcome: 'refused', running: null }); } catch { log(`org-admin: ${id} bus restart: trail write failed after refused`); }
return json(res, 409, { error: 'bus-refused', step, message: `${plainMessage(ctx, e.message)}. Inspect it with mosaic bus status ${id}; mosaic bus recover ${id} clears a host that died.` });
@@ -600,3 +608,33 @@ async function restartBus(req, res, { sub, write, id, ctx, service, admin, log,
try { healthy = (await bus.busStatus(ctx, id)).healthy; } catch { /* the receipt still stands */ }
return json(res, 200, { outcome: 'done', step, restarted: running, healthy, receipt: forReceipt(ctx, receipt), verb: running ? verbFor('bus-restart', id) : verbFor('bus', id) });
}
// The dev-local owner exactly (packages/cli/src/dev-local-owner.mjs). A signed-in session carries
// its provider's issuer, so no SSO user matches.
const isDevOwner = by => by?.provider === DEV_OWNER.provider && by.iss === DEV_OWNER.iss && by.sub === DEV_OWNER.sub;
// D2 (P0-3): the dev Console's Restart bus on a bus whose state did not check out. A Console unit
// stop kills the hosts the Console started and leaves their state behind; busRevive clears that
// state only when busRecover proves the host dead, then starts the bus on its recorded options.
// A live or unprovable host, or a lock that is not the dead host's, refuses and nothing changes.
async function reviveBus(res, { id, ctx, bus, log, trail, by, step }) {
try { trail({ by, action: 'bus.restart', org: id, outcome: 'requested', running: false, recover: true }); }
catch { return json(res, 500, { error: 'trail-unavailable', step, message: 'the action trail could not be written, so nothing was recovered' }); }
const note = outcome => { try { trail({ by, action: 'bus.restart', org: id, outcome, recover: true }); } catch { log(`org-admin: ${id} bus recover: trail write failed after ${outcome}`); } };
let receipt;
try { receipt = await bus.busRevive(ctx, id); }
catch (e) {
// busRevive marks a failure after busRecover cleared the dead host's state (e.recovered).
const cleared = Boolean(e?.recovered);
note(cleared ? 'failed' : 'refused');
if (!cleared && e?.code === 'org-locked') return json(res, 409, { error: 'bus-busy', step, message: `another bus or lifecycle action on ${id} is running, and nothing was changed. Press Restart bus again when it ends.` });
if (cleared) return json(res, 409, { error: 'bus-failed', step, message: isCliError(e) ? plainMessage(ctx, `The dead host's state was cleared, but ${e.message}`) : `The dead host's state was cleared, but the bus did not start. Check it with mosaic bus status ${id}.` });
if (!isCliError(e)) return refusal(res, e, { step, id, unknownOk: true });
return json(res, 409, { error: 'bus-refused', step, message: `${plainMessage(ctx, e.message)}. Inspect it with mosaic bus status ${id}.` });
}
note('done');
log(`org-admin: ${id} bus recovered and started`);
let healthy = null;
try { healthy = (await bus.busStatus(ctx, id)).healthy; } catch { /* the receipt still stands */ }
return json(res, 200, { outcome: 'done', step, restarted: false, recovered: true, healthy, receipt: forReceipt(ctx, receipt), verb: verbFor('bus-revive', id) });
}
+3 -1
View File
@@ -146,7 +146,9 @@
if (response.type === 'opaqueredirect') throw new Error('Sign in again, then retry.');
const body = await response.json();
if (!response.ok || body.outcome !== 'done' || body.step !== 'bus-restart') throw new Error(body.message || 'The restart was refused. Nothing was retried.');
busReceipt = `${body.restarted ? 'Bus restarted' : 'Bus started'} at ${clock()}. ${body.healthy ? 'It answers.' : `It has not answered yet. Check it with ${body.verb.replace('restart', 'status')}.`}`;
// D2: a recovered bus's verb is two commands; its check is the status verb.
const check = body.recovered ? `mosaic bus status ${id}` : body.verb.replace('restart', 'status');
busReceipt = `${body.recovered ? 'Dead bus host cleared; bus started' : body.restarted ? 'Bus restarted' : 'Bus started'} at ${clock()}. ${body.healthy ? 'It answers.' : `It has not answered yet. Check it with ${check}.`}`;
// The page's send flags were read while the bus was in its old state; read them again. This
// read goes through the page's fetch, which display.js scopes to the selected organization.
// The organization list goes first, so the view the config event redraws shows the bus as it is now.
+95
View File
@@ -0,0 +1,95 @@
// D2 (P0-3): the dev Console's start-up revive (bus-revive.mjs), on fakes. The real-host paths
// (a SIGKILLed host, a live host, a live writer lock) are in packages/cli/tests/org-revive.test.mjs.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { reviveBuses, revivePending } from '../src/bus-revive.mjs';
import { reviveStartOptions } from '../../cli/src/org-bus.mjs';
const cliError = (message, extra = {}) => Object.assign(new Error(message), { name: 'CliError', exitCode: 3 }, extra);
const orgs = ids => async () => ids.map(id => ({ id, state: 'active' }));
function run({ ids = ['alpha'], status, revive, trail, env = {} }) {
const calls = [], lines = [], trailed = [];
const done = reviveBuses({
ctx: {}, listOrgs: orgs(ids),
busStatus: async (ctx, id) => status(id),
busRevive: async (ctx, id, opts) => { calls.push([id, opts]); return revive(id); },
trail: trail ?? (e => trailed.push(e)),
log: line => lines.push(line), env,
});
return { done, calls, lines, trailed };
}
test('the start-up revive asks busRevive for pm:false, so no chief launches without a press', async () => {
const r = run({ status: () => { throw cliError('stale bus host lock'); }, revive: () => ({ pid: 77 }) });
assert.deepEqual(await r.done, [{ id: 'alpha', outcome: 'revived', pid: 77 }]);
assert.deepEqual(r.calls, [['alpha', { pm: false }]]);
});
test('reviveStartOptions keeps the recorded options, never wider, and pm only when both allow it', () => {
assert.deepEqual(reviveStartOptions({ devLocalOwner: true, pm: true, noLaunch: false }, { pm: false }), { detach: true, devLocalOwner: true, pm: false, noLaunch: false });
assert.deepEqual(reviveStartOptions({ devLocalOwner: true, pm: true, noLaunch: false }), { detach: true, devLocalOwner: true, pm: true, noLaunch: false });
assert.deepEqual(reviveStartOptions({ pm: false, noLaunch: true }, { pm: true }), { detach: true, devLocalOwner: false, pm: false, noLaunch: true });
// Anything but a literal true is false: a malformed record widens nothing.
assert.deepEqual(reviveStartOptions({ devLocalOwner: 'yes', pm: 1 }), { detach: true, devLocalOwner: false, pm: false, noLaunch: false });
assert.deepEqual(reviveStartOptions(null), { detach: true, devLocalOwner: false, pm: false, noLaunch: false });
});
test('a dropped --pm is logged with the way back', async () => {
const r = run({ status: () => { throw cliError('stale bus host lock'); }, revive: () => Object.defineProperty({ pid: 9 }, 'pmDropped', { value: true }) });
await r.done;
assert.ok(r.lines.some(l => l === 'bus revive: alpha: the dead host was started with --pm; the chief was not relaunched. mosaic bus stop alpha; mosaic bus start alpha --pm brings it back'), r.lines.join('\n'));
});
test('running, not-answering, stopped and unreadable buses are left alone and never trailed', async () => {
const states = { a: { healthy: true, pid: 1 }, b: { healthy: false, pid: 2 }, c: { healthy: false, pid: null } };
const r = run({ ids: ['a', 'b', 'c', 'd'], status: id => { if (id === 'd') throw new Error('EACCES /secret/path'); return states[id]; }, revive: () => assert.fail('revived') });
assert.deepEqual(await r.done, [{ id: 'a', outcome: 'running' }, { id: 'b', outcome: 'not-answering' }, { id: 'c', outcome: 'stopped' }, { id: 'd', outcome: 'uncertain' }]);
assert.deepEqual(r.trailed, []);
assert.ok(!r.lines.join('\n').includes('/secret/path'));
});
test('a trail that cannot be written skips the revive', async () => {
const r = run({ status: () => { throw cliError('stale bus host lock'); }, revive: () => assert.fail('revived without a trail line'), trail: () => { throw new Error('disk full'); } });
assert.deepEqual(await r.done, [{ id: 'alpha', outcome: 'trail-unavailable' }]);
assert.deepEqual(r.calls, []);
});
test('a start that fails after the dead state was cleared is "failed", not "refused"', async () => {
const r = run({ status: () => { throw cliError('stale bus host lock'); }, revive: () => { throw cliError('bus failed to start; see bus-host/host.log in the runtime directory of organization alpha', { recovered: { pid: 4 } }); } });
assert.deepEqual(await r.done, [{ id: 'alpha', outcome: 'failed' }]);
assert.deepEqual(r.trailed.map(e => [e.outcome, e.recovered]), [['requested', undefined], ['failed', true]]);
assert.match(r.lines.at(-1), /^bus revive: alpha: dead bus host cleared, but the bus did not start \(bus failed to start;/);
// A non-CliError shows its code, never its text.
const q = run({ status: () => { throw cliError('stale bus host lock'); }, revive: () => { throw Object.assign(new Error('EIO /home/someone/x'), { code: 'org-locked' }); } });
assert.deepEqual(await q.done, [{ id: 'alpha', outcome: 'refused' }]);
assert.equal(q.lines.at(-1), 'bus revive: alpha: refused (unexpected error org-locked); left for mosaic bus status alpha');
});
test('revivePending holds each organization until its own revive ends', async () => {
let release;
const gate = new Promise(resolve => { release = resolve; });
const r = run({ ids: ['alpha', 'beta'], status: id => { if (id === 'alpha') throw cliError('stale bus host lock'); return { healthy: true, pid: 3 }; }, revive: async () => { await gate; return { pid: 5 }; } });
await new Promise(resolve => setImmediate(resolve));
const alpha = revivePending('alpha'), beta = revivePending('beta');
assert.ok(alpha instanceof Promise); assert.ok(beta instanceof Promise);
assert.equal(revivePending('gamma'), null);
let settled = false; alpha.then(() => { settled = true; });
await new Promise(resolve => setImmediate(resolve));
assert.equal(settled, false);
release();
await r.done;
assert.equal(settled, true);
assert.equal(revivePending('alpha'), null); assert.equal(revivePending('beta'), null);
});
test('a Console that runs inside an agent run revives nothing and reads no organization', async () => {
for (const marker of ['MOSAIC_AGENT_RUN', 'CLAUDECODE', 'CODEX_THREAD_ID']) {
let listed = false;
const lines = [];
const results = await reviveBuses({ ctx: {}, listOrgs: async () => { listed = true; return []; }, busStatus: () => assert.fail('read'), busRevive: () => assert.fail('revived'), trail: () => assert.fail('trailed'), log: line => lines.push(line), env: { [marker]: '1' } });
assert.deepEqual(results, []);
assert.equal(listed, false);
assert.deepEqual(lines, ['bus revive: this Console runs inside an agent run; no bus was checked']);
}
});
+109
View File
@@ -565,3 +565,112 @@ test('a bus that refuses to restart is a visible failure and the trail says refu
assert.deepEqual(tr.lines().slice(2), [{ by: DEV, action: 'bus.restart', org: 'acme', outcome: 'refused', running: null }]);
assert.equal(c.calls.slice(before).filter(y => ['busStart', 'busRestart'].includes(y[0])).length, 0);
});
// D2 (P0-3): on the dev Console, Restart bus on a host that died (its state left behind by a
// Console unit stop) clears that state through busRevive and starts the bus. busRevive's
// busRecover is the proof; a live or unprovable host refuses and nothing starts.
const cliError = message => Object.assign(new Error(message), { name: 'CliError', exitCode: 3 });
const withRevive = (c, revive) => c.admin.bus().then(bus => {
bus.busStatus = async () => { throw cliError('stale bus host lock'); };
bus.busRevive = async (ctx, id, opts) => { c.calls.push(['busRevive', id, opts]); return revive(bus, id); };
return bus;
});
test('D2: the dev owner\'s Restart bus brings back a host that died, and the trail records the recovery', async t => {
const tr = trailIn(t);
const c = await console_(t, { trail: tr.trail });
await c.call('POST', '/api/orgs', { id: 'acme', doc: browserDoc('acme') });
await withRevive(c, (bus, id) => {
bus.busStatus = async () => ({ org: id, healthy: true, pid: 5151 });
return { org: id, pid: 5151, socket: '/data/org-runtime/acme/bus/broker.sock', root: '/data/org-runtime/acme', recovered: { pid: 4242, receipt: '/data/org-runtime/acme/bus-host/recovery-1.json' } };
});
const r = await c.call('POST', '/api/orgs/acme/restart/bus', {});
assert.equal(r.status, 200, r.raw);
assert.deepEqual(r.body, { outcome: 'done', step: 'bus-restart', restarted: false, recovered: true, healthy: true,
receipt: { org: 'acme', pid: 5151, socket: '<data root>/org-runtime/acme/bus/broker.sock', root: '<data root>/org-runtime/acme', recovered: { pid: 4242, receipt: '<data root>/org-runtime/acme/bus-host/recovery-1.json' } },
verb: 'mosaic bus recover acme; mosaic bus start acme --dev-local-owner --detach' });
assert.deepEqual(c.calls.filter(x => ['busRevive', 'busStart', 'busRestart'].includes(x[0])), [['busRevive', 'acme', undefined]]);
assert.deepEqual(tr.lines(), [
{ by: DEV, action: 'bus.restart', org: 'acme', outcome: 'requested', running: false, recover: true },
{ by: DEV, action: 'bus.restart', org: 'acme', outcome: 'done', recover: true },
]);
});
test('D2: a host that is alive or unprovable still refuses on the dev Console, and nothing starts', async t => {
const tr = trailIn(t);
const c = await console_(t, { trail: tr.trail });
await c.call('POST', '/api/orgs', { id: 'acme', doc: browserDoc('acme') });
for (const reason of ['bus host pid 4242 is alive', 'bus host lock does not match host state', 'bus socket answers or is uncertain', 'bus host identity is uncertain']) {
await withRevive(c, () => { throw cliError(reason); });
const r = await c.call('POST', '/api/orgs/acme/restart/bus', {});
assert.equal(r.status, 409, reason); assert.equal(r.body.error, 'bus-refused');
assert.equal(r.body.message, `${reason}. Inspect it with mosaic bus status acme.`);
}
assert.equal(c.calls.filter(x => ['busStart', 'busRestart'].includes(x[0])).length, 0);
assert.deepEqual(tr.lines().map(e => e.outcome), ['requested', 'refused', 'requested', 'refused', 'requested', 'refused', 'requested', 'refused']);
// A host that was cleared but did not come back says so, without host paths.
await withRevive(c, () => { throw Object.assign(cliError('bus failed to start; see bus-host/host.log in the runtime directory of organization acme'), { recovered: { pid: 4242 } }); });
const f = await c.call('POST', '/api/orgs/acme/restart/bus', {});
assert.equal(f.status, 409); assert.equal(f.body.error, 'bus-failed');
assert.equal(f.body.message, "The dead host's state was cleared, but bus failed to start; see bus-host/host.log in the runtime directory of organization acme");
assert.deepEqual(tr.lines().slice(-1), [{ by: DEV, action: 'bus.restart', org: 'acme', outcome: 'failed', recover: true }]);
// Another lifecycle action holds the organization's lock: nothing changed, and the page says to press again.
await withRevive(c, () => { throw Object.assign(new Error('organization lifecycle lock is held; inspect the lock and resume'), { code: 'org-locked' }); });
const l = await c.call('POST', '/api/orgs/acme/restart/bus', {});
assert.equal(l.status, 409); assert.equal(l.body.error, 'bus-busy');
assert.equal(l.body.message, 'another bus or lifecycle action on acme is running, and nothing was changed. Press Restart bus again when it ends.');
assert.deepEqual(tr.lines().slice(-1).map(e => e.outcome), ['refused']);
// No trail, no recovery.
const broken = await console_(t, { trail: () => { throw new Error('disk full'); } });
await broken.call('POST', '/api/orgs', { id: 'acme', doc: browserDoc('acme') });
await withRevive(broken, () => assert.fail('busRevive ran without a trail line'));
const b = await broken.call('POST', '/api/orgs/acme/restart/bus', {});
assert.equal(b.status, 500); assert.equal(b.body.error, 'trail-unavailable');
});
test('D2: a signed-in (non-dev) owner keeps the stale-state refusal and never reaches busRevive', async t => {
const { Readable } = await import('node:stream');
const tr = trailIn(t);
const f = fakes();
await f.service.createOrg({}, 'acme', template('acme'));
const bus = await f.admin.bus();
bus.busStatus = async () => { throw cliError('stale bus host lock'); };
bus.busRevive = async () => assert.fail('busRevive reached from a signed-in session');
const call = async session => {
const req = Object.assign(Readable.from([Buffer.from('{}')]), { method: 'POST', headersDistinct: { origin: ['http://console.example'], 'sec-fetch-site': ['same-origin'], 'content-type': ['application/json'] } });
const res = { writeHead(status) { this.status = status; }, end(body) { this.body = JSON.parse(body); } };
await orgAdminHTTP(req, res, { path: '/api/orgs/acme/restart/bus', admin: f.admin, session, owner: true, send: true, origin: 'http://console.example', trail: tr.trail });
return res;
};
// An SSO owner, and a session that names the local-dev provider under another issuer.
for (const session of [{ provider: 'fixture', iss: 'https://issuer.example', sub: 'owner', name: 'jason', rootOwner: true }, { provider: 'local-dev', iss: 'https://issuer.example', sub: 'local-dev-owner', rootOwner: true }]) {
const r = await call(session);
assert.equal(r.status, 409); assert.equal(r.body.error, 'bus-refused');
assert.equal(r.body.message, 'stale bus host lock. Inspect it with mosaic bus status acme; mosaic bus recover acme clears a host that died.');
}
assert.deepEqual(tr.lines().map(e => [e.outcome, e.running, e.recover]), [['refused', null, undefined], ['refused', null, undefined]]);
});
test('D2: a press during the Console start-up revive waits for it and acts on the bus it brought back', async t => {
const { reviveBuses } = await import('../src/bus-revive.mjs');
const tr = trailIn(t);
const c = await console_(t, { trail: tr.trail });
await c.call('POST', '/api/orgs', { id: 'acme', doc: browserDoc('acme') });
const bus = await c.admin.bus();
let up = false, release;
const gate = new Promise(resolve => { release = resolve; });
bus.busStatus = async () => { if (!up) throw cliError('stale bus host lock'); return { org: 'acme', healthy: true, pid: 6161 }; };
bus.busRevive = async () => assert.fail('the press revived a bus the Console was already bringing back');
bus.busRestart = async (ctx, id) => { c.calls.push(['busRestart', id]); return { org: id, pid: 6262 }; };
const startup = reviveBuses({ ctx: {}, env: {}, listOrgs: async () => [{ id: 'acme', state: 'active' }], busStatus: bus.busStatus,
busRevive: async () => { await gate; up = true; return { pid: 6161 }; }, trail: () => {}, log: () => {} });
await new Promise(resolve => setImmediate(resolve));
const press = c.call('POST', '/api/orgs/acme/restart/bus', {});
await new Promise(resolve => setTimeout(resolve, 50));
assert.equal(c.calls.filter(x => x[0] === 'busRestart').length, 0);
release();
await startup;
const r = await press;
assert.equal(r.status, 200, r.raw); assert.equal(r.body.restarted, true);
assert.deepEqual(c.calls.filter(x => x[0] === 'busRestart'), [['busRestart', 'acme']]);
});