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
jason.woltje b1fc96f4ae fix(routes): name Disable only when it makes the save apply (P0-4 review)
The catalogue refusals no longer tell every user to press Disable model
proxy catalogue. The server text names the proxy setup step. The Routing
page adds the Disable step only when cliproxy is listed and every moved
model comes from another listed provider, and sends a tester to the
owner. The switch mismatch message no longer claims a setting changed.

Tests: a fallback thinking-only change saves with no catalogue; a move
between model-proxy models refuses before and after cliproxy is removed.
2026-11-11 13:35:56 -06:00
jason.woltje 967025f749 fix(routes): fixture routes, thinking and limits save without a model catalogue (P0-4)
A fixture-only organization reads a ready catalogue with no model proxy
client: the catalogue asks the proxy only when a module calls it.
A thinking-only change, and a limits patch that keeps harness, model and
entry, no longer asks the catalogue. A change that truly needs it still
refuses, and the WebUI names the next step: keep the model and change
thinking, or press Disable model proxy catalogue. The Routing switch
follows whether cliproxy is listed and removes or adds only cliproxy.
2026-11-11 13:35:55 -06:00
27 changed files with 1004 additions and 48 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();
+5 -2
View File
@@ -22,9 +22,12 @@ export function createModelCatalogue({ client, settings, pluginDataRoot = null,
const loader = makeLoader({ log: () => {}, dataRoot:pluginDataRoot });
const result = await loader.load({ 'model-provider': names });
if (result.refused.length || loader.snapshot('model-provider').length !== names.length) throw Error();
const c = client(); if (!c) throw Error();
// A module that needs no model proxy (fixture) answers without one; a module
// that asks the proxy while none is configured gets a refusal and reads unavailable.
const c = client();
const proxy = method => async (...args) => { if (!c) throw Error('model proxy not configured'); return c[method](...args); };
const answers = await Promise.all(loader.snapshot('model-provider').map(async ({ name, module }) => {
const answer = await module.catalogue({ listAccounts: () => c.getCredentials(), listModels: name => c.getCredentialModels(name), listGroups: () => c.getUpstreamGroups(), now: () => new Date(now()).toISOString() });
const answer = await module.catalogue({ listAccounts: proxy('getCredentials'), listModels: proxy('getCredentialModels'), listGroups: proxy('getUpstreamGroups'), now: () => new Date(now()).toISOString() });
if (answer?.status === 'unavailable' && answer.reason === 'model-key-collision') throw Object.assign(Error(), { code: 'model-key-collision' });
if (answer?.status !== 'ready' || !Array.isArray(answer.models) || answer.models.length > 10000 || !Array.isArray(answer.notices) || answer.notices.length > 100) throw Error();
for (const m of answer.models) {
+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');}
});
@@ -0,0 +1,143 @@
// D3 (P0-4, readiness-1111): an organization on the fixture model provider, with no model proxy
// configured, changes routes and thinking, takes limits.set and starts a hire. The real catalogue
// service runs here (no modelCatalogue stand-in), so a fixture-only organization reads ready
// without a proxy client. A thinking-only change saves while the catalogue is unavailable, a
// change that needs the proxy still refuses, and removing cliproxy (the Routing page's Disable)
// makes the next save apply.
import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdirSync, mkdtempSync, readFileSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { busOrganization } from '../src/organization-context.mjs';
import { startBroker } from '../src/runtime.mjs';
import { createModelCatalogue } from '../src/model-catalogue.mjs';
import { loadBusiness } from '../../organization/src/organization-config.mjs';
import { resolveInstance } from '../../organization/src/resolve.mjs';
import { businessDoc, rolesCopy, systemFor, writeJson } from '../../organization/tests/helpers.mjs';
import { FIXTURE_ENTRY, FIXTURE_KEY_NAME } from '../../plugins/model-provider/fixture/index.mjs';
import { approveTemplateHire } from './hire-fixtures.mjs';
function busForm(root, rolesDir, id = 'demo') {
const b = loadBusiness(id, { dir: root, rolesDir });
const resolved = {};
for (const instance of Object.keys(b.roles)) resolved[instance] = resolveInstance({ system: systemFor(root), business: b, instance });
return busOrganization(b, resolved);
}
async function setup(t, providers = ['fixture'], edit = () => {}) {
const root = mkdtempSync(join(tmpdir(), 'd3-fixture-'));
t.after(() => rmSync(root, { recursive: true, force: true }));
const rolesDir = rolesCopy(root);
const doc = businessDoc(root, 'demo');
doc.plugins = { harness: ['pi'], 'model-provider': providers };
doc.launch.models[FIXTURE_KEY_NAME] = { ...FIXTURE_ENTRY };
doc.roles.coder.vars = { harness: 'pi', model: FIXTURE_KEY_NAME, thinking: 'off', 'limits.network': 'none' };
// The audit's instance: the assistant on the fixture route.
doc.roles.assistant = { definition: 'assistant', vars: { harness: 'pi', model: FIXTURE_KEY_NAME, thinking: 'off' } };
edit(doc);
const file = writeJson(join(root, 'businesses/demo.json'), doc);
mkdirSync(join(root, 'data'), { mode: 0o700 });
// No proxy option: the runtime has no model proxy client and no proxy lock.
const bus = await startBroker({ dataRoot: join(root, 'data'), businesses: { demo: busForm(root, rolesDir) }, organization: { dir: root, rolesDir, system: systemFor(root) } });
t.after(() => bus.close());
const human = bus.broker.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
const ask = (verb, args) => Promise.resolve().then(() => bus.broker.request(human, { verb, args }));
const saved = () => JSON.parse(readFileSync(file, 'utf8'));
return { root, rolesDir, file, bus, ask, saved };
}
const route = (thinking, model = FIXTURE_KEY_NAME) => ({ harness: 'pi', model, thinking, fallback: null });
// readiness-1111 evidence patch-low.json: `mosaic limits set --file patch-low.json`.
const patch = (revision, thinking) => ({ revision, assistants: { assistant: { definition: 'assistant', vars: { harness: 'pi', model: FIXTURE_KEY_NAME, thinking }, tracker: null } } });
test('D3 a fixture-only organization with no model proxy reads a ready catalogue and saves a thinking change and a fixture route', async t => {
const f = await setup(t);
const view = await f.ask('routes');
assert.equal(view.catalogue.status, 'ready');
assert.deepEqual(view.catalogue.models.map(m => m.key), [FIXTURE_KEY_NAME]);
const thinking = await f.ask('routes.set', { revision: view.revision, routes: { coder: route('low') } });
assert.ok(thinking.transaction);
assert.equal(f.saved().roles.coder.vars.thinking, 'low');
// A fixture route change: the reviewer moves onto the fixture model.
const moved = await f.ask('routes.set', { revision: thinking.revision, routes: { reviewer: route('off') } });
assert.ok(moved.transaction);
assert.equal(f.saved().roles.reviewer.vars.model, FIXTURE_KEY_NAME);
});
test('D3 limits.set takes the audit thinking patch on a fixture-only organization with no model proxy', async t => {
const f = await setup(t);
const { revision } = await f.ask('limits');
const result = await f.ask('limits.set', patch(revision, 'low'));
assert.ok(result.transaction);
assert.equal(f.saved().roles.assistant.vars.thinking, 'low');
});
test('D3 hire start on a fixture-only organization with no model proxy writes the hire', async t => {
const f = await setup(t);
const hired = await approveTemplateHire(f.bus.broker, f.rolesDir, { name: 'sec-1', definition: 'chief-information-security-officer', model: FIXTURE_KEY_NAME });
assert.ok(hired.transaction);
assert.equal(f.saved().roles['sec-1'].vars.model, FIXTURE_KEY_NAME);
});
test('D3 with cliproxy enabled and no model proxy, a thinking change saves, a proxy model refuses, and removing cliproxy lets a route change save', async t => {
const f = await setup(t, ['fixture', 'cliproxy']);
let view = await f.ask('routes');
assert.equal(view.catalogue.status, 'unavailable');
// Thinking only: same harness, model key and entry, so no catalogue is asked.
const thinking = await f.ask('routes.set', { revision: view.revision, routes: { coder: route('low') } });
assert.equal(f.saved().roles.coder.vars.thinking, 'low');
const limits = await f.ask('limits');
await f.ask('limits.set', patch(limits.revision, 'medium'));
assert.equal(f.saved().roles.assistant.vars.thinking, 'medium');
// A route change truly needs the catalogue, which the absent proxy makes unavailable.
const before = readFileSync(f.file, 'utf8');
view = await f.ask('routes');
await assert.rejects(f.ask('routes.set', { revision: view.revision, routes: { reviewer: route('off') } }), { code: 'model-catalogue-unavailable' });
await assert.rejects(f.ask('routes.set', { revision: view.revision, routes: { reviewer: route('off', 'zai-not-offered') } }), { code: 'model-catalogue-unavailable' });
assert.equal(readFileSync(f.file, 'utf8'), before);
assert.ok(thinking.transaction);
// Disable model proxy catalogue: plugins.set without cliproxy.
const settings = await f.ask('plugins');
await f.ask('plugins.set', { revision: settings.revision, plugins: { 'model-provider': ['fixture'] } });
view = await f.ask('routes');
assert.equal(view.catalogue.status, 'ready');
await f.ask('routes.set', { revision: view.revision, routes: { reviewer: route('off') } });
assert.equal(f.saved().roles.reviewer.vars.model, FIXTURE_KEY_NAME);
});
test('D3 the catalogue asks a module that needs no proxy even with no proxy client, and a module that asks the proxy reads unavailable', async () => {
let names = ['fixture'];
const service = createModelCatalogue({ client: () => null, settings: () => ({ 'model-provider': names }) });
const ready = await service.read('demo', { fresh: true });
assert.equal(ready.status, 'ready');
assert.deepEqual(ready.models.map(m => [m.key, m.available]), [[FIXTURE_KEY_NAME, true]]);
names = ['cliproxy'];
assert.equal((await service.read('demo', { fresh: true })).status, 'unavailable');
names = ['fixture', 'cliproxy'];
const both = await service.read('demo', { fresh: true });
assert.equal(both.status, 'unavailable');
assert.ok(both.models.every(m => m.available === false));
});
// D3 review (finding 1): Disable can't make a move between two model-proxy models apply, which is
// why the Routing page names Disable only for a move onto a model from another listed module.
test('D3 a move between model-proxy models refuses before and after cliproxy is removed, and writes nothing', async t => {
const proxyEntry = model => ({ provider: 'proxy', model, module: 'cliproxy', upstream: 'claude' });
const f = await setup(t, ['fixture', 'cliproxy'], doc => {
doc.launch.models['claude-a'] = proxyEntry('claude-a');
doc.launch.models['claude-b'] = proxyEntry('claude-b');
doc.roles.reviewer.vars = { ...doc.roles.reviewer.vars, harness: 'pi', model: 'claude-a', thinking: 'off' };
});
const before = readFileSync(f.file, 'utf8');
let view = await f.ask('routes');
await assert.rejects(f.ask('routes.set', { revision: view.revision, routes: { reviewer: route('off', 'claude-b') } }), { code: 'model-catalogue-unavailable' });
assert.equal(readFileSync(f.file, 'utf8'), before);
const settings = await f.ask('plugins');
await f.ask('plugins.set', { revision: settings.revision, plugins: { 'model-provider': ['fixture'] } });
const disabled = readFileSync(f.file, 'utf8');
view = await f.ask('routes');
assert.equal(view.catalogue.status, 'ready');
await assert.rejects(f.ask('routes.set', { revision: view.revision, routes: { reviewer: route('off', 'claude-b') } }), { code: 'model-provider-unavailable' });
assert.equal(readFileSync(f.file, 'utf8'), disabled);
assert.notEqual(disabled, before);
});
+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??{});
+12 -5
View File
@@ -132,15 +132,18 @@ export function checkRoutes({ proposed, old = null, levels, catalogue = null })
}
const currentEntry = proposed.launch?.models?.[vars[model]];
const previousEntry = old?.launch?.models?.[before[model]];
const changed = [harness, model, thinking].some(field => vars[field] !== before[field]) || canonicalJson(currentEntry ?? null) !== canonicalJson(previousEntry ?? null);
if (changed && previousEntry?.module && !currentEntry?.module && vars[model] === before[model]) throw new OrganizationError('model-provider-unavailable');
if (changed && currentEntry?.module) {
// D3 (P0-4): the catalogue answers which harness a model runs on, not its thinking level, so
// a change that keeps the harness, the model key and its entry needs no catalogue; the
// harness checks below still run on the new thinking level.
const routeChanged = [harness, model].some(field => vars[field] !== before[field]) || canonicalJson(currentEntry ?? null) !== canonicalJson(previousEntry ?? null);
if (routeChanged && previousEntry?.module && !currentEntry?.module && vars[model] === before[model]) throw new OrganizationError('model-provider-unavailable');
if (routeChanged && currentEntry?.module) {
if (catalogue?.status !== 'ready') throw new OrganizationError('model-provider-unavailable');
const offered = catalogue.models?.find(m => m.key === vars[model]);
if (!offered || !offered.available || !proposed.plugins?.['model-provider']?.includes(currentEntry.module) || canonicalJson(currentEntry) !== canonicalJson({ provider: offered.provider, model: offered.model, module: offered.module, upstream: offered.upstream })) throw new OrganizationError('model-provider-unavailable');
if (!offered.harnesses.includes(vars[harness])) throw new OrganizationError('model-harness-incompatible');
}
if (changed && currentEntry && !currentEntry.module && proposed.plugins?.['model-provider']?.length) {
if (routeChanged && currentEntry && !currentEntry.module && proposed.plugins?.['model-provider']?.length) {
if (catalogue?.status !== 'ready') throw new OrganizationError('model-provider-unavailable');
const mapping = catalogue.resolveLegacy ? catalogue.resolveLegacy(currentEntry) : catalogue.legacy?.[vars[model]];
const offered = mapping && catalogue.models?.find(m => m.module === mapping.module && m.upstream === mapping.upstream && m.model === currentEntry.model);
@@ -369,6 +372,10 @@ export function launchSettings({ dir, rolesDir, system, sessions = () => [], now
for (const [choice, prefix] of [[route, ''], [fallback, 'fallback.']]) {
if (!choice) continue;
const unchanged = ROUTE_FIELDS.every(field => priorVars[prefix + field] === choice[field]);
// D3 (P0-4): a choice that keeps the harness and model key needs no catalogue, as in
// checkRoutes; only its thinking level may differ. A key the catalogue offers as a
// different model still refuses model-key-collision below on any change.
const sameRoute = ['harness', 'model'].every(field => priorVars[prefix + field] === choice[field]);
const offered = catalogue?.models?.find(m => m.key === choice.model);
const existing = next.launch.models[choice.model];
const entry = offered ? { provider: offered.provider, model: offered.model, module: offered.module, upstream: offered.upstream } : null;
@@ -377,7 +384,7 @@ export function launchSettings({ dir, rolesDir, system, sessions = () => [], now
if (!existing && !offered) throw new OrganizationError('unknown-model', `${instance} variable ${prefix}model isn't a launch.models key`);
if (existing && offered && canonicalJson(existing) !== canonicalJson(entry) && !unchanged) throw new OrganizationError('model-key-collision');
if (!existing || existing.module) {
if (unchanged) continue;
if (sameRoute) continue;
if (catalogue?.status !== 'ready') throw new OrganizationError('model-catalogue-unavailable');
if (!offered || !offered.available || !Array.isArray(next.plugins?.['model-provider']) || !next.plugins['model-provider'].includes(offered.module)) throw new OrganizationError('model-provider-unavailable');
if (!offered.harnesses.includes(choice.harness)) throw new OrganizationError('model-harness-incompatible');
+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);
});
@@ -267,3 +267,35 @@ test('a role can add a legacy model when the model-provider module is off', asyn
const saved=settings.set('acme',{revision:view.revision,models,instances:{coder:'legacy-new'}},'chief-of-staff.chief');
assert.equal(saved.instances.coder.modelKey,'legacy-new');
});
test('D3 P0-4: a thinking-only change on a provider model saves with no catalogue, and a model or harness change still refuses', async () => {
const f=fixture();const {settings}=await routeSettings(f);const doc=JSON.parse(readFileSync(f.file,'utf8'));
doc.plugins['model-provider']=['fixture','cliproxy'];doc.launch.models['fixture-deterministic']={provider:'proxy',model:'fixture-deterministic',module:'fixture',upstream:'fixture'};
doc.roles.coder.vars={...doc.roles.coder.vars,harness:'pi',model:'fixture-deterministic',thinking:'off'};writeJson(f.file,doc);
const pick=(thinking,extra={})=>({coder:{harness:'pi',model:'fixture-deterministic',thinking,fallback:null,...extra}});
for(const catalogue of [null,{status:'unavailable',models:[],notices:[]}]){
const view=settings.routes('acme'),next=view.instances.coder.thinking==='low'?'high':'low';
settings.setRoutes('acme',{revision:view.revision,routes:pick(next)},'jason',catalogue);
assert.equal(JSON.parse(readFileSync(f.file,'utf8')).roles.coder.vars.thinking,next);
}
// A level the harness doesn't apply still refuses on a thinking-only change.
assert.throws(()=>settings.setRoutes('acme',{revision:settings.routes('acme').revision,routes:pick('medium')},'jason',null),code('thinking-unsupported'));
// A harness or model change truly needs the catalogue.
const rev=settings.routes('acme').revision,before=readFileSync(f.file,'utf8');
assert.throws(()=>settings.setRoutes('acme',{revision:rev,routes:pick('low',{harness:'claude-code'})},'jason',{status:'unavailable',models:[],notices:[]}),code('model-catalogue-unavailable'));
assert.throws(()=>settings.set('acme',{revision:rev,instances:{reviewer:'fixture-deterministic'}},'jason',{status:'unavailable',models:[],notices:[]}),code('model-provider-unavailable'));
assert.equal(readFileSync(f.file,'utf8'),before);
});
test('D3 review: a fallback thinking-only change on a provider model saves with no catalogue, and a fallback harness change still refuses', async () => {
const f=fixture();const {settings}=await routeSettings(f);const doc=JSON.parse(readFileSync(f.file,'utf8'));
doc.plugins['model-provider']=['fixture','cliproxy'];doc.launch.models['fixture-deterministic']={provider:'proxy',model:'fixture-deterministic',module:'fixture',upstream:'fixture'};
doc.roles.coder.vars={...doc.roles.coder.vars,harness:'pi',model:'fixture-deterministic',thinking:'off','fallback.harness':'pi','fallback.model':'fixture-deterministic','fallback.thinking':'off'};writeJson(f.file,doc);
const unavailable={status:'unavailable',models:[],notices:[]};
const pick=fallback=>({coder:{harness:'pi',model:'fixture-deterministic',thinking:'off',fallback:{harness:'pi',model:'fixture-deterministic',...fallback}}});
settings.setRoutes('acme',{revision:settings.routes('acme').revision,routes:pick({thinking:'low'})},'jason',unavailable);
assert.equal(JSON.parse(readFileSync(f.file,'utf8')).roles.coder.vars['fallback.thinking'],'low');
const before=readFileSync(f.file,'utf8');
assert.throws(()=>settings.setRoutes('acme',{revision:settings.routes('acme').revision,routes:pick({harness:'claude-code',thinking:'low'})},'jason',unavailable),code('model-catalogue-unavailable'));
assert.equal(readFileSync(f.file,'utf8'),before);
});
+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) });
}
+24 -4
View File
@@ -1190,8 +1190,13 @@
return select(name,key,[...new Set([...legacy,...offered.map(m=>m.key)])],c[key],(v.providerModels?.[c[key]]||enabled&&v.models.includes(c[key]))?'unavailable':'not in the model table',labels);};
const thinking=(key,h)=>select(name,key,levelsOf(c[h]),c[key],c[h]?`not applied by ${c[h]}`:'no harness');
return `<tr class="route-row${['fallback.harness','fallback.model','fallback.thinking'].some(k=>c[k])?' fallback-open':''}"><th scope="row">${txt(name)}${instanceInfo?.instances?.[name]?.definition?`<small class="route-definition">${txt(instanceInfo.instances[name].definition)}</small>`:''}</th><td data-label="Harness">${harness('harness')}</td><td data-label="Model">${model('model')}</td><td data-label="Thinking">${thinking('thinking','harness')}</td><td class="fallback-cell" data-label="Fallback harness"><button type="button" class="fallback-toggle">Set fallback</button>${harness('fallback.harness')}</td><td class="fallback-cell" data-label="Fallback model">${model('fallback.model')}</td><td class="fallback-cell" data-label="Fallback thinking">${thinking('fallback.thinking','fallback.harness')}</td></tr>`;}).join('');
const catalogueInfo=v.catalogue?.status==='ready'?`<p class="small muted">Model proxy catalogue, observed ${txt(window.MosaicLocalTime(v.catalogue.observedAt))}.</p>${(v.catalogue.notices??[]).map(n=>`<p class="small warn">${txt(n)}</p>`).join('')}`:v.catalogue?.status==='unavailable'?'<p class="small warn">Model catalogue unavailable.</p>':'<p class="small muted">Model catalogue not enabled.</p>';
const providerControl=canSend&&!isTester?`<button type="button" id="provider-switch" data-enabled="${v.catalogue?.status!=='disabled'}">${v.catalogue?.status!=='disabled'?'Disable':'Enable'} model proxy catalogue</button><p role="status" id="provider-switch-status"></p>`:'';
// D3 (P0-4): the switch adds or removes the cliproxy module only, so its label follows that
// module in the organization's model-provider list, not the whole catalogue's status (a
// fixture-only catalogue is ready with no model proxy). Before the setting is read it falls
// back to the catalogue status, as before.
const proxyOn=harnessSettings?(Array.isArray(harnessSettings.plugins?.['model-provider'])?harnessSettings.plugins['model-provider']:[]).includes('cliproxy'):v.catalogue?.status!=='disabled';
const catalogueInfo=v.catalogue?.status==='ready'?`<p class="small muted">Model catalogue, observed ${txt(window.MosaicLocalTime(v.catalogue.observedAt))}.</p>${(v.catalogue.notices??[]).map(n=>`<p class="small warn">${txt(n)}</p>`).join('')}`:v.catalogue?.status==='unavailable'?`<p class="small warn">Model catalogue unavailable. A change to thinking alone still saves.${proxyOn&&canSend&&!isTester?' A new model needs the catalogue: set up the model proxy, or, when this organization has none, press Disable model proxy catalogue.':''}</p>`:'<p class="small muted">Model catalogue not enabled.</p>';
const providerControl=canSend&&!isTester?`<button type="button" id="provider-switch" data-enabled="${proxyOn}">${proxyOn?'Disable':'Enable'} model proxy catalogue</button><p role="status" id="provider-switch-status"></p>`:'';
const harnessControl=canSend&&!isTester?`<fieldset class="harness-control"><legend>Harnesses</legend>${harnessSettings ? [...new Set([...harnessSettings.available.harness,...(harnessSettings.plugins.harness??[])])].map(name=>`<label><input type="checkbox" data-harness="${esc(name)}"${(harnessDraft??harnessSettings.plugins.harness??[]).includes(name)?' checked':''}${harnessBusy?' disabled':''}> ${txt(name)}</label>`).join('')+` <button type="button" id="harness-save"${harnessBusy?' disabled':''}>Save harnesses</button>` : '<p>Harnesses unavailable.</p><button type="button" id="harness-read">Read harnesses again</button>'}<p role="status" id="harness-status">${txt(harnessReceipt)}</p></fieldset>`:'';
const off=v.harnesses.length?'':'<p class="small warn">No harness is switched on in the business file\'s plugins.harness, so every launch refuses harness-off and a save that changes a harness refuses too. Switch one on under Harnesses.</p>';
return `<section class="panel routing-panel" aria-labelledby="routes-title"><span class="sr-only">${title}</span>${catalogueInfo}${providerControl}${harnessControl}${off}<form id="routes-form"><div class="table-wrap" tabindex="0" role="region" aria-label="Routes, scroll for all columns"><table id="routes-table"><thead><tr><th scope="col" rowspan="2">Agent</th><th scope="colgroup" colspan="3">Primary</th><th scope="colgroup" colspan="3">Fallback</th></tr><tr>${ROUTE_CELLS.map(([,l])=>`<th scope="col">${l.replace('Fallback ','')}</th>`).join('')}</tr></thead><tbody>${rows}</tbody></table></div><p class="small muted">${canSend || isTester?'Changes are checked and logged with your identity.':'Saving requires signed-in send mode.'}</p><div class="settings-actions route-actions${routesDraft?' has-draft':''}"><button type="submit"${disabled}>${routesBusy?'<span class="settings-spinner" aria-hidden="true"></span> Saving…':'Save routes'}</button> <button type="button" id="routes-reload"${routesDraft?'':' hidden'}${routesBusy||!routesDraft?' disabled':''}>Discard draft</button><p role="status" id="routes-receipt" class="${routesReceipt.startsWith('Outcome unknown')?'banner ref':routesReceipt.includes('Draft kept')?'banner err':routesReceipt.includes('saved')?'settings-success':''}">${txt(routesReceipt)}</p></div>${routesApplyMarkup()}</form></section>`;
@@ -1208,6 +1213,19 @@
return {revision:routesView.revision,routes};
}
const unknownRoutes='Outcome unknown. Read the routes again before trying. Draft kept.';
// D3 (P0-4): after a catalogue refusal, Disable is named only when it can make this save apply:
// cliproxy is listed and every model the change moves to comes from another listed module.
// A viewer without the switch is sent to the owner.
function catalogueStep(code,routes){
if(!['model-catalogue-unavailable','model-provider-unavailable'].includes(code))return '';
if(!canSend||isTester)return ' Ask the organization\'s owner to set up the model proxy, or to disable the model proxy catalogue when the change needs no proxy.';
const names=Array.isArray(harnessSettings?.plugins?.['model-provider'])?harnessSettings.plugins['model-provider']:[];
if(!names.includes('cliproxy'))return '';
const moved=Object.entries(routes).flatMap(([name,r])=>{const was=routesView?.instances?.[name]??{};
return [[r.harness,r.model,was.harness,was.model],[r.fallback?.harness,r.fallback?.model,was.fallback?.harness,was.fallback?.model]].filter(([h,m,wh,wm])=>m&&(h!==wh||m!==wm)).map(([,m])=>m);});
const module=key=>routesView?.providerModels?.[key]?.module;
return moved.length&&moved.every(key=>module(key)&&module(key)!=='cliproxy'&&names.includes(module(key)))?' This change needs no model proxy: press Disable model proxy catalogue, then save again.':'';
}
const routesAnswer=r=>/^[a-f0-9]{64}$/.test(r?.revision??'')&&Array.isArray(r.harnesses)&&Array.isArray(r.models)&&[r.levels,r.instances].every(o=>o&&typeof o==='object'&&!Array.isArray(o));
for(const kind of ['input','change'])document.addEventListener(kind,e=>{const form=e.target?.closest?.('#routes-form');if(!form||!routesView)return;routesDraft=routesFields(form);routesEditVersion++;markSettingsDraft('routing',true);form.querySelector('.settings-actions')?.classList.add('has-draft');const discard=$('routes-reload');if(discard){discard.hidden=false;discard.disabled=false;}
// A harness change redraws the row so its thinking cell offers that harness's levels.
@@ -1230,7 +1248,9 @@
const button=e.target,status=$('provider-switch-status');if(routesDraft){status.textContent='Save or discard the route draft first.';return;}button.disabled=true;
try {const read=await fetch('/api/proxy/plugins',{cache:'no-store'});const settings=await read.json();if(!read.ok||!/^[a-f0-9]{64}$/.test(settings.revision??''))throw Error('Could not read provider settings.');
const names=Array.isArray(settings.plugins?.['model-provider'])?settings.plugins['model-provider']:[];
const next=button.dataset.enabled==='true'?names.filter(n=>n!=='cliproxy'):[...new Set([...names,'cliproxy'])];
// The fresh read decides: a stale label never turns the proxy module on when it meant off.
const on=names.includes('cliproxy');if(String(on)!==button.dataset.enabled){button.dataset.enabled=String(on);button.textContent=`${on?'Disable':'Enable'} model proxy catalogue`;button.disabled=false;status.textContent='The switch now shows the current setting. Press it again to change it.';return;}
const next=on?names.filter(n=>n!=='cliproxy'):[...names,'cliproxy'];
const response=await fetch('/api/proxy/plugins/set',{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify({revision:settings.revision,plugins:{'model-provider':next}}),cache:'no-store',redirect:'manual'});
if(!response.ok)throw Error('Provider setting was refused.');harnessSettings=null;routesView=null;last.delete('routes');render(true);
}catch(error){status.textContent=error.message;button.disabled=false;}
@@ -1243,7 +1263,7 @@
routesBusy=true;routesReceipt='';routesApply=null;render(false);
try{const response=await fetch('/api/routes',{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify(payload),cache:'no-store',redirect:'manual'});if(response.type==='opaqueredirect'){routesReceipt='Sign in again. Draft kept.';signOut();}else{const result=await response.json();
if(response.ok&&routesAnswer(result)){routesDraft=null;routesView=result;harnessSettings=null;const prior=last.get('routes');if(prior)last.set('routes',{...prior,rows:[result]});routesReceipt=`Routes saved for ${Object.keys(payload.routes).join(', ')}. The next launch of each uses the new route.`;routesApply=await routesApplyRead(Object.keys(payload.routes));}
else routesReceipt=response.ok||result?.sent==='unknown'?unknownRoutes:`${result?.message||'Refused.'} Draft kept.`;}}
else routesReceipt=response.ok||result?.sent==='unknown'?unknownRoutes:`${result?.message||'Refused.'}${catalogueStep(result?.error,payload.routes)} Draft kept.`;}}
catch{routesReceipt=unknownRoutes;}finally{routesBusy=false;if(!out())render(false);}
});
// Row 194 (#1692, D7): the Agents section lives in agents.js (window.MosaicAgents) and draws from
+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.
+7
View File
@@ -583,6 +583,13 @@ const ROUTE_REASONS = {
'organization-unknown-instance': ['400', 'the business has no such instance'],
'organization-no-change': ['400', 'the business file already says that'],
'revision-conflict': ['409', 'the business file changed since you read it. Read the routes again'],
// D3 (P0-4): the model refusals name the next step. Status stays the bus status these codes
// already had (502). The Routing page adds Disable model proxy catalogue only when that switch
// can make the save apply (bus.js catalogueStep).
'model-catalogue-unavailable': ['502', 'the model catalogue is unavailable, so a model it supplies can\'t be checked. A change to thinking alone still saves; a model from the model proxy needs the proxy set up first'],
'model-provider-unavailable': ['502', 'the model provider doesn\'t offer that model now. A change to thinking alone still saves; otherwise pick a model the catalogue lists, or set up the model proxy when the model comes from it'],
'model-harness-incompatible': ['502', 'that model doesn\'t run on that harness. Pick a harness the model lists, or another model'],
'model-key-collision': ['502', 'that model key already names a different model in the model table. Pick the table\'s entry or another key'],
};
const routeObject = o => o && typeof o === 'object' && !Array.isArray(o);
const routesAnswer = v => /^[a-f0-9]{64}$/.test(v?.revision ?? '') && Array.isArray(v.harnesses) && Array.isArray(v.models) && routeObject(v.levels) && routeObject(v.instances);
+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']]);
});
@@ -264,3 +264,76 @@ test('quick fallback edits survive the harness redraw', {timeout:60000}, async t
await b.evaluate('new Promise(r=>setTimeout(r,500))');
assert.deepEqual(await b.evaluate(`[${cell('fallback.harness')}.value,${cell('fallback.model')}.value,${cell('fallback.thinking')}.value]`),['claude-code','second','high']);
});
// D3 (P0-4): an organization on the fixture provider with cliproxy listed and no model proxy reads
// an unavailable catalogue. The page says a thinking change still saves, the switch reads Disable
// from the cliproxy module itself, and pressing it removes cliproxy only. A fixture-only
// organization's catalogue is ready with no proxy, and its switch reads Enable.
test('D3 Routing: with cliproxy listed and no model proxy the switch reads Disable and removes cliproxy only; a fixture-only org reads Enable', {timeout:90000}, async t => {
for (const [providers, label, line, next] of [
[['fixture','cliproxy'], 'Disable', /Model catalogue unavailable\. A change to thinking alone still saves\. A new model needs the catalogue: set up the model proxy, or, when this organization has none, press Disable model proxy catalogue\./, ['fixture']],
[['fixture'], 'Enable', /^Model catalogue, observed /, ['fixture','cliproxy']],
]) {
const root=tmp(t,'d3-switch-'), f=fixture(root,'acme',doc=>{doc.plugins['model-provider']=providers;return doc;});
const dir=join(root,'config'),file=join(dir,'businesses','acme.json');
// No modelCatalogue stand-in and no proxy: the broker's own catalogue service answers.
const host=await startBroker({dataRoot:f.dataRoot,businesses:{acme:{id:'acme',human:'jason',arbiters:{technical:'chief',delivery:'chief'},roles:{chief: { definition: 'chief-of-staff', authority: { withinRole: ['decision.resolve.technical', 'decision.resolve.delivery'], crossRole: [] } }, coder:{authority:{withinRole:[],crossRole:[]}}}}},organization:{dir,rolesDir:f.roles,system:systemFor(root)}});t.after(()=>host.close());
const human=host.broker.bindHuman({business:'acme',human:'jason',via:'cli',outsideAgent:true});
const call=async(verb,args)=>structuredClone(await host.broker.request(human,{verb,args}));
const bus={messages:async()=>({messages:[],unread:0}),usage:async()=>{throw Error('no usage');},routes:()=>call('routes')};
const web=await startServer({port:0,board:'http://127.0.0.1:9',bus,reads:{settings:()=>({business:'acme',release:'fixture',notifier:{}})},log:()=>{}});t.after(()=>close(web));
const b=await browser();t.after(()=>b.close());
const settings=await call('plugins');
await b.call('Page.addScriptToEvaluateOnNewDocument',{source:`const original=window.fetch;window.posts=[];window.fetch=async(url,options={})=>{if(url==='/api/config')return new Response(JSON.stringify({send:true}));if(url==='/api/proxy/plugins')return new Response(JSON.stringify(${JSON.stringify(settings)}));if(url==='/api/proxy/plugins/set'){posts.push(JSON.parse(options.body));return new Promise(r=>window.answer=(status,body)=>r(new Response(JSON.stringify(body),{status})));}return original(url,options);};`});
await b.navigate(`http://127.0.0.1:${web.address().port}/#/settings/routing`);
const wait=expr=>b.evaluate(`(async()=>{for(let n=0;n<200;n++){if(${expr})return true;await new Promise(r=>setTimeout(r,30));}throw Error('wait failed');})()`);
await wait(`document.querySelector("#provider-switch")?.textContent.startsWith(${JSON.stringify(label)})&&!!document.querySelector('[data-harness]')`);
const info=await b.evaluate('document.querySelector(".routing-panel > p.small")?.textContent');
assert.match(info,line,providers.join(','));
await b.evaluate('document.querySelector("#provider-switch").click();true');
await wait('posts.length===1');const payload=await b.evaluate('posts[0]');
assert.deepEqual(payload.plugins,{'model-provider':next},providers.join(','));
const result=await call('plugins.set',payload);await b.evaluate(`window.answer(200,${JSON.stringify(result)});true`);
assert.deepEqual(JSON.parse(readFileSync(file,'utf8')).plugins['model-provider'],next);
await b.close();
}
});
// D3 review (finding 1, 3): after a catalogue refusal the receipt names Disable model proxy
// catalogue only when pressing it can make the save apply: cliproxy is listed and the model the
// change moves to comes from another listed module. A model from the proxy, or a list without
// cliproxy (Disable already pressed), gets no Disable step; a tester is sent to the owner.
test('D3 Routing: a catalogue refusal names Disable only when it can make the save apply, and sends a tester to the owner', {timeout:90000}, async t => {
const refusal=code=>({error:code,message:`${code}: the reason. Nothing changed.`,sent:false});
for (const [providers, tester, instance, code, expect] of [
[['fixture','cliproxy'], false, 'coder', 'model-catalogue-unavailable', /Nothing changed\. This change needs no model proxy: press Disable model proxy catalogue, then save again\. Draft kept\.$/],
[['fixture','cliproxy'], false, 'reviewer', 'model-catalogue-unavailable', /Nothing changed\. Draft kept\.$/],
[['fixture'], false, 'reviewer', 'model-provider-unavailable', /Nothing changed\. Draft kept\.$/],
[['fixture','cliproxy'], true, 'coder', 'model-catalogue-unavailable', /Nothing changed\. Ask the organization's owner to set up the model proxy, or to disable the model proxy catalogue when the change needs no proxy\. Draft kept\.$/],
]) {
const label=`${providers.join(',')} tester=${tester} ${instance}`;
const view={business:'acme',revision:'a'.repeat(64),harnesses:['pi','claude-code'],models:['fixture-deterministic','proxy-x'],
providerModels:{'fixture-deterministic':{module:'fixture',upstream:'fixture',model:'fixture-deterministic'},'proxy-x':{module:'cliproxy',upstream:'u',model:'x'}},
levels:{pi:['off','low'],'claude-code':['off','low']},
instances:{coder:{harness:'pi',model:'fixture-deterministic',thinking:'off',fallback:null},reviewer:{harness:'pi',model:'proxy-x',thinking:'off',fallback:null}},
catalogue:{status:'unavailable',models:[],notices:[]}};
const plugins={revision:'b'.repeat(64),plugins:{harness:['pi','claude-code'],'model-provider':providers},available:{harness:['pi','claude-code']}};
const web=await startServer({port:0,board:'http://127.0.0.1:9',bus:{messages:async()=>({messages:[],unread:0}),usage:async()=>{throw Error('fixture')},routes:async()=>view},reads:{settings:()=>({business:'acme',release:'fixture',notifier:{}})},log:()=>{}});t.after(()=>close(web));
const b=await browser();t.after(()=>b.close());
await b.call('Page.addScriptToEvaluateOnNewDocument',{source:`const original=window.fetch;window.posts=[];window.fetch=async(url,options={})=>{if(url==='/api/config')return new Response(JSON.stringify({send:${!tester},tester:${tester}}));if(url==='/api/proxy/plugins')return new Response(JSON.stringify(${JSON.stringify(plugins)}));if(url==='/api/routes'&&options.method==='POST'){posts.push(JSON.parse(options.body));return new Response(JSON.stringify(${JSON.stringify(refusal(code))}),{status:502});}return original(url,options);};`});
await b.navigate(`http://127.0.0.1:${web.address().port}/#/settings/routing`);
const wait=expr=>b.evaluate(`(async()=>{for(let n=0;n<200;n++){if(${expr})return true;await new Promise(r=>setTimeout(r,30));}throw Error('wait failed: '+${JSON.stringify(expr)});})()`);
const cell=key=>`document.querySelector(${JSON.stringify(`[data-route='${JSON.stringify([instance,key])}']`)})`;
await wait(`${cell('harness')}&&!document.querySelector('#routes-form button[type=submit]').disabled${tester?'':"&&!!document.querySelector('[data-harness]')"}`);
// The harness change moves the instance's route, so the bus would ask the catalogue.
await b.evaluate(`{const e=${cell('harness')};e.value='claude-code';e.dispatchEvent(new Event('change',{bubbles:true}));}true`);
await wait(`${cell('harness')}.value==='claude-code'`);
await b.evaluate(`document.querySelector('#routes-form').requestSubmit();true`);
await wait('posts.length===1&&document.querySelector("#routes-receipt").textContent.includes("Draft kept")');
assert.deepEqual(await b.evaluate('posts[0].routes'),{[instance]:{harness:'claude-code',model:view.instances[instance].model,thinking:'off',fallback:null}},label);
const text=await b.evaluate('document.querySelector("#routes-receipt").textContent');
assert.match(text,new RegExp(`^${code}: the reason\\. `),label);
assert.match(text,expect,label);
await b.close();await close(web);
}
});
+18
View File
@@ -173,3 +173,21 @@ test('row 65 Console: the routes read takes no arguments, needs a bus and refuse
const refused = await setup(t, { bus: { routes: async () => { throw new BusReadError('human-required'); } } });
assert.equal((await get(refused, '/api/bus/routes')).status, 403);
});
// The Disable step is the Routing page's to add (bus.js catalogueStep), since only the page knows
// whether pressing it can help; the server text never names it.
test('D3 P0-4: a model refusal keeps its status and names the next step that makes the save apply', async t => {
for (const [code, step] of [
['model-catalogue-unavailable', /A change to thinking alone still saves; a model from the model proxy needs the proxy set up first\. Nothing changed\.$/],
['model-provider-unavailable', /A change to thinking alone still saves; otherwise pick a model the catalogue lists, or set up the model proxy when the model comes from it\. Nothing changed\.$/],
['model-harness-incompatible', /Pick a harness the model lists, or another model/],
['model-key-collision', /Pick the table's entry or another key/],
]) {
const s = await setup(t, { write: async () => { throw new BusReadError(code); } });
const r = await send(s, '/api/routes', { revision: REV, routes: { coder: ROUTE } });
assert.deepEqual([r.status, r.body.error, r.body.sent], [502, code, false], code);
assert.match(r.body.message, new RegExp(`^${code}: .*Nothing changed\\.$`), code);
assert.match(r.body.message, step, code);
assert.doesNotMatch(r.body.message, /refused by the bus|Disable/, code);
}
});