jason.woltje
f65faf9df8
Ruling 1: the bus runtime passes launchSettings a key lineage lookup read from its own credential records (ids only). A new daily key quota takes the earliest stored since among the old document's daily key quotas in its key's lineage, and now only when there is none. A reset still sets now, a caller-supplied since still refuses, and a quota removed and added back in a later write is new. A failed or malformed lookup refuses proxy-key-refused; a write that adds no new daily key quota never calls it. Ruling 2: every limits refresh updates the remembered orphan set, an empty set when no daily key quota is left, so a quota removed and added back logs again. In memory only: a restart is a fresh observation. Ruling 3: pollSettledAfterTick yields on the real setImmediate, captured at load, inside its deadline. Tests: p5-key-lineage (writer move counts 160, reset restarts, lookup failure, orphan remove and re-add), p5-runtime (exhausted on A, moved to B through the real runtime, refuses), p5-background (barrier with every timer mocked), token-limit-settings (earliest, no match, re-add, caller since, malformed answers, default refusal). READMEs: business, plugins, daily-quota, cliproxy-queue.
969 lines
57 KiB
JavaScript
969 lines
57 KiB
JavaScript
import { lstatSync, readdirSync, readFileSync } from 'node:fs';
|
|
import { join } from 'node:path';
|
|
import { fileURLToPath } from 'node:url';
|
|
import { Store } from './store.mjs';
|
|
import * as projectModule from '../../project/src/index.mjs';
|
|
import * as libraryModule from '../../library/src/index.mjs';
|
|
import { Broker, BusError, LIBRARY_FETCH_FIX } from './broker.mjs';
|
|
import { Credentials } from './credentials.mjs';
|
|
import { serve } from './server.mjs';
|
|
import { launchSettings } from '../../business/src/launch-settings.mjs';
|
|
import { readSessions, sessionsFile, stopSession } from '../../seat/src/session.mjs';
|
|
import { compactionCount, storeDir } from '../../seat/src/compactions.mjs';
|
|
import { publishRemote, workingRepo, WorkingError } from '../../seat/src/working.mjs';
|
|
import { createLoader, hasModule } from '../../plugins/src/loader.mjs';
|
|
import { pluginSettings } from '../../business/src/plugin-settings.mjs';
|
|
import { loadManagementClient } from '../../model-proxy/src/client.mjs';
|
|
import { ProxyPage } from './proxy-page.mjs';
|
|
import { sha256 } from '../../model-proxy/src/common.mjs';
|
|
import { initializeProxy } from '../../model-proxy/src/settings.mjs';
|
|
import { ModelKeys, modelOrganization, stopModelRuns, MODEL_STOP_TIMEOUT_MS } from './model-keys.mjs';
|
|
import { instanceSettings } from '../../business/src/instance-settings.mjs';
|
|
import { organizationWriter, OrganizationError, setReplyLimits } from '../../business/src/organization.mjs';
|
|
import { credentialVerbs } from './credential-verbs.mjs';
|
|
import { loadBusiness } from '../../business/src/business.mjs';
|
|
import { loadRole } from '../../business/src/role.mjs';
|
|
import { resolveInstance } from '../../business/src/resolve.mjs';
|
|
import { canonicalJson } from '../../business/src/util.mjs';
|
|
import { busBusiness } from './business.mjs';
|
|
import { webService } from './web.mjs';
|
|
import { hostExecutor, sweepRunDir } from './host-run.mjs';
|
|
import { grantKeys, hostGrant, hostReference, sweepHostKeys } from './host-grant.mjs';
|
|
import { giteaGrant, giteaKeys, giteaStart, giteaReference } from './gitea-grant.mjs';
|
|
import { verifyPrincipal } from './principal.mjs';
|
|
import { loadAccess } from '../../business/src/access.mjs';
|
|
import { inspectInterpreter, recordInterpreterStart } from './interpreter.mjs';
|
|
import { createPeerReader } from './peercred.mjs';
|
|
import { createUsage } from './usage.mjs';
|
|
import { createProxyUsage } from './proxy-usage.mjs';
|
|
import { createLimits, modelUsageKey, modelKeyLineages, modelKeyLineage } from './limits.mjs';
|
|
import { createProcessing } from '../../processing-control/src/coordinator.mjs';
|
|
import { discordProcessingHooks } from '../../discord/src/processing.mjs';
|
|
import { createContainerStopper } from '../../processing-control/src/containers.mjs';
|
|
|
|
// Row 194 (D3): the verbs the instance writer commits, whose answers carry restart.
|
|
const INSTANCE_VERBS = new Set(['instances.set', 'veto.set', 'instance.hire', 'instance.fire']);
|
|
// FED R4 (#1801, design 3.7): the revocation timer's period.
|
|
export const ACCESS_SWEEP_MS = 30_000;
|
|
// Trusted host API. S1 supplies resolved definitions, S6 supplies launch records and ends them.
|
|
// Human limits writes use the shared organization writer behind the verified transport.
|
|
// `tasks` is S3's adapter factory: ({broker, credentials, businesses}) => {handle, timeout, close}.
|
|
// REQ-HARN-2, REQ-PLUG-2 (decision 0b81b0c5): one web service per business. Its search source is
|
|
// the "search-source" plug-in the business's organization file switches on: searxng-url (the
|
|
// business var web.searxng) or searxng-bundled. With none, web.search refuses "not configured" and
|
|
// web.fetch still works. A plug-in that refuses to load is logged and leaves search off; it never
|
|
// stops the broker. Under harness-builtin the harness serves web itself, so the broker gets no
|
|
// service and refuses the web verbs.
|
|
export async function webServices(businesses, { log = (e) => process.stderr.write(`plugins: ${JSON.stringify(e)}\n`), deps } = {}) {
|
|
const services = new Map();
|
|
for (const [name, b] of Object.entries(businesses)) {
|
|
const loader = createLoader({ log });
|
|
await loader.load(b.plugins ?? {});
|
|
const [source] = loader.modules('search-source');
|
|
// "harness-builtin" is served by the harness itself, not the broker: no broker web verbs.
|
|
if (loader.loaded()['search-source'][0] === 'harness-builtin') continue;
|
|
// The module takes the business var web.searxng; "searxng-bundled" ignores it.
|
|
const settings = { searxng: b.vars?.['web.searxng'] };
|
|
const search = source ? { search: async (query, opts) => source.search(settings, query, deps, opts) } : null;
|
|
services.set(name, webService({ search, deps }));
|
|
}
|
|
return services;
|
|
}
|
|
// Row 65 (REQ-PLUG-2): the thinking levels each harness module in this tree applies, for the route
|
|
// 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.
|
|
export async function harnessLevels({ log = (e) => process.stderr.write(`plugins: ${JSON.stringify(e)}\n`) } = {}) {
|
|
const levels = new Map();
|
|
let names = [];
|
|
try {
|
|
names = readdirSync(fileURLToPath(new URL('../../plugins/harness/', import.meta.url))).filter((name) => hasModule('harness', name));
|
|
} catch (e) {
|
|
log({ event: 'harness-levels-unread', message: e.message });
|
|
}
|
|
for (const name of names) {
|
|
const loader = createLoader({ log });
|
|
if ((await loader.load({ harness: [name] })).refused.length) continue;
|
|
try {
|
|
const list = loader.modules('harness')[0].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 });
|
|
}
|
|
}
|
|
return (name) => levels.get(name) ?? null;
|
|
}
|
|
// Row 224 C2-05b: a host-only adapter. A failed fetch keeps serving immutable old
|
|
// objects, as C2-03a requires; no caller may select a repo, commit or sync --accept.
|
|
export function createLibraryLookup({ dataRoot, sync, now = () => Date.now() }) {
|
|
const methods = { find: libraryModule.find, get: libraryModule.get, search: libraryModule.search,
|
|
manuscript: libraryModule.manuscript, universeInfo: libraryModule.universeInfo };
|
|
return async (context, { method, args, authorize }) => {
|
|
if (!Object.hasOwn(methods, method)) throw new libraryModule.LibraryError('query-argument-invalid', 'unknown lookup');
|
|
authorize();
|
|
const source = { ...context, dataRoot };
|
|
let state = await libraryModule.libraryStatus(source);
|
|
authorize();
|
|
// sage-2819 / dewey-0852: failed attempts also pace inline retries. Otherwise an
|
|
// outage makes every lookup refetch while lastFetch remains old or null.
|
|
const recentAttempt = Math.max(state.lastFetch === null ? -Infinity : Date.parse(state.lastFetch),
|
|
state.lastError === null ? -Infinity : Date.parse(state.lastError.at));
|
|
const age = now() - recentAttempt;
|
|
if (age > libraryModule.SYNC_INTERVAL_MS || age < 0) {
|
|
try { await sync(source); }
|
|
catch (error) {
|
|
// A timer may already be fetching this mirror. Its current served state is still safe.
|
|
if (!(error instanceof libraryModule.LibraryError) || error.code !== 'library-sync-busy') throw error;
|
|
}
|
|
authorize();
|
|
state = await libraryModule.libraryStatus(source);
|
|
}
|
|
if (!state.served) throw new libraryModule.LibraryError('library-commit', 'library has never been synced');
|
|
authorize();
|
|
return methods[method]({ repo: libraryModule.mirrorPath(source), commit: state.served,
|
|
mapping: libraryModule.checkMapping(context.business.projects[context.project]) }, args);
|
|
};
|
|
}
|
|
|
|
// B1: discovery is evidence of a process, never authority to stop it. Keep this
|
|
// host-only inventory across the launcher's Registry[] recovery gap and endLaunch.
|
|
// A failed boot read cannot be repaired by a later empty registry: that would lose
|
|
// the very prior runs whose existence this inventory must establish.
|
|
function modelRunInventory(dataRoot, bound, credentials) {
|
|
const known = new Map(), terminal = new Map(), validated = new Map();
|
|
const refuse = () => { throw new BusError('proxy-stop-failed'); };
|
|
const same = (a, b) => ['business', 'role', 'run', 'pid', 'startTime', 'runnerPid', 'runnerStartTime', 'keyId'].every(k => a[k] === b[k]);
|
|
const shape = s => {
|
|
if (!s || ['business', 'instance', 'run'].some(k => typeof s[k] !== 'string' || !s[k])
|
|
|| ['pid', 'runnerPid'].some(k => !Number.isSafeInteger(s[k]) || s[k] < 2)
|
|
|| ['startTime', 'runnerStartTime'].some(k => typeof s[k] !== 'string' || !/^\d+$/.test(s[k]))
|
|
|| (s.modelKeyId !== undefined && (typeof s.modelKeyId !== 'string' || !/^[a-f0-9]{12}$/.test(s.modelKeyId)))) refuse();
|
|
return { business: s.business, role: s.instance, instance: s.business + '/' + s.instance,
|
|
run: s.run, pid: s.pid, startTime: s.startTime, runnerPid: s.runnerPid,
|
|
runnerStartTime: s.runnerStartTime, keyId: s.modelKeyId };
|
|
};
|
|
const refresh = () => {
|
|
const file = sessionsFile(dataRoot), stat = lstatSync(file);
|
|
if (!stat.isFile() || stat.isSymbolicLink()) refuse();
|
|
const entries = readSessions(dataRoot).map(shape), seen = new Set();
|
|
const after = lstatSync(file);
|
|
if (!['dev', 'ino', 'size', 'mtimeMs', 'ctimeMs'].every(k => stat[k] === after[k])) refuse();
|
|
// Validate the whole snapshot before merging any of it. Missing entries do
|
|
// not erase known identities; the host may have just rewritten Registry[].
|
|
for (const r of entries) {
|
|
const key = r.business + ':' + r.run;
|
|
if (seen.has(key) || (known.has(key) && !same(known.get(key), r))) refuse();
|
|
seen.add(key);
|
|
}
|
|
for (const r of entries) known.set(r.business + ':' + r.run, r);
|
|
};
|
|
let bootChecked = false;
|
|
try { refresh(); bootChecked = true; } catch { /* hold retirement, not the human/control surface */ }
|
|
// startTimeOf/live alone conflate absence and read errors. Only a missing proc
|
|
// directory, or a zombie with the SAME recorded start ticks, confirms an exit.
|
|
const liveIdentity = (pid, ticks, proof, part) => {
|
|
if (proof[part]) return false;
|
|
let raw;
|
|
try { raw = readFileSync(`/proc/${pid}/stat`, 'utf8'); }
|
|
catch (error) {
|
|
if (error.code !== 'ENOENT') refuse();
|
|
try { lstatSync(`/proc/${pid}`); } catch (missing) { if (missing.code === 'ENOENT') { proof[part] = true; return false; } }
|
|
refuse();
|
|
}
|
|
const fields = raw.slice(raw.lastIndexOf(')') + 2).trim().split(/\s+/);
|
|
if (!/^[A-Z]$/.test(fields[0]) || !/^\d+$/.test(fields[19]) || fields[19] !== ticks) refuse();
|
|
if (fields[0] === 'Z') { proof[part] = true; return false; }
|
|
return true;
|
|
};
|
|
const runs = instance => {
|
|
try {
|
|
if (!bootChecked) refuse();
|
|
refresh();
|
|
const live = [], checked = [];
|
|
for (const r of known.values()) {
|
|
if (r.instance !== instance) continue;
|
|
const key = r.business + ':' + r.run;
|
|
const proof = terminal.get(key) ?? { outer: false, runner: false };
|
|
terminal.set(key, proof);
|
|
let associated = false;
|
|
if (!proof.outer || !proof.runner) {
|
|
const record = credentials.modelRecord(instance);
|
|
associated = !!record && [record.keyId, ...record.retiring.map(k => k.keyId)].includes(r.keyId);
|
|
const prior = validated.get(key);
|
|
if (!associated && (!prior || !same(prior, r))) refuse();
|
|
}
|
|
const outerLive = liveIdentity(r.pid, r.startTime, proof, 'outer'), runnerLive = liveIdentity(r.runnerPid, r.runnerStartTime, proof, 'runner');
|
|
// Never interpret an unknown/wrong key on a surviving prior run as no
|
|
// old-key users. A confirmed exit may be retired without key guessing.
|
|
if (outerLive || runnerLive) {
|
|
// Provenance permits strict exit confirmation, NEVER filtering away a
|
|
// surviving run whose model-key association has disappeared.
|
|
if (!associated) refuse();
|
|
const current = bound.get(r.business + ':' + r.run);
|
|
if (current && (current.instance !== r.instance || current.keyId !== r.keyId
|
|
|| current.pid !== r.pid || current.startTime !== r.startTime)) refuse();
|
|
live.push(r);
|
|
}
|
|
if (associated && (outerLive || runnerLive)) checked.push(r);
|
|
}
|
|
// A newly bound launch whose registry publication has not happened is
|
|
// uncertain, not absent. Do not grant retirement through that startup gap.
|
|
for (const r of bound.values()) if (r.instance === instance && !known.has(r.business + ':' + r.run)) refuse();
|
|
// Commit ONLY after the entire same-instance view passed both identities,
|
|
// live binding consistency, later known entries and unpublished bindings.
|
|
// Existing M1 terminal bits remain independently monotonic and unchanged.
|
|
for (const r of checked) validated.set(r.business + ':' + r.run, { ...r });
|
|
return live;
|
|
} catch { refuse(); }
|
|
};
|
|
return {
|
|
runs,
|
|
authorizeStop(requests) {
|
|
for (const r of requests) {
|
|
const current = bound.get(r.business + ':' + r.run);
|
|
if (!current || current.instance !== r.instance || current.keyId !== r.keyId
|
|
|| current.pid !== r.pid || current.startTime !== r.startTime) refuse();
|
|
if (!runs(r.instance).some(found => same(found, r))) refuse();
|
|
}
|
|
},
|
|
};
|
|
}
|
|
|
|
export async function startBroker({
|
|
dataRoot,
|
|
businesses,
|
|
launches = [],
|
|
readers = [],
|
|
repoRoots = [],
|
|
tasks = null,
|
|
organization = null,
|
|
// Row 199: the tests' scratch CA for publish's git (`-c http.sslCAInfo=...`). process.mjs drops
|
|
// it from the boot config, so only an in-process caller sets it.
|
|
publishTls = [],
|
|
// C2-03a: the tests' stand-in for syncCommand, a shorter timer and a scratch config dir for the
|
|
// slot, { command, intervalMs, configDir }. A boot config is JSON and can't carry a function, so
|
|
// a librarySync without one is refused.
|
|
librarySync = null,
|
|
// FED R3 (#414): the access file and its loader context, { file, context } (R1's loadAccess).
|
|
// Without it no user binds. R6 and R8 pass it; the live boot config doesn't yet. FED R4 (#1801):
|
|
// intervalMs, the revocation timer's period (design 3.7, 30 s), in-process only like librarySync's.
|
|
access = null,
|
|
// Row 252 (#1753, design section 2): the checked console.sso subjects, [{provider, iss, sub}], each an
|
|
// owner of the business (Webby N1). The boot config carries them; no socket verb sets them. Empty, no
|
|
// Console owner binds and the terminal owner is unchanged.
|
|
consoleOwners = [],
|
|
// P1 validated sidecar from the trusted host; never a socket request setting.
|
|
proxy = null,
|
|
}) {
|
|
if (!Array.isArray(consoleOwners) || consoleOwners.length > 512 || consoleOwners.some((o) => o === null || typeof o !== 'object' || Array.isArray(o) ||
|
|
Object.keys(o).length !== 3 || ['provider', 'iss', 'sub'].some((k) => typeof o[k] !== 'string' || !o[k] || o[k].length > 512)))
|
|
throw new BusError('invalid-host-request');
|
|
const owners = Object.freeze(consoleOwners.map((o) => Object.freeze({ provider: o.provider, iss: o.iss, sub: o.sub })));
|
|
if (access !== null && (typeof access?.file !== 'string' || access.context === null || typeof access.context !== 'object'
|
|
|| (Object.hasOwn(access, 'intervalMs') && !(Number.isSafeInteger(access.intervalMs) && access.intervalMs > 0))))
|
|
throw new BusError('invalid-host-request');
|
|
if (librarySync !== null && (typeof librarySync?.command !== 'function'
|
|
|| (Object.hasOwn(librarySync, 'intervalMs') && !(Number.isSafeInteger(librarySync.intervalMs) && librarySync.intervalMs > 0))
|
|
|| (Object.hasOwn(librarySync, 'configDir') && typeof librarySync.configDir !== 'string')))
|
|
throw new BusError('invalid-host-request');
|
|
let modelProxyClient = null, proxyPage = null, proxyUsage = null, usageTimer = null;
|
|
let modelKeys = null, modelTimer = null, modelSweep = null;
|
|
const modelRuns = new Map();
|
|
let store, credentials, server, adapter, processing, syncTimer = null, syncPass = null, accessTimer = null;
|
|
try {
|
|
const proxyReceipt = entry => process.stderr.write(`model-proxy: ${JSON.stringify(entry)}\n`);
|
|
const references = {};
|
|
for (const [business, b] of Object.entries(businesses)) {
|
|
for (const [role, r] of Object.entries(b.roles))
|
|
if (r.credentials) references[business + '/' + role] = r.credentials;
|
|
if (b.tracker?.sync?.credentials) references[business + '/@sync'] = b.tracker.sync.credentials;
|
|
}
|
|
const protectedRoots = [
|
|
fileURLToPath(new URL('../../../', import.meta.url)),
|
|
...repoRoots,
|
|
...Object.values(businesses).flatMap((b) => Object.values(b.projects ?? {}).filter(p => p.kind !== 'library').map((p) => p.root)),
|
|
];
|
|
// Row 199 (#1697, design D1): the per-agent store is <configDir>/secrets/agents.
|
|
const storeRoot = organization ? join(organization.dir, 'secrets', 'agents') : null;
|
|
store = new Store(dataRoot);
|
|
// Store's writer lock admits one bus per data root before it holds the management client.
|
|
if (proxy) {
|
|
if (!organization) throw new BusError('not-configured');
|
|
modelProxyClient = await loadManagementClient({ configDir: organization.dir, proxy, receipt: proxyReceipt });
|
|
}
|
|
|
|
// Row 234 (#1740, r6 X4 third bullet): each host grant record gives its agent a store reference to
|
|
// host.<id>, so a restart finds the keys the grant executor wrote. The loader refuses host.<id> in a
|
|
// role's credentials, so these never meet a business-file reference.
|
|
const granted = new Set();
|
|
for (const g of store.all('SELECT business, agent, host, rotate_by FROM host_grants ORDER BY business, agent, host')) {
|
|
granted.add(g.business + '/' + g.agent + '/' + g.host);
|
|
if (!Object.hasOwn(businesses, g.business)) continue;
|
|
const refs = (references[g.business + '/' + g.agent] = { ...(references[g.business + '/' + g.agent] ?? {}) });
|
|
refs['host.' + g.host] = hostReference(storeRoot, g.business, g.agent, g.host, g.rotate_by);
|
|
}
|
|
// Row 235 (r6 section 6, N1): a host.<id> key file with no record goes before Credentials loads.
|
|
for (const entry of sweepHostKeys(storeRoot, Object.keys(businesses), granted))
|
|
process.stderr.write(`hosts: ${JSON.stringify({ event: 'host-key-swept', ...entry })}\n`);
|
|
// Row 236 (#1742, #436b r5 section 5): each gitea grant record gives its agent a store reference to
|
|
// gitea, and `held` names the files memory must not load until reconcile or a revoke.
|
|
const { held } = storeRoot ? giteaStart(store, storeRoot, businesses, references) : { held: [] };
|
|
credentials = new Credentials({ references, repoRoots: protectedRoots, dataRoot, storeRoot, held,
|
|
modelAgents: proxy ? Object.entries(businesses).flatMap(([id, b]) => Object.keys(b.roles).map(role => id + '/' + role)) : [] });
|
|
if (proxy) {
|
|
const inventory = modelRunInventory(dataRoot, modelRuns, credentials);
|
|
modelKeys = new ModelKeys({ credentials, client: modelProxyClient, configDir: organization.dir,
|
|
baseUrl: `http://127.0.0.1:${proxy.port}`, receipt: proxyReceipt, rotateGraceMinutes: proxy.rotateGraceMinutes ?? 120,
|
|
runs: inventory.runs,
|
|
stopRuns: async runs => {
|
|
inventory.authorizeStop(runs);
|
|
await stopModelRuns({ runs, sessions: () => readSessions(dataRoot), stop: run => stopSession(dataRoot, run, { timeoutMs: MODEL_STOP_TIMEOUT_MS }), receipt: proxyReceipt });
|
|
for (const r of runs) {
|
|
if (inventory.runs(r.instance).some(found => found.run === r.run)) throw new BusError('proxy-stop-failed');
|
|
}
|
|
for (const r of runs) modelRuns.delete(r.business + ':' + r.run);
|
|
} });
|
|
// Keep the control/human surface available, but refuse every launch on failed reconcile.
|
|
try { await modelKeys.reconcile(); } catch (error) { proxyReceipt({ action: 'proxy.boot-refused', code: error.code }); }
|
|
}
|
|
const configureProxyPage = () => {
|
|
proxyPage = new ProxyPage({ client: modelProxyClient, instanceTag: sha256(organization.dir).slice(0, 12),
|
|
// P2 supplies the model metadata; only the matching business's record
|
|
// participates in this owner view. No credential value is read here.
|
|
keyRecords: business => credentials.status().filter(r => r.service === 'model' && r.instance?.startsWith(business + '/'))
|
|
.map(r => ({ keyId: r.keyId, agent: r.instance.slice(business.length + 1), issuedAt: r.issuedAt, state: r.state })),
|
|
});
|
|
};
|
|
if (modelProxyClient) configureProxyPage();
|
|
|
|
// vikunja:19/305: the plug-in switch through the same human-only path as the limits: the verbs
|
|
// plugins and plugins.set, logged in changes.jsonl with the actor, a missing module refused.
|
|
const plugins=organization?pluginSettings({dir:organization.dir,rolesDir:organization.rolesDir,exists:hasModule}):null;
|
|
const levels=organization?await harnessLevels():()=>null;
|
|
// Row 333 R4 (#1831): the writer's key lineage lookup, ids only, from this bus's own credential
|
|
// records. A daily key quota moved within one agent's lineage keeps its since and its count.
|
|
const keyLineage=(business,keyId)=>modelKeyLineage(credentials.status(),business,keyId);
|
|
const settings=organization?launchSettings({...organization,sessions:()=>readSessions(dataRoot),levels,keyLineage}):null;
|
|
// Row 194 (#1692, brief item 8, row 199 D13): a fire or a delete revokes the agent's credentials
|
|
// after the writer's checks and before its write, under the writer's lock (row 239). Fire revokes
|
|
// each service of the entry and keeps the store directory, delete calls remove (sage-1593). A
|
|
// service the running bus doesn't know (an agent created since the start) has nothing in memory,
|
|
// so it goes on, and then no token or tmp file may remain in the agent's store directory
|
|
// (sage-1592). Any failure stops the verb with the file unchanged.
|
|
const revokeFailed = (message) => new OrganizationError('credential-revoke-failed', message);
|
|
function revoke(business, agent, how, services) {
|
|
const instance = business + '/' + agent;
|
|
const outside = [];
|
|
try {
|
|
if (how === 'delete' && storeRoot) credentials.remove(instance);
|
|
else
|
|
for (const service of services) {
|
|
try {
|
|
if (!credentials.revoke(instance, service).file) outside.push(service);
|
|
} catch (e) {
|
|
if (e.code !== 'credential-unknown') throw e;
|
|
}
|
|
}
|
|
} catch (e) {
|
|
throw revokeFailed(`${agent}: ${e.code ?? 'credential-file'}`);
|
|
}
|
|
if (storeRoot) {
|
|
let left = [];
|
|
try {
|
|
left = readdirSync(join(storeRoot, business, agent));
|
|
} catch (e) {
|
|
if (e.code !== 'ENOENT') throw revokeFailed(`${agent}: store directory can't be read`);
|
|
}
|
|
if (left.length > 0) throw revokeFailed(`${agent}: ${left.sort().join(' and ')} still in the store`);
|
|
}
|
|
// A file or env reference outside the store keeps its token for a restart (row 199 section 6).
|
|
return { revoked: services, outsideStore: how === 'fire' ? outside : [] };
|
|
}
|
|
// Row 165 (#1662): the human adds and changes role instances through instances and instances.set.
|
|
const proxyOrganization = modelKeys ? modelOrganization({ modelKeys, writer: organizationWriter({ dir: organization.dir, rolesDir: organization.rolesDir }) }) : null;
|
|
const instances=organization?instanceSettings({dir:organization.dir,rolesDir:organization.rolesDir,system:organization.system,levels,revoke,
|
|
...(proxyOrganization ? { writer: proxyOrganization.writer, revoke: proxyOrganization.revoke(revoke) } : {})}):null;
|
|
const usageLog = entry => process.stderr.write(`usage: ${JSON.stringify(entry)}\n`);
|
|
const keyForRun = (business, run) => modelRuns.get(business + ':' + run)?.keyId ?? null;
|
|
// Row 333 R3 (#1831): the usage view marks key quotas that match no hired agent; limits answers.
|
|
let orphanedKeys = () => null;
|
|
const usage = await createUsage({ store, dataRoot, businesses, log: usageLog, keyForRun,
|
|
proxyView: business => proxyUsage?.view(business) ?? null, keyOrphans: business => orphanedKeys(business) });
|
|
proxyUsage = createProxyUsage({ store, log: usageLog, client: async () => {
|
|
if (!modelKeys) return null;
|
|
await modelKeys.verifyIdentity();
|
|
return modelProxyClient;
|
|
},
|
|
records: () => credentials.status(),
|
|
read: () => Object.fromEntries(Object.keys(businesses).map(id => [id, settings ? settings.current(id) : businesses[id]])) });
|
|
// Row 125 (#1619, REQ-LIMIT-3, REQ-LIMIT-4): the limit checks read the organization file at
|
|
// each ask, so a limit, a reset or a switched-on check takes effect without a restart.
|
|
const limitLog = entry => process.stderr.write(`limits: ${JSON.stringify(entry)}\n`);
|
|
const limits = createLimits({ store, read: id => settings ? settings.current(id) : businesses[id], log: limitLog,
|
|
keyForSubject: (business, subject) => modelUsageKey(credentials.status(), modelRuns, business, subject),
|
|
keyLineages: business => modelKeyLineages(credentials.status(), business) });
|
|
orphanedKeys = business => limits.orphanedKeys(business);
|
|
for (const id of Object.keys(businesses)) await limits.refresh(id);
|
|
const withTokens = (id, result) => ({ ...result, tokens: limits.report(id) });
|
|
// Row 210 (#1707, design D7 step 6) and row 234 (#1740, r6 E2): the run dir and the host enrollment half.
|
|
const runRoot = organization ? join(organization.dir, 'secrets', 'run') : null;
|
|
let broker;
|
|
processing = createProcessing({ dataRoot, ...discordProcessingHooks(dataRoot), containerStop: createContainerStopper({ dataRoot }),
|
|
onTransition: () => broker.reconcileProcessing(),
|
|
onStop: stop => broker.recordProcessingStop(stop),
|
|
log: line => process.stderr.write(`${line}\n`),
|
|
});
|
|
// C2-04: mirrors are host state, never session checkouts. An unsafe
|
|
// existing mirror refuses startup before a socket can become ready.
|
|
for (const business of Object.values(businesses))
|
|
for (const [project, entry] of Object.entries(business.projects ?? {}))
|
|
if (entry.kind === 'library') {
|
|
try { await processing.work(() => libraryModule.ensureMirror({ business, project, dataRoot })); }
|
|
catch (e) { if (!e.code?.startsWith('processing-')) throw e; }
|
|
}
|
|
// C2-03a: the slot is <configDir>/secrets/<business>/library-gitea (design 10 step 4). Sync only
|
|
// checks that it is there; the askpass helper reads it.
|
|
// C2-03b: the broker chooses. The chief's grant store file once a chief has Gitea grant evidence,
|
|
// the slot before that, or a refusal sync records. The answer carries no file to sync but this one.
|
|
const slotDir = librarySync?.configDir ?? organization?.dir ?? null;
|
|
const fetchCredential = business => {
|
|
const choice = broker.libraryFetchCredential(business.id);
|
|
if (choice.kind === 'chief')
|
|
return { kind: 'chief', user: choice.user, file: storeRoot ? giteaReference(storeRoot, business.id, choice.agent).file : null };
|
|
if (choice.kind === 'slot') return { kind: 'slot', file: slotDir ? join(slotDir, 'secrets', business.id, libraryModule.SLOT) : null };
|
|
return { refused: choice.refused };
|
|
};
|
|
const syncOptions = business => ({
|
|
credential: fetchCredential(business),
|
|
...(librarySync ? { command: librarySync.command } : {}),
|
|
});
|
|
const humanWriteOptions = context => {
|
|
const mapping = libraryModule.checkMapping(context.business.projects[context.project]);
|
|
if (!mapping.editor) throw new libraryModule.LibraryError('library-mapping-invalid', 'editor is required');
|
|
return { credential: libraryModule.humanCredential(context, { user: mapping.editor.user, configDir: slotDir }),
|
|
...(librarySync ? { command: librarySync.command } : {}) };
|
|
};
|
|
// Row 194 (#1692, D3): after every write that commits, the broker takes the business as a restart
|
|
// would load it. The credential references, the project roots the Credentials store protects
|
|
// and the tracker bots are fixed at start (the Credentials store, the tasks adapter), so a file
|
|
// whose references, roots or bots differ from the boot copy isn't reloaded, and the write
|
|
// answers restart: true. Hire adds neither. A human create of a credentialed definition adds
|
|
// references, so it answers restart: true, and issue works after the restart (brief item 7).
|
|
// A role not in the boot copy matches only with no references and no bot. A reference compares
|
|
// by its kind and path only: row 199's issue rewrites the date in the file outside this reload,
|
|
// Credentials keeps its own copy of the date, and the broker never reads it.
|
|
const fixedRef = ({ file, env, store }) => ({ file, env, store });
|
|
const fixedRole = r => canonicalJson({
|
|
credentials: Object.fromEntries(Object.entries(r?.credentials ?? {}).map(([service, ref]) => [service, fixedRef(ref)])),
|
|
tracker: r?.tracker ?? null,
|
|
});
|
|
const fixedAtStart = b => ({
|
|
roles: Object.fromEntries(Object.entries(b.roles).map(([role, r]) => [role, fixedRole(r)])),
|
|
sync: canonicalJson(b.tracker?.sync ?? null),
|
|
roots: canonicalJson(Object.values(b.projects ?? {}).filter(p => p.kind !== 'library').map(p => p.root).sort()),
|
|
// A library entry has no root, and its mirror is ensured at start (C2-04), so each one is fixed whole.
|
|
libraries: canonicalJson(Object.fromEntries(Object.entries(b.projects ?? {}).filter(([, p]) => p.kind === 'library'))),
|
|
});
|
|
const sameFixed = (boot, b) => fixedAtStart(b).sync === boot.sync && fixedAtStart(b).roots === boot.roots
|
|
&& fixedAtStart(b).libraries === boot.libraries
|
|
&& Object.entries(fixedAtStart(b).roles).every(([role, f]) => f === (boot.roles[role] ?? fixedRole(null)));
|
|
const booted = new Map(Object.entries(businesses).map(([id, b]) => [id, fixedAtStart(b)]));
|
|
const reloadLog = entry => process.stderr.write(`organization: ${JSON.stringify(entry)}\n`);
|
|
function reload(id) {
|
|
try {
|
|
const b = loadBusiness(id, { dir: organization.dir, rolesDir: organization.rolesDir });
|
|
const resolved = {};
|
|
for (const instance of Object.keys(b.roles)) resolved[instance] = resolveInstance({ system: organization.system, business: b, instance });
|
|
const next = busBusiness(b, resolved);
|
|
if (!sameFixed(booted.get(id), next)) {
|
|
reloadLog({ event: 'reload-needs-restart', business: id });
|
|
return false;
|
|
}
|
|
broker.reloadBusiness(id, next);
|
|
return true;
|
|
} catch (e) {
|
|
reloadLog({ event: 'reload-failed', business: id, message: e.message });
|
|
return false;
|
|
}
|
|
}
|
|
broker = new Broker({
|
|
// FED R3: the access file is read per user request; a refusal leaves the user no grant.
|
|
access: access ? () => loadAccess(access.file, access.context) : null,
|
|
// FED R4 (#1801, design 3.7 trigger 1): a request stats the file and reads it only when this
|
|
// changes. A failed stat is a stamp of its own, so a removed file reads (and refuses) once.
|
|
accessStamp: access ? () => {
|
|
try {
|
|
const st = lstatSync(access.file, { bigint: true });
|
|
return `${st.dev}:${st.ino}:${st.size}:${st.mtimeNs}:${st.ctimeNs}:${st.mode}:${st.uid}`;
|
|
} catch (e) {
|
|
return `unreadable:${e.code ?? 'error'}`;
|
|
}
|
|
} : null,
|
|
// Row 270 (264c): session.list's compaction count, read from the session store's ledger.
|
|
compactions: ({ business, thread, session, legacy }) => {
|
|
const dir = storeDir(dataRoot, business, thread, session, { legacy });
|
|
return dir ? compactionCount(dir) : { count: null, known: false, reason: 'store-unreadable', at: null };
|
|
},
|
|
hostGrant: runRoot ? hostGrant({ configDir: organization.dir, runRoot }) : null,
|
|
// Row 235 (#1741, r6 X4, X5): the grant executor's claim, register and issue, never use().
|
|
hostKeys: storeRoot ? grantKeys({ credentials, storeRoot }) : null,
|
|
processing,
|
|
// Row 236 (#1742): the Gitea grant path. The admin slot is <configDir>/secrets/grants/<business>/gitea.
|
|
giteaGrant: organization ? giteaGrant({ configDir: organization.dir }) : null,
|
|
giteaKeys: storeRoot ? giteaKeys({ credentials, storeRoot }) : null,
|
|
usage,
|
|
limits,
|
|
store,
|
|
businesses,
|
|
library: {
|
|
lookup: createLibraryLookup({ dataRoot, sync: context =>
|
|
processing.work(() => libraryModule.sync(context, syncOptions(context.business))) }),
|
|
log: entry => process.stderr.write(`library: ${JSON.stringify(entry)}\n`),
|
|
readFile: context => libraryModule.readLibraryFile({ ...context, dataRoot }),
|
|
status: context => libraryModule.libraryStatus({ ...context, dataRoot }),
|
|
sync: (context, { accept }) => processing.work(() => libraryModule.sync({ ...context, dataRoot }, { ...syncOptions(context.business), accept })),
|
|
// C2-02: trusted references only. The askpass helper alone opens a credential file.
|
|
writeDoc: (context, options) => processing.work(() => libraryModule.writeLibraryDoc({ ...context, dataRoot }, {
|
|
...options, credential: { ...options.credential,
|
|
file: storeRoot ? giteaReference(storeRoot, context.business.id, options.credential.agent).file : null },
|
|
...(librarySync ? { command: librarySync.command } : {}),
|
|
})),
|
|
saveFile: context => processing.work(() => libraryModule.saveLibraryFile({ ...context, dataRoot }, humanWriteOptions(context))),
|
|
revert: context => processing.work(() => libraryModule.revertLibraryCommit({ ...context, dataRoot }, humanWriteOptions(context))),
|
|
LibraryError: libraryModule.LibraryError,
|
|
// C2-06: the citation guard, the reply limits read from the business file at each check (none
|
|
// configured reads as unreadable, so requireCitation is true), and thread.messages' `last`.
|
|
guard: (context, body, options) => libraryModule.guardReply({ ...context, dataRoot }, body, options),
|
|
replyLimits: id => libraryModule.resolveReplyLimits(organization
|
|
? organizationWriter({ dir: organization.dir, rolesDir: organization.rolesDir }).readRaw(id) : { readable: false }),
|
|
last: (context, paths) => libraryModule.lastCommits({ ...context, dataRoot }, paths),
|
|
GuardError: libraryModule.GuardError,
|
|
},
|
|
project: {
|
|
readSpec: context => projectModule.readSpec({ ...context, dataRoot }),
|
|
writeRecord: context => processing.work(() => projectModule.writeRecord({ ...context, dataRoot })),
|
|
listFiles: context => projectModule.listProjectFiles({ ...context, dataRoot }),
|
|
readFile: context => projectModule.readProjectFile({ ...context, dataRoot }),
|
|
writeFile: context => projectModule.writeHumanFile({ ...context, dataRoot }),
|
|
ProjectError: projectModule.ProjectError,
|
|
},
|
|
// limits and limits.set (organization.set is limits.set) carry each token limit's count; a
|
|
// plugins.set reloads the business's limit checks in the background.
|
|
organization:settings?(proxyOrganization?.wrap ?? (operation => operation))((id,verb,args,actor,context)=>{
|
|
if(verb==='limits')return withTokens(id,settings.read(id));
|
|
if(verb==='plugins')return plugins.read(id);
|
|
// Row 256 (#1761): the actor is the reader, so endProtected is the delete check for that reader.
|
|
if(verb==='instances')return instances.read(id,actor);
|
|
// Row 65 (mos-0840): the Console's routing table.
|
|
if(verb==='routes')return settings.routes(id);
|
|
// Row 298 (#1804, Q228): the chief's grant scope as the file stands, the fallback shown.
|
|
if(verb==='grants.scope')return instances.grantScopeRead(id);
|
|
// Row 194 (#1692, V5): a delete's checks at the verb, with no write and no reload.
|
|
if(verb==='instances.check')return instances.checkDelete(id,args.instance,actor);
|
|
// Row 194 b6 (webby-0094 B1): a fire's checks against the file, before step A.
|
|
if(verb==='instance.fireCheck')return instances.checkFire(id,args.name);
|
|
// Internal read-only path: no reload, journal, credential or publication.
|
|
if(verb==='instance.hireCheck')return instances.checkHire(id,args);
|
|
let result;
|
|
if(verb==='limits.set')result=withTokens(id,settings.set(id,args,actor));
|
|
else if(verb==='instances.set')result=instances.set(id,args,actor);
|
|
else if(verb==='routes.set')result=settings.setRoutes(id,args,actor);
|
|
else if(verb==='plugins.set')result=plugins.set(id,args,actor);
|
|
// Row 194 (#1692, V6): the human's keep and noHire lists, through the instance writer.
|
|
else if(verb==='veto.set')result=instances.veto(id,args,actor);
|
|
// Row 298: the human's grant scope, through the instance writer.
|
|
else if(verb==='grants.scope.set')result=instances.grantScope(id,args,actor);
|
|
// C2-06 M11: the human's replyLimits, through the organization writer's fence.
|
|
else if(verb==='library.limits.set')result=setReplyLimits(organizationWriter({dir:organization.dir,rolesDir:organization.rolesDir}),id,args,actor);
|
|
// Row 194 (D2.1, D2.2): the chief's hire and fire, with the reason in the change context.
|
|
else if(verb==='instance.hire')result=instances.hire(id,args,{actor,context});
|
|
else if(verb==='instance.fire')result=instances.fire(id,{name:args.name},{actor,context});
|
|
else throw new BusError('invalid-request');
|
|
// Row 194 (D3 item 2): one reload, after every verb that committed, then the limit checks.
|
|
const live=reload(id);
|
|
limits.refresh(id).catch(e=>limitLog({event:'limit-refresh-failed',business:id,message:e.message}));
|
|
// Row 194 b6 (webby-0094 B1, sage-2241): every instance-writer verb says whether the reload
|
|
// took, so a refused one (credentials, roots, ...) shows the restart line for a keep or a
|
|
// hire too. The other writers' answers carry no restart field.
|
|
return INSTANCE_VERBS.has(verb)?{...result,restart:!live}:result;
|
|
}):null,
|
|
// Row 194 (D2.3): the role files as they stand, for the hire checks.
|
|
roleFile: organization ? (name) => loadRole(organization.rolesDir, name) : null,
|
|
secretCheck: (value) => credentials.assertClean(value),
|
|
credentialStatus: () => credentials.status(),
|
|
// Row 199 (#1697, design section 6): issue, rotate, revoke and delete, live with no restart; each
|
|
// rewrites or records through the organization writer, so changes.jsonl holds the receipt.
|
|
proxyPage: async (verb, args, scope) => {
|
|
if (!proxyPage) throw new BusError('not-configured');
|
|
return proxyPage.request(verb, args, scope);
|
|
},
|
|
proxyInit: organization ? async args => {
|
|
const settings = await initializeProxy({ configDir: organization.dir, dataRoot, ...args });
|
|
modelProxyClient = await loadManagementClient({ configDir: organization.dir, proxy: settings, receipt: proxyReceipt });
|
|
configureProxyPage();
|
|
return settings;
|
|
} : null,
|
|
credentialVerbs: organization
|
|
? credentialVerbs({ credentials, writer: organizationWriter({ dir: organization.dir, rolesDir: organization.rolesDir }) })
|
|
: null,
|
|
// Row 65: a new launch binds only on a harness its business switches on, read from the file
|
|
// as it stands now; an unreadable file switches none on.
|
|
harnessOn: (id, name) => {
|
|
try {
|
|
const b = settings ? settings.current(id) : businesses[id];
|
|
return Array.isArray(b?.plugins?.harness) && b.plugins.harness.includes(name);
|
|
} catch {
|
|
return false;
|
|
}
|
|
},
|
|
// Row 298 (#1804, Q228): one read binds launch.grantScope to the organization revision.
|
|
// A missing organization reader or unreadable file cannot authorize an automatic grant.
|
|
grantScope: (id) => {
|
|
if (!settings) return { value: undefined, revision: null };
|
|
const { document, revision } = instances.writer.read(id);
|
|
return { value: document.launch?.grantScope, revision };
|
|
},
|
|
// Row 194 (#1692, D4.1, B8): bus.tester as the file sets it now; an unreadable file is off.
|
|
testerOn: (id) => {
|
|
try {
|
|
const b = settings ? settings.current(id) : businesses[id];
|
|
return b?.vars?.['bus.tester'] === true;
|
|
} catch {
|
|
return false;
|
|
}
|
|
},
|
|
});
|
|
broker.reconcileProcessing();
|
|
processing.start();
|
|
const processingWork = () => {
|
|
const state = processing.read();
|
|
if (state.invalid) throw new BusError(state.code);
|
|
if (state.halted) throw new BusError('processing-halted');
|
|
};
|
|
// Process identities must be supplied for every managed run when using the human transport.
|
|
const launchRecords = [];
|
|
function bindLaunch(r) {
|
|
if (
|
|
!Number.isSafeInteger(r.pid) ||
|
|
r.pid < 2 ||
|
|
typeof r.startTime !== 'string' ||
|
|
!/^\d+$/.test(r.startTime)
|
|
)
|
|
throw new BusError('invalid-launch-process');
|
|
const { modelKeyId, ...record } = r;
|
|
if (modelKeys) {
|
|
modelKeys.assertLaunchable(r.business + '/' + r.role);
|
|
const m = credentials.modelRecord(r.business + '/' + r.role);
|
|
if (!m || ![m.keyId, ...m.retiring.map(k => k.keyId)].includes(modelKeyId)) throw new BusError('proxy-key-refused');
|
|
}
|
|
const cap = broker.bindLaunch(record);
|
|
launchRecords.push({ ...r });
|
|
if (modelKeys) modelRuns.set(r.business + ':' + r.run, { ...r, instance: r.business + '/' + r.role, keyId: modelKeyId });
|
|
return { business: r.business, run: r.run, cap };
|
|
}
|
|
const bound = launches.map(bindLaunch);
|
|
const readCaps = readers.map((business) => ({ business, cap: broker.bindReader({ business }) }));
|
|
if (tasks) {
|
|
adapter = await tasks({ broker, credentials, businesses, processing });
|
|
if (typeof adapter?.handle !== 'function' || !Number.isSafeInteger(adapter.timeout) || adapter.timeout < 1)
|
|
throw new BusError('invalid-adapter');
|
|
}
|
|
const web = await webServices(businesses);
|
|
// Row 210 (#1707, design D7 step 6): clear what a killed bus child left, before anyone can call.
|
|
if (runRoot) sweepRunDir(runRoot);
|
|
// Row 234 (r6 X3): end the enrollments a killed bus child left, before anyone can call.
|
|
broker.reconcileGrants();
|
|
const interpreter = inspectInterpreter();
|
|
recordInterpreterStart(store.directory, interpreter);
|
|
const peerReader = createPeerReader({ interpreter });
|
|
const principalPaths = {
|
|
human: fileURLToPath(new URL('./human-cli.mjs', import.meta.url)),
|
|
tester: fileURLToPath(new URL('./tester-cli.mjs', import.meta.url)),
|
|
user: fileURLToPath(new URL('./user-cli.mjs', import.meta.url)),
|
|
console: fileURLToPath(new URL('../../webui/src/cli.mjs', import.meta.url)),
|
|
consoleCli: fileURLToPath(new URL('./console-cli.mjs', import.meta.url)),
|
|
};
|
|
const path = join(store.directory, 'broker.sock');
|
|
// The existing host executor owns SSH and its key lifecycle. Combine its revocation
|
|
// signal with processing cancellation without changing its proof or credential entry points.
|
|
const host = runRoot ? hostExecutor({ root: runRoot, credentials: {
|
|
busy: (...args) => credentials.busy(...args),
|
|
useFor: (session, name, fn, options) => credentials.useFor(session, name, (token, meta) =>
|
|
fn(token, { ...meta, signal: meta.signal
|
|
? AbortSignal.any([meta.signal, processing.signal()]) : processing.signal() }), options),
|
|
} }) : null;
|
|
server = await serve({
|
|
web: (business) => web.get(business) ?? null,
|
|
host: host ? { busy: (...args) => host.busy(...args),
|
|
run: (...args) => processing.work(() => host.run(...args)) } : null,
|
|
tasks: adapter ? { handle: adapter.handle, timeout: adapter.timeout } : null,
|
|
broker,
|
|
path,
|
|
peerReader,
|
|
authenticateHuman: (proof, peer) => {
|
|
const p = verifyPrincipal({ human: proof }, { peer, paths: principalPaths, businesses, launches: launchRecords });
|
|
return broker.bindHuman({ business: p.business, human: p.id, via: 'cli', outsideAgent: true });
|
|
},
|
|
authenticateTester: (proof, peer) => {
|
|
const p = verifyPrincipal({ tester: proof }, { peer, paths: principalPaths, businesses, launches: launchRecords });
|
|
return broker.bindTester({ business: p.business });
|
|
},
|
|
// FED R3 (design 3.6): checks 1 and 2 are row 196's user kind; check 3, the live grant, is bindUser's.
|
|
authenticateUser: (proof, peer) => {
|
|
const p = verifyPrincipal({ user: proof }, { peer, paths: principalPaths, businesses, launches: launchRecords });
|
|
return broker.bindUser({ business: p.business, user: p.id });
|
|
},
|
|
// Row 252 (#1753): a fresh proof per connection, a cap for the Console verbs only.
|
|
authenticateConsole: (proof, peer) => {
|
|
const p = verifyPrincipal({ console: proof }, { peer, paths: principalPaths, businesses, launches: launchRecords, consoleOwners: owners });
|
|
return broker.bindConsole({ business: p.business, human: p.id, subject: p.subject });
|
|
},
|
|
});
|
|
// Row 235 (r6 X3): the remote undo of interrupted host grants, once the bus is up, so a start doesn't
|
|
// wait on the network.
|
|
broker.reconcileHosts();
|
|
// Row 236 (#436b r5 section 5): the Gitea grants' remote undo, settle and open revokes, likewise.
|
|
broker.reconcileGitea();
|
|
// S6 launcher operations. The host decides capacity and the instance list; the broker
|
|
// authorizes role.launch and keeps the evidence, refusals included.
|
|
// A refusal is recorded against the caller, as request() records one.
|
|
function refusing(cap, fn, subject = null) {
|
|
try {
|
|
return fn();
|
|
} catch (e) {
|
|
if (!(e instanceof BusError)) throw e;
|
|
broker.refuse(cap, e.code, subject);
|
|
throw e;
|
|
}
|
|
}
|
|
const authorizeLaunch = (cap, instance) =>
|
|
refusing(cap, () => broker.authorize(cap, 'role.launch', { target: instance }));
|
|
// Row 93 (#1587): the host's git socket. A push is checked before the host takes the bundle
|
|
// and recorded once the ref moved; a reviewer's fetch is checked against its review.requested.
|
|
const checkPush = (cap, task_ref) => refusing(cap, () => broker.checkPush(cap, task_ref), task_ref);
|
|
const reviewFetch = (cap, args) => refusing(cap, () => broker.reviewFetch(cap, args), args?.task_ref);
|
|
// Row 137 (#1631): the git socket's land op. checkLand's refusal is the run's, like checkPush's.
|
|
// A landing then ends in recordLand or landRefused under its ticket, which outlive the run's cap.
|
|
const checkLand = (cap, args) => refusing(cap, () => broker.checkLand(cap, args), args?.task_ref);
|
|
// Row 199 (#1697, design D9b): the git socket's publish op. checkPublish's refusal is the run's.
|
|
// The gitea token is used only under the open ticket, keyed from the agent checkPublish checked,
|
|
// and the op table returns no token. A publish then ends in recordPublish or publishRefused,
|
|
// which files any refusal once, withGitea's included.
|
|
const checkPublish = (cap, args) => refusing(cap, () => broker.checkPublish(cap, args), args?.task_ref);
|
|
const withGitea = async (ticket, fn) => credentials.useFor(broker.publishSession(ticket), 'gitea', fn);
|
|
// The push itself runs here, where the token is (D9 step 5), and the git socket's host polls
|
|
// for its end: the host's op channel answers one op at a time within 10 s, and ls-remote and
|
|
// push can take longer. Fork, user, source and dst all come from the ticket. A WorkingError's
|
|
// code is the refusal; git's stderr stays here.
|
|
const remotes = new Map();
|
|
function publishPush(ticket) {
|
|
if (remotes.has(ticket)) throw new BusError('publish-busy');
|
|
const { business, project, fork, user, source, dst } = broker.publishTarget(ticket);
|
|
const entry = { state: 'running' };
|
|
remotes.set(ticket, entry);
|
|
processing.work(() => withGitea(ticket, (token) =>
|
|
publishRemote({ repo: workingRepo(dataRoot, business, project), fork, user, source, dst, token, tls: publishTls }).catch((e) => {
|
|
throw e instanceof WorkingError ? new BusError(e.code) : e;
|
|
}),
|
|
)).then(
|
|
(r) => Object.assign(entry, { state: 'pushed', ...r }),
|
|
(e) => {
|
|
// Not a refusal: the error's name and code only, never a message that could carry git's output.
|
|
if (!(e instanceof BusError)) process.stderr.write(`bus: publish git-failed: ${JSON.stringify({ name: e?.name, code: e?.code })}\n`);
|
|
Object.assign(entry, { state: 'refused', code: e instanceof BusError ? e.code : 'git-failed' });
|
|
},
|
|
).then(() => {
|
|
if (entry.held) settle(ticket, entry, entry.held);
|
|
});
|
|
return { state: 'running' };
|
|
}
|
|
// Row 242 (#1748): the git socket gave up on a publish (publish-timeout) whose push it never
|
|
// collected. Its publishRefused waits for the push: until that ends the ticket stays open, so
|
|
// checkPublish refuses another publish on the decision with publish-busy (row 237). A push that
|
|
// reached the fork is then recorded, which spends the decision; one that didn't is filed with
|
|
// the socket's code. Either way the decision can't push a second ref.
|
|
function settle(ticket, entry, code) {
|
|
remotes.delete(ticket);
|
|
try {
|
|
if (entry.state === 'pushed') {
|
|
try {
|
|
broker.recordPublish(ticket, { ref: entry.ref, commit: entry.commit, previous: entry.previous });
|
|
process.stderr.write(`bus: publish ${ticket}: pushed after the git socket refused ${code}; recorded\n`);
|
|
return;
|
|
} catch (e) {
|
|
if (!(e instanceof BusError)) throw e;
|
|
code = e.code;
|
|
}
|
|
}
|
|
broker.publishRefused(ticket, { code });
|
|
} catch (e) {
|
|
process.stderr.write(`bus: publish ${ticket}: not settled: ${JSON.stringify({ name: e?.name, code: e?.code })}\n`);
|
|
}
|
|
}
|
|
function publishRefused(ticket, args) {
|
|
const entry = remotes.get(ticket);
|
|
if (!entry) return broker.publishRefused(ticket, args);
|
|
if (!/^[a-z-]{1,64}$/.test(args?.code ?? '')) throw new BusError('invalid-request');
|
|
if (entry.state === 'running') entry.held = args.code;
|
|
else settle(ticket, entry, args.code);
|
|
return args.code;
|
|
}
|
|
function publishResult(ticket) {
|
|
const entry = remotes.get(ticket);
|
|
if (!entry) throw new BusError('publish-unknown');
|
|
if (entry.state !== 'running') remotes.delete(ticket);
|
|
return { ...entry };
|
|
}
|
|
// Metadata only (service, date, state); the token stays in Credentials.
|
|
function credentialStatus(business, role) {
|
|
return credentials
|
|
.status()
|
|
.filter((c) => c.instance === business + '/' + role)
|
|
.map(({ service, date, state, keyId, issuedAt }) => ({ service, date, state, ...(service === 'model' ? { keyId, issuedAt } : {}) }));
|
|
}
|
|
// C2-03a: each library entry with a sync user fetches once at start, then every 60 s. A pass
|
|
// still running skips the tick. Only codes are logged, never git's output.
|
|
const timed = Object.values(businesses).flatMap(business => Object.entries(business.projects ?? {})
|
|
.filter(([, entry]) => entry.kind === 'library' && Object.hasOwn(entry, 'user')).map(([project]) => ({ business, project })));
|
|
const syncLog = entry => process.stderr.write(`library: ${JSON.stringify(entry)}\n`);
|
|
const syncAll = async () => {
|
|
for (const { business, project } of timed) {
|
|
try {
|
|
const { code } = await libraryModule.sync({ business, project, dataRoot }, syncOptions(business));
|
|
if (code !== null) syncLog({ event: 'library-sync-held', business: business.id, project, code,
|
|
...(code === 'library-fetch-no-grant' ? { fix: LIBRARY_FETCH_FIX } : {}) });
|
|
} catch (e) {
|
|
if (e?.code === 'library-sync-busy') continue;
|
|
syncLog({ event: 'library-sync-refused', business: business.id, project,
|
|
code: e instanceof libraryModule.LibraryError ? e.code : 'library-sync-failed' });
|
|
}
|
|
}
|
|
};
|
|
const tick = () => {
|
|
const state = processing.read();
|
|
if (!syncPass && !state.halted && !state.invalid)
|
|
syncPass = processing.work(syncAll).catch(e => {
|
|
if (!e.code?.startsWith('processing-')) syncLog({ event: 'library-sync-refused', code: 'library-sync-failed' });
|
|
}).finally(() => { syncPass = null; });
|
|
};
|
|
if (timed.length) {
|
|
tick();
|
|
syncTimer = setInterval(tick, librarySync?.intervalMs ?? libraryModule.SYNC_INTERVAL_MS);
|
|
syncTimer.unref();
|
|
}
|
|
const stopSync = async () => { clearInterval(syncTimer); syncTimer = null; await syncPass; };
|
|
// FED R4 (#1801, design 3.7 trigger 2): every 30 s the broker re-reads the access file, so a hand
|
|
// edit or an expiry with no request ends the lost runs within one period.
|
|
if (access) {
|
|
accessTimer = setInterval(() => {
|
|
try {
|
|
broker.accessSweep();
|
|
} catch (e) {
|
|
process.stderr.write(`access: ${JSON.stringify({ event: 'access-sweep-failed', code: e.code ?? 'error' })}\n`);
|
|
}
|
|
}, access.intervalMs ?? ACCESS_SWEEP_MS);
|
|
accessTimer.unref();
|
|
}
|
|
if (modelKeys) {
|
|
modelTimer = setInterval(() => {
|
|
if (modelSweep) return;
|
|
modelSweep = modelKeys.sweep().catch(error => proxyReceipt({ action: 'proxy.sweep-refused', code: error.code ?? 'proxy-key-issue-failed' })).finally(() => { modelSweep = null; });
|
|
}, 1000);
|
|
modelTimer.unref();
|
|
}
|
|
// One bounded queue read at a time. The service applies the configured
|
|
// 1..59 second poll deadline; this tick only checks whether a read is due.
|
|
const usageTick = () => { proxyUsage.tick().catch(() => usageLog({ action: 'proxy.usage', code: 'proxy-usage-read-failed' })); };
|
|
usageTimer = setInterval(usageTick, 1000);
|
|
usageTimer.unref();
|
|
usageTick(); // Optional telemetry never blocks the native ready answer.
|
|
return {
|
|
broker,
|
|
credentials,
|
|
// P2 consumes this inside the bus only. No process IPC/verb returns it.
|
|
get modelProxy() { return modelProxyClient; },
|
|
path,
|
|
bindLaunch,
|
|
recoverLaunch: ({ modelKeyId, ...record }) => broker.recoverLaunch(record),
|
|
endLaunch: (r) => { const out = broker.endLaunch(r); modelRuns.delete(r.business + ':' + r.run); return out; },
|
|
modelPreflight: async ({ business, role }) => {
|
|
const current = settings ? settings.current(business) : businesses[business];
|
|
if (!Object.hasOwn(current?.roles ?? {}, role)) throw new BusError('unknown-role');
|
|
return modelKeys ? modelKeys.preflight(business + '/' + role) : null;
|
|
},
|
|
modelStall: async ({ business, role, run, keyId }) => {
|
|
const record = modelRuns.get(business + ':' + run);
|
|
if (!record || record.role !== role || record.keyId !== keyId) throw new BusError('proxy-stop-failed');
|
|
await stopModelRuns({ runs: [record], reason: 'proxy-stall', sessions: () => readSessions(dataRoot),
|
|
stop: id => stopSession(dataRoot, id, { timeoutMs: MODEL_STOP_TIMEOUT_MS }), receipt: proxyReceipt });
|
|
modelRuns.delete(business + ':' + run);
|
|
return { stopped: true };
|
|
},
|
|
threadsState: (business) => broker.threadsState(business),
|
|
threadIdle: (r) => broker.threadIdle(r),
|
|
// Row 125 (#1619, REQ-LIMIT-4): the launcher asks the limit checks before every launch.
|
|
mayLaunch: async ({ business, instance, provider, model, account = null }) => {
|
|
processingWork();
|
|
return limits.mayLaunch(business, { instance, provider, model, account });
|
|
},
|
|
identity: (cap) => broker.identity(cap),
|
|
authorizeLaunch,
|
|
checkPush,
|
|
recordPush: (cap, push) => refusing(cap, () => broker.recordPush(cap, push), push?.task_ref),
|
|
reviewFetch,
|
|
checkLand,
|
|
recordLand: (ticket, land) => broker.recordLand(ticket, land),
|
|
landRefused: (ticket, args) => broker.landRefused(ticket, args),
|
|
checkPublish,
|
|
withGitea,
|
|
publishPush,
|
|
publishResult,
|
|
recordPublish: (ticket, pushed) => broker.recordPublish(ticket, pushed),
|
|
publishRefused,
|
|
landHeld: (record) => broker.landHeld(record),
|
|
// Row 156 (#1651): the launcher's landing pass.
|
|
landHolders: (business) => broker.landHolders(business),
|
|
landRelease: (r) => broker.landRelease(r),
|
|
// Row 194 (#1692, D1.1): the launcher's lifecycle pass.
|
|
lifecycleRequests: (business) => broker.lifecycleRequests(business),
|
|
lifecycleTake: (r) => broker.lifecycleTake(r),
|
|
lifecycleDelete: (r) => broker.lifecycleDelete(r),
|
|
lifecycleDone: (r) => broker.lifecycleDone(r),
|
|
// Row 271 (264d, #1776): the launcher's clear pass.
|
|
sessionClears: (business) => broker.sessionClears(business),
|
|
sessionClearExited: (r) => broker.sessionClearExited(r),
|
|
sessionClearUncertain: (r) => broker.sessionClearUncertain(r),
|
|
sessionClearRetire: (r) => broker.sessionClearRetire(r),
|
|
sessionClearStart: (r) => broker.sessionClearStart(r),
|
|
refuse: (cap, code, subject) => broker.refuse(cap, code, subject),
|
|
credentialStatus,
|
|
launches: bound,
|
|
readers: readCaps,
|
|
async close() {
|
|
clearInterval(usageTimer);
|
|
await proxyUsage.close();
|
|
clearInterval(modelTimer);
|
|
await modelSweep;
|
|
clearInterval(accessTimer);
|
|
await stopSync();
|
|
await server.close();
|
|
await adapter?.close?.();
|
|
await processing?.close();
|
|
store.close();
|
|
credentials.close();
|
|
},
|
|
};
|
|
} catch (e) {
|
|
clearInterval(usageTimer);
|
|
await proxyUsage?.close();
|
|
clearInterval(modelTimer);
|
|
clearInterval(accessTimer);
|
|
await server?.close();
|
|
await adapter?.close?.();
|
|
await processing?.close();
|
|
store?.close();
|
|
credentials?.close();
|
|
throw e;
|
|
}
|
|
}
|