813 lines
39 KiB
JavaScript
813 lines
39 KiB
JavaScript
// The S6 launcher, inside the trusted bus host: it starts managed role
|
|
// sessions (packages/seat/src/session.mjs) and serves the launch socket the
|
|
// PM's `launch` tool calls. One launch at a time.
|
|
//
|
|
// A session asks over `<dataRoot>/bus-host/launch.sock` (0600) with one line,
|
|
// `{"cap": "<its capability>", "instance": "<role instance>"}`, and gets one
|
|
// line back: `{"ok": true, "result": {run, instance, harness, model}}` or
|
|
// `{"ok": false, "error": "<code>"}`. The host checks, in order:
|
|
//
|
|
// 1. identity: the capability names a role of this business
|
|
// 2. the business file's launch.instances lists the instance
|
|
// (instance-not-listed)
|
|
// 3. the instance isn't already running (instance-running)
|
|
// 4. it resolves, has a role contract (no-role-contract), sets the
|
|
// harness, model and thinking vars (no-route names the missing one)
|
|
// and its model is a launch.models key (unsupported-harness)
|
|
// 5. the route's harness (row 65): plugins.harness switches it on, and a
|
|
// stated fallback's too (harness-off); the plug-in loader loads it
|
|
// (harness-refused names the module and the loader's reason); it takes
|
|
// the thinking level (thinking-unsupported); its preflight passes
|
|
// (the module's code, e.g. no-session-credential or
|
|
// harness-binary-missing). A failed primary preflight takes the
|
|
// stated fallback through the same checks; there is no other retry.
|
|
// A harness whose launches count against launch.account gets the
|
|
// row 126 account checks.
|
|
// 6. bwrap runs here (sandbox-unavailable)
|
|
// 7. its model names an exact provider/model table entry
|
|
// 8. total, provider and provider/model have room, including the launcher
|
|
// (capacity-full)
|
|
// 9. every switched-on limit check allows it (limit-reached,
|
|
// limit-check-failed), asked through the bus's plug-in loader only
|
|
// 10. the broker authorizes role.launch for the caller (its own codes:
|
|
// launch-revoked, gated, ...)
|
|
//
|
|
// Steps 2 to 9 refuse through the broker's refuse(), so every refusal is an
|
|
// action.refused event; step 10's refusals the broker records itself. Then
|
|
// the host makes sure the workspace has a clone of each project
|
|
// (packages/seat/src/checkout.mjs): one it can't make refuses
|
|
// checkout-failed, one it fetches but doesn't fast-forward is a
|
|
// checkout-fetched record in the launch log, and one it leaves alone is a
|
|
// checkout-kept record. Then it builds the bundle, spawns the runner in its
|
|
// sandbox (packages/seat/src/sandbox.mjs) with the paths it may see, binds
|
|
// the run's process identity (the outer bwrap pid) and writes the new
|
|
// capability on the runner's stdin. When the runner exits, the host ends the launch.
|
|
//
|
|
// `bus start <business> --pm` launches launch.by the same way with
|
|
// launchedBy = the business's human and no caller capability.
|
|
//
|
|
// listen() also serves the git socket (src/git-socket.mjs, row 93), which
|
|
// the sandbox binds for a role that holds git.push.working or
|
|
// review.verdict, or a merge decision to land (landHeld, row 137);
|
|
// session.json names it as gitSocket.
|
|
//
|
|
// Thread sessions (row 121, #1615). A thread instance (an instance whose
|
|
// role definition is the assistant) runs once per thread, never on its own:
|
|
// a launch from a session or --pm refuses thread-required. Every tick
|
|
// (threadTickMs) the host asks the broker for the business's threads
|
|
// (launch op threadsState) and:
|
|
//
|
|
// - starts a session for a thread with no running session that has an
|
|
// undelivered message to its instance, or an open newer than its last
|
|
// claim (REQ-LAUNCH-1). The start goes through check(), so it refuses
|
|
// as any launch does, and the next tick tries again;
|
|
// - stops a thread session once the thread has had no message or open
|
|
// for launch.threadIdleSeconds, the way close() stops a run. The broker
|
|
// decides (launch op threadIdle): it releases the thread only when the
|
|
// run is between turns, with nothing delivered and unread and nothing
|
|
// waiting, and the SIGTERM goes only after it releases. A message sent
|
|
// after the snapshot gets a turn from the same run or waits for the
|
|
// next one, instead of going to the stopping run. An idle thread holds
|
|
// no session and counts against no limit.
|
|
//
|
|
// A thread run's sessionDir is the thread's store (session.mjs
|
|
// threadStore()), so the next run for the thread resumes the harness
|
|
// session (REQ-THREAD-2).
|
|
//
|
|
// Landing runs (row 156, #1651). Every launch of a non-thread instance
|
|
// reads the broker's landHolders (launch op): the approved, unspent merge
|
|
// decisions each non-thread instance raised. The instance's entries go into
|
|
// session.json as `landing: [{decision, task_ref}]`, and the same read
|
|
// decides land_work, so a landing turn always comes with the tool. The
|
|
// runner takes one turn per entry before its receive loop. After the
|
|
// thread pass, every tick also:
|
|
//
|
|
// - starts a holder that launch.instances lists and that isn't running,
|
|
// through check() as any launch, once per decision. A run that ended
|
|
// without landing isn't started again, and a refusal is logged once per
|
|
// code and isn't a start;
|
|
// - stops a running holder whose landing doesn't name the decision (it
|
|
// started before the resolution, so it has no land_work for it). The
|
|
// broker decides (launch op landRelease): it releases the role only
|
|
// between turns, with nothing delivered and unread, and leaves waiting
|
|
// messages undelivered for the next run. The SIGTERM goes only after it
|
|
// releases, and the next tick starts the instance.
|
|
//
|
|
// Nothing here is persisted. After a restart recover() ends the earlier
|
|
// host's runs, so the first tick starts each holder.
|
|
|
|
import { createServer } from "node:net";
|
|
import { chmodSync, existsSync, lstatSync, mkdirSync, readFileSync, rmSync } from "node:fs";
|
|
import { join } from "node:path";
|
|
import { configDir, loadBusiness, resolveInstance, systemVars } from "../../business/src/index.mjs";
|
|
import { limitHit } from "../../business/src/launch.mjs";
|
|
import { BundleError, buildBundle, hasFallback, readSkills, sessionModel } from "../../harness/src/bundle.mjs";
|
|
import { createLoader } from "../../plugins/src/loader.mjs";
|
|
import {
|
|
appendLaunchLog,
|
|
endReason,
|
|
gitSocketPath,
|
|
launchSocketPath,
|
|
newRun,
|
|
readSessions,
|
|
Registry,
|
|
runnerOf,
|
|
SESSION_CREDENTIALS,
|
|
sessionEnv,
|
|
threadStore,
|
|
sessionExit,
|
|
spawnSession,
|
|
workspaceDir,
|
|
writeSessionFile,
|
|
} from "../../seat/src/session.mjs";
|
|
import { BWRAP, sandboxProblem } from "../../seat/src/sandbox.mjs";
|
|
import { CheckoutError, ensureCheckouts } from "../../seat/src/checkout.mjs";
|
|
import { createGitSocket } from "./git-socket.mjs";
|
|
import { cmdlineOf, startTimeOf } from "../../seat/src/proc.mjs";
|
|
import { rolesDir } from "./config.mjs";
|
|
import { CliError } from "./errors.mjs";
|
|
|
|
const STOP_TIMEOUT_MS = 30000;
|
|
const LINE_MAX = 4096;
|
|
// How often the host asks for the threads' state; a poll period, not a limit.
|
|
const THREAD_TICK_MS = 2000;
|
|
// The broker's thread instance definition (packages/bus/src/broker.mjs).
|
|
const THREAD_DEFINITION = "assistant";
|
|
|
|
// A refusal the host decided; `code` goes back to the caller and to the broker.
|
|
class Refusal extends Error {
|
|
constructor(code, message = code) {
|
|
super(message);
|
|
this.code = code;
|
|
}
|
|
}
|
|
|
|
// A harness module runs with the host's authority, and anything it throws refuses the launch
|
|
// and names the module rather than escaping the host.
|
|
function call(name, what, fn) {
|
|
try {
|
|
return fn();
|
|
} catch (e) {
|
|
throw new Refusal("harness-failed", `harness ${name} ${what}: ${e.message}`);
|
|
}
|
|
}
|
|
|
|
// host: startHost()'s handle. tracker: true when the broker has task verbs for
|
|
// the business. launches: false for `bus start --no-launch` (row 163, the D2
|
|
// rehearsal): start() refuses launches-disabled, listen() starts no ticker, a
|
|
// session's launch request is refused, recover() kills and ends nothing, and
|
|
// sessions.json stays as it was.
|
|
// makeLoader (the plug-in loader's factory, which tests point at a
|
|
// scratch tree), bwrap, readOnly, sessionDefaults, harnessVersion,
|
|
// stopTimeoutMs, threadTickMs and git (createGitSocket options) exist for
|
|
// the tests; the command line never sets them.
|
|
export function createLauncher({
|
|
host,
|
|
system,
|
|
env = process.env,
|
|
tracker = false,
|
|
launches = true,
|
|
makeLoader = createLoader,
|
|
bwrap = BWRAP,
|
|
readOnly = [],
|
|
sessionDefaults = {},
|
|
harnessVersion = undefined,
|
|
stopTimeoutMs = STOP_TIMEOUT_MS,
|
|
threadTickMs = THREAD_TICK_MS,
|
|
git = {},
|
|
log = (l) => process.stderr.write(`mosaic-bus: ${l}\n`),
|
|
}) {
|
|
const dataRoot = system.dataRoot;
|
|
// Row 64 (#1557): where a role's skills are read from; scripts/skill.sh activate fills it.
|
|
const skillsDir = join(dataRoot, "skills-enabled");
|
|
let b = loadBusiness(host.business, { dir: configDir(env), rolesDir: rolesDir(env) });
|
|
const vars = systemVars(system);
|
|
|
|
// Sessions an earlier host left in sessions.json: it died without ending
|
|
// them, so their runs still hold their roles. listen() ends them. A file
|
|
// that can't be read refuses the host rather than drop them unseen.
|
|
let leftover;
|
|
try {
|
|
leftover = readSessions(dataRoot).filter((s) => s.business === b.id);
|
|
} catch (e) {
|
|
throw new CliError(`${e.message}; check which sessions still run, then move the file aside`, 3);
|
|
}
|
|
const registry = new Registry(dataRoot, { readOnly: !launches });
|
|
const children = new Map();
|
|
let chain = Promise.resolve();
|
|
let closed = false;
|
|
let server = null;
|
|
let ticker = null;
|
|
let ticking = false;
|
|
const stopping = new Set();
|
|
const refusedThreads = new Map();
|
|
const startedFor = new Map();
|
|
// Row 156: run -> the decisions its landing names; decision -> started; decision -> last
|
|
// refusal code; run -> last landRelease refusal code; and landHolders' last refusal.
|
|
const landingOf = new Map();
|
|
const landStarted = new Set();
|
|
const refusedLand = new Map();
|
|
const refusedRelease = new Map();
|
|
let refusedHolders = null;
|
|
const sockets = new Set();
|
|
const gitSocket = createGitSocket({ host, business: b, dataRoot, env, log, ...git });
|
|
|
|
const record = (entry) => {
|
|
try {
|
|
appendLaunchLog(dataRoot, b.id, entry);
|
|
} catch (e) {
|
|
log(`launch log: ${e.message}`);
|
|
}
|
|
};
|
|
|
|
function serialize(fn) {
|
|
const run = chain.then(fn);
|
|
chain = run.catch(() => {});
|
|
return run;
|
|
}
|
|
|
|
// Steps 2 to 9. Returns what the launch needs. Only `--pm` may name
|
|
// launch.by itself. A thread instance launches only for a thread, and
|
|
// nothing else launches for one.
|
|
async function check(instance, { pm = false, thread = null } = {}) {
|
|
try { b=loadBusiness(host.business,{dir:configDir(env),rolesDir:rolesDir(env)}); } catch(e){throw new Refusal("organization-refused",e.message);}
|
|
if (!b.launch) throw new Refusal("launch-not-configured", `business ${b.id} has no launch block`);
|
|
if (!(pm ? instance === b.launch.by : b.launch.instances.includes(instance))) throw new Refusal("instance-not-listed");
|
|
const threaded = b.roles[instance]?.definition === THREAD_DEFINITION;
|
|
if (threaded && !thread) throw new Refusal("thread-required", `${instance} runs only for a thread`);
|
|
if (!threaded && thread) throw new Refusal("thread-not-allowed", `${instance} isn't a thread instance`);
|
|
if (registry.running(b.id, instance, thread)) throw new Refusal("instance-running");
|
|
let resolved;
|
|
try {
|
|
resolved = resolveInstance({ system: vars, business: b, instance });
|
|
} catch (e) {
|
|
throw new Refusal("unsupported-harness", e.message);
|
|
}
|
|
// Row 65 (#1558): the route's harness, model and thinking come from the instance's vars only.
|
|
const routeOf = (which) => {
|
|
try {
|
|
return sessionModel(resolved, b.launch.models, which);
|
|
} catch (e) {
|
|
throw new Refusal(e instanceof BundleError && e.code === "no-route" ? "no-route" : "unsupported-harness", e.message);
|
|
}
|
|
};
|
|
const primary = routeOf("primary");
|
|
const fallback = hasFallback(resolved) ? routeOf("fallback") : null;
|
|
if (!resolved.contract) throw new Refusal("no-role-contract", `role ${resolved.definition} has no contract file`);
|
|
// Row 64 (#1557): every skill on the role's list is enabled and readable before a run exists.
|
|
try {
|
|
readSkills(resolved.skills, skillsDir);
|
|
} catch (e) {
|
|
if (e instanceof BundleError) throw new Refusal(e.code === "skill-not-enabled" ? "skill-not-enabled" : "skill-refused", e.message);
|
|
throw e;
|
|
}
|
|
// Both of a launch's harnesses must be switched on, whichever it takes.
|
|
const on = Array.isArray(b.plugins?.harness) ? b.plugins.harness : [];
|
|
for (const m of [primary, fallback].filter(Boolean)) {
|
|
if (!on.includes(m.harness)) throw new Refusal("harness-off", `harness ${m.harness} isn't switched on in plugins.harness`);
|
|
}
|
|
// One harness through a fresh loader, so a refusal never leaves an earlier load's module in use.
|
|
const harnessOf = async (m) => {
|
|
const loader = makeLoader({ log: (entry) => record({ ...entry, instance }) });
|
|
const { refused } = await loader.load({ harness: [m.harness] });
|
|
if (refused.length) throw new Refusal("harness-refused", `harness ${m.harness}: ${refused.map((r) => `${r.code}: ${r.message}`).join("; ")}`);
|
|
const module = loader.modules("harness")[0];
|
|
const levels = call(m.harness, "levels", () => module.levels());
|
|
if (!levels.includes(m.thinking)) throw new Refusal("thinking-unsupported", `harness ${m.harness} takes thinking ${levels.join(", ")}, not ${m.thinking}`);
|
|
return module;
|
|
};
|
|
let model = primary;
|
|
let harness = await harnessOf(primary);
|
|
let route = "primary";
|
|
let routeReason = null;
|
|
const failed = call(primary.harness, "preflight", () => harness.preflight({ env }));
|
|
if (failed) {
|
|
if (!fallback) throw new Refusal(failed.code, failed.message);
|
|
// The fallback only on a failed primary preflight, and only at launch (row 65). Any refusal of
|
|
// the fallback names the primary's failure too, so the reason it was tried stays visible.
|
|
const second = await harnessOf(fallback).catch((e) => {
|
|
if (e instanceof Refusal) e.message += ` (primary ${primary.harness}: ${failed.message})`;
|
|
throw e;
|
|
});
|
|
const also = call(fallback.harness, "preflight", () => second.preflight({ env }));
|
|
if (also) throw new Refusal(also.code, `fallback ${fallback.harness}: ${also.message} (primary ${primary.harness}: ${failed.message})`);
|
|
[model, harness, route, routeReason] = [fallback, second, "fallback", failed.code];
|
|
}
|
|
// Row 126 (#1620): only a launch on a harness whose module counts the account spends the named
|
|
// account. The bus environment must hold the named set and no other credential variable, so the
|
|
// session can't log in some other way. Messages name variables, never values.
|
|
const account = call(model.harness, "countsAccount", () => harness.countsAccount()) === true ? b.launch.account?.name ?? null : null;
|
|
if (account !== null) {
|
|
// The usage records keep this name (bus schema 3i), so it must never be a credential's value.
|
|
const holds = [...new Set(SESSION_CREDENTIALS.flat())].filter((k) => env[k] === account);
|
|
if (holds.length) throw new Refusal("account-is-credential", `launch.account.name is the value of ${holds.join(", ")}; it names the account, never its credential`);
|
|
const named = b.launch.account.credential.split("+");
|
|
if (!SESSION_CREDENTIALS.some((set) => set.join("+") === named.join("+")) || !named.every((k) => Boolean(env[k])))
|
|
throw new Refusal("account-credential-missing", `account ${account} names ${named.join(" with ")}, which the bus host doesn't hold as a session login`);
|
|
const other = [...new Set(SESSION_CREDENTIALS.flat())].filter((k) => !named.includes(k) && Boolean(env[k]));
|
|
if (other.length) throw new Refusal("account-credential-ambiguous", `account ${account} names ${named.join(" with ")}, but the bus host also sets ${other.join(", ")}`);
|
|
}
|
|
const unavailable = sandboxProblem({ bwrap });
|
|
if (unavailable) throw new Refusal("sandbox-unavailable", unavailable);
|
|
const hit=limitHit(registry.sessions,b.id,b.launch.limits,model);
|
|
if(hit){const error=new Refusal("capacity-full",JSON.stringify(hit));error.limit=hit;throw error;}
|
|
// Row 125 (#1619, REQ-LIMIT-4): the limit checks. An answer that isn't an allow refuses,
|
|
// and so does a bus that can't answer (fail closed, lead decision 150).
|
|
let may;
|
|
try {
|
|
may = await host.op({ op: "mayLaunch", business: b.id, instance, provider: model.provider, model: model.model, account });
|
|
} catch (e) {
|
|
throw new Refusal("limit-check-failed", `bus: ${e.code ?? "launch-op-refused"}`);
|
|
}
|
|
if (may?.allow !== true) {
|
|
if (may?.hit) { const error = new Refusal("limit-reached", JSON.stringify(may.hit)); error.limit = { check: may.check, ...may.hit }; throw error; }
|
|
throw new Refusal("limit-check-failed", `${may?.check ?? "bus"}: ${may?.code ?? "malformed-answer"}`);
|
|
}
|
|
return { resolved, model, account, harness, route, routeReason };
|
|
}
|
|
|
|
// Bundle, spawn, bind, register, hand over the capability.
|
|
async function start({ instance, launchedBy, resolved, model, account = null, thread = null }) {
|
|
// Every launch path ends here, so a --no-launch host spawns nothing whichever path asked.
|
|
if (!launches) throw new Refusal("launches-disabled", "this host was started with --no-launch");
|
|
// A thread's store is checked first, so a bad id or an unsafe directory
|
|
// launches nothing.
|
|
let store = null;
|
|
if (thread) {
|
|
try {
|
|
store = threadStore(dataRoot, b.id, thread);
|
|
} catch (e) {
|
|
throw new Refusal("thread-store-refused", e.message);
|
|
}
|
|
}
|
|
const credentials = await host.op({ op: "credentialStatus", business: b.id, role: instance });
|
|
// The clones come before the run, so one that can't be made leaves no
|
|
// run behind. The host's git treats each clone as hostile, and its
|
|
// checks hold only while no session writes the workspace: a workspace
|
|
// is one instance's, check() refused instance-running, and launches are
|
|
// serialized, so none does.
|
|
const workspace = workspaceDir(dataRoot, b.id, instance);
|
|
mkdirSync(workspace, { recursive: true, mode: 0o700 });
|
|
let checkouts;
|
|
try {
|
|
checkouts = await ensureCheckouts({ workspace, projects: b.projects, env });
|
|
} catch (e) {
|
|
if (e instanceof CheckoutError) throw new Refusal("checkout-failed", `${e.project}: ${e.message}`);
|
|
throw e;
|
|
}
|
|
for (const c of checkouts) {
|
|
if (c.outcome === "kept") record({ event: "checkout-kept", instance, project: c.project, reason: c.reason });
|
|
else if (c.outcome === "fetched") record({ event: "checkout-fetched", instance, project: c.project, reason: c.reason });
|
|
}
|
|
let harness, route, routeReason;
|
|
({resolved,model,account,harness,route,routeReason}=await check(instance,{pm:instance===b.launch.by,thread}));
|
|
const { run, runDir } = newRun(dataRoot, b.id);
|
|
const sessionDir = store ? store.session : join(runDir, "session");
|
|
if (!store) mkdirSync(sessionDir, { mode: 0o700 });
|
|
const toolSocket = join(runDir, "tools.sock");
|
|
const turnMarker = join(runDir, "turn.done");
|
|
const version = harnessVersion !== undefined ? harnessVersion : call(model.harness, "version", () => harness.version({ env }));
|
|
// Row 137 (#1631): a role and thread holding an approved, unspent merge decision get land_work.
|
|
// The broker checks the decision again on every land; this only decides what the bundle shows.
|
|
// Row 156 (#1651): a non-thread instance's land and landing come from one landHolders read, so a
|
|
// decision resolved between two reads can't give a landing turn without land_work.
|
|
const holders = thread ? null : await host.op({ op: "landHolders", business: b.id });
|
|
if (holders !== null && !Array.isArray(holders)) throw new Refusal("launch-failed", "landHolders: malformed answer");
|
|
const holding = holders?.filter((h) => h.instance === instance) ?? null;
|
|
const land = thread ? await host.op({ op: "landHeld", record: { business: b.id, role: instance, thread } }) : holding.length > 0;
|
|
const bundle = buildBundle({
|
|
dir: join(runDir, "bundle"),
|
|
resolved,
|
|
models: b.launch.models,
|
|
contract: readFileSync(resolved.contract, "utf8"),
|
|
arbiter: Object.values(b.arbiters).includes(instance),
|
|
tracker,
|
|
credentials,
|
|
session: { run, launchedBy, workspace, toolSocket, turnMarker, checkouts },
|
|
harness,
|
|
route,
|
|
routeReason,
|
|
harnessVersion: version,
|
|
land,
|
|
skillsDir,
|
|
});
|
|
const { manifest, ...files } = bundle.files;
|
|
const adapter = call(model.harness, "adapter", () => harness.adapter());
|
|
const turnEnd = call(model.harness, "turnEnd", () => harness.turnEnd());
|
|
const transcript = call(model.harness, "transcript", () => harness.transcript({ runDir, store }));
|
|
const harnessMounts = call(model.harness, "mounts", () => harness.mounts({ env }));
|
|
// Row 93: the git socket, for a role that pushes work or reviews it. Row 137: or lands it.
|
|
const held = [...resolved.limits.authority.withinRole, ...resolved.limits.authority.crossRole];
|
|
const git = land || held.includes("git.push.working") || held.includes("review.verdict") ? gitSocketPath(dataRoot) : null;
|
|
writeSessionFile(runDir, {
|
|
...sessionDefaults,
|
|
business: b.id,
|
|
instance,
|
|
...(thread ? { thread } : {}),
|
|
run,
|
|
launchedBy,
|
|
harness: model.harness,
|
|
provider: model.provider,
|
|
model: model.model,
|
|
brokerSocket: host.path,
|
|
launchSocket: launchSocketPath(dataRoot),
|
|
gitSocket: git,
|
|
workspace,
|
|
sessionDir,
|
|
adapter,
|
|
bundle: files,
|
|
harnessEnv: bundle.env,
|
|
turnEnd,
|
|
toolSocket,
|
|
turnMarker,
|
|
...(holding?.length ? { landing: holding.map(({ decision, task_ref }) => ({ decision, task_ref })) } : {}),
|
|
});
|
|
|
|
const paths = {
|
|
workspace,
|
|
adapter,
|
|
harnessMounts,
|
|
brokerSocket: host.path,
|
|
launchSocket: resolved.launch ? launchSocketPath(dataRoot) : null,
|
|
gitSocket: git,
|
|
threadStore: store,
|
|
readOnly,
|
|
};
|
|
const child = spawnSession({ runDir, env: sessionEnv(env, run), paths, business: b.id, role: instance, bwrap });
|
|
let bound = false;
|
|
let entry = null;
|
|
const exited = new Promise((resolve) => child.once("close", (code, signal) => resolve(sessionExit(code, signal))));
|
|
child.on("error", (e) => log(`run ${run}: ${e.message}`));
|
|
child.stdin.on("error", () => {});
|
|
try {
|
|
if (child.pid === undefined) throw new Error("the session process did not start");
|
|
const outerStart = startTimeOf(child.pid);
|
|
const runner = await runnerOf(child);
|
|
if (!outerStart) throw new Error("the session process exited at once");
|
|
const launch = await host.bindLaunch({ business: b.id, role: instance, run, harness: model.harness, provider: model.provider, model: model.model, account, transcript, pid: child.pid, startTime: outerStart, ...(thread ? { thread } : {}) });
|
|
bound = true;
|
|
entry = {
|
|
business: b.id, instance, ...(thread ? { thread } : {}), run, launchedBy, harness: model.harness, model: model.model, provider: model.provider, modelKey: model.modelKey,
|
|
pid: child.pid, startTime: outerStart, runnerPid: runner.pid, runnerStartTime: runner.startTime, runDir, startedAt: new Date().toISOString(),
|
|
};
|
|
registry.add(entry);
|
|
// Row 156: one start per decision, whichever launch wrote it into landing.
|
|
landingOf.set(run, new Set(holding?.map((h) => h.decision) ?? []));
|
|
for (const h of holding ?? []) landStarted.add(h.decision);
|
|
child.stdin.end(`${JSON.stringify({ cap: launch.cap })}\n`);
|
|
} catch (e) {
|
|
child.kill("SIGKILL");
|
|
await exited;
|
|
if (bound) await endRun(run, "failed", null);
|
|
throw new Refusal("launch-failed", e.message);
|
|
}
|
|
children.set(run, exited);
|
|
exited.then(async ({ code, signal }) => {
|
|
const reason = endReason(code, signal);
|
|
try {
|
|
const ended = await endRun(run, reason, signal ? null : code);
|
|
registry.remove(b.id, run);
|
|
// Row 125 (#1619, REQ-LIMIT-4): a run a limit check stopped names the limit, from the claim row.
|
|
record({ event: "end", instance, ...(thread ? { thread } : {}), run, reason, exitCode: code, signal, ...(ended?.limit ? { limit: ended.limit } : {}) });
|
|
log(`run ${run} (${instance}) ended: ${reason}`);
|
|
} catch (e) {
|
|
log(`run ${run}: ${e.message}`);
|
|
} finally {
|
|
children.delete(run);
|
|
stopping.delete(run);
|
|
landingOf.delete(run);
|
|
refusedRelease.delete(run);
|
|
}
|
|
});
|
|
record({ event: "launch", instance, ...(thread ? { thread } : {}), run, launchedBy, harness: model.harness, route, routeReason, model: model.model, provider: model.provider, modelKey: model.modelKey, account, pid: child.pid, runnerPid: entry.runnerPid, manifest });
|
|
return { run, instance, harness: model.harness, model: model.model };
|
|
}
|
|
|
|
// Row 121: one tick of thread starts and idle stops, then (row 156) the
|
|
// landing pass. A tick that is still running when the next is due skips it.
|
|
async function tick() {
|
|
if (ticking || closed) return;
|
|
ticking = true;
|
|
try {
|
|
await threadPass();
|
|
await landPass();
|
|
} finally {
|
|
ticking = false;
|
|
}
|
|
}
|
|
|
|
async function threadPass() {
|
|
let threads;
|
|
try {
|
|
threads = await host.op({ op: "threadsState", business: b.id });
|
|
} catch (e) {
|
|
log(`threadsState refused: ${e.code ?? e.message}`);
|
|
return;
|
|
}
|
|
const now = Date.now();
|
|
const ms = (at) => (at ? Date.parse(at) : 0);
|
|
for (const t of threads) {
|
|
if (closed) return;
|
|
const s = registry.running(b.id, t.instance, t.thread);
|
|
if (s) {
|
|
const last = Math.max(ms(s.startedAt), ms(t.lastMessage), ms(t.lastOpen));
|
|
if (!stopping.has(s.run) && now - last >= b.launch.threadIdleSeconds * 1000) {
|
|
let idle;
|
|
try {
|
|
idle = await host.op({ op: "threadIdle", record: { business: b.id, thread: t.thread, run: s.run } });
|
|
} catch (e) {
|
|
log(`run ${s.run}: threadIdle refused: ${e.code ?? e.message}`);
|
|
continue;
|
|
}
|
|
// A turn in progress, or a message since the snapshot: the run keeps going.
|
|
if (!idle.released) continue;
|
|
stopping.add(s.run);
|
|
record({ event: "thread-idle", instance: t.instance, thread: t.thread, run: s.run });
|
|
try {
|
|
if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM");
|
|
} catch {}
|
|
}
|
|
continue;
|
|
}
|
|
const opened = ms(t.lastOpen) > ms(t.lastClaim);
|
|
if (t.holder || !(t.pending > 0 || opened)) continue;
|
|
// One start per message or open: a session that ended without taking
|
|
// the work (it died before its claim, say) waits for the next one
|
|
// rather than start again every tick. A refused start isn't a start.
|
|
const signature = `${t.lastMessage}|${t.lastOpen}`;
|
|
if (startedFor.get(t.thread) === signature) continue;
|
|
await serialize(async () => {
|
|
if (closed || registry.running(b.id, t.instance, t.thread)) return;
|
|
try {
|
|
await start({ instance: t.instance, launchedBy: b.human, thread: t.thread, ...(await check(t.instance, { thread: t.thread })) });
|
|
startedFor.set(t.thread, signature);
|
|
refusedThreads.delete(t.thread);
|
|
} catch (e) {
|
|
const code = e instanceof Refusal ? e.code : "launch-failed";
|
|
// Logged once per code, not once per tick.
|
|
if (refusedThreads.get(t.thread) === code) return;
|
|
refusedThreads.set(t.thread, code);
|
|
log(`thread ${t.thread} (${t.instance}): start refused: ${code}`);
|
|
record({ event: "refused", by: b.human, instance: t.instance, thread: t.thread, code, why: e.message, ...(e.limit?{limit:e.limit}:{}) });
|
|
}
|
|
});
|
|
}
|
|
}
|
|
|
|
// Row 156 (#1651): start each listed merge decision holder once, and stop a
|
|
// running holder that started before its decision's resolution.
|
|
async function landPass() {
|
|
if (!b.launch) return;
|
|
let holders;
|
|
try {
|
|
holders = await host.op({ op: "landHolders", business: b.id });
|
|
if (!Array.isArray(holders)) throw new Refusal("malformed-answer");
|
|
refusedHolders = null;
|
|
} catch (e) {
|
|
const code = e.code ?? e.message;
|
|
if (refusedHolders !== code) log(`landHolders refused: ${code}`);
|
|
refusedHolders = code;
|
|
return;
|
|
}
|
|
for (const h of holders) {
|
|
if (closed) return;
|
|
if (!b.launch.instances.includes(h.instance)) continue;
|
|
const s = registry.running(b.id, h.instance, null);
|
|
if (s) {
|
|
if (stopping.has(s.run) || landingOf.get(s.run)?.has(h.decision)) continue;
|
|
let r;
|
|
try {
|
|
r = await host.op({ op: "landRelease", record: { business: b.id, run: s.run, decision: h.decision } });
|
|
} catch (e) {
|
|
const code = e.code ?? e.message;
|
|
if (refusedRelease.get(s.run) !== code) log(`run ${s.run}: landRelease refused: ${code}`);
|
|
refusedRelease.set(s.run, code);
|
|
continue;
|
|
}
|
|
// A landing turn or a message turn in progress: a later tick.
|
|
if (!r?.released) continue;
|
|
stopping.add(s.run);
|
|
record({ event: "land-relaunch", instance: h.instance, run: s.run, decision: h.decision });
|
|
try {
|
|
if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM");
|
|
} catch {}
|
|
continue;
|
|
}
|
|
if (landStarted.has(h.decision)) continue;
|
|
await serialize(async () => {
|
|
if (closed || registry.running(b.id, h.instance, null)) return;
|
|
try {
|
|
await start({ instance: h.instance, launchedBy: b.human, ...(await check(h.instance)) });
|
|
refusedLand.delete(h.decision);
|
|
} catch (e) {
|
|
const code = e instanceof Refusal ? e.code : "launch-failed";
|
|
if (refusedLand.get(h.decision) === code) return;
|
|
refusedLand.set(h.decision, code);
|
|
log(`decision ${h.decision} (${h.instance}): landing start refused: ${code}`);
|
|
record({ event: "refused", by: b.human, instance: h.instance, decision: h.decision, code, why: e.message, ...(e.limit ? { limit: e.limit } : {}) });
|
|
}
|
|
});
|
|
}
|
|
}
|
|
|
|
// endLaunch's answer ({ released, limit? }), or null when it refused.
|
|
async function endRun(run, reason, exitCode) {
|
|
try {
|
|
return await host.op({ op: "endLaunch", record: { business: b.id, run, reason, exitCode } });
|
|
} catch (e) {
|
|
log(`run ${run}: endLaunch refused: ${e.code ?? e.message}`);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
// A launch a session asked for. Resolves to the reply line's object.
|
|
function fromSession(cap, instance) {
|
|
return serialize(async () => {
|
|
if (closed) return { ok: false, error: "host-closing" };
|
|
let who;
|
|
try {
|
|
who = await host.op({ op: "identity", cap });
|
|
} catch (e) {
|
|
return { ok: false, error: e.code ?? "unauthenticated" };
|
|
}
|
|
if (!who.role || who.business !== b.id) {
|
|
await host.op({ op: "refuse", cap, code: "not-a-role" }).catch(() => {});
|
|
return { ok: false, error: "not-a-role" };
|
|
}
|
|
if (!launches) {
|
|
await host.op({ op: "refuse", cap, code: "launches-disabled" }).catch(() => {});
|
|
record({ event: "refused", by: who.role, instance, code: "launches-disabled" });
|
|
return { ok: false, error: "launches-disabled" };
|
|
}
|
|
let checked;
|
|
try {
|
|
if (typeof instance !== "string") throw new Refusal("invalid-request");
|
|
checked = await check(instance);
|
|
} catch (e) {
|
|
if (!(e instanceof Refusal)) throw e;
|
|
await host.op({ op: "refuse", cap, code: e.code }).catch(() => {});
|
|
record({ event: "refused", by: who.role, instance, code: e.code, why: e.message, ...(e.limit?{limit:e.limit}:{}) });
|
|
return { ok: false, error: e.code };
|
|
}
|
|
try {
|
|
await host.op({ op: "authorizeLaunch", cap, instance });
|
|
} catch (e) {
|
|
record({ event: "refused", by: who.role, instance, code: e.code ?? "not-authorized" });
|
|
return { ok: false, error: e.code ?? "not-authorized" };
|
|
}
|
|
try {
|
|
return { ok: true, result: await start({ instance, launchedBy: who.role, ...checked }) };
|
|
} catch (e) {
|
|
const code = e instanceof Refusal ? e.code : "launch-failed";
|
|
log(`launch of ${instance} for ${who.role} failed: ${e.message}`);
|
|
await host.op({ op: "refuse", cap, code }).catch(() => {});
|
|
record({ event: "refused", by: who.role, instance, code, why: e.message, ...(e.limit?{limit:e.limit}:{}) });
|
|
return { ok: false, error: code };
|
|
}
|
|
});
|
|
}
|
|
|
|
// `bus start --pm`: the business's launcher instance, launched by the human.
|
|
function launchPm() {
|
|
return serialize(async () => {
|
|
if (!b.launch) throw new CliError(`business ${b.id} has no launch block, so --pm has nothing to launch`, 3);
|
|
try {
|
|
return await start({ instance: b.launch.by, launchedBy: b.human, ...(await check(b.launch.by, { pm: true })) });
|
|
} catch (e) {
|
|
if (!(e instanceof Refusal)) throw e;
|
|
record({ event: "refused", by: b.human, instance: b.launch.by, code: e.code, why: e.message, ...(e.limit?{limit:e.limit}:{}) });
|
|
throw new CliError(`PM launch refused: ${e.code}${e.message !== e.code ? ` (${e.message})` : ""}`, 3);
|
|
}
|
|
});
|
|
}
|
|
|
|
// The earlier host's sessions. Their capabilities died with its broker, so
|
|
// one that still runs can do nothing: kill it (the namespace takes its
|
|
// turn with it). Then end each run in the bus, which releases its role.
|
|
// This broker never bound those runs, so each is bound again first with
|
|
// its identical launch record, which the broker accepts.
|
|
async function recover() {
|
|
// Row 163 (mos-0961): a --no-launch host never signals or ends an earlier host's session. Its
|
|
// data root may hold a sessions.json that names live processes, so it only says what it left.
|
|
if (!launches) {
|
|
for (const s of leftover) log(`run ${s.run} (${s.instance}) from an earlier host: left as it is, this host was started with --no-launch`);
|
|
leftover = [];
|
|
return;
|
|
}
|
|
for (const s of leftover) {
|
|
if (startTimeOf(s.pid) === s.startTime && (cmdlineOf(s.pid) ?? [])[0] === bwrap) {
|
|
try {
|
|
process.kill(s.pid, "SIGKILL");
|
|
} catch {}
|
|
const end = Date.now() + 5000;
|
|
while (startTimeOf(s.pid) === s.startTime && Date.now() < end) await new Promise((r) => setTimeout(r, 20));
|
|
}
|
|
try {
|
|
await host.bindLaunch({ business: b.id, role: s.instance, run: s.run, harness: s.harness, pid: s.pid, startTime: s.startTime, ...(s.thread ? { thread: s.thread } : {}) });
|
|
} catch (e) {
|
|
// A host that died after ending the run but before updating sessions.json.
|
|
if (e.code === "run-ended") log(`run ${s.run} (${s.instance}) from an earlier host was already ended`);
|
|
else log(`run ${s.run} (${s.instance}) from an earlier host: not ended, bindLaunch refused: ${e.code ?? e.message}`);
|
|
continue;
|
|
}
|
|
await endRun(s.run, "host-lost", null);
|
|
record({ event: "end", instance: s.instance, ...(s.thread ? { thread: s.thread } : {}), run: s.run, reason: "host-lost", exitCode: null, signal: null });
|
|
log(`run ${s.run} (${s.instance}) from an earlier host ended: host-lost`);
|
|
}
|
|
leftover = [];
|
|
}
|
|
|
|
async function listen() {
|
|
await recover();
|
|
const path = launchSocketPath(dataRoot);
|
|
if (existsSync(path)) {
|
|
if (!lstatSync(path).isSocket()) throw new CliError(`${path} exists and isn't a socket`, 3);
|
|
rmSync(path);
|
|
}
|
|
server = createServer((socket) => {
|
|
sockets.add(socket);
|
|
socket.once("close", () => sockets.delete(socket));
|
|
let buf = "";
|
|
let answered = false;
|
|
const reply = (obj) => {
|
|
if (answered) return;
|
|
answered = true;
|
|
socket.end(`${JSON.stringify(obj)}\n`);
|
|
};
|
|
socket.setTimeout(10000, () => reply({ ok: false, error: "invalid-request" }));
|
|
socket.on("error", () => {});
|
|
socket.on("data", (chunk) => {
|
|
if (answered) return;
|
|
buf += chunk;
|
|
if (buf.length > LINE_MAX) return reply({ ok: false, error: "invalid-request" });
|
|
const nl = buf.indexOf("\n");
|
|
if (nl < 0) return;
|
|
socket.setTimeout(0);
|
|
let req;
|
|
try {
|
|
req = JSON.parse(buf.slice(0, nl));
|
|
} catch {
|
|
return reply({ ok: false, error: "invalid-request" });
|
|
}
|
|
if (typeof req?.cap !== "string" || !/^[0-9a-f]{64}$/.test(req.cap)) return reply({ ok: false, error: "unauthenticated" });
|
|
fromSession(req.cap, req.instance).then(reply, (e) => {
|
|
log(`launch socket: ${e.message}`);
|
|
reply({ ok: false, error: "launch-failed" });
|
|
});
|
|
});
|
|
});
|
|
// bus-host/ is 0700, so the moment before the chmod exposes nothing.
|
|
await new Promise((resolve, reject) => {
|
|
server.once("error", reject);
|
|
server.listen(path, resolve);
|
|
});
|
|
chmodSync(path, 0o600);
|
|
await gitSocket.listen();
|
|
if (b.launch && launches) ticker = setInterval(() => tick().catch((e) => log(`thread tick: ${e.message}`)), threadTickMs);
|
|
}
|
|
|
|
// SIGTERM every runner, wait, then end what's left. Runs before the host
|
|
// stops the broker, so endLaunch still works.
|
|
async function close() {
|
|
closed = true;
|
|
if (ticker) clearInterval(ticker);
|
|
if (server) {
|
|
// server.close waits for every connection, and a client that never
|
|
// ends its side (or never sends) would hold it; so they go first.
|
|
const stopped = new Promise((r) => server.close(r));
|
|
for (const socket of sockets) socket.destroy();
|
|
await stopped;
|
|
rmSync(launchSocketPath(dataRoot), { force: true });
|
|
}
|
|
await gitSocket.close();
|
|
await chain;
|
|
// Again every second: a runner still starting has no handler yet, and as
|
|
// pid 1 of its namespace it never sees a signal sent before it has one.
|
|
const term = () => {
|
|
for (const s of registry.sessions) {
|
|
try {
|
|
if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM");
|
|
} catch {}
|
|
}
|
|
};
|
|
term();
|
|
const again = setInterval(term, 1000);
|
|
const pending = [...children.values()];
|
|
const timer = setTimeout(() => {
|
|
for (const s of registry.sessions) {
|
|
try {
|
|
if (startTimeOf(s.pid) === s.startTime) process.kill(s.pid, "SIGKILL");
|
|
} catch {}
|
|
}
|
|
}, stopTimeoutMs);
|
|
await Promise.all(pending);
|
|
// The exit handlers run on the same tick as the close events; let them finish.
|
|
while (children.size) await new Promise((r) => setTimeout(r, 20));
|
|
clearInterval(again);
|
|
clearTimeout(timer);
|
|
}
|
|
|
|
host.beforeClose(close);
|
|
return { business: b, registry, listen, launchPm, fromSession, tick, close };
|
|
}
|