jason.woltje
bfd2cc5e79
pack.state gave a pack hire only the chief of staff's resolved route, so an org whose chief has no model refused every hire pack-hire-route-missing. The route is now the chief's (launch.by), else the org assistant's (the instance whose definition is assistant; the one named assistant when there are several, else the first in key order), else null. Both resolve through resolveInstance, need harness and model, and carry thinking when set. pack.state adds routeFrom, and each hired or awaiting-owner receipt line carries it. Refusal next step, CLI hint, WebUI hire reason and the bus, CLI and TOOLS docs say the refusal now means neither has a route.
1861 lines
123 KiB
JavaScript
1861 lines
123 KiB
JavaScript
import { autonomyHost, authorityHost } from './autonomy-host.mjs';
|
|
import { credentialLines, hireDecisions } from '../../organization/src/hire-decisions.mjs';
|
|
import { organizationPlugins } from '../../plugins/src/organization.mjs';
|
|
import { pluginService } from '../../plugins/src/tools.mjs';
|
|
import { readPluginLock } from '../../plugins/src/pinned.mjs';
|
|
import { createRequirementsHost } from '../../tasks/src/builtin.mjs';
|
|
import { createRequirementCatalog } from './requirement-catalog.mjs';
|
|
import { readPrdMaterial } from '../../project/src/prd.mjs';
|
|
import { lstatSync, mkdirSync, readdirSync, readFileSync } from 'node:fs';
|
|
import { join } from 'node:path';
|
|
import { fileURLToPath } from 'node:url';
|
|
import { Store } from './store.mjs';
|
|
import { createRequestStore } from './request-store.mjs';
|
|
import { createRequestLifecycle } from './request-lifecycle.mjs';
|
|
import { requestWorkspace } from './request-workspace.mjs';
|
|
import { readReviewedArtifact } from '../../seat/src/artifact.mjs';
|
|
import { executeBootstrap } from './request-bootstrap.mjs';
|
|
import { hostPreparation } from './request-host.mjs';
|
|
import { managedPaths, managedState } from '../../organization/src/project-bootstrap.mjs';
|
|
import * as projectModule from '../../project/src/index.mjs';
|
|
import * as libraryModule from '../../library/src/index.mjs';
|
|
import { Broker, projectArchived, BusError, LIBRARY_FETCH_FIX } from './broker.mjs';
|
|
import { Credentials } from './credentials.mjs';
|
|
import { serve } from './server.mjs';
|
|
import { launchSettings } from '../../organization/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 '../../organization/src/plugin-settings.mjs';
|
|
import { loadManagementClient } from '../../model-proxy/src/client.mjs';
|
|
import { ProxyPage } from './proxy-page.mjs';
|
|
import { createModelCatalogue } from './model-catalogue.mjs';
|
|
import { sha256 } from '../../model-proxy/src/common.mjs';
|
|
import { initializeProxy } from '../../model-proxy/src/settings.mjs';
|
|
import { createProxyLock } from './proxy-lock.mjs';
|
|
import { verifyRestart } from '../../model-proxy/src/admin.mjs';
|
|
import { ModelKeys, modelOrganization, stopModelRuns, MODEL_STOP_TIMEOUT_MS } from './model-keys.mjs';
|
|
import { instanceSettings } from '../../organization/src/instance-settings.mjs';
|
|
import { organizationWriter, OrganizationError, setReplyLimits } from '../../organization/src/organization.mjs';
|
|
import { definitionVerbs, DEFINITION_BUS_VERBS } from '../../organization/src/definition-verbs.mjs';
|
|
import { ORG_PREFIX, orgDefinitionName, compositionOf, readTemplate, entryFromTemplate, lineDiff, PRIMITIVES } from '../../organization/src/definitions.mjs';
|
|
import { installedStarterPacks, readSavedTemplate } from '../../plugins/src/starter-pack.mjs';
|
|
import { TemplateError, templateList, templatePreview, templateSave } from '../../organization/src/template-save.mjs';
|
|
import { checkComposition } from '../../organization/src/composition.mjs';
|
|
import { validateBusinessDocument } from '../../organization/src/organization-config.mjs';
|
|
import { DEFINITION_HOLDS_CREDENTIAL, heldPath, scanDefinitions } from './definition-scan.mjs';
|
|
import { ROLE_RIGHTS } from '../../organization/src/org-rights.mjs';
|
|
import { getProjectRemote } from '../../organization/src/project-remotes.mjs';
|
|
import { credentialVerbs } from './credential-verbs.mjs';
|
|
import { COPY_BINDABLE, loadWithCopies, promoteCopies } from '../../organization/src/agent-copies.mjs';
|
|
import { copyGrant, sweepCopySlots } from './copy-grant.mjs';
|
|
import { copyLaunchGate } from './copy-launch.mjs';
|
|
import { copyRevoker as makeCopyRevoker, giteaCopyRemote } from './copy-revoke.mjs';
|
|
import { readOrgEntrySync } from '../../organization/src/org-registry.mjs';
|
|
import { loadRole } from '../../organization/src/role.mjs';
|
|
import { resolveInstance } from '../../organization/src/resolve.mjs';
|
|
import { canonicalJson, sha256 as textSha256 } from '../../organization/src/util.mjs';
|
|
import { selectModelProvider } from '../../plugins/src/model-provider-selection.mjs';
|
|
import { busOrganization } from './organization-context.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 '../../organization/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.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}.
|
|
// U4a (#1873): it serves the businesses whose task-tracker module is Vikunja; the others get theirs
|
|
// through the plug-in loader, the built-in store by default (see trackerOf in startBroker).
|
|
// 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, dataRoot = null } = {}) {
|
|
const services = new Map();
|
|
for (const [name, b] of Object.entries(businesses)) {
|
|
const loader = createLoader({ log, dataRoot });
|
|
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.
|
|
// 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 = [];
|
|
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 });
|
|
}
|
|
let pins = {};
|
|
if (dataRoot !== null) {
|
|
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 = 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
|
|
// 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();
|
|
}
|
|
},
|
|
};
|
|
}
|
|
|
|
// #1873 CB4: how often the bus re-checks the copies holding a grant, and retries a pending revoke.
|
|
const COPY_SWEEP_MS = 60000;
|
|
|
|
// #1873 SP4: the human's pack, board and workflow ops (design section 3). preview reads the pack and
|
|
// the organization file and returns the plan with each item's mark, the lines each proposed hire
|
|
// would carry against the chief, the noHire marks and the skills this machine has not enabled, plus
|
|
// a digest of the plan and the file revision. apply rebuilds the plan, refuses pack-plan-stale when
|
|
// the digest differs and commits it in one applyPack. board.set and workflow.set are one-key plans.
|
|
// Nothing here reads a saved template but the caller's own (readSavedTemplate, account = actor), and
|
|
// a pack's origin comes from where it was found, never from the request.
|
|
const PACK_AS = /^(definitions|workflows)\.([a-z0-9][a-z0-9._-]{0,63})=([a-z0-9][a-z0-9._-]{0,63})$/;
|
|
const SAVED_PREFIX = 'saved:';
|
|
// Each op, and whether it writes.
|
|
const PACK_OPS = Object.freeze({ packs: false, 'pack.preview': false, 'pack.state': false, 'pack.apply': true, 'board.set': true, 'workflow.set': true });
|
|
const EMPTY_CHIEF = Object.freeze({ tools: [], network: 'none', authority: { withinRole: [], crossRole: [] }, credentials: [] });
|
|
function packOps({ writer, rolesDir, dataRoot, system, registryFor, assertClean, configDir }) {
|
|
const bad = () => { throw new BusError('invalid-request'); };
|
|
const isObject = v => v !== null && typeof v === 'object' && !Array.isArray(v);
|
|
const scan = value => {
|
|
if (typeof value === 'string') assertClean(value);
|
|
else if (value !== null && typeof value === 'object') for (const [k, v] of Object.entries(value)) { assertClean(k); scan(v); }
|
|
};
|
|
const installed = async () => (await installedStarterPacks({ dataRoot, rolesDir })).packs;
|
|
async function loadPack(name, actor) {
|
|
if (typeof name !== 'string' || name.length > 128) bad();
|
|
if (name.startsWith(SAVED_PREFIX)) return readSavedTemplate({ configDir, account: actor, name: name.slice(SAVED_PREFIX.length), rolesDir });
|
|
const found = (await installed()).find(p => p.id === name);
|
|
if (!found) throw new BusError('starter-pack-not-found');
|
|
return found;
|
|
}
|
|
function renames(as) {
|
|
const out = { definitions: new Map(), workflows: new Map() };
|
|
if (as === undefined) return out;
|
|
if (!Array.isArray(as) || as.length > 32) bad();
|
|
for (const value of as) {
|
|
const m = typeof value === 'string' ? PACK_AS.exec(value) : null;
|
|
// `human` is the reserved step role: no rename may write a definition by that name.
|
|
if (!m || out[m[1]].has(m[2]) || [...out[m[1]].values()].includes(m[3]) || (m[1] === 'definitions' && m[3] === 'human')) bad();
|
|
out[m[1]].set(m[2], m[3]);
|
|
}
|
|
return out;
|
|
}
|
|
// The entry a pack definition writes: a template reference becomes the template's own entry, pinned
|
|
// by its hash (a moved hash refuses), and a full entry is copied as it is. Neither carries rev.
|
|
function entryOf(value) {
|
|
if (Object.hasOwn(value, 'template') && Object.hasOwn(value, 'sha256') && Object.keys(value).length === 2) {
|
|
const template = readTemplate(rolesDir, value.template);
|
|
if (template.sha256 !== value.sha256) throw new BusError('starter-pack-template-changed');
|
|
const { rev: _rev, ...entry } = entryFromTemplate(template);
|
|
return { entry, ref: { template: value.template, sha256: value.sha256 } };
|
|
}
|
|
return { entry: structuredClone(value), ref: null };
|
|
}
|
|
function chiefOf(doc, file) {
|
|
const loaded = validateBusinessDocument(doc, file, { rolesDir, dataRoot });
|
|
const chief = loaded.definitions[doc.roles?.[doc.launch?.by]?.definition];
|
|
if (!chief) return null;
|
|
return { ...chief, ...resolveInstance({ system, business: loaded, instance: doc.launch.by }).limits };
|
|
}
|
|
// Each earlier packs entry of this id and origin (and, for a project pack, this project) lists what
|
|
// it added; an item absent now that one of them added was removed on purpose, so it stays removed.
|
|
// A saved template may reuse a shipped id, so the origin is part of the key.
|
|
const sameApply = (e, id, origin, project) => e.id === id && e.origin === origin && (e.project ?? null) === project;
|
|
const tombstones = (doc, id, origin, project) => new Set((doc.packs ?? []).filter(e => sameApply(e, id, origin, project)).flatMap(e => e.added ?? []));
|
|
// #1873 SP5 (design section 3, "The chief applying a pack"): context.agent is an
|
|
// organization.change holder's call. It names no saved template and no definitions rename
|
|
// (human-required), and its plan writes no definition: a definition a human apply would add is
|
|
// marked not-written, and its hire proposal pins the template or the inline entry, which the chief
|
|
// raises as a hire decision. A workflow step naming a pack definition the organization does not
|
|
// hold and that is not a V2 template is left out and listed; a workflow left with no step is
|
|
// marked left-out and not written. applyPack refuses an agent's definitions or saved origin
|
|
// whatever this builds.
|
|
async function build(id, args, actor, context) {
|
|
const agent = context?.agent === true;
|
|
if (agent && typeof args.pack === 'string' && args.pack.startsWith(SAVED_PREFIX)) throw new BusError('human-required');
|
|
if (agent && Array.isArray(args.as) && args.as.some(v => typeof v === 'string' && v.startsWith('definitions.'))) throw new BusError('human-required');
|
|
const { document: doc, revision } = writer.read(id);
|
|
const found = await loadPack(args.pack, actor);
|
|
const pack = found.pack;
|
|
const origin = found.origin;
|
|
let project = null;
|
|
if (pack.scope === 'org') { if (args.project !== undefined) bad(); }
|
|
else {
|
|
project = args.project;
|
|
const entry = typeof project === 'string' && Object.hasOwn(doc.projects ?? {}, project) ? doc.projects[project] : null;
|
|
if (!isObject(entry) || entry.kind === 'library') bad();
|
|
}
|
|
const as = renames(args.as);
|
|
for (const name of as.definitions.keys()) if (!Object.hasOwn(pack.definitions, name)) bad();
|
|
for (const name of as.workflows.keys()) if (!Object.hasOwn(pack.workflows, name)) bad();
|
|
// Two pack items of one kind never land on one name: a rename onto another item's own name
|
|
// would show both as add while the later one replaced the earlier in the plan.
|
|
for (const kind of ['definitions', 'workflows']) {
|
|
const targets = Object.keys(pack[kind]).map(name => as[kind].get(name) ?? name);
|
|
if (new Set(targets).size !== targets.length) bad();
|
|
}
|
|
scan(pack);
|
|
const gone = tombstones(doc, pack.id, origin, project);
|
|
const scope = project === null ? doc : doc.projects[project];
|
|
const prefix = project === null ? '' : `projects.${project}.`;
|
|
const chief = chiefOf(doc, `${id}.json`);
|
|
const registry = registryFor(doc);
|
|
const noHire = doc.launch?.noHire ?? [];
|
|
const templates = new Map();
|
|
const isTemplate = name => {
|
|
if (!templates.has(name)) {
|
|
try { readTemplate(rolesDir, name); templates.set(name, true); } catch (error) {
|
|
if (error?.name !== 'BusinessError') throw error;
|
|
templates.set(name, false);
|
|
}
|
|
}
|
|
return templates.get(name);
|
|
};
|
|
const items = [];
|
|
const plan = {};
|
|
// Definitions are organization-wide in either scope: a project's agents name org definitions.
|
|
const definitions = {};
|
|
const checked = {};
|
|
for (const [name, value] of Object.entries(pack.definitions)) {
|
|
const target = as.definitions.get(name) ?? name;
|
|
const { entry, ref } = entryOf(value);
|
|
const { composition, flagged } = checkComposition(compositionOf(entry), registry, chief ?? EMPTY_CHIEF, { assertClean });
|
|
checked[name] = { target, entry, ref, flagged: chief ? flagged : null, credentialLines: credentialLines(composition.credentials) };
|
|
const path = `definitions.${target}`;
|
|
const current = doc.definitions?.[target];
|
|
let mark;
|
|
if (current === undefined) mark = gone.has(path) ? 'removed' : 'add';
|
|
else mark = canonicalJson(compositionOf(current)) === canonicalJson(compositionOf(entry)) ? 'same' : 'kept';
|
|
const item = { item: 'definition', name, target, path, mark };
|
|
if (mark === 'kept') {
|
|
const before = compositionOf(current), after = compositionOf(entry);
|
|
const primitives = PRIMITIVES.filter(k => canonicalJson(before[k] ?? null) !== canonicalJson(after[k] ?? null));
|
|
item.diff = { primitives, instructions: primitives.includes('instructions') ? lineDiff(before.instructions, after.instructions) : [] };
|
|
}
|
|
if (agent && mark === 'add') item.mark = mark = 'not-written';
|
|
if (mark === 'add') definitions[target] = entry;
|
|
checked[name].added = mark === 'add';
|
|
items.push(item);
|
|
}
|
|
if (Object.keys(definitions).length) plan.definitions = definitions;
|
|
{
|
|
const path = `${prefix}board`;
|
|
let mark;
|
|
if (scope.board === undefined) mark = gone.has(path) ? 'removed' : 'add';
|
|
else mark = canonicalJson(scope.board) === canonicalJson(pack.board) ? 'same' : 'kept';
|
|
items.push({ item: 'board', path, mark });
|
|
if (mark === 'add') plan[path] = structuredClone(pack.board);
|
|
}
|
|
const workflows = {};
|
|
for (const [name, value] of Object.entries(pack.workflows)) {
|
|
const target = as.workflows.get(name) ?? name;
|
|
// The step ids are the pack's own; the organization's workflow keeps title, state and roles.
|
|
// A role naming one of this pack's definitions follows that definition's --as rename; the
|
|
// reserved `human` role never moves. A step whose mapped roles collide refuses here, not at apply.
|
|
const role = r => r !== 'human' && Object.hasOwn(pack.definitions, r) ? as.definitions.get(r) ?? r : r;
|
|
const workflow = { description: value.description, steps: value.steps.map(s => ({ title: s.title, state: s.state, roles: s.roles.map(role) })) };
|
|
if (workflow.steps.some(s => new Set(s.roles).size !== s.roles.length)) bad();
|
|
const path = `${prefix}workflows.${target}`;
|
|
const leftOut = [];
|
|
if (agent) {
|
|
// A step keeps a pack role the organization already defines, or one the pack itself pins to the
|
|
// template of that name. An inline definition that only shares a template's name is left out.
|
|
const unwritable = r => r !== 'human' && Object.hasOwn(pack.definitions, r) && !Object.hasOwn(doc.definitions ?? {}, role(r))
|
|
&& !(checked[r].ref?.template === role(r) && isTemplate(role(r)));
|
|
workflow.steps = workflow.steps.filter((s, i) => {
|
|
const missing = value.steps[i].roles.filter(unwritable).map(role);
|
|
if (missing.length) leftOut.push({ step: i + 1, title: s.title, roles: missing });
|
|
return missing.length === 0;
|
|
});
|
|
}
|
|
const current = scope.workflows?.[target];
|
|
let mark;
|
|
if (agent && workflow.steps.length === 0) mark = 'left-out';
|
|
else if (current === undefined) mark = gone.has(path) ? 'removed' : 'add';
|
|
else mark = canonicalJson(current) === canonicalJson(workflow) ? 'same' : 'kept';
|
|
items.push({ item: 'workflow', name, target, path, mark, ...(leftOut.length ? { leftOut } : {}) });
|
|
if (mark === 'add') workflows[target] = workflow;
|
|
}
|
|
if (Object.keys(workflows).length) plan[`${prefix}workflows`] = workflows;
|
|
// A proposal is filled when an instance by that name exists, or an open hire decision names it.
|
|
const instances = project === null ? doc.roles ?? {} : doc.projects[project].agents ?? {};
|
|
const openHires = context?.openHires ?? [];
|
|
const hires = [];
|
|
const proposedHires = [];
|
|
for (const hire of pack.hires) {
|
|
const d = checked[hire.definition];
|
|
const dropped = Object.hasOwn(instances, hire.name) || openHires.some(h => h.name === hire.name && (h.project ?? null) === project);
|
|
const pin = d.added ? { definition: d.target, rev: 1 } : d.ref ? { ...d.ref } : { inline: d.target, sha256: textSha256(canonicalJson(d.entry)) };
|
|
const marked = [d.target, hire.definition, d.ref?.template, d.entry.from?.template].some(n => n !== undefined && noHire.includes(n));
|
|
hires.push({ name: hire.name, definition: d.target, mark: dropped ? 'dropped' : 'propose', pin, flagged: d.flagged, credentialLines: d.credentialLines, noHire: marked });
|
|
if (!dropped) proposedHires.push({ name: hire.name, reason: hire.reason, pin, ...(Object.hasOwn(pin, 'inline') ? { entry: structuredClone(d.entry) } : {}) });
|
|
}
|
|
plan.packs = { id: pack.id, version: pack.version, sha256: textSha256(canonicalJson(pack)), origin, scope: pack.scope, project, proposedHires };
|
|
scan(plan);
|
|
const notEnabled = pack.skills.filter(name => { try { lstatSync(join(dataRoot, 'skills-enabled', name)); return false; } catch { return true; } });
|
|
const digest = textSha256(canonicalJson({ plan, revision }));
|
|
return {
|
|
pack: { id: pack.id, version: pack.version, label: pack.label, origin, scope: pack.scope }, project, revision,
|
|
items, hires, skills: { notEnabled }, chief: chief !== null, plan, digest,
|
|
};
|
|
}
|
|
// context.beforeWrite (SP5) is the broker's recheck of an agent's organization.change grant,
|
|
// run once more before the writer commits.
|
|
const write = (id, plan, revision, actor, detail, context) => {
|
|
context?.beforeWrite?.();
|
|
const out = writer.applyPack(id, plan, { revision, actor, assertClean, detail });
|
|
return { revision: out.revision, added: out.added };
|
|
};
|
|
const scoped = (args, key) => {
|
|
if (args.project === undefined) return key;
|
|
if (typeof args.project !== 'string' || !/^[a-z0-9][a-z0-9._-]{0,63}$/.test(args.project)) bad();
|
|
return `projects.${args.project}.${key}`;
|
|
};
|
|
const revisionOf = args => { if (typeof args.revision !== 'string' || !/^[0-9a-f]{64}$/.test(args.revision)) bad(); return args.revision; };
|
|
// Ruling 2026-11-14 ("If I choose a starter pack and apply it, that is a decision!"): the
|
|
// organization facts the broker's pack hire loop and the packs read need, as the file stands.
|
|
// `pack` names a packs entry the way apply names the pack: saved:NAME is the owner's saved
|
|
// template, any other name the latest shipped or installed entry of that id. The route is a
|
|
// resolved harness, model and thinking (system, organization and agent vars): the chief's (the
|
|
// launch.by instance), else the org assistant's (lane pr2, lead ruling 2026-11-14), else null.
|
|
// routeFrom names which one gave it.
|
|
const latestEntries = doc => {
|
|
const latest = new Map();
|
|
for (const e of doc.packs ?? []) if (isObject(e) && typeof e.id === 'string') latest.set(canonicalJson([e.id, e.origin ?? null, e.project ?? null]), e);
|
|
return [...latest.values()];
|
|
};
|
|
const agentsOf = doc => {
|
|
const launched = doc.launch?.instances ?? [];
|
|
const one = (entry, isLaunched) => ({ hireDecision: typeof entry?.hireDecision === 'string' ? entry.hireDecision : null, launched: isLaunched });
|
|
return {
|
|
org: Object.fromEntries(Object.entries(doc.roles ?? {}).map(([name, e]) => [name, one(e, launched.includes(name))])),
|
|
projects: Object.fromEntries(Object.entries(doc.projects ?? {}).filter(([, p]) => isObject(p) && p.kind !== 'library')
|
|
.map(([project, p]) => [project, Object.fromEntries(Object.entries(p.agents ?? {}).map(([name, e]) => [name, one(e, true)]))])),
|
|
};
|
|
};
|
|
// The org assistant is the instance whose definition is `assistant` (broker.mjs continuationEffect);
|
|
// with several, the one named `assistant`, else the first in key order.
|
|
const packRoute = (doc, id) => {
|
|
const roles = doc.roles ?? {};
|
|
let loaded = null;
|
|
const routeOf = instance => {
|
|
loaded ??= validateBusinessDocument(doc, `${id}.json`, { rolesDir, dataRoot });
|
|
const { vars } = resolveInstance({ system, business: loaded, instance });
|
|
const pick = k => (typeof vars?.[k] === 'string' && vars[k] ? vars[k] : null);
|
|
if (!pick('harness') || !pick('model')) return null;
|
|
return { harness: pick('harness'), model: pick('model'), ...(pick('thinking') ? { thinking: pick('thinking') } : {}) };
|
|
};
|
|
const by = doc.launch?.by;
|
|
const chief = typeof by === 'string' && Object.hasOwn(roles, by) ? routeOf(by) : null;
|
|
if (chief) return { route: chief, routeFrom: 'chief' };
|
|
const assistants = Object.keys(roles).filter(name => isObject(roles[name]) && roles[name].definition === 'assistant');
|
|
const assistant = assistants.includes('assistant') ? 'assistant' : assistants[0];
|
|
const route = assistant === undefined ? null : routeOf(assistant);
|
|
return route ? { route, routeFrom: 'assistant' } : { route: null, routeFrom: null };
|
|
};
|
|
const applied = doc => latestEntries(doc).map(e => ({ id: e.id, origin: e.origin ?? null, project: e.project ?? null, version: e.version ?? null,
|
|
proposedHires: (Array.isArray(e.proposedHires) ? e.proposedHires : []).filter(isObject).map(h => ({ name: h.name, reason: h.reason ?? null })) }));
|
|
return {
|
|
// Shipped and installed packs only: a saved template is its owner's, read by name. `applied` is
|
|
// the latest packs entry of each id, origin and project, and `agents` each instance's hire record.
|
|
packs: async (id) => {
|
|
const doc = writer.read(id).document;
|
|
return { packs: (await installed()).map(({ id: pid, origin, pack }) => ({ id: pid, origin, pack })), applied: applied(doc), agents: agentsOf(doc) };
|
|
},
|
|
'pack.state': (id, args) => {
|
|
if (typeof args.pack !== 'string' || !args.pack) bad();
|
|
const project = args.project ?? null;
|
|
if (project !== null && (typeof project !== 'string' || !/^[a-z0-9][a-z0-9._-]{0,63}$/.test(project))) bad();
|
|
const { document: doc, revision } = writer.read(id);
|
|
const saved = args.pack.startsWith(SAVED_PREFIX);
|
|
const packId = saved ? args.pack.slice(SAVED_PREFIX.length) : args.pack;
|
|
const entry = (doc.packs ?? []).filter(e => isObject(e) && e.id === packId && (e.project ?? null) === project && (saved ? e.origin === 'saved' : e.origin !== 'saved')).at(-1) ?? null;
|
|
const scope = project === null ? doc.roles ?? {} : doc.projects?.[project]?.agents ?? {};
|
|
return { revision, entry: entry ? structuredClone(entry) : null, ...packRoute(doc, id), names: Object.keys(scope), noHire: [...(doc.launch?.noHire ?? [])] };
|
|
},
|
|
'pack.preview': (id, args, actor, context) => build(id, args, actor, context),
|
|
'pack.apply': async (id, args, actor, context) => {
|
|
if (typeof args.digest !== 'string') bad();
|
|
const preview = await build(id, args, actor, context);
|
|
if (preview.digest !== args.digest) throw new BusError('pack-plan-stale');
|
|
// An apply of the same pack (same sha256) that adds nothing and proposes what the last apply of
|
|
// it proposed writes nothing, so a retry loop does not fill the packs list to its cap. A changed
|
|
// pack still records its entry. The digest pins the revision.
|
|
const { plan } = preview;
|
|
const last = (writer.read(id).document.packs ?? []).filter(e => sameApply(e, plan.packs.id, plan.packs.origin, plan.packs.project)).at(-1);
|
|
if (Object.keys(plan).every(k => k === 'packs') && last && last.sha256 === plan.packs.sha256
|
|
&& canonicalJson(last.proposedHires ?? []) === canonicalJson(plan.packs.proposedHires))
|
|
return { revision: preview.revision, added: [], pack: preview.pack.id };
|
|
return { ...write(id, plan, preview.revision, actor, {}, context), pack: preview.pack.id };
|
|
},
|
|
'board.set': (id, args, actor, context) => {
|
|
if (args.board !== null && !isObject(args.board)) bad();
|
|
return write(id, { [scoped(args, 'board')]: args.board }, revisionOf(args), actor, {}, context);
|
|
},
|
|
'workflow.set': (id, args, actor, context) => {
|
|
if (typeof args.name !== 'string' || (args.workflow !== null && !isObject(args.workflow))) bad();
|
|
return write(id, { [scoped(args, 'workflows')]: { [args.name]: args.workflow } }, revisionOf(args), actor, {}, context);
|
|
},
|
|
};
|
|
}
|
|
|
|
export async function startBroker({
|
|
dataRoot,
|
|
instanceRoot = null,
|
|
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,
|
|
org = null, instanceDataRoot = null, instanceConfigDir = null,
|
|
// Host-only preparation/admission capabilities. The host builds its own from the organization writer and `trackers`
|
|
// (request-host.mjs); anything given here replaces the host's, and a capability nothing supplies leaves a setup wait.
|
|
requestServices = {},
|
|
// The boot config's plain-data tracker settings, { business: { project, ... } }. The tasks adapter reads its own copy;
|
|
// the host's new-project preparation reads each business's tracker root board from this one.
|
|
trackers = null,
|
|
builtinBinding = null,
|
|
modelCatalogue = null, // in-process fixture seam; boot JSON cannot supply a service
|
|
makeProviderLoader = createLoader, // in-process fixture seam; boot JSON cannot supply a function
|
|
// #1873 A3: the boot's root and project agent copies (cli config.mjs copiesAtBoot), plain data. Every
|
|
// reload of the organization file below re-applies this snapshot, never the root file.
|
|
copies = null,
|
|
// #1873 CB3: the tests' Gitea transport and CA for the copy launch gate and copyGrant, { send, ca }.
|
|
// In-process only, like librarySync; production passes none.
|
|
copyTransport = null,
|
|
}) {
|
|
if (copies !== null && (typeof copies !== 'object' || Array.isArray(copies) || typeof copies.org !== 'string'
|
|
|| !Object.hasOwn(businesses ?? {}, copies.org) || copies.copies === null || typeof copies.copies !== 'object' || Array.isArray(copies.copies)))
|
|
throw new BusError('invalid-host-request');
|
|
const loadOrg = id => loadWithCopies(id, { dir: organization.dir, rolesDir: organization.rolesDir, dataRoot:instanceRoot ?? dataRoot, snapshot: copies?.org === id ? copies : 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, proxyLock = null, proxyPort = null;
|
|
let modelKeys = null, modelTimer = null, modelSweep = null, modelCatalogues = null;
|
|
const modelRuns = new Map();
|
|
let store, credentials, server, adapter, processing, syncTimer = null, syncPass = null, accessTimer = null, requestTimer = null, copyLaunch = null, copyRevoker = null, copyTimer = null;
|
|
// #1873 CB4: a revoke's stop waits for the broker; a start that fails before it exists stops nothing.
|
|
let brokerUp = null;
|
|
const brokerReady = new Promise((resolve) => { brokerUp = resolve; });
|
|
try {
|
|
// Lifecycle Task 2 (#797): a product whose mapping is published but not yet activated is not part of
|
|
// this process. A managed product is part of it only with both a committed activate receipt in the
|
|
// request journal and a complete marker on disk (#802: a receipt alone, with the marker gone or never
|
|
// published, is not enough); every other managed entry waits for the bootstrap executor's
|
|
// activateManagedProject, so the broker, the Credentials roots and the boot copy never see a
|
|
// half-built product. `businesses` becomes this process's own copy.
|
|
store = new Store(dataRoot);
|
|
const requestJournal = createRequestStore(store);
|
|
const activatedProjects = business => new Set(requestJournal.committed(business, 'bootstrap')
|
|
.filter(r => r.op.endsWith(':activate') && typeof r.result?.project === 'string').map(r => r.result.project));
|
|
const pendingProject = (business, project, entry, activated) => {
|
|
if (entry?.kind === 'library') return false;
|
|
const state = managedState({ dataRoot, business, project, entry });
|
|
return state !== 'unmanaged' && !(state === 'complete' && activated.has(project));
|
|
};
|
|
const withoutPending = (business, b) => {
|
|
const activated = activatedProjects(business);
|
|
const projects = Object.fromEntries(Object.entries(b.projects ?? {}).filter(([project, entry]) => !pendingProject(business, project, entry, activated)));
|
|
return Object.keys(projects).length === Object.keys(b.projects ?? {}).length ? b : { ...b, projects };
|
|
};
|
|
businesses = Object.fromEntries(Object.entries(businesses).map(([id, b]) => [id, withoutPending(id, b)]));
|
|
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;
|
|
let requestLifecycle;
|
|
// 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');
|
|
if (org !== null) {
|
|
if (!org || typeof org.id !== 'string' || !/^[a-z0-9][a-z0-9._-]{0,63}$/.test(org.id) || typeof instanceDataRoot !== 'string' || typeof instanceConfigDir !== 'string') throw new BusError('invalid-host-request');
|
|
proxyLock = createProxyLock({ stateDir: join(instanceDataRoot, 'proxy'), identity: `${instanceConfigDir}:${proxy.port}` });
|
|
}
|
|
modelProxyClient = await loadManagementClient({ configDir: org ? instanceConfigDir : organization.dir, proxy, receipt: proxyReceipt });
|
|
proxyPort = proxy.port;
|
|
}
|
|
|
|
// 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: [] };
|
|
// #1873 CB2 (security review P2): a copy slot an issue staged before a crash, whose binding holds
|
|
// no grant for it, goes before Credentials loads and before any issue can run.
|
|
if (storeRoot) {
|
|
const reader = organizationWriter({ dir: organization.dir, rolesDir: organization.rolesDir, dataRoot: instanceRoot ?? dataRoot });
|
|
const documents = Object.keys(businesses).map((id) => {
|
|
try {
|
|
return [id, reader.read(id).document];
|
|
} catch {
|
|
return [id, null];
|
|
}
|
|
});
|
|
for (const entry of sweepCopySlots(storeRoot, documents, references))
|
|
process.stderr.write(`copies: ${JSON.stringify({ event: 'copy-slot-swept', ...entry })}\n`);
|
|
// #1873 CB4: a named instance shares its slot path with a former copy of the same name. While
|
|
// that copy's record is not a finished tombstone, the slot is held (never loaded as the named
|
|
// instance's) until the shadowed revoke discards it.
|
|
for (const [id, doc] of documents)
|
|
for (const [name, r] of Object.entries(doc?.copyBindings ?? {}))
|
|
if (Object.hasOwn(doc.roles ?? {}, name) && !(r?.state === 'unbound' && r.revoke?.done === true))
|
|
for (const service of new Set([...COPY_BINDABLE, ...Object.keys(r?.services ?? {})])) held.push(id + '/' + name + ':' + service);
|
|
}
|
|
// #1886 (F3 defect l): a store reference with no file yet reads as no key (design D3a), but Credentials
|
|
// refuses a missing store root, and only issue() made it. Start makes it the same way issue() does when
|
|
// any reference names the store. A root that exists with loose modes or as a symlink still refuses.
|
|
if (storeRoot && Object.values(references).some(refs => Object.values(refs ?? {}).some(ref => ref?.store === true)))
|
|
mkdirSync(storeRoot, { recursive: true, mode: 0o700 });
|
|
credentials = new Credentials({ references, repoRoots: protectedRoots, dataRoot, storeRoot, held,
|
|
modelAgents: proxy ? Object.entries(businesses).flatMap(([id, b]) => Object.keys(b.roles).map(role => id + '/' + role)) : [] });
|
|
// #1873 T1 follow-up (org-defscan): no organization definition's text holds a credential this store
|
|
// holds. The boot copy's org: definitions first, then every entry of the file itself, named or not.
|
|
// A hit refuses the start; the refusal names the definition and key, never the text.
|
|
const clean = value => credentials.assertClean(value);
|
|
for (const [id, b] of Object.entries(businesses)) {
|
|
scanDefinitions(b.definitions, clean, { prefix: ORG_PREFIX });
|
|
if (!organization) continue;
|
|
const raw = organizationWriter({ dir: organization.dir, rolesDir: organization.rolesDir, dataRoot: instanceRoot ?? dataRoot }).readRaw(id);
|
|
if (raw.readable) scanDefinitions(raw.document.definitions, clean);
|
|
}
|
|
// The boot copy snapshot's definitions (the root agents file's, A4): every reload re-applies them.
|
|
if (copies?.definitions !== undefined) scanDefinitions(copies.definitions, clean, { boot: true });
|
|
// #1873 CB3: each unbound copy whose binding is bound is re-resolved and its slot proved live at
|
|
// the service; only those that pass join the organization, now, before anything takes a copy of
|
|
// `businesses`. A binding approved later answers restart-required. copyGrant keeps the boot's own
|
|
// snapshot (bootCopies), and so does the gate: what it promotes is decided at boot only.
|
|
const bootCopies = copies;
|
|
// #1873 CB4: the revoke of a copy's binding, for every organization this bus serves. The gate (one
|
|
// organization's) closes and drains the copy; the broker stops its runs once it is up.
|
|
if (organization && storeRoot) {
|
|
const revokeWriter = organizationWriter({ dir: organization.dir, rolesDir: organization.rolesDir, dataRoot: instanceRoot ?? dataRoot });
|
|
const grant = giteaGrant({ configDir: organization.dir, ...(copyTransport ? { transport: copyTransport.send, ca: copyTransport.ca } : {}) });
|
|
copyRevoker = makeCopyRevoker({
|
|
writer: revokeWriter, credentials,
|
|
gate: (id) => (copyLaunch && bootCopies?.org === id ? copyLaunch : null),
|
|
stopRuns: async (id, instance, opts) => {
|
|
const b = await brokerReady;
|
|
return b ? b.stopCopyRuns(id, instance, opts) : { confirmed: false };
|
|
},
|
|
remote: giteaCopyRemote({ grant, read: (id) => revokeWriter.read(id).document, incarnation: (id) => readOrgEntrySync(organization.dir, id)?.incarnation ?? null }),
|
|
definitions: (id) => (bootCopies?.org === id ? bootCopies.definitions ?? {} : {}),
|
|
});
|
|
}
|
|
if (organization && storeRoot && copies) {
|
|
copyLaunch = copyLaunchGate({ credentials, writer: organizationWriter({ dir: organization.dir, rolesDir: organization.rolesDir, dataRoot: instanceRoot ?? dataRoot }),
|
|
configDir: organization.dir, rolesDir: organization.rolesDir, snapshot: copies, ...(copyTransport ? { transport: copyTransport.send, ca: copyTransport.ca } : {}),
|
|
onLoss: (id, instance, cause) => copyRevoker?.start(id, instance, { cause, actor: 'bus' }) });
|
|
await copyLaunch.boot(copies.org);
|
|
const verified = copyLaunch.bound();
|
|
if (Object.keys(verified).length) {
|
|
copies = promoteCopies(copies, verified);
|
|
const b = loadOrg(copies.org);
|
|
const resolved = {};
|
|
for (const instance of Object.keys(b.roles)) resolved[instance] = resolveInstance({ system: organization.system, business: b, instance });
|
|
businesses[copies.org] = withoutPending(copies.org, busOrganization(b, resolved));
|
|
if (proxy) for (const instance of Object.keys(verified)) if (Object.hasOwn(businesses[copies.org].roles, instance)) credentials.registerModel(copies.org + '/' + instance);
|
|
}
|
|
}
|
|
if (proxy) {
|
|
const inventory = modelRunInventory(dataRoot, modelRuns, credentials);
|
|
modelKeys = new ModelKeys({ credentials, client: modelProxyClient, configDir: org ? instanceConfigDir : organization.dir, org: org?.id ?? null, lock: proxyLock,
|
|
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(org ? instanceConfigDir : organization.dir).slice(0, 12), org: org?.id ?? null, lock: proxyLock,
|
|
// P2 supplies the model metadata; only the matching business's record
|
|
// participates in this owner view. No credential value is read here.
|
|
onChange: phase => { if (phase === 'begin') modelCatalogues?.begin?.(); else if (phase === 'end') modelCatalogues?.end?.(); else modelCatalogues?.invalidate(); },
|
|
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,dataRoot:instanceRoot ?? dataRoot,exists:(kind,name)=>hasModule(kind,name,instanceRoot ?? dataRoot)}):null;
|
|
const levels=organization?await harnessLevels({dataRoot:instanceRoot ?? dataRoot}):()=>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;
|
|
modelCatalogues = modelCatalogue ?? (settings ? createModelCatalogue({ pluginDataRoot:instanceRoot ?? dataRoot, client: () => modelProxyClient, settings: id => settings.current(id).plugins ?? {}, proxyLock }) : null);
|
|
// Resolve legacy table entries with the enabled modules' declarations. A
|
|
// missing or ambiguous mapping never becomes an available route.
|
|
async function catalogueFor(id, options = {}) {
|
|
const snapshot = settings.writer.read(id);
|
|
const current = settings.current(id);
|
|
const names = current.plugins?.['model-provider'] ?? [];
|
|
const loader = makeProviderLoader({dataRoot:instanceRoot ?? dataRoot,log:()=>{}});
|
|
const loaded = await loader.load({'model-provider': names});
|
|
const catalogue = await modelCatalogues.read(id, options);
|
|
if (loaded.refused.length || settings.writer.read(id).revision !== snapshot.revision) return {...catalogue, status:'unavailable', legacy:{}};
|
|
const modules = loader.snapshot('model-provider');
|
|
const resolveLegacy = entry => {
|
|
const selected = selectModelProvider(modules, entry);
|
|
if (!selected) return null;
|
|
let result; try { result = selected.module.route({selection:entry,harness:null}); } catch { return null; }
|
|
return typeof result?.upstream === 'string' && /^[a-z0-9][a-z0-9._-]{0,63}$/.test(result.upstream)
|
|
? {module:selected.name,upstream:result.upstream,model:entry.model} : null;
|
|
};
|
|
const legacy = {};
|
|
for (const [key, entry] of Object.entries(current.launch?.models ?? {})) {
|
|
if (!entry.module) {
|
|
const mapping = resolveLegacy(entry);
|
|
if (mapping) legacy[key] = mapping;
|
|
}
|
|
}
|
|
const result = {...catalogue, legacy};
|
|
// This trusted in-process helper is never serialized into the client view.
|
|
Object.defineProperty(result, 'resolveLegacy', {value:resolveLegacy});
|
|
return result;
|
|
}
|
|
// 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, dataRoot:instanceRoot ?? dataRoot }) }) : 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, pluginDataRoot:instanceRoot ?? dataRoot, businesses, log: usageLog, keyForRun,
|
|
proxyView: business => proxyUsage?.view(business) ?? null, keyOrphans: business => orphanedKeys(business) });
|
|
proxyUsage = createProxyUsage({ store, pluginDataRoot:instanceRoot ?? dataRoot, 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, pluginDataRoot:instanceRoot ?? dataRoot, 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;
|
|
const recordArgs = context => ({ ...context, ...projectModule.recordContext({ ...context, dataRoot: instanceDataRoot ?? dataRoot, orgRoot: org ? dataRoot : 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)]));
|
|
// #1873 CB3: a copy the boot snapshot had unbound (whether the gate promoted it or not).
|
|
const copyNeedsAuthority = (id, instance) => copyLaunch !== null && bootCopies?.org === id
|
|
&& Object.hasOwn(bootCopies.copies, instance) && bootCopies.copies[instance].state === 'unbound';
|
|
const reloadLog = entry => process.stderr.write(`organization: ${JSON.stringify(entry)}\n`);
|
|
function reload(id) {
|
|
try {
|
|
const b = loadOrg(id);
|
|
const resolved = {};
|
|
for (const instance of Object.keys(b.roles)) resolved[instance] = resolveInstance({ system: organization.system, business: b, instance });
|
|
const next = withoutPending(id, busOrganization(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;
|
|
}
|
|
}
|
|
// Lifecycle Task 2 (#797): the only way a project joins a running bus. The bootstrap executor calls it
|
|
// synchronously after every earlier step passed, the complete marker is published and the activate receipt commits. It admits exactly the
|
|
// receipt-backed addition under the organization lock: the file must still be this business plus that entry,
|
|
// with the credentials, bots, libraries, roles and prior mappings the bus booted with. Anything else leaves
|
|
// the product pending. Both readiness records must exist before even the broker's reload begins.
|
|
const writer = organization ? organizationWriter({ dir: organization.dir, rolesDir: organization.rolesDir, dataRoot:instanceRoot ?? dataRoot }) : null;
|
|
const known = new Map(Object.entries(businesses).map(([id, b]) => [id, canonicalJson(b.projects ?? {})]));
|
|
function activateManagedProject({ business: id, project, receipt, organization_revision }) {
|
|
if (!writer || !booted.has(id) || !Object.hasOwn(businesses, id)) throw new BusError('unknown-business');
|
|
if (Object.hasOwn(businesses[id].projects ?? {}, project)) return; // already admitted: activation is idempotent
|
|
const refuse = code => { throw new BusError(code); };
|
|
writer.hold(id, ({ revision }) => {
|
|
const found = writer.findAddition(id, receipt);
|
|
if (!found || found.after !== organization_revision) refuse('bootstrap-receipt-mismatch');
|
|
// Only this bootstrap's own post-add revision may be admitted: any later organization write means the confirmed world moved (#800).
|
|
if (revision !== found.after) refuse('bootstrap-revision-stale');
|
|
const b = loadOrg(id);
|
|
const resolved = {};
|
|
for (const instance of Object.keys(b.roles)) resolved[instance] = resolveInstance({ system: organization.system, business: b, instance });
|
|
const full = busOrganization(b, resolved);
|
|
const entry = full.projects?.[project];
|
|
const here = managedPaths({ dataRoot, business: id, project });
|
|
if (!entry || entry.kind === 'library' || entry.root !== here.root || entry.repo !== here.repo
|
|
|| Object.keys(entry).some(k => !['root', 'repo', 'remote'].includes(k))) refuse('bootstrap-mapping-mismatch');
|
|
if (managedState({ dataRoot, business: id, project, entry }) !== 'complete') refuse('bootstrap-marker-incomplete');
|
|
if (!activatedProjects(id).has(project)) refuse('bootstrap-receipt-mismatch');
|
|
const next = withoutPending(id, full);
|
|
const { [project]: _added, ...before } = next.projects;
|
|
if (canonicalJson(before) !== known.get(id)) refuse('bootstrap-mapping-mismatch');
|
|
const expected = { ...booted.get(id), roots: canonicalJson([...JSON.parse(booted.get(id).roots), entry.root].sort()) };
|
|
if (!sameFixed(expected, next)) refuse('bootstrap-restart-required');
|
|
broker.reloadBusiness(id, next);
|
|
protectedRoots.push(entry.root);
|
|
businesses[id] = next;
|
|
booted.set(id, expected);
|
|
known.set(id, canonicalJson(next.projects));
|
|
});
|
|
}
|
|
// Host IPC request.admission: may a project's code checkout be cloned and its tools used. A project the
|
|
// broker does not hold is never ready, and a declared one that is not yet held says it is pending. A managed
|
|
// product the broker holds is ready only while it has a complete marker and a committed activate receipt: the
|
|
// answer is read from both on every call, so a marker removed after the activation closes it again, and the
|
|
// broker's reload now follows both readiness records (#802).
|
|
// With `instance` (#1873 A3), the launch boundary also asks whether that instance may work in each
|
|
// project as the business stands now: a project that excludes it, or whose union refuses, is not ready.
|
|
// `instance` comes with `revision`, the organization file revision the launch resolved its bundle
|
|
// from; a file that has changed since (a narrow, a project define's vars, an exclude) is not ready.
|
|
// #1873 CB3: with `instance`, a copy the boot left needing a binding is admitted only on
|
|
// `authority`, the launch authority digest its prepare returned, still current now. Checked
|
|
// before the projects, so a launch with no project is held to it as well.
|
|
function admission({ business: id, project, projects, instance, revision: launched, thread, session, authority }) {
|
|
if (projects !== undefined) {
|
|
const no = code => ({ ready: false, revision: null, code });
|
|
if (typeof id !== 'string' || !Object.hasOwn(businesses, id) || !Array.isArray(projects) || projects.some(p => typeof p !== 'string')) return no('project-unknown');
|
|
if (thread != null) {
|
|
const row = broker.threadsState(id).find(r => r.thread === thread && r.session === session);
|
|
if (!row) return no('session-not-found');
|
|
if (row.archived) return no('project-archived');
|
|
}
|
|
if (instance !== undefined && (typeof instance !== 'string' || typeof launched !== 'string')) return no('project-unknown');
|
|
if (instance !== undefined) {
|
|
let now;
|
|
try { now = writer?.read(id); } catch { return no('organization-unreadable'); }
|
|
if (now?.revision === undefined) return no('organization-unreadable');
|
|
if (now.revision !== launched) return no('organization-changed');
|
|
// org-defscan (#1873): the launch boundary. A credential issued since the launcher's scan
|
|
// runs on this same loop, so the org: definition the instance names is checked again here,
|
|
// in the reply the launcher spawns after with no further wait.
|
|
const name = orgDefinitionName(now.document.roles?.[instance]?.definition);
|
|
if (name !== null) {
|
|
const clean = value => credentials.assertClean(value);
|
|
let held;
|
|
try { clean(name); held = heldPath(now.document.definitions?.[name], clean); } catch { held = 'name'; }
|
|
if (held !== null) return no(DEFINITION_HOLDS_CREDENTIAL);
|
|
}
|
|
if (copyNeedsAuthority(id, instance)) {
|
|
try { copyLaunch.admit(id, instance, authority); } catch (e) { return no(e?.code ?? 'copy-authority-changed'); }
|
|
}
|
|
}
|
|
for (const p of projects) {
|
|
const result = admission({ business: id, project: p });
|
|
if (!result.ready) return result;
|
|
if (instance !== undefined) {
|
|
// The broker holds the document the last writer reload swapped in; `businesses` is the boot one.
|
|
const barred = broker.projectBarred(id, instance, p);
|
|
if (barred) return no(barred);
|
|
}
|
|
}
|
|
return { ready: true, revision: null, code: null };
|
|
}
|
|
const no = code => ({ ready: false, revision: null, code });
|
|
if (typeof id !== 'string' || typeof project !== 'string' || !Object.hasOwn(businesses, id)) return no('project-unknown');
|
|
if (projectArchived(store, id, project)) return no('project-archived');
|
|
let revision = null, declared = false;
|
|
try {
|
|
if (writer) {
|
|
const now = writer.read(id);
|
|
revision = now.revision; declared = Object.hasOwn(now.document.projects ?? {}, project);
|
|
}
|
|
} catch { return no('organization-unreadable'); }
|
|
if (Object.hasOwn(businesses[id].projects ?? {}, project)) {
|
|
try { if (pendingProject(id, project, businesses[id].projects[project], activatedProjects(id))) return no('project-pending'); }
|
|
catch { return no('project-pending'); }
|
|
return { ready: true, revision, code: null };
|
|
}
|
|
return no(declared ? 'project-pending' : 'project-unknown');
|
|
}
|
|
const requirementsHost = createRequirementsHost({ store, businesses, builtinBinding, journal: requestJournal });
|
|
// U4a (#1873, REQ-TASK-1, REQ-PLUG-1): the task-tracker module each business uses, fixed for this
|
|
// start. The organization file's "task-tracker" plug-in names it. With none, a business the boot
|
|
// config gives a Vikunja tracker keeps Vikunja (so does every business when an in-process caller
|
|
// hands `tasks` and no `trackers`), and any other business gets the built-in store. A business that
|
|
// names another module while the boot config also gives it a Vikunja tracker refuses the boot:
|
|
// two trackers can't both hold its tasks.
|
|
const trackerOf = new Map();
|
|
for (const [id, b] of Object.entries(businesses)) {
|
|
const named = b?.plugins?.['task-tracker'];
|
|
const tracked = trackers !== null && typeof trackers === 'object' && Object.hasOwn(trackers, id);
|
|
if (named !== undefined && named !== 'vikunja' && tracked) throw new BusError('tracker-config');
|
|
trackerOf.set(id, named ?? ((trackers ? tracked : tasks !== null) ? 'vikunja' : 'store'));
|
|
}
|
|
broker = new Broker({
|
|
pluginService: pluginService({dataRoot:instanceRoot ?? dataRoot,read:id=>organization ? loadOrg(id) : businesses[id]}),
|
|
taskTracker: (id) => trackerOf.get(id),
|
|
workspaceRead: state => requestWorkspace({ dataRoot, state, events: store.all("SELECT kind,actor_role,subject,body FROM events WHERE business=? AND kind IN ('task.created','task.assigned','session.launched') ORDER BY seq LIMIT 10001", state.business) }),
|
|
requirementsHost,
|
|
requestLifecycle: services => {
|
|
// The PRD reader goes through the same record resolver as the file adapters, so both roots present refuses here too.
|
|
const readCandidate = async (candidate, business) => {
|
|
const { request, project, prd_version, records_commit, prd_path, mode_revision } = candidate;
|
|
const records = recordArgs({ business: services.prdBusiness(business), project });
|
|
return readPrdMaterial({ business: records.business, dataRoot: records.dataRoot, request, project, prd_version, records_commit, prd_path, mode_revision,
|
|
requirement_ids: candidate.requirement_ids ?? candidate.requirements.map(r => r.id) });
|
|
};
|
|
const catalog = createRequirementCatalog({ journal: { ...requestJournal, readCandidate, authority: services.prdAuthority, approval: services.prdApproval },
|
|
readSpec: ({ business, project }) => projectModule.readSpec(recordArgs({ business: services.prdBusiness(business), project })),
|
|
builtinBinding: services.builtinBinding, projectArchived: services.projectArchived, threadProject: services.threadProject });
|
|
return (requestLifecycle = createRequestLifecycle({ journal: requestJournal, services: { catalog, readCandidate, readReviewedArtifact: (state, commit, path) => readReviewedArtifact({ repo: workingRepo(dataRoot, state.business, state.project), commit, path }), readVerification: (state, record) => projectModule.readProjectFile(recordArgs({ business: services.prdBusiness(state.business), project: state.project, path: record.path, commit: record.commit })), requestAdmission: state => admission({ business: state.business, project: state.project }), ...(writer ? hostPreparation({ dataRoot, organization: writer, trackers, journal: requestJournal, catalog, admission, work: fn => processing.work(fn), readSpec: ({ business, project }) => projectModule.readSpec(recordArgs({ business: services.prdBusiness(business), project })) }) : {}), ...requestServices, ...services,
|
|
// The new-project executor needs the organization writer; without one a confirmed selection stays at its setup wait.
|
|
...(writer ? { bootstrap: ({ binding, state }) => processing.work(() => executeBootstrap({ binding, owner: state.owner, journal: requestJournal, services: {
|
|
dataRoot, organization: writer, activate: activateManagedProject, barrier: requestServices.barrier, projectArchived: services.projectArchived, threadProject: services.threadProject,
|
|
chat: { state: args => broker.chatProjectState(args), associate: args => broker.associateChat(args) } } })) } : {}) } }));
|
|
},
|
|
// 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,
|
|
// #1873 CB2: credentials.issue for an agent copy's bound Gitea service, verified live at the server.
|
|
copyGrant: organization && storeRoot
|
|
? copyGrant({ credentials, writer: organizationWriter({ dir: organization.dir, rolesDir: organization.rolesDir, dataRoot: instanceRoot ?? dataRoot }),
|
|
copies: bootCopies, organizations: () => Object.keys(businesses), ...(copyTransport ? { transport: copyTransport.send, ca: copyTransport.ca } : {}) })
|
|
: null,
|
|
copyLaunch,
|
|
copyRevoker,
|
|
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, dataRoot:instanceRoot ?? dataRoot }).readRaw(id) : { readable: false }),
|
|
last: (context, paths) => libraryModule.lastCommits({ ...context, dataRoot }, paths),
|
|
GuardError: libraryModule.GuardError,
|
|
},
|
|
project: {
|
|
readSpec: context => projectModule.readSpec(recordArgs(context)),
|
|
writeRecord: context => processing.work(() => projectModule.writeRecord(recordArgs(context))),
|
|
listFiles: context => projectModule.listProjectFiles(recordArgs(context)),
|
|
readFile: context => projectModule.readProjectFile(recordArgs(context)),
|
|
writeFile: context => projectModule.writeHumanFile(recordArgs(context)),
|
|
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 === 'project.remote.get.console' || verb === 'project.remote.set.console') {
|
|
const current = loadOrg(id);
|
|
const resolved = {};
|
|
for (const instance of Object.keys(current.roles)) resolved[instance] = resolveInstance({ system: organization.system, business: current, instance });
|
|
const live = withoutPending(id, busOrganization(current, resolved));
|
|
if (!sameFixed(booted.get(id), live) || canonicalJson(live.projects) !== known.get(id)) throw new BusError('project-remote-protected');
|
|
if (!Object.hasOwn(live.projects, args.project)) throw new BusError('project-remote-unknown');
|
|
if (verb === 'project.remote.get.console') return getProjectRemote(writer, id, args.project);
|
|
writer.setProjectRemote(id, args, actor);
|
|
const next = structuredClone(live);
|
|
next.projects[args.project].remote = args.remote;
|
|
broker.reloadBusiness(id, next);
|
|
businesses[id] = next;
|
|
known.set(id, canonicalJson(next.projects));
|
|
return getProjectRemote(writer, id, args.project);
|
|
}
|
|
// #1873 SP4: the human's pack, board and workflow ops. The broker passes only the business's human;
|
|
// the reads return here, and each write reloads the organization and refreshes its limit checks.
|
|
// SP5: or an organization.change holder's actor, with context.agent and context.beforeWrite.
|
|
if (Object.hasOwn(PACK_OPS, verb)) {
|
|
const ops = packOps({ writer: organizationWriter({dir:organization.dir,rolesDir:organization.rolesDir}), rolesDir: organization.rolesDir, dataRoot: instanceRoot ?? dataRoot,
|
|
system: organization.system, registryFor: doc => organizationPlugins(instanceRoot ?? dataRoot, doc.plugins).registry, assertClean: value => credentials.assertClean(value), configDir: organization.dir });
|
|
const out = ops[verb](id, args, actor, context);
|
|
if (!PACK_OPS[verb]) return out;
|
|
const committed = value => { reload(id); limits.refresh(id).catch(e=>limitLog({event:'limit-refresh-failed',business:id,message:e.message})); return value; };
|
|
return out?.then ? out.then(committed) : committed(out);
|
|
}
|
|
// #1873 SP8: saving a template and listing the owner's own. The broker passes only the owner's
|
|
// human; both read the organization file and neither writes it, so nothing reloads.
|
|
if (verb === 'template.save' || verb === 'template.list') {
|
|
const base = { rolesDir: organization.rolesDir, configDir: organization.dir, account: actor };
|
|
if (verb === 'template.list') return templateList(base);
|
|
const { name, project, preview, digest } = args;
|
|
const document = organizationWriter({dir:organization.dir,rolesDir:organization.rolesDir}).read(id).document;
|
|
const assertClean = value => credentials.assertClean(value);
|
|
if (preview === true && digest === undefined) return templatePreview({ document, name, project, rolesDir: organization.rolesDir, assertClean });
|
|
if (preview === undefined && typeof digest === 'string') return templateSave({ ...base, document, name, project, digest, assertClean });
|
|
throw new TemplateError('invalid-request', 'template.save takes preview: true or the preview digest');
|
|
}
|
|
if(verb==='limits')return withTokens(id,settings.read(id));
|
|
if(verb==='plugins')return {...plugins.read(id), available:{harness:levels.names ?? []}};
|
|
// 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') {
|
|
const view = settings.routes(id);
|
|
if (!settings.current(id).plugins?.['model-provider']?.length) return { ...view, catalogue: { status: 'disabled', observedAt: null, models: [], notices: [] } };
|
|
return catalogueFor(id).then(catalogue => ({ ...settings.routes(id), catalogue }));
|
|
}
|
|
// 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);
|
|
let result;
|
|
if (['limits.set', 'instances.set'].includes(verb) && settings.current(id).plugins?.['model-provider']?.length) {
|
|
return catalogueFor(id, { fresh: true }).then(catalogue => {
|
|
context?.beforeWrite?.();
|
|
const result = verb === 'limits.set' ? withTokens(id, settings.set(id,args,actor,catalogue)) : instances.set(id,args,actor,catalogue);
|
|
const live=reload(id);
|
|
limits.refresh(id).catch(e=>limitLog({event:'limit-refresh-failed',business:id,message:e.message}));
|
|
return INSTANCE_VERBS.has(verb) ? {...result,restart:!live} : 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') {
|
|
const current = settings.current(id);
|
|
if (current.plugins?.['model-provider']?.length) return catalogueFor(id, { fresh: true }).then(catalogue => {
|
|
const result = settings.setRoutes(id,args,actor,catalogue);
|
|
reload(id); limits.refresh(id).catch(e=>limitLog({event:'limit-refresh-failed',business:id,message:e.message}));
|
|
return { ...result, catalogue };
|
|
});
|
|
result=settings.setRoutes(id,args,actor);
|
|
}
|
|
else if(verb==='plugins.set') {
|
|
if(actor==='tester' && Object.entries(args.plugins ?? {}).some(([kind,value])=>(Array.isArray(value)?value:[value]).some(name=>name!==null && !hasModule(kind,name)))) throw new BusError('human-required');
|
|
result=plugins.set(id,args,actor);modelCatalogues?.invalidate();
|
|
}
|
|
// 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);
|
|
// #1873 T1: the human's definition writes. The broker passes only the business's human (the
|
|
// owner, so ROLE_RIGHTS.owner); the store's assertClean is the held-secret check each write needs.
|
|
else if(Object.hasOwn(DEFINITION_BUS_VERBS,verb))result=definitionVerbs({writer:organizationWriter({dir:organization.dir,rolesDir:organization.rolesDir}),rolesDir:organization.rolesDir,assertClean:value=>credentials.assertClean(value)})[DEFINITION_BUS_VERBS[verb]](id,args,{actor,rights:[...ROLE_RIGHTS.owner],agent:false});
|
|
// Row 194 (D2.1, D2.2): the chief's hire and fire, with the reason in the change context.
|
|
else if(verb==='instance.fire')result=instances.fire(id,{name:args.name,revision:args.revision},{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.
|
|
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,
|
|
// #1873 CB1: a copy binding names a copy this bus booted with; no snapshot, no binding.
|
|
copies:id=>copies?.org===id?copies:null});
|
|
return authorityHost(service,{modelKeys,reload,proxyOrganization});
|
|
})() : null,
|
|
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: instanceRoot ?? dataRoot, ...args });
|
|
modelProxyClient = await loadManagementClient({ configDir: organization.dir, proxy: settings, receipt: proxyReceipt });
|
|
proxyPort = settings.port;
|
|
// #1873: an org bus that initializes the shared proxy also holds its write lock from now on.
|
|
if (org && !proxyLock) proxyLock = createProxyLock({ stateDir: join(instanceDataRoot, 'proxy'), identity: `${instanceConfigDir}:${settings.port}` });
|
|
configureProxyPage();
|
|
return settings;
|
|
} : null,
|
|
// #1873: clear the uncertain-write fence only on restart evidence this host re-proves: the
|
|
// process on the proxy port is new, alive and started after the fence was set.
|
|
proxyFenceClear: organization ? async ({ evidence }) => {
|
|
if (!proxyLock || !modelProxyClient) throw new BusError('not-configured');
|
|
return proxyLock.clear({ restarted: true, read: () => modelProxyClient.getKeys(), verify: previous => verifyRestart(evidence, { fence: previous, port: proxyPort }) });
|
|
} : null,
|
|
credentialVerbs: organization
|
|
? credentialVerbs({ credentials, writer: organizationWriter({ dir: organization.dir, rolesDir: organization.rolesDir, dataRoot:instanceRoot ?? dataRoot }),
|
|
organizations: () => Object.keys(businesses), boot: copies?.definitions ? { org: copies.org, definitions: copies.definitions } : null })
|
|
: 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 }) }));
|
|
// U4a: one adapter per task-tracker module in use, behind one handle that routes each call by the
|
|
// caller's business. The host's `tasks` factory is the Vikunja module. Any other module, and Vikunja
|
|
// when the host gave no factory but an organization file names it, loads through the plug-in loader
|
|
// with the boot config's `trackers`. A module that refuses to load or has no factory refuses
|
|
// startup with tracker-config. A module whose factory makes no
|
|
// adapter refuses the boot (invalid-adapter), and every adapter made so far is closed. The host's
|
|
// factory is always made when given, used or not, so its own checks of `trackers` still run.
|
|
const made = new Map();
|
|
adapter = { async close() { for (const a of made.values()) await a?.close?.(); } };
|
|
const pluginLog = (e) => process.stderr.write(`plugins: ${JSON.stringify(e)}\n`);
|
|
for (const name of new Set([...(tasks ? ['vikunja'] : []), ...trackerOf.values()])) {
|
|
let factory = null;
|
|
if (name === 'vikunja' && tasks) factory = tasks;
|
|
else if (name !== 'vikunja' || [...Object.values(businesses)].some((b) => b?.plugins?.['task-tracker'] === 'vikunja')) {
|
|
const loader = createLoader({ dataRoot:instanceRoot ?? dataRoot, log: pluginLog });
|
|
const { refused } = await loader.load({ 'task-tracker': name });
|
|
const [module] = loader.modules('task-tracker');
|
|
if (!refused.length && module) factory = module.adapter({ trackers: trackers ?? {} });
|
|
}
|
|
// A host may omit its implicit Vikunja adapter, retaining the legacy unconfigured path.
|
|
// An explicitly selected module must always supply a factory.
|
|
if (!factory && name === 'vikunja' && !tasks && !Object.values(businesses).some(b => b?.plugins?.['task-tracker'] === 'vikunja')) continue;
|
|
if (typeof factory !== 'function') throw new BusError('tracker-config');
|
|
const one = await factory({ broker, credentials, businesses, processing });
|
|
made.set(name, one);
|
|
if (typeof one?.handle !== 'function' || !Number.isSafeInteger(one.timeout) || one.timeout < 1)
|
|
throw new BusError('invalid-adapter');
|
|
}
|
|
if (made.size) {
|
|
adapter.timeout = Math.max(...[...made.values()].map((a) => a.timeout));
|
|
adapter.handle = async (cap, verb, args) => {
|
|
const one = made.get(trackerOf.get(broker.identity(cap).business));
|
|
if (!one) throw new BusError('tracker-unconfigured');
|
|
return one.handle(cap, verb, args);
|
|
};
|
|
}
|
|
const web = await webServices(businesses,{dataRoot:instanceRoot ?? dataRoot});
|
|
// 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 ? { 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, project) => refusing(cap, () => broker.checkPush(cap, task_ref, project), 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 } : {}) }));
|
|
}
|
|
// #1873 T1 follow-up (org-defscan): the launcher's check of the org: definition it is about to
|
|
// launch, as it read it from the organization file. Answers where a held credential sits, or null;
|
|
// never the text. A malformed ask refuses, so the launch refuses.
|
|
function definitionScan({ business, definition, entry } = {}) {
|
|
if (typeof business !== 'string' || !Object.hasOwn(businesses, business) || typeof definition !== 'string'
|
|
|| !definition.startsWith(ORG_PREFIX) || definition.length > 128 || entry === null || typeof entry !== 'object' || Array.isArray(entry))
|
|
throw new BusError('invalid-request');
|
|
let size;
|
|
try { size = JSON.stringify(entry).length; } catch { throw new BusError('invalid-request'); }
|
|
if (size > 4 * 1024 * 1024) throw new BusError('invalid-request');
|
|
const clean = value => credentials.assertClean(value);
|
|
try { clean(definition); } catch { return { held: 'name' }; }
|
|
return { held: heldPath(entry, clean) };
|
|
}
|
|
// 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.
|
|
if (instanceRoot === null) {
|
|
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.
|
|
}
|
|
// Startup recovery waits while processing is halted or invalid, like the periodic pass below;
|
|
// processing.resume schedules it again.
|
|
const startState = processing.read();
|
|
if (!startState.halted && !startState.invalid) await requestLifecycle.recover();
|
|
requestTimer = setInterval(() => {
|
|
const state = processing.read();
|
|
if (state.halted || state.invalid) return;
|
|
requestLifecycle.recover({ limit: 64 }).catch(e => process.stderr.write(`request: ${JSON.stringify({ event: 'reconcile-refused', code: e.code ?? 'request-failed' })}\n`));
|
|
}, 30000);
|
|
requestTimer.unref();
|
|
// #1873 CB4: a revoke a previous run left pending finishes (or reports) with its own op id before
|
|
// anything else can use the copy, then each copy holding a grant is checked now and every minute.
|
|
brokerUp(broker);
|
|
if (copyRevoker) {
|
|
const pass = () => {
|
|
for (const id of Object.keys(businesses)) {
|
|
copyRevoker.resume(id).catch(() => {});
|
|
copyRevoker.sweep(id);
|
|
}
|
|
};
|
|
pass();
|
|
copyTimer = setInterval(pass, COPY_SWEEP_MS);
|
|
copyTimer.unref();
|
|
}
|
|
return {
|
|
broker,
|
|
requestLifecycle,
|
|
requirementCatalog: requestLifecycle.catalog,
|
|
admission,
|
|
requestLaunches: business => requestLifecycle.launches(business),
|
|
requestLaunchResult: record => requestLifecycle.launchResult(record),
|
|
requestManifest: record => requestLifecycle.manifestNeeded(record),
|
|
requestRelease: (cap, record) => refusing(cap, () => broker.requestRelease(cap, record)),
|
|
requestChief: business => broker.requestChief(business),
|
|
credentials,
|
|
// P2 consumes this inside the bus only. No process IPC/verb returns it.
|
|
get modelProxy() { return modelProxyClient; },
|
|
// #1873 CB3: the copies the boot proved live, { instance: [service] }; names only.
|
|
copyBound: Object.freeze(copyLaunch ? copyLaunch.bound() : {}),
|
|
path,
|
|
bindLaunch,
|
|
// #1873 CB3: the launcher's prepare for a copy the boot left needing a binding: the launch
|
|
// authority { authority, services } after a live proof at the service, or the refusal. Any
|
|
// other instance needs none and gets { authority: null }.
|
|
// `revision` is the organization file revision the launch resolved its bundle from (A3); a
|
|
// copy's prepare holds the file to it.
|
|
copyLaunch: async ({ business, instance, revision }) => {
|
|
if (typeof business !== 'string' || typeof instance !== 'string') throw new BusError('invalid-request');
|
|
if (!copyNeedsAuthority(business, instance)) return { authority: null, services: [] };
|
|
if (typeof revision !== 'string') throw new BusError('invalid-request');
|
|
return copyLaunch.prepare(business, instance, revision);
|
|
},
|
|
// #1873 CB3: one checked Gitea request for a bound copy's run. Who it is for comes from the
|
|
// capability's session, never from the request. Returns { status, json }; nothing holds the token.
|
|
copyCall: async (cap, request) => {
|
|
if (!copyLaunch) throw new BusError('credential-unavailable');
|
|
return copyLaunch.call(broker.copySession(cap), request);
|
|
},
|
|
recoverLaunch: ({ modelKeyId, ...record }) => broker.recoverLaunch(record),
|
|
endLaunch: (r) => { const out = broker.endLaunch(r); modelRuns.delete(r.business + ':' + r.run); return out; },
|
|
modelAdmission: async ({ business, instance, modelKey, harness, revision, entry: expected, provider, model, account, route, transport }) => {
|
|
const refused = code => ({ allow: false, code });
|
|
if (!settings || !modelCatalogues) return refused('model-catalogue-unavailable');
|
|
try { proxyLock?.assertCertain(); } catch { return refused('proxy-uncertain'); }
|
|
const snapshot = settings.writer.read(business);
|
|
if (snapshot.revision !== revision) return refused('model-provider-unavailable');
|
|
const current = settings.current(business), role = current.roles?.[instance];
|
|
if (!role) return refused('model-provider-unavailable');
|
|
const resolved = resolveInstance({ system: organization.system, business: current, instance });
|
|
const vars = resolved.vars;
|
|
if (!['primary', 'fallback'].includes(route)) return refused('model-provider-unavailable');
|
|
const prefix = route === 'primary' ? '' : 'fallback.';
|
|
if (vars[prefix + 'model'] !== modelKey || vars[prefix + 'harness'] !== harness) return refused('model-provider-unavailable');
|
|
const entry = current.launch?.models?.[modelKey];
|
|
if (!entry || canonicalJson(entry) !== canonicalJson(expected ?? null) || entry.provider !== provider || entry.model !== model) return refused('model-provider-unavailable');
|
|
const harnessLoader = createLoader({dataRoot:instanceRoot ?? dataRoot,log:()=>{}});
|
|
if ((await harnessLoader.load({harness:[harness]})).refused.length) return refused('model-provider-unavailable');
|
|
const countsAccount = harnessLoader.modules('harness')[0].countsAccount();
|
|
if (typeof countsAccount !== 'boolean' || account !== (countsAccount ? current.launch?.account?.name ?? null : null)) return refused('model-provider-unavailable');
|
|
const names = current.plugins?.['model-provider'];
|
|
if (!Array.isArray(names) || !names.length || (entry.module && !names.includes(entry.module))) return refused('model-provider-unavailable');
|
|
const loader = makeProviderLoader({dataRoot:instanceRoot ?? dataRoot,log:()=>{}}), loaded = await loader.load({'model-provider':entry.module ? [entry.module] : names});
|
|
if (loaded.refused.length) return refused('model-provider-unavailable');
|
|
const selected = selectModelProvider(loader.snapshot('model-provider'), entry);
|
|
if (!selected) return refused('model-provider-unavailable');
|
|
const {name:moduleName,module} = selected;
|
|
const descriptor = module.route({selection:entry,harness});
|
|
if (descriptor?.code) return refused(descriptor.code === 'model-harness-incompatible' ? descriptor.code : 'model-provider-unavailable');
|
|
const expectedTransport = (({routeId, endpoint, api, piProvider, compat}) => ({routeId, endpoint, api, piProvider, compat}))(descriptor);
|
|
if (canonicalJson(expectedTransport) !== canonicalJson(transport ?? null)) return refused('model-provider-unavailable');
|
|
const catalogue = await modelCatalogues.read(business,{fresh:true});
|
|
if (settings.writer.read(business).revision !== snapshot.revision) return refused('model-provider-unavailable');
|
|
const selection = {upstream:entry.upstream ?? descriptor.upstream,model:entry.model};
|
|
const answer = module.preflight({selection,harness,catalogue:{...catalogue,models:(catalogue.models ?? []).filter(m => m.module === moduleName)}});
|
|
if (answer?.allow === true) return {allow:true};
|
|
return refused(['model-provider-unavailable','model-catalogue-unavailable','model-harness-incompatible'].includes(answer?.code) ? answer.code : 'model-provider-unavailable');
|
|
},
|
|
modelPreflight: async ({ business, role }) => {
|
|
const current = organization ? loadOrg(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,
|
|
definitionScan,
|
|
launches: bound,
|
|
readers: readCaps,
|
|
async close() {
|
|
clearInterval(copyTimer);
|
|
brokerUp(null);
|
|
clearInterval(requestTimer);
|
|
clearInterval(usageTimer);
|
|
await proxyUsage.close();
|
|
clearInterval(modelTimer);
|
|
await modelSweep;
|
|
clearInterval(accessTimer);
|
|
await stopSync();
|
|
await server.close();
|
|
await adapter?.close?.();
|
|
await requestLifecycle.close();
|
|
await processing?.close();
|
|
store.close();
|
|
credentials.close();
|
|
},
|
|
};
|
|
} catch (e) {
|
|
clearInterval(copyTimer);
|
|
brokerUp(null);
|
|
clearInterval(requestTimer);
|
|
clearInterval(usageTimer);
|
|
await proxyUsage?.close();
|
|
clearInterval(modelTimer);
|
|
clearInterval(accessTimer);
|
|
await server?.close();
|
|
await adapter?.close?.();
|
|
await processing?.close();
|
|
store?.close();
|
|
credentials?.close();
|
|
throw e;
|
|
}
|
|
}
|