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 /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., so a restart finds the keys the grant executor wrote. The loader refuses host. 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. 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 /secrets//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 /secrets/grants//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; } }