feat(harness): meta-harness and PM launching off T3 (#1523, row 41)
Slice 1 S6. packages/harness runs a role session under the pi or claude
adapter behind a tool gate, and packages/seat holds the session and its
process. The bus host gains a launch socket: `mosaic bus start
<business> --pm <harness>:<model>` runs the PM without a window, the PM
launches role sessions through the broker within the business file's
`launch.max`, and every launch, refusal and end is a launch-log line and
a broker event. `mosaic talk`, `mosaic launches off|on|list` and
`mosaic stop <run>` are new CLI verbs. packages/bus gains the IPC ops
identity, authorize, refuse, endLaunch and credentialStatus, plus
Broker.endLaunch (decision 77); no socket verb, event kind or schema
changed. The session environment is an allowlist, and the runner exits
before claiming if a founder credential reaches it.
Filbert built it (agents/filbert/work/s6). Darkwing reviewed on #1523:
round 1 changes (26991/26995), round 2 changes (27004/27010), round 3
changes (27012/27032), round 4 approve (27033/27034) on manifest
08a78972, 42 paths.
Sage's gate on 64b0f9f1 (12:32Z), sequential and teed: node suites
business 60, bus 74, cli 83, control-board 124, conversation 182,
discord 178, harness 56, ledger 78, mosaic 69, queue 149, runs 41,
seat 27, tasks 51, webui 39, every one 0 failed; test-auth 15,
test-conductor 17, test-config 24, test-discord 66,
test-extension-package 18, test-foundation 44, test-queue 27,
test-release 14, test-task 98 and 26 without Docker, all 0 failed.
64b0f9f1..ee7f909b touches none of the 42 paths. It landed after the
Q14 hold ended at 2026-10-11T17:19:23.857Z. The recorded PM-to-coder
run on a scratch data root is still owed before row 41 moves to done.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Claude Opus 5.5
+120
-10
@@ -2,9 +2,10 @@
|
||||
|
||||
Jason's front door to the bus (slice 1 row S4, #1521): `mosaic inbox`,
|
||||
`decide`, `tasks`, `agents` and `trail`, and the bus host `mosaic bus
|
||||
start|stop|status` with its Discord notifier. Brief:
|
||||
`docs/plans/2026-10-04_slice-1.md`, row S4. Rulings: lead decision 70 in
|
||||
`docs/plans/2026-09-26_lead-decisions.md`.
|
||||
start|stop|status` with its Discord notifier (row S4, #1521); the session
|
||||
launcher in the host and `mosaic talk`, `stop` and `launches` (row S6,
|
||||
#1523). Brief: `docs/plans/2026-10-04_slice-1.md`, rows S4 and S6. Rulings:
|
||||
lead decisions 70 and 77 in `docs/plans/2026-09-26_lead-decisions.md`.
|
||||
|
||||
Every command reads through the broker. None opens the SQLite file.
|
||||
|
||||
@@ -24,6 +25,10 @@ mosaic decide <decision> <option> [--note <text>] [--yes] [--business <id>]
|
||||
mosaic tasks [--business <id>] [--json]
|
||||
mosaic agents [--business <id>] [--json]
|
||||
mosaic trail <task|decision> [--business <id>] [--json]
|
||||
mosaic talk <role> <text> [--wait <seconds>] [--business <id>]
|
||||
mosaic stop <run>
|
||||
mosaic launches off|on [--business <id>]
|
||||
mosaic launches list [--json]
|
||||
```
|
||||
|
||||
- The business is `--business`, or else the business of the bus host running
|
||||
@@ -49,11 +54,28 @@ mosaic trail <task|decision> [--business <id>] [--json]
|
||||
- `trail` prints rows in the broker's order (at, table, seq). A decision's
|
||||
trail ends with its `task_ref` and `follow with: mosaic trail <task>`. It
|
||||
does not pull in the task's rows.
|
||||
- `talk` sends `<text>` to the role as a REQUEST, then polls the human's
|
||||
inbox every 2 s for up to `--wait` seconds (default 900). It prints every
|
||||
message that arrives and marks it read, and stops at the reply to its
|
||||
REQUEST. No reply in time exits 1; the reply still lands in the inbox.
|
||||
`--wait 0` only sends.
|
||||
- `launches off` and `on` are the broker's `launch.revoke` and
|
||||
`launch.restore`. While off, every `role.launch` refuses with
|
||||
`launch-revoked`; running sessions keep running. Stop them with `stop`.
|
||||
`bus start --pm` doesn't ask the broker (`launchPm` skips
|
||||
`authorizeLaunch`): the human launches the PM, so `launches off` doesn't
|
||||
stop it. The other checks in "Sessions" below still apply.
|
||||
- `stop` and `launches list` read `<dataRoot>/bus-host/sessions.json`, not
|
||||
the bus (see "Sessions" below). `stop` refuses inside an agent run;
|
||||
`launches list` does not, because the file is readable to the same user
|
||||
anyway.
|
||||
|
||||
## Bus host
|
||||
|
||||
```
|
||||
mosaic bus start <business> the host; run by the unit, not by hand
|
||||
mosaic bus start <business> [--pm]
|
||||
the host; run by the unit, not by hand.
|
||||
--pm also launches the business's PM
|
||||
mosaic bus stop SIGTERM to the recorded host, then wait
|
||||
mosaic bus status [--json] host state file, socket, writer.lock
|
||||
```
|
||||
@@ -76,12 +98,23 @@ mosaic bus status [--json] host state file, socket, writer.lock
|
||||
5. It writes `<dataRoot>/bus-host/host.json` (0600): the pid, the process
|
||||
start time, the business, the start time as text and the notifier
|
||||
binding. The file holds no capability.
|
||||
6. It starts the session launcher (`src/launcher.mjs`, "Sessions" below)
|
||||
and, with `--pm`, launches the business's `launch.by` instance. A PM
|
||||
launch that refuses closes the host with the refusal's exit code.
|
||||
|
||||
No capability appears in argv, stdout, the environment or a file.
|
||||
`startHost()` in `src/host.mjs` is also the in-process API S6 uses:
|
||||
`bindLaunch(record)` binds a launched run's process identity
|
||||
(`pid`, `startTime`) and returns `{business, run, cap}`. Requests to the
|
||||
broker process go one at a time, because its replies carry no request id.
|
||||
`startHost()` in `src/host.mjs` is also the in-process API the launcher
|
||||
uses: `bindLaunch(record)` binds a launched run's process identity
|
||||
(`pid`, `startTime`) and returns `{business, run, cap}`; `op(message)` sends
|
||||
one of the broker process's launch ops (`identity`, `authorizeLaunch`,
|
||||
`refuse`, `endLaunch`, `credentialStatus`); `beforeClose(fn)` runs `fn`
|
||||
before the broker closes, so the launcher stops its sessions first.
|
||||
Requests to the broker process go one at a time. Each carries an id, and
|
||||
the broker echoes it on the reply. If a reply doesn't arrive within 10 s, or
|
||||
arrives with an id no request is waiting for, the channel is broken: the
|
||||
waiting request refuses with `broker-channel-broken`, so does every later
|
||||
one, and the host stops with exit 1 for the unit to restart. A late reply
|
||||
never answers a later request, and a launch waiting on one refuses.
|
||||
|
||||
SIGTERM or SIGINT stops the notifier after its poll in flight, then closes
|
||||
the broker and removes `host.json`. If either child dies on its own, the
|
||||
@@ -96,6 +129,70 @@ and whose command line is `…/packages/cli/src/cli.mjs bus start`.
|
||||
a broker was killed hard. Check, then remove the lock by hand
|
||||
(`packages/bus/README.md`).
|
||||
|
||||
### Sessions
|
||||
|
||||
The launcher (`src/launcher.mjs`) starts managed role sessions
|
||||
(`packages/seat/src/session.mjs`, `packages/harness`). It serves
|
||||
`<dataRoot>/bus-host/launch.sock` (0600), which the PM's `launch` tool
|
||||
calls through its runner. One request is one line,
|
||||
`{"cap": "…", "instance": "<role instance>"}`, at most 4096 bytes; the reply
|
||||
is `{"ok": true, "result": {run, instance, harness, model}}` or
|
||||
`{"ok": false, "error": "<code>"}`. Launches run one at a time. In order:
|
||||
|
||||
| Check | Refusal |
|
||||
|---|---|
|
||||
| the request is one JSON line with a string instance | `invalid-request` |
|
||||
| the capability is live and names a role of this business | `unauthenticated`, `not-a-role` |
|
||||
| the business file has a `launch` block | `launch-not-configured` |
|
||||
| `launch.instances` lists the instance (for `--pm`: it is `launch.by`) | `instance-not-listed` |
|
||||
| the instance isn't running | `instance-running` |
|
||||
| the instance resolves to `pi` or `claude-code` and a model | `unsupported-harness` |
|
||||
| its role has a contract file | `no-role-contract` |
|
||||
| its model names exactly one `launch.max` family | `unknown-model-family` |
|
||||
| that family has room; every running session counts, the PM's too | `capacity-full` |
|
||||
| the broker authorizes `role.launch` for the caller | the broker's code (`launch-revoked`, `decision-required`, …) |
|
||||
| the bundle builds and the runner starts | `launch-failed` |
|
||||
|
||||
Every refusal is an `action.refused` event (the broker's `refuse`, or its
|
||||
own refusal for the last broker check) and a line in the launch log. Then
|
||||
the launcher reads the instance's credential status from the broker
|
||||
(metadata only), builds the bundle in the run directory, spawns the runner
|
||||
under `unshare`, binds the outer process's pid and start time, adds the
|
||||
session to `sessions.json`, and writes the capability as one line on the
|
||||
runner's stdin. The capability is never on disk, in argv or in the
|
||||
environment. When the runner exits, the launcher sends `endLaunch` with the
|
||||
end reason from its exit code (`packages/harness/README.md`), which writes
|
||||
`session.ended`, releases the role and kills the run's capabilities.
|
||||
|
||||
The manifest records the Pi version from the repository's pin (the
|
||||
`@earendil-works/pi-coding-agent` dependency in the root `package.json`)
|
||||
and, for Claude Code, `claude --version`.
|
||||
|
||||
On disk, under the data root: `launches/<business>/<run>/` (0700) is the run
|
||||
directory; `launches/<business>.jsonl` (0600, append-only) logs every launch,
|
||||
refusal and end; `workspaces/<business>/<instance>/` (0700) is kept across
|
||||
launches; `bus-host/sessions.json` (0600) lists the running sessions.
|
||||
|
||||
When the host closes, the launcher stops taking requests and drops every
|
||||
open `launch.sock` connection, so a client that never ends its side can't
|
||||
hold the close. It sends SIGTERM to
|
||||
every runner (again each second, see "Limits"), waits up to 30 s, then
|
||||
SIGKILLs what is left and ends those launches as `killed`.
|
||||
|
||||
A host that starts reads the `sessions.json` the last host left. An entry
|
||||
for this business means that host died without ending the launch, and its
|
||||
run would keep holding the role. Before the launch socket opens, the
|
||||
launcher SIGKILLs each such session that still runs (same start time,
|
||||
`unshare` on its command line), binds the run again with the recorded
|
||||
identity and ends it as `host-lost`, which releases the role. A
|
||||
`sessions.json` that can't be read refuses the host start with exit 3; move
|
||||
it aside after checking which sessions still run.
|
||||
|
||||
`mosaic stop <run>` checks that the runner in `sessions.json` still runs with
|
||||
the recorded start time and that its command line is the runner's, then
|
||||
sends SIGTERM until it exits (30 s). A stale entry prints "not running" and
|
||||
exits 0; an unknown run exits 2.
|
||||
|
||||
### Trackers
|
||||
|
||||
`config.trackers[<business>]` is `{baseUrl, project, pollSeconds,
|
||||
@@ -223,7 +320,7 @@ exit 3, so a host never starts until someone decides whether it DMs.
|
||||
|---|---|
|
||||
| 0 | ok |
|
||||
| 1 | failed, or outcome unknown: check before retrying |
|
||||
| 2 | invalid input: unknown option, no such open decision, decision already closed |
|
||||
| 2 | invalid input: unknown option, no such open decision, decision already closed, no such run, not a runner |
|
||||
| 3 | refused or config problem: inside an agent run, human proof failed, unknown business, bad config, broker or notifier refused to start |
|
||||
| 4 | usage |
|
||||
|
||||
@@ -238,9 +335,22 @@ exit 3, so a host never starts until someone decides whether it DMs.
|
||||
- **The real human transport is untested here.** No test runs the real
|
||||
`human-cli.mjs`, because its proof needs a human shell. The live run
|
||||
covers it.
|
||||
- **`stop` is same-UID process control.** It checks the start time and the
|
||||
command line, not authority. Any process of the same user can signal a
|
||||
runner directly.
|
||||
- **Signals before the runner's handlers.** A session's runner is pid 1 of
|
||||
its PID namespace, and pid 1 ignores a signal it has no handler for. The
|
||||
runner installs its handlers first, but Node's startup leaves a short
|
||||
window, so the launcher and `stop` resend SIGTERM every second.
|
||||
- **If the host dies hard,** its sessions keep running until their broker
|
||||
polls fail, then exit 23, and their launches stay open in the bus until
|
||||
the next host starts and ends them as `host-lost` ("Sessions" above).
|
||||
With no next host, nothing ends them.
|
||||
- **`talk` and the launches verbs are tested only with the in-process
|
||||
transport,** as the other human commands are.
|
||||
- **Tracker boot is tested only without trackers.** The `trackers` boot case
|
||||
is tested once S3's broker change lands.
|
||||
|
||||
## Not in this piece
|
||||
|
||||
`mosaic talk` (S6) and the WebUI (S5).
|
||||
The WebUI (S5).
|
||||
|
||||
+113
-6
@@ -1,10 +1,12 @@
|
||||
#!/usr/bin/env node
|
||||
// mosaic inbox | decide | tasks | agents | trail, and mosaic bus start|stop|status.
|
||||
// See packages/cli/README.md. Exit codes are in src/errors.mjs.
|
||||
// mosaic inbox | decide | tasks | agents | trail | talk | stop | launches,
|
||||
// and mosaic bus start|stop|status. See packages/cli/README.md. Exit codes
|
||||
// are in src/errors.mjs.
|
||||
//
|
||||
// The human commands read and resolve through the broker's human transport
|
||||
// (src/transport.mjs), never the SQLite file. `bus start` is the trusted
|
||||
// host (src/host.mjs) the systemd unit mosaic-bus@<business> runs.
|
||||
// host (src/host.mjs) the systemd unit mosaic-bus@<business> runs; with
|
||||
// --pm it also launches the business's PM (src/launcher.mjs).
|
||||
|
||||
import { lstatSync, readFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
@@ -16,6 +18,9 @@ import { bootConfig, loadSystem, socketPath } from "./config.mjs";
|
||||
import { humanTransport, refuseInsideAgent } from "./transport.mjs";
|
||||
import { formatAgents, formatDecision, formatInbox, formatTasks, formatTrail, shortId } from "./format.mjs";
|
||||
import { hostStatus, readHostState, startHost, stopHost } from "./host.mjs";
|
||||
import { createLauncher } from "./launcher.mjs";
|
||||
import { readSessions, stopSession } from "../../seat/src/session.mjs";
|
||||
import { SeatError } from "../../seat/src/seat.mjs";
|
||||
|
||||
export const USAGE = `usage:
|
||||
mosaic inbox [--business <id>] [--json]
|
||||
@@ -23,7 +28,11 @@ export const USAGE = `usage:
|
||||
mosaic tasks [--business <id>] [--json]
|
||||
mosaic agents [--business <id>] [--json]
|
||||
mosaic trail <task|decision> [--business <id>] [--json]
|
||||
mosaic bus start <business>
|
||||
mosaic talk <role> <text> [--wait <seconds>] [--business <id>]
|
||||
mosaic stop <run>
|
||||
mosaic launches off|on [--business <id>]
|
||||
mosaic launches list [--json]
|
||||
mosaic bus start <business> [--pm]
|
||||
mosaic bus stop
|
||||
mosaic bus status [--json]`;
|
||||
|
||||
@@ -146,17 +155,112 @@ async function human(verb, rest, io, deps) {
|
||||
return print(await call("agents"), formatAgents);
|
||||
}
|
||||
|
||||
const TALK_WAIT_S = 900;
|
||||
const TALK_POLL_MS = 2000;
|
||||
|
||||
function formatMessage(m) {
|
||||
const meta = [m.class, m.in_reply_to ? `re ${shortId(m.in_reply_to)}` : null].filter(Boolean).join(", ");
|
||||
return `${m.from_role} (${meta}) ${shortId(m.id)}:\n${m.body.trimEnd()}\n`;
|
||||
}
|
||||
|
||||
// `mosaic talk <role> <text>`: a REQUEST to the role, then the human's inbox
|
||||
// until the reply arrives or --wait runs out. Every message received on the
|
||||
// way is printed and marked read, so a late reply shows up on the next talk.
|
||||
async function talk(rest, io, deps) {
|
||||
const parsed = parse(rest, { values: ["--business", "--wait"] });
|
||||
if (parsed.positional.length !== 2) throw usage("talk takes a role and one quoted text");
|
||||
const wait = parsed.values["--wait"] === undefined ? TALK_WAIT_S : Number(parsed.values["--wait"]);
|
||||
if (!Number.isInteger(wait) || wait < 0) throw usage("--wait takes whole seconds, 0 or more");
|
||||
refuseInsideAgent(io.env);
|
||||
const system = deps.system ?? loadSystem({ env: io.env });
|
||||
const business = businessFor(parsed, system.dataRoot);
|
||||
const call = deps.transport ? deps.transport(business) : humanTransport({ socket: socketPath(system.dataRoot), business, env: io.env });
|
||||
const [to, body] = parsed.positional;
|
||||
const sent = await call("message.send", { to, body, class: "REQUEST" });
|
||||
io.stdout.write(`sent ${shortId(sent.id)} to ${to}\n`);
|
||||
if (wait === 0) return;
|
||||
const pollMs = deps.talkPollMs ?? TALK_POLL_MS;
|
||||
const deadline = Date.now() + wait * 1000;
|
||||
for (;;) {
|
||||
let answered = false;
|
||||
for (const m of await call("message.receive")) {
|
||||
io.stdout.write(formatMessage(m));
|
||||
await call("message.read", { id: m.id });
|
||||
if (m.in_reply_to === sent.id) answered = true;
|
||||
}
|
||||
if (answered) return;
|
||||
if (Date.now() >= deadline) throw new CliError(`no reply from ${to} within ${wait} s; it may still come, and the next mosaic talk or mosaic inbox shows it`, 1);
|
||||
await new Promise((r) => setTimeout(r, pollMs));
|
||||
}
|
||||
}
|
||||
|
||||
// Seat errors carry the seat's exit codes; here they become CLI errors.
|
||||
async function seat(fn) {
|
||||
try {
|
||||
return await fn();
|
||||
} catch (e) {
|
||||
if (e instanceof SeatError) throw new CliError(e.message, e.exitCode === 2 ? 2 : 1);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
async function stop(rest, io, deps) {
|
||||
const parsed = parse(rest, {});
|
||||
if (parsed.positional.length !== 1) throw usage("stop takes one run id; see mosaic launches list");
|
||||
refuseInsideAgent(io.env);
|
||||
const system = deps.system ?? loadSystem({ env: io.env });
|
||||
const r = await seat(() => stopSession(system.dataRoot, parsed.positional[0], deps.stopTimeoutMs ? { timeoutMs: deps.stopTimeoutMs } : {}));
|
||||
io.stdout.write(r.stopped ? `stopped ${r.business}/${r.instance} run ${r.run}\n` : `${r.reason}\n`);
|
||||
}
|
||||
|
||||
async function launches(rest, io, deps) {
|
||||
const [sub, ...args] = rest;
|
||||
if (sub === "off" || sub === "on") {
|
||||
const parsed = parse(args, { values: ["--business"] });
|
||||
if (parsed.positional.length !== 0) throw usage(`launches ${sub} takes no arguments`);
|
||||
refuseInsideAgent(io.env);
|
||||
const system = deps.system ?? loadSystem({ env: io.env });
|
||||
const business = businessFor(parsed, system.dataRoot);
|
||||
const call = deps.transport ? deps.transport(business) : humanTransport({ socket: socketPath(system.dataRoot), business, env: io.env });
|
||||
await call(sub === "off" ? "launch.revoke" : "launch.restore");
|
||||
io.stdout.write(sub === "off" ? `launches off for ${business}: the PM's launch requests are refused until mosaic launches on; running sessions keep running\n` : `launches on for ${business}\n`);
|
||||
return;
|
||||
}
|
||||
if (sub === "list") {
|
||||
const parsed = parse(args, { flags: ["--json"] });
|
||||
if (parsed.positional.length !== 0) throw usage("launches list takes no arguments");
|
||||
const system = deps.system ?? loadSystem({ env: io.env });
|
||||
const list = await seat(() => readSessions(system.dataRoot));
|
||||
if (parsed.flags.has("--json")) return io.stdout.write(`${JSON.stringify(list, null, 2)}\n`);
|
||||
if (list.length === 0) return io.stdout.write("no sessions\n");
|
||||
const lines = list.map((s) => `${s.run} ${s.business}/${s.instance} ${s.harness} ${s.model} by ${s.launchedBy} since ${s.startedAt} ${s.live ? "running" : "not running (stale)"}`);
|
||||
return io.stdout.write(`${lines.join("\n")}\n`);
|
||||
}
|
||||
throw usage();
|
||||
}
|
||||
|
||||
async function bus(rest, io, deps) {
|
||||
const [sub, ...args] = rest;
|
||||
if (sub === "start") {
|
||||
const parsed = parse(args, {});
|
||||
const parsed = parse(args, { flags: ["--pm"] });
|
||||
if (parsed.positional.length !== 1) throw usage("bus start takes one business");
|
||||
const businessId = parsed.positional[0];
|
||||
const system = deps.system ?? loadSystem({ env: io.env });
|
||||
const boot = bootConfig({ system, businessId, env: io.env, warn: (w) => io.stderr.write(`mosaic-bus: warning: ${w}\n`) });
|
||||
const binding = readNotifyConfig(system.dataRoot, businessId);
|
||||
const host = await startHost({ boot, business: businessId, notifier: binding ? { binding } : null });
|
||||
io.stdout.write(`bus host up: business ${businessId}, socket ${host.path}, notifier ${binding ?? "off"}\n`);
|
||||
try {
|
||||
const launcher = createLauncher({ host, system, env: io.env, tracker: Boolean(boot.trackers?.[businessId]), ...(deps.launcher ?? {}) });
|
||||
await launcher.listen();
|
||||
io.stdout.write(`bus host up: business ${businessId}, socket ${host.path}, notifier ${binding ?? "off"}\n`);
|
||||
if (parsed.flags.has("--pm")) {
|
||||
const pm = await launcher.launchPm();
|
||||
io.stdout.write(`launched ${pm.instance}: run ${pm.run}, ${pm.harness} ${pm.model}\n`);
|
||||
}
|
||||
} catch (e) {
|
||||
await host.close(1);
|
||||
throw e;
|
||||
}
|
||||
const stop = () => host.close(0);
|
||||
process.on("SIGTERM", stop);
|
||||
process.on("SIGINT", stop);
|
||||
@@ -195,6 +299,9 @@ async function bus(rest, io, deps) {
|
||||
export async function main(argv, io = { env: process.env, stdin: process.stdin, stdout: process.stdout, stderr: process.stderr }, deps = {}) {
|
||||
const [verb, ...rest] = argv;
|
||||
if (["inbox", "decide", "tasks", "agents", "trail"].includes(verb)) return human(verb, rest, io, deps);
|
||||
if (verb === "talk") return talk(rest, io, deps);
|
||||
if (verb === "stop") return stop(rest, io, deps);
|
||||
if (verb === "launches") return launches(rest, io, deps);
|
||||
if (verb === "bus") return bus(rest, io, deps);
|
||||
if (verb === "-h" || verb === "--help") return io.stdout.write(`${USAGE}\n`);
|
||||
throw usage();
|
||||
|
||||
+79
-34
@@ -5,8 +5,12 @@
|
||||
// host runs, `<dataRoot>/bus-host/host.json` (0600) names it for `mosaic bus
|
||||
// stop|status` and for the human commands' default business.
|
||||
//
|
||||
// startHost() is also the in-process API S6 uses: bindLaunch(record) binds
|
||||
// a launched run's process identity and returns its capability.
|
||||
// startHost() is also the in-process API the S6 launcher (src/launcher.mjs)
|
||||
// uses: bindLaunch(record) binds a launched run's process identity and
|
||||
// returns its capability; op(message) runs one of the broker's launch ops
|
||||
// (identity, authorizeLaunch, refuse, endLaunch, credentialStatus); and
|
||||
// beforeClose(fn) runs fn before the host stops its children, while those
|
||||
// ops still work.
|
||||
|
||||
import { fork } from "node:child_process";
|
||||
import { once } from "node:events";
|
||||
@@ -14,35 +18,19 @@ import { existsSync, mkdirSync, readFileSync, renameSync, rmSync, writeFileSync
|
||||
import { join } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { CliError } from "./errors.mjs";
|
||||
import { cmdlineOf, startTimeOf } from "../../seat/src/proc.mjs";
|
||||
|
||||
const BROKER = fileURLToPath(new URL("../../bus/src/process.mjs", import.meta.url));
|
||||
const NOTIFIER = fileURLToPath(new URL("./notifier-process.mjs", import.meta.url));
|
||||
export const BOOT_TIMEOUT_MS = 30000;
|
||||
const START_TIMEOUT_MS = 15000;
|
||||
const CLOSE_TIMEOUT_MS = 20000;
|
||||
const REQUEST_TIMEOUT_MS = 10000;
|
||||
|
||||
export const hostDir = (dataRoot) => join(dataRoot, "bus-host");
|
||||
export const hostFile = (dataRoot) => join(hostDir(dataRoot), "host.json");
|
||||
|
||||
// `/proc/<pid>/stat` field 22, the start time in clock ticks. Null when the
|
||||
// process is gone or has exited and waits to be reaped (state Z).
|
||||
export function startTimeOf(pid) {
|
||||
try {
|
||||
const raw = readFileSync(`/proc/${pid}/stat`, "utf8");
|
||||
const fields = raw.slice(raw.lastIndexOf(")") + 2).split(" ");
|
||||
return fields[0] === "Z" ? null : fields[19];
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export function cmdlineOf(pid) {
|
||||
try {
|
||||
return readFileSync(`/proc/${pid}/cmdline`, "utf8").split("\0").filter(Boolean);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
export { cmdlineOf, startTimeOf };
|
||||
|
||||
// The state file, or null. `live` is true only when the pid still runs with
|
||||
// the recorded start time, so a recycled pid never counts as the host.
|
||||
@@ -107,8 +95,9 @@ export function watchChildren(children, onDeath) {
|
||||
|
||||
// boot: the {op:'boot'} config from bootConfig(). notifier: null, or
|
||||
// {binding, base?, pollMs?}; base and pollMs exist for the tests, and the
|
||||
// command line never sets them. Resolves once both children are up.
|
||||
export async function startHost({ boot, business, notifier = null, bootTimeoutMs = BOOT_TIMEOUT_MS, log = (l) => process.stderr.write(`mosaic-bus: ${l}\n`) }) {
|
||||
// command line never sets them, nor requestTimeoutMs. Resolves once both
|
||||
// children are up.
|
||||
export async function startHost({ boot, business, notifier = null, bootTimeoutMs = BOOT_TIMEOUT_MS, requestTimeoutMs = REQUEST_TIMEOUT_MS, log = (l) => process.stderr.write(`mosaic-bus: ${l}\n`) }) {
|
||||
const dataRoot = boot.dataRoot;
|
||||
const prior = readHostState(dataRoot);
|
||||
if (prior?.live) throw new CliError(`a bus host already runs for ${prior.business} (pid ${prior.pid})`, 3);
|
||||
@@ -157,27 +146,83 @@ export async function startHost({ boot, business, notifier = null, bootTimeoutMs
|
||||
rmSync(hostFile(dataRoot), { force: true });
|
||||
writeHostState(dataRoot, state);
|
||||
|
||||
// stopping: close() has started (no new close, child deaths are expected).
|
||||
// closing: the before-close hooks have run, so no more broker requests.
|
||||
let stopping = false;
|
||||
let closing = false;
|
||||
let finish;
|
||||
const done = new Promise((r) => (finish = r));
|
||||
const hooks = [];
|
||||
|
||||
// Replies from process.mjs carry no request id, so requests go one at a time.
|
||||
// Requests go one at a time, each with an id that process.mjs echoes. A
|
||||
// reply that misses the wait, or carries an id no request is waiting for,
|
||||
// breaks the channel: the waiting request and every later one refuse, and
|
||||
// the host stops with exit 1 so the unit restarts it. A late reply can
|
||||
// never answer a later request.
|
||||
let queue = Promise.resolve();
|
||||
function bindLaunch(record) {
|
||||
let seq = 0;
|
||||
let pending = null;
|
||||
let broken = null;
|
||||
const fail = (code, text) => Object.assign(new CliError(text, 1), { code });
|
||||
function breakChannel(why) {
|
||||
if (broken) return;
|
||||
broken = why;
|
||||
if (pending) {
|
||||
clearTimeout(pending.timer);
|
||||
pending.reject(fail("broker-channel-broken", `broker channel broken: ${why}`));
|
||||
pending = null;
|
||||
}
|
||||
if (stopping) return;
|
||||
log(`broker channel broken (${why}); stopping the host`);
|
||||
close(1);
|
||||
}
|
||||
broker.on("message", (m) => {
|
||||
if (pending && m?.id === pending.id) {
|
||||
clearTimeout(pending.timer);
|
||||
const { resolve } = pending;
|
||||
pending = null;
|
||||
resolve(m);
|
||||
} else breakChannel(pending ? "reply for another request" : "reply with no request waiting");
|
||||
});
|
||||
broker.once("exit", () => {
|
||||
if (!pending) return;
|
||||
clearTimeout(pending.timer);
|
||||
pending.reject(fail("broker-exited", "broker exited before it replied"));
|
||||
pending = null;
|
||||
});
|
||||
function request(message, { what, pick }) {
|
||||
const run = queue.then(async () => {
|
||||
if (closing) throw new CliError("bus host is closing", 1);
|
||||
const r = firstReply(broker, 10000, "broker");
|
||||
broker.send({ op: "bindLaunch", record });
|
||||
const m = await r;
|
||||
if (m?.ok !== true) throw new CliError(`launch bind refused: ${m?.error ?? "bind-refused"}`, 3);
|
||||
return m.launch;
|
||||
if (broken) throw fail("broker-channel-broken", `broker channel broken: ${broken}`);
|
||||
if (closing || !broker.connected) throw fail("host-closing", "bus host is closing");
|
||||
const id = ++seq;
|
||||
const m = await new Promise((resolve, reject) => {
|
||||
const timer = setTimeout(() => breakChannel(`no reply within ${Math.round(requestTimeoutMs / 1000)} s`), requestTimeoutMs);
|
||||
pending = { id, resolve, reject, timer };
|
||||
broker.send({ ...message, id });
|
||||
});
|
||||
if (m?.ok !== true) {
|
||||
const code = typeof m?.error === "string" ? m.error : `${what}-refused`;
|
||||
throw Object.assign(new CliError(`${what} refused: ${code}`, 3), { code });
|
||||
}
|
||||
return pick(m);
|
||||
});
|
||||
queue = run.catch(() => {});
|
||||
return run;
|
||||
}
|
||||
const bindLaunch = (record) => request({ op: "bindLaunch", record }, { what: "launch bind", pick: (m) => m.launch });
|
||||
const op = (message) => request(message, { what: message.op, pick: (m) => m.result });
|
||||
const beforeClose = (fn) => void hooks.push(fn);
|
||||
|
||||
async function close(code = 0) {
|
||||
if (closing) return done;
|
||||
if (stopping) return done;
|
||||
stopping = true;
|
||||
for (const fn of hooks) {
|
||||
try {
|
||||
await fn();
|
||||
} catch (e) {
|
||||
log(`before close: ${e.message}`);
|
||||
}
|
||||
}
|
||||
closing = true;
|
||||
let result = code;
|
||||
if (notify) {
|
||||
@@ -197,12 +242,12 @@ export async function startHost({ boot, business, notifier = null, bootTimeoutMs
|
||||
// restarts the pair rather than run a broker without its notifier.
|
||||
// Runs after `queue` exists, since close() awaits it.
|
||||
watchChildren({ broker, notifier: notify }, (name, code, signal) => {
|
||||
if (closing) return;
|
||||
if (stopping) return;
|
||||
log(`${name} exited (${signal ?? code}); stopping the host`);
|
||||
close(1);
|
||||
});
|
||||
|
||||
return Object.freeze({ path: ready.path, business, bindLaunch, close, done, pids: Object.freeze({ broker: broker.pid, notifier: notify?.pid ?? null }) });
|
||||
return Object.freeze({ path: ready.path, dataRoot, business, bindLaunch, op, beforeClose, close, done, pids: Object.freeze({ broker: broker.pid, notifier: notify?.pid ?? null }) });
|
||||
}
|
||||
|
||||
// `mosaic bus stop`: SIGTERM to the recorded host, after checking that the
|
||||
|
||||
@@ -0,0 +1,420 @@
|
||||
// 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) and a supported
|
||||
// harness and model (unsupported-harness)
|
||||
// 5. its model is in exactly one launch.max family (unknown-model-family)
|
||||
// 6. that family has room; every running session counts, the PM's too
|
||||
// (capacity-full)
|
||||
// 7. the broker authorizes role.launch for the caller (its own codes:
|
||||
// launch-revoked, gated, ...)
|
||||
//
|
||||
// Steps 2 to 6 refuse through the broker's refuse(), so every refusal is an
|
||||
// action.refused event; step 7's refusals the broker records itself. Then
|
||||
// the host builds the bundle, spawns the runner, binds the run's process
|
||||
// identity (the outer unshare 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.
|
||||
|
||||
import { spawnSync } from "node:child_process";
|
||||
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 { buildBundle, sessionModel } from "../../harness/src/bundle.mjs";
|
||||
import {
|
||||
ADAPTERS,
|
||||
appendLaunchLog,
|
||||
endReason,
|
||||
family,
|
||||
launchSocketPath,
|
||||
newRun,
|
||||
readSessions,
|
||||
Registry,
|
||||
runnerOf,
|
||||
sessionEnv,
|
||||
spawnSession,
|
||||
UNSHARE,
|
||||
workspaceDir,
|
||||
writeSessionFile,
|
||||
} from "../../seat/src/session.mjs";
|
||||
import { cmdlineOf, startTimeOf } from "../../seat/src/proc.mjs";
|
||||
import { REPO, rolesDir } from "./config.mjs";
|
||||
import { CliError } from "./errors.mjs";
|
||||
|
||||
const PI_PACKAGE = "@earendil-works/pi-coding-agent";
|
||||
const STOP_TIMEOUT_MS = 30000;
|
||||
const LINE_MAX = 4096;
|
||||
|
||||
// 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;
|
||||
}
|
||||
}
|
||||
|
||||
// The exact pi pin from the repository's package.json.
|
||||
export function piVersion(repo = REPO) {
|
||||
try {
|
||||
const pkg = JSON.parse(readFileSync(join(repo, "package.json"), "utf8"));
|
||||
return pkg.dependencies?.[PI_PACKAGE] ?? pkg.devDependencies?.[PI_PACKAGE] ?? null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
// `claude --version`, or null when it doesn't answer.
|
||||
export function claudeVersion(env = process.env) {
|
||||
const r = spawnSync("claude", ["--version"], { encoding: "utf8", timeout: 10000, env });
|
||||
return r.status === 0 ? r.stdout.trim().split("\n")[0] || null : null;
|
||||
}
|
||||
|
||||
// host: startHost()'s handle. tracker: true when the broker has task verbs for
|
||||
// the business. adapters, namespace, sessionDefaults, versions and
|
||||
// stopTimeoutMs exist for the tests; the command line never sets them.
|
||||
export function createLauncher({
|
||||
host,
|
||||
system,
|
||||
env = process.env,
|
||||
tracker = false,
|
||||
adapters = ADAPTERS,
|
||||
namespace = true,
|
||||
sessionDefaults = {},
|
||||
versions = null,
|
||||
stopTimeoutMs = STOP_TIMEOUT_MS,
|
||||
log = (l) => process.stderr.write(`mosaic-bus: ${l}\n`),
|
||||
}) {
|
||||
const dataRoot = system.dataRoot;
|
||||
const 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);
|
||||
const children = new Map();
|
||||
let chain = Promise.resolve();
|
||||
let closed = false;
|
||||
let server = null;
|
||||
const sockets = new Set();
|
||||
|
||||
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 6. Returns what the launch needs. Only `--pm` may name
|
||||
// launch.by itself.
|
||||
function check(instance, { pm = false } = {}) {
|
||||
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");
|
||||
if (registry.running(b.id, instance)) throw new Refusal("instance-running");
|
||||
let resolved, model;
|
||||
try {
|
||||
resolved = resolveInstance({ system: vars, business: b, instance });
|
||||
model = sessionModel(resolved);
|
||||
} catch (e) {
|
||||
throw new Refusal("unsupported-harness", e.message);
|
||||
}
|
||||
if (!resolved.contract) throw new Refusal("no-role-contract", `role ${resolved.definition} has no contract file`);
|
||||
const fam = family(model.model, b.launch.max);
|
||||
if (!fam) throw new Refusal("unknown-model-family", `model ${model.model} is in none, or more than one, of ${Object.keys(b.launch.max).join(", ")}`);
|
||||
if (registry.count(b.id, fam) >= b.launch.max[fam]) throw new Refusal("capacity-full", `${fam} sessions: ${registry.count(b.id, fam)} of ${b.launch.max[fam]}`);
|
||||
return { resolved, model, fam };
|
||||
}
|
||||
|
||||
// Bundle, spawn, bind, register, hand over the capability.
|
||||
async function start({ instance, launchedBy, resolved, model, fam }) {
|
||||
const credentials = await host.op({ op: "credentialStatus", business: b.id, role: instance });
|
||||
const { run, runDir } = newRun(dataRoot, b.id);
|
||||
const workspace = workspaceDir(dataRoot, b.id, instance);
|
||||
mkdirSync(workspace, { recursive: true, mode: 0o700 });
|
||||
const sessionDir = join(runDir, "session");
|
||||
mkdirSync(sessionDir, { mode: 0o700 });
|
||||
const toolSocket = join(runDir, "tools.sock");
|
||||
const turnMarker = join(runDir, "turn.done");
|
||||
const v = versions ?? { pi: piVersion(), claude: model.harness === "claude-code" ? claudeVersion(env) : null };
|
||||
const bundle = buildBundle({
|
||||
dir: join(runDir, "bundle"),
|
||||
resolved,
|
||||
contract: readFileSync(resolved.contract, "utf8"),
|
||||
arbiter: Object.values(b.arbiters).includes(instance),
|
||||
tracker,
|
||||
credentials,
|
||||
session: { run, launchedBy, workspace, toolSocket, turnMarker },
|
||||
piVersion: v.pi,
|
||||
claudeVersion: v.claude,
|
||||
});
|
||||
const { manifest, ...files } = bundle.files;
|
||||
writeSessionFile(runDir, {
|
||||
...sessionDefaults,
|
||||
business: b.id,
|
||||
instance,
|
||||
run,
|
||||
launchedBy,
|
||||
harness: model.harness,
|
||||
provider: model.provider,
|
||||
model: model.model,
|
||||
brokerSocket: host.path,
|
||||
launchSocket: launchSocketPath(dataRoot),
|
||||
workspace,
|
||||
sessionDir,
|
||||
adapter: adapters[model.harness],
|
||||
bundle: files,
|
||||
toolSocket,
|
||||
turnMarker,
|
||||
});
|
||||
|
||||
const child = spawnSession({ runDir, env: sessionEnv(env, run), namespace });
|
||||
let bound = false;
|
||||
let entry = null;
|
||||
const exited = new Promise((resolve) => child.once("close", (code, signal) => resolve({ 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, { namespace });
|
||||
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, pid: child.pid, startTime: outerStart });
|
||||
bound = true;
|
||||
entry = {
|
||||
business: b.id, instance, run, launchedBy, harness: model.harness, model: model.model, family: fam,
|
||||
pid: child.pid, startTime: outerStart, runnerPid: runner.pid, runnerStartTime: runner.startTime, runDir, startedAt: new Date().toISOString(),
|
||||
};
|
||||
registry.add(entry);
|
||||
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 {
|
||||
await endRun(run, reason, signal ? null : code);
|
||||
registry.remove(b.id, run);
|
||||
record({ event: "end", instance, run, reason, exitCode: code, signal });
|
||||
log(`run ${run} (${instance}) ended: ${reason}`);
|
||||
} catch (e) {
|
||||
log(`run ${run}: ${e.message}`);
|
||||
} finally {
|
||||
children.delete(run);
|
||||
}
|
||||
});
|
||||
record({ event: "launch", instance, run, launchedBy, harness: model.harness, model: model.model, family: fam, pid: child.pid, runnerPid: entry.runnerPid, manifest });
|
||||
return { run, instance, harness: model.harness, model: model.model };
|
||||
}
|
||||
|
||||
async function endRun(run, reason, exitCode) {
|
||||
try {
|
||||
await host.op({ op: "endLaunch", record: { business: b.id, run, reason, exitCode } });
|
||||
} catch (e) {
|
||||
log(`run ${run}: endLaunch refused: ${e.code ?? e.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
// 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" };
|
||||
}
|
||||
let checked;
|
||||
try {
|
||||
if (typeof instance !== "string") throw new Refusal("invalid-request");
|
||||
checked = 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 });
|
||||
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 });
|
||||
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, ...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 });
|
||||
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() {
|
||||
for (const s of leftover) {
|
||||
if (startTimeOf(s.pid) === s.startTime && (cmdlineOf(s.pid) ?? [])[0] === UNSHARE[0]) {
|
||||
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 });
|
||||
} 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, 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);
|
||||
}
|
||||
|
||||
// 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 (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 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, close };
|
||||
}
|
||||
+23
@@ -0,0 +1,23 @@
|
||||
// A bus host with its launcher in a process of its own, so a test can kill
|
||||
// it hard. It launches the PM, prints the launch result as one JSON line and
|
||||
// runs until it is killed. argv: the adapter to use for both harnesses.
|
||||
|
||||
import { bootConfig, loadSystem } from "../../src/config.mjs";
|
||||
import { startHost } from "../../src/host.mjs";
|
||||
import { createLauncher } from "../../src/launcher.mjs";
|
||||
|
||||
const [adapter] = process.argv.slice(2);
|
||||
const system = loadSystem({ env: process.env });
|
||||
const boot = bootConfig({ system, businessId: "acme", env: process.env });
|
||||
const host = await startHost({ boot, business: "acme", log: () => {} });
|
||||
const launcher = createLauncher({
|
||||
host,
|
||||
system,
|
||||
env: process.env,
|
||||
adapters: { pi: adapter, "claude-code": adapter },
|
||||
sessionDefaults: { pollInterval: 100 },
|
||||
versions: { pi: "0.85.1", claude: null },
|
||||
log: () => {},
|
||||
});
|
||||
await launcher.listen();
|
||||
process.stdout.write(`${JSON.stringify(await launcher.launchPm())}\n`);
|
||||
@@ -203,6 +203,88 @@ test("a second host for the same data root refuses with exit 3 while the first r
|
||||
assert.equal(await host.close(0), 0);
|
||||
});
|
||||
|
||||
test("a broker reply that misses the wait breaks the channel: the late reply answers nothing, later requests refuse, the host exits 1", { timeout: 30000 }, async (t) => {
|
||||
const root = tmp(t);
|
||||
const f = fixture(root);
|
||||
makeDeployment(root);
|
||||
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
||||
const logs = [];
|
||||
const host = await startHost({ boot, business: "acme", requestTimeoutMs: 1000, log: (l) => logs.push(l) });
|
||||
// Hooks run in order: the broker resumes before the close, which a
|
||||
// stopped broker would hold.
|
||||
t.after(() => {
|
||||
try {
|
||||
process.kill(host.pids.broker, "SIGCONT");
|
||||
} catch {}
|
||||
});
|
||||
t.after(() => host.close(0));
|
||||
const record = (role, run) => ({ business: "acme", role, run, harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) });
|
||||
const coder = await host.bindLaunch(record("coder", "coder-run"));
|
||||
assert.equal((await host.op({ op: "identity", cap: coder.cap })).role, "coder");
|
||||
|
||||
// Stall the broker with a request waiting and a bind queued behind it.
|
||||
process.kill(host.pids.broker, "SIGSTOP");
|
||||
const code = (p) => p.then((v) => ({ answered: v }), (e) => e.code);
|
||||
const first = code(host.op({ op: "identity", cap: "bogus" }));
|
||||
const second = code(host.bindLaunch(record("reviewer", "reviewer-run")));
|
||||
assert.equal(await first, "broker-channel-broken");
|
||||
assert.equal(await second, "broker-channel-broken", "a queued request never reaches the broker");
|
||||
process.kill(host.pids.broker, "SIGCONT");
|
||||
|
||||
// The broker answers the stalled request late; nothing takes that reply.
|
||||
assert.equal(await host.done, 1);
|
||||
assert.ok(logs.some((l) => /^broker channel broken \(no reply within 1 s\); stopping the host$/.test(l)), logs.join("\n"));
|
||||
assert.equal(await code(host.op({ op: "identity", cap: coder.cap })), "broker-channel-broken");
|
||||
});
|
||||
|
||||
test("a broker reply with another request's id, or none, breaks the channel and the host exits 1", { timeout: 30000 }, async (t) => {
|
||||
// The broker echoes whatever id it's sent, so changing the id on the way
|
||||
// out gives the host the reply a confused broker would send.
|
||||
let what, tamper;
|
||||
spySends(t, (m) => m?.cap === "tamper" && tamper(m));
|
||||
for ([what, tamper] of [
|
||||
["another id", (m) => (m.id += 1000)],
|
||||
["no id", (m) => delete m.id],
|
||||
]) {
|
||||
const root = tmp(t);
|
||||
const f = fixture(root);
|
||||
makeDeployment(root);
|
||||
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
||||
const logs = [];
|
||||
const host = await startHost({ boot, business: "acme", log: (l) => logs.push(l) });
|
||||
t.after(() => host.close(0));
|
||||
const record = (role, run) => ({ business: "acme", role, run, harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) });
|
||||
const coder = await host.bindLaunch(record("coder", "coder-run"));
|
||||
await assert.rejects(host.op({ op: "identity", cap: "tamper" }), (e) => e.code === "broker-channel-broken", what);
|
||||
await assert.rejects(host.op({ op: "identity", cap: coder.cap }), (e) => e.code === "broker-channel-broken", what);
|
||||
assert.equal(await host.done, 1, what);
|
||||
assert.ok(logs.includes("broker channel broken (reply for another request); stopping the host"), `${what}: ${logs.join("\n")}`);
|
||||
}
|
||||
});
|
||||
|
||||
test("a broker reply with no request waiting breaks the channel and the host exits 1", { timeout: 30000 }, async (t) => {
|
||||
// A request sent to the real broker past the host's queue: its reply
|
||||
// arrives when nothing is waiting (Darkwing round 2, Mr).
|
||||
const children = [];
|
||||
const onChild = ({ process: child }) => children.push(child);
|
||||
subscribe("child_process", onChild);
|
||||
t.after(() => unsubscribe("child_process", onChild));
|
||||
const root = tmp(t);
|
||||
const f = fixture(root);
|
||||
makeDeployment(root);
|
||||
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
||||
const logs = [];
|
||||
const host = await startHost({ boot, business: "acme", log: (l) => logs.push(l) });
|
||||
t.after(() => host.close(0));
|
||||
const record = { business: "acme", role: "coder", run: "coder-run", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) };
|
||||
const coder = await host.bindLaunch(record);
|
||||
children.find((c) => c.pid === host.pids.broker).send({ op: "identity", cap: coder.cap, id: 1_000_000 });
|
||||
const late = new Promise((r) => setTimeout(r, 5000, "still running").unref());
|
||||
assert.equal(await Promise.race([host.done, late]), 1);
|
||||
assert.ok(logs.includes("broker channel broken (reply with no request waiting); stopping the host"), logs.join("\n"));
|
||||
await assert.rejects(host.op({ op: "identity", cap: coder.cap }), (e) => e.code === "broker-channel-broken");
|
||||
});
|
||||
|
||||
test("a notifier that refuses stops the broker and the host refuses with exit 3", async (t) => {
|
||||
const root = tmp(t);
|
||||
const f = fixture(root);
|
||||
|
||||
@@ -0,0 +1,394 @@
|
||||
// The S6 launcher inside a real bus host: real broker child, real runners
|
||||
// under unshare, the fake adapter from packages/harness standing in for the
|
||||
// harness. The PM launches a coder through its `launch` tool; every host
|
||||
// refusal comes back over the launch socket and lands in the launch log.
|
||||
|
||||
import { test } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { spawn, spawnSync } from "node:child_process";
|
||||
import { once } from "node:events";
|
||||
import { createInterface } from "node:readline";
|
||||
import { connect } from "node:net";
|
||||
import { existsSync, mkdirSync, readFileSync, readdirSync, statSync, writeFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { Client } from "../../bus/src/client.mjs";
|
||||
import { launchLogFile, launchSocketPath, readSessions, Registry, sessionsFile, UNSHARE } from "../../seat/src/session.mjs";
|
||||
import { bootConfig, loadSystem } from "../src/config.mjs";
|
||||
import { startHost, startTimeOf } from "../src/host.mjs";
|
||||
import { createLauncher } from "../src/launcher.mjs";
|
||||
import { fixture, tmp } from "./helpers.mjs";
|
||||
|
||||
const FAKE = fileURLToPath(new URL("../../harness/tests/fixtures/fake-adapter.mjs", import.meta.url));
|
||||
const HOST = fileURLToPath(new URL("./fixtures/launch-host.mjs", import.meta.url));
|
||||
const NAMESPACES = spawnSync(UNSHARE[0], [...UNSHARE.slice(1), "true"]).status === 0;
|
||||
const skip = NAMESPACES ? false : "unprivileged user and PID namespaces are unavailable here";
|
||||
|
||||
// pm opus, coder and reviewer sonnet, one of each family; cto is listed but
|
||||
// runs the system model, which is in no family.
|
||||
function edit(doc) {
|
||||
doc.roles.coder.vars.model = "claude-sonnet-5-5";
|
||||
doc.roles.reviewer.vars = { harness: "pi", model: "claude-sonnet-5-5" };
|
||||
doc.launch = { by: "pm", instances: ["coder", "reviewer", "cto"], max: { opus: 1, sonnet: 1 } };
|
||||
return doc;
|
||||
}
|
||||
|
||||
async function until(fn, ms = 15000) {
|
||||
const end = Date.now() + ms;
|
||||
while (Date.now() < end) {
|
||||
const v = await fn();
|
||||
if (v) return v;
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
}
|
||||
throw new Error("timed out waiting");
|
||||
}
|
||||
|
||||
// A config and data root with the fake adapter; setup() starts a host on it.
|
||||
function prepare(t, more = (doc) => doc) {
|
||||
const root = tmp(t, "mosaic-launch-");
|
||||
const f = fixture(root, "acme", (doc) => more(edit(doc)));
|
||||
mkdirSync(f.dataRoot, { mode: 0o700 });
|
||||
const adapter = join(root, "adapter.sh");
|
||||
writeFileSync(adapter, `exec '${process.execPath}' '${FAKE}'\n`, { mode: 0o644 });
|
||||
return { root, f, adapter };
|
||||
}
|
||||
|
||||
async function setup(t, more = (doc) => doc, { root, f, adapter } = prepare(t, more), options = {}) {
|
||||
const system = loadSystem({ env: f.env });
|
||||
const boot = bootConfig({ system, businessId: "acme", env: f.env });
|
||||
const logs = [];
|
||||
const host = await startHost({ boot, business: "acme", log: (l) => logs.push(l) });
|
||||
let launcher;
|
||||
try {
|
||||
launcher = createLauncher({
|
||||
host,
|
||||
system,
|
||||
env: f.env,
|
||||
adapters: { pi: adapter, "claude-code": adapter },
|
||||
sessionDefaults: { pollInterval: 100 },
|
||||
versions: { pi: "0.85.1", claude: null },
|
||||
log: (l) => logs.push(l),
|
||||
...options,
|
||||
});
|
||||
await launcher.listen();
|
||||
} catch (e) {
|
||||
await host.close(1);
|
||||
throw e;
|
||||
}
|
||||
let closed = false;
|
||||
const close = async () => {
|
||||
if (closed) return host.done;
|
||||
closed = true;
|
||||
return host.close(0);
|
||||
};
|
||||
t.after(close);
|
||||
// A test caller: a cto launch bound to this process, claimed.
|
||||
const cto = await host.bindLaunch({ business: "acme", role: "cto", run: "cto-test", harness: "claude-code", pid: process.pid, startTime: startTimeOf(process.pid) });
|
||||
const client = new Client({ path: host.path, cap: cto.cap });
|
||||
await client.call("role.claim");
|
||||
return { root, f, host, launcher, cto, client, logs, close };
|
||||
}
|
||||
|
||||
// One request on the launch socket, one reply line.
|
||||
function ask(path, line) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const s = connect(path);
|
||||
let buf = "";
|
||||
s.on("connect", () => s.write(line));
|
||||
s.on("data", (b) => (buf += b));
|
||||
s.on("end", () => resolve(JSON.parse(buf)));
|
||||
s.on("error", reject);
|
||||
});
|
||||
}
|
||||
|
||||
const log = (f) => readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l));
|
||||
|
||||
test("bus start --pm: the PM runs under its own PID namespace, registered, with a manifest", { skip, timeout: 30000 }, async (t) => {
|
||||
const { f, launcher, host, cto, close } = await setup(t);
|
||||
const sock = launchSocketPath(f.dataRoot);
|
||||
assert.equal(statSync(sock).mode & 0o777, 0o600);
|
||||
|
||||
const pm = await launcher.launchPm();
|
||||
assert.deepEqual({ ...pm, run: undefined }, { run: undefined, instance: "pm", harness: "pi", model: "claude-opus-5-5" });
|
||||
assert.match(pm.run, /^r[0-9a-f]{12}$/);
|
||||
|
||||
const [s] = readSessions(f.dataRoot);
|
||||
assert.equal(s.run, pm.run);
|
||||
assert.equal(s.live, true);
|
||||
assert.equal(s.family, "opus");
|
||||
assert.equal(s.launchedBy, "jason");
|
||||
assert.notEqual(s.pid, s.runnerPid, "the runner is the unshare process's child");
|
||||
assert.equal(statSync(sessionsFile(f.dataRoot)).mode & 0o777, 0o600);
|
||||
|
||||
const runDir = s.runDir;
|
||||
assert.equal(statSync(runDir).mode & 0o777, 0o700);
|
||||
const session = JSON.parse(readFileSync(join(runDir, "session.json"), "utf8"));
|
||||
assert.equal(session.launchedBy, "jason");
|
||||
assert.equal(session.brokerSocket, host.path);
|
||||
assert.equal(session.launchSocket, sock);
|
||||
const manifest = JSON.parse(readFileSync(join(runDir, "bundle", "manifest.json"), "utf8"));
|
||||
assert.equal(manifest.piVersion, "0.85.1");
|
||||
assert.equal(manifest.claudeVersion, null);
|
||||
const [launched] = log(f).filter((e) => e.event === "launch");
|
||||
assert.equal(launched.run, pm.run);
|
||||
assert.equal(launched.manifest, join(runDir, "bundle", "manifest.json"));
|
||||
assert.equal(manifest.model, "claude-opus-5-5");
|
||||
|
||||
// A second PM is refused before anything spawns.
|
||||
await assert.rejects(launcher.launchPm(), /PM launch refused: instance-running/);
|
||||
|
||||
// The session's environment carries the run id and no capability; the
|
||||
// capability is nowhere on disk or on any session process's command line.
|
||||
await until(() => existsSync(join(runDir, "runner.log")) && readSessions(f.dataRoot)[0]?.live);
|
||||
const proc = (pid, what) => {
|
||||
try {
|
||||
return readFileSync(`/proc/${pid}/${what}`, "utf8");
|
||||
} catch {
|
||||
return "";
|
||||
}
|
||||
};
|
||||
const environ = proc(s.runnerPid, "environ").split("\0");
|
||||
assert.ok(environ.includes(`MOSAIC_RUN_ID=${pm.run}`));
|
||||
assert.ok(!environ.some((v) => v.startsWith("MOSAIC_CONFIG=")), "only allowlisted variables pass");
|
||||
const files = (dir) => readdirSync(dir, { recursive: true, withFileTypes: true }).filter((d) => d.isFile()).map((d) => join(d.parentPath, d.name));
|
||||
const capOf = cto.cap; // a known capability shape, to prove the scan finds nothing like it
|
||||
for (const file of files(f.dataRoot)) {
|
||||
const text = readFileSync(file, "latin1");
|
||||
assert.ok(!text.includes(capOf), `${file} holds no capability`);
|
||||
}
|
||||
for (const pid of [s.pid, s.runnerPid]) {
|
||||
assert.notEqual(proc(pid, "cmdline"), "");
|
||||
assert.ok(!/[0-9a-f]{64}/.test(proc(pid, "cmdline")), "no capability in argv");
|
||||
assert.ok(!/[0-9a-f]{64}/.test(proc(pid, "environ")), "no capability in the environment");
|
||||
}
|
||||
|
||||
// Closing the host stops the session and ends the launch.
|
||||
assert.equal(await close(), 0);
|
||||
assert.equal(startTimeOf(s.runnerPid), null);
|
||||
assert.deepEqual(readSessions(f.dataRoot), []);
|
||||
assert.equal(existsSync(sock), false);
|
||||
const end = log(f).find((e) => e.event === "end");
|
||||
assert.equal(end.run, pm.run);
|
||||
assert.equal(end.reason, "stopped");
|
||||
assert.equal(end.exitCode, 0);
|
||||
});
|
||||
|
||||
test("the PM launches a coder through its launch tool; the coder answers; refusals name their code", { skip, timeout: 30000 }, async (t) => {
|
||||
const { f, launcher, client, cto } = await setup(t);
|
||||
const sock = launchSocketPath(f.dataRoot);
|
||||
const pm = await launcher.launchPm();
|
||||
|
||||
// cto asks pm to launch a coder; pm's turn calls the launch tool.
|
||||
const sent = await client.call("message.send", { to: "pm", class: "REQUEST", body: 'TOOL launch {"instance":"coder"}' });
|
||||
const reply = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === sent.id));
|
||||
assert.equal(reply.from_role, "pm");
|
||||
assert.equal(reply.class, "RESULT");
|
||||
const result = JSON.parse(reply.body);
|
||||
assert.equal(result.instance, "coder");
|
||||
assert.equal(result.model, "claude-sonnet-5-5");
|
||||
const coder = readSessions(f.dataRoot).find((s) => s.instance === "coder");
|
||||
assert.equal(coder.run, result.run);
|
||||
assert.equal(coder.launchedBy, "pm");
|
||||
assert.equal(coder.live, true);
|
||||
|
||||
// The coder claimed its role and answers its own messages.
|
||||
const hello = await client.call("message.send", { to: "coder", class: "REQUEST", body: "hello" });
|
||||
const echo = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === hello.id));
|
||||
assert.equal(echo.from_role, "coder");
|
||||
assert.match(echo.body, /^echo: Message .* from cto, class REQUEST:\n\nhello/);
|
||||
|
||||
// Host refusals, asked with the cto capability (steps 2 to 6 come before
|
||||
// the broker's role.launch check, which cto would fail).
|
||||
const refusals = [
|
||||
["pm", "instance-not-listed"],
|
||||
["nobody", "instance-not-listed"],
|
||||
["coder", "instance-running"],
|
||||
["reviewer", "capacity-full"],
|
||||
["cto", "unknown-model-family"],
|
||||
];
|
||||
for (const [instance, code] of refusals) {
|
||||
assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: cto.cap, instance })}\n`), { ok: false, error: code }, instance);
|
||||
}
|
||||
// A listed, free instance with room still needs role.launch: cto lacks it.
|
||||
// Stop the coder first so sonnet has room again.
|
||||
const coderRun = coder.run;
|
||||
process.kill(coder.runnerPid, "SIGTERM");
|
||||
await until(() => !readSessions(f.dataRoot).some((s) => s.run === coderRun) && log(f).some((e) => e.event === "end" && e.run === coderRun));
|
||||
const denied = await ask(sock, `${JSON.stringify({ cap: cto.cap, instance: "reviewer" })}\n`);
|
||||
assert.equal(denied.ok, false);
|
||||
assert.notEqual(denied.error, "capacity-full");
|
||||
|
||||
// Malformed requests.
|
||||
assert.deepEqual(await ask(sock, "not json\n"), { ok: false, error: "invalid-request" });
|
||||
assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: "short", instance: "coder" })}\n`), { ok: false, error: "unauthenticated" });
|
||||
assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: "0".repeat(64), instance: "coder" })}\n`), { ok: false, error: "unauthenticated" });
|
||||
assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: cto.cap, instance: 7 })}\n`), { ok: false, error: "invalid-request" });
|
||||
|
||||
// The launch log has the launches, every refusal, and the coder's end.
|
||||
const entries = log(f);
|
||||
assert.deepEqual(entries.filter((e) => e.event === "launch").map((e) => [e.instance, e.launchedBy]), [["pm", "jason"], ["coder", "pm"]]);
|
||||
const refused = entries.filter((e) => e.event === "refused").map((e) => [e.instance, e.code]);
|
||||
for (const r of refusals) assert.ok(refused.some(([i, c]) => i === r[0] && c === r[1]), `${r} is logged`);
|
||||
const end = entries.find((e) => e.event === "end" && e.run === coderRun);
|
||||
assert.equal(end.reason, "stopped");
|
||||
assert.ok(readSessions(f.dataRoot).some((s) => s.run === pm.run));
|
||||
});
|
||||
|
||||
test("a runner that stops at once ends its launch with the runner's reason", { skip, timeout: 30000 }, async (t) => {
|
||||
// pm's tracker token has expired: the runner's founder-credential stop
|
||||
// exits 20 before role.claim, and the host records founder-credentials.
|
||||
const { f, launcher } = await setup(t, (doc) => {
|
||||
doc.roles.pm.credentials.vikunja.expires = "2000-01-01";
|
||||
return doc;
|
||||
});
|
||||
const pm = await launcher.launchPm();
|
||||
const end = await until(() => existsSync(launchLogFile(f.dataRoot, "acme")) && log(f).find((e) => e.event === "end" && e.run === pm.run));
|
||||
assert.equal(end.reason, "founder-credentials");
|
||||
assert.equal(end.exitCode, 20);
|
||||
assert.deepEqual(readSessions(f.dataRoot), []);
|
||||
assert.match(readFileSync(join(f.dataRoot, "launches", "acme", pm.run, "runner.log"), "utf8"), /vikunja/);
|
||||
});
|
||||
|
||||
for (const exited of [false, true]) {
|
||||
test(`a host that died hard leaves its sessions to the next host, which ${exited ? "finds one already exited (23)" : "kills them"} and ends their runs so the role can be launched again`, { skip, timeout: 30000 }, async (t) => {
|
||||
const given = prepare(t);
|
||||
const { f } = given;
|
||||
const old = spawn(process.execPath, [HOST, given.adapter], { env: f.env, stdio: ["ignore", "pipe", "inherit"] });
|
||||
t.after(() => old.kill("SIGKILL"));
|
||||
const [line] = await once(createInterface({ input: old.stdout }), "line");
|
||||
const first = JSON.parse(line);
|
||||
const [left] = readSessions(f.dataRoot);
|
||||
assert.equal(left.run, first.run);
|
||||
// Killed before the claim, the runner would exit 23 on its own.
|
||||
const runnerLog = join(f.dataRoot, "launches", "acme", first.run, "runner.log");
|
||||
await until(() => existsSync(runnerLog) && readFileSync(runnerLog, "utf8").includes("claimed by run"));
|
||||
assert.equal(left.live, true);
|
||||
old.kill("SIGKILL");
|
||||
await once(old, "exit");
|
||||
// The broker child exits on the lost IPC channel; the session runs on.
|
||||
await until(() => !existsSync(join(f.dataRoot, "bus", "writer.lock")));
|
||||
assert.equal(startTimeOf(left.runnerPid), left.runnerStartTime);
|
||||
if (!exited) {
|
||||
// Stopped, the leftover neither polls the dead broker nor answers
|
||||
// SIGTERM: only the next host's SIGKILL ends it.
|
||||
for (const pid of [left.pid, left.runnerPid]) process.kill(pid, "SIGSTOP");
|
||||
t.after(() => {
|
||||
for (const [pid, start] of [[left.pid, left.startTime], [left.runnerPid, left.runnerStartTime]]) {
|
||||
if (startTimeOf(pid) === start) process.kill(pid, "SIGKILL");
|
||||
}
|
||||
});
|
||||
}
|
||||
if (exited) {
|
||||
await until(() => startTimeOf(left.pid) === null);
|
||||
assert.match(readFileSync(runnerLog, "utf8"), /broker unreachable after 30 tries[\s\S]*exiting 23/);
|
||||
assert.equal(readSessions(f.dataRoot)[0].live, false);
|
||||
}
|
||||
|
||||
const s = await setup(t, undefined, given);
|
||||
assert.equal(startTimeOf(left.pid), null, "the old session process is gone");
|
||||
assert.equal(startTimeOf(left.runnerPid), null, "and its runner with it");
|
||||
assert.ok(s.logs.includes(`run ${first.run} (pm) from an earlier host ended: host-lost`), s.logs.join("\n"));
|
||||
const ended = (await s.client.call("trail", { subject: first.run })).filter((r) => r.kind === "session.ended");
|
||||
assert.equal(ended.length, 1);
|
||||
assert.deepEqual(ended[0].body, { reason: "host-lost", exitCode: null, released: true });
|
||||
const log = readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l));
|
||||
assert.deepEqual(log.filter((e) => e.event === "end").map((e) => [e.run, e.reason]), [[first.run, "host-lost"]]);
|
||||
assert.deepEqual(readSessions(f.dataRoot), []);
|
||||
|
||||
// Without the end, the old run would still hold pm and this runner would exit 21.
|
||||
const again = await s.launcher.launchPm();
|
||||
await until(async () => (await s.client.call("agents")).some((a) => a.role === "pm" && a.holder_run === again.run));
|
||||
await s.client.call("role.release"); // the test's own cto claim, so the next host's can claim again
|
||||
assert.equal(await s.close(), 0);
|
||||
|
||||
// Died after the end but before sessions.json changed: the run stays ended.
|
||||
const { live, ...entry } = left;
|
||||
new Registry(f.dataRoot).add(entry);
|
||||
const third = await setup(t, undefined, given);
|
||||
assert.ok(third.logs.includes(`run ${first.run} (pm) from an earlier host was already ended`), third.logs.join("\n"));
|
||||
assert.equal((await third.client.call("trail", { subject: first.run })).filter((r) => r.kind === "session.ended").length, 1);
|
||||
assert.equal(await third.close(), 0);
|
||||
});
|
||||
}
|
||||
|
||||
test("a malformed sessions.json refuses the host start with exit 3 and stays as it was", { timeout: 30000 }, async (t) => {
|
||||
const given = prepare(t);
|
||||
const file = sessionsFile(given.f.dataRoot);
|
||||
mkdirSync(join(given.f.dataRoot, "bus-host"), { mode: 0o700 });
|
||||
writeFileSync(file, "{", { mode: 0o600 });
|
||||
await assert.rejects(setup(t, undefined, given), (e) => e.exitCode === 3 && /sessions file is unreadable.*move the file aside/.test(e.message));
|
||||
assert.equal(readFileSync(file, "utf8"), "{");
|
||||
assert.equal(existsSync(join(given.f.dataRoot, "bus", "writer.lock")), false, "the host closed");
|
||||
});
|
||||
|
||||
const within = (p, ms, what) => Promise.race([p, new Promise((_, reject) => setTimeout(() => reject(new Error(`${what} took over ${ms} ms`)), ms).unref())]);
|
||||
|
||||
test("an over-long launch request is refused at once, not at the 10 s idle timeout", { timeout: 30000 }, async (t) => {
|
||||
const { f } = await setup(t);
|
||||
const s = connect(launchSocketPath(f.dataRoot));
|
||||
t.after(() => s.destroy());
|
||||
let buf = "";
|
||||
s.on("data", (b) => (buf += b));
|
||||
s.write("x".repeat(5000));
|
||||
await within(once(s, "end"), 3000, "the refusal");
|
||||
assert.deepEqual(JSON.parse(buf), { ok: false, error: "invalid-request" });
|
||||
});
|
||||
|
||||
test("a launch client that never closes its side doesn't hold the host's close", { timeout: 30000 }, async (t) => {
|
||||
const { f, close } = await setup(t);
|
||||
const path = launchSocketPath(f.dataRoot);
|
||||
// One got its reply and never ends; one never sends anything.
|
||||
const answered = connect({ path, allowHalfOpen: true });
|
||||
const silent = connect({ path, allowHalfOpen: true });
|
||||
const connected = once(silent, "connect");
|
||||
const gone = Promise.all([once(answered, "close"), once(silent, "close")]);
|
||||
const drop = () => {
|
||||
answered.destroy();
|
||||
silent.destroy();
|
||||
};
|
||||
t.after(drop);
|
||||
let buf = "";
|
||||
answered.on("data", (b) => (buf += b));
|
||||
answered.write("not json\n");
|
||||
await once(answered, "end");
|
||||
assert.deepEqual(JSON.parse(buf), { ok: false, error: "invalid-request" });
|
||||
await within(connected, 2000, "connect");
|
||||
// On a failure the clients go, so the host can still close and the test
|
||||
// fails instead of hanging.
|
||||
const code = await within(close(), 5000, "close").catch((e) => {
|
||||
drop();
|
||||
throw e;
|
||||
});
|
||||
assert.equal(code, 0);
|
||||
assert.equal(existsSync(path), false);
|
||||
answered.end();
|
||||
silent.end();
|
||||
await within(gone, 2000, "the clients' close");
|
||||
});
|
||||
|
||||
test("a runner that ignores SIGTERM is killed when the host closes", { skip, timeout: 60000 }, async (t) => {
|
||||
const { f, launcher, close } = await setup(t, undefined, undefined, { stopTimeoutMs: 1000 });
|
||||
const pm = await launcher.launchPm();
|
||||
const runnerLog = join(f.dataRoot, "launches", "acme", pm.run, "runner.log");
|
||||
await until(() => existsSync(runnerLog) && readFileSync(runnerLog, "utf8").includes("claimed by run"));
|
||||
const [s] = readSessions(f.dataRoot);
|
||||
// Stopped from outside its namespace, it can't act on SIGTERM.
|
||||
process.kill(s.runnerPid, "SIGSTOP");
|
||||
const kill = () => {
|
||||
if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGKILL");
|
||||
};
|
||||
t.after(kill);
|
||||
const started = Date.now();
|
||||
const code = await within(close(), 15000, "close").catch((e) => {
|
||||
kill();
|
||||
throw e;
|
||||
});
|
||||
assert.equal(code, 0);
|
||||
assert.ok(Date.now() - started >= 1000, "it waited for the stop timeout");
|
||||
assert.equal(startTimeOf(s.pid), null);
|
||||
assert.equal(startTimeOf(s.runnerPid), null);
|
||||
assert.deepEqual(readSessions(f.dataRoot), []);
|
||||
const ends = log(f).filter((e) => e.event === "end" && e.run === pm.run);
|
||||
assert.equal(ends.length, 1);
|
||||
assert.equal(ends[0].signal, "SIGKILL");
|
||||
});
|
||||
@@ -0,0 +1,132 @@
|
||||
// The S6 human verbs: talk, stop and launches. The human transport is the
|
||||
// in-process broker from helpers.mjs, because the real one refuses to run
|
||||
// under an agent (src/transport.mjs). A real stop of a real runner is in
|
||||
// packages/seat/tests/session.test.mjs and tests/launcher.test.mjs.
|
||||
|
||||
import { test } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { main } from "../src/cli.mjs";
|
||||
import { CliError } from "../src/errors.mjs";
|
||||
import { startTimeOf } from "../src/host.mjs";
|
||||
import { Registry } from "../../seat/src/session.mjs";
|
||||
import { broker, io, tmp } from "./helpers.mjs";
|
||||
|
||||
function setup(t, deps = {}) {
|
||||
const bus = broker(t);
|
||||
const dataRoot = tmp(t);
|
||||
const run = async (argv, x = io(), more = {}) => {
|
||||
try {
|
||||
await main(argv, x, { system: { dataRoot }, transport: bus.transport, talkPollMs: 20, ...deps, ...more });
|
||||
return { code: 0, io: x };
|
||||
} catch (e) {
|
||||
if (!(e instanceof CliError)) throw e;
|
||||
return { code: e.exitCode, error: e.message, io: x };
|
||||
}
|
||||
};
|
||||
return { ...bus, dataRoot, run };
|
||||
}
|
||||
|
||||
const call = (b, cap, verb, args = {}) => b.request(cap, { verb, args });
|
||||
|
||||
test("talk sends a REQUEST, prints and reads everything that arrives, and stops at the reply", async (t) => {
|
||||
const s = setup(t);
|
||||
const cto = s.b.bindLaunch({ business: "demo", role: "cto", run: "cto-run", harness: "pi" });
|
||||
call(s.b, cto, "role.claim");
|
||||
// The coder answers on the second poll; a cto note arrives first and is printed too.
|
||||
let polls = 0;
|
||||
const transport = (business) => {
|
||||
const inner = s.transport(business);
|
||||
return async (verb, args) => {
|
||||
if (verb === "message.receive" && ++polls === 2) {
|
||||
const [m] = call(s.b, s.coder, "message.receive");
|
||||
call(s.b, cto, "message.send", { to: m.from_role, body: "fyi", class: "INFO" });
|
||||
call(s.b, s.coder, "message.send", { to: m.from_role, body: "done: 3 files", class: "RESULT", in_reply_to: m.id });
|
||||
}
|
||||
return inner(verb, args);
|
||||
};
|
||||
};
|
||||
const r = await s.run(["talk", "coder", "fix the build", "--business", "demo"], io(), { transport });
|
||||
assert.equal(r.code, 0, r.error);
|
||||
const sent = s.calls.find((c) => c.verb === "message.send");
|
||||
assert.deepEqual(sent.args, { to: "coder", body: "fix the build", class: "REQUEST" });
|
||||
assert.match(r.io.out.text, /^sent [0-9a-f]{8} to coder\n/);
|
||||
assert.match(r.io.out.text, /cto \(INFO\) [0-9a-f]{8}:\nfyi\n/);
|
||||
assert.match(r.io.out.text, /coder \(RESULT, re [0-9a-f]{8}\) [0-9a-f]{8}:\ndone: 3 files\n$/);
|
||||
assert.equal(s.calls.filter((c) => c.verb === "message.read").length, 2);
|
||||
assert.deepEqual(await s.transport("demo")("message.receive"), [], "both were marked read");
|
||||
});
|
||||
|
||||
test("talk --wait 0 only sends; no reply within --wait exits 1 and says where it will show", async (t) => {
|
||||
const s = setup(t);
|
||||
const quick = await s.run(["talk", "coder", "hello", "--wait", "0", "--business", "demo"]);
|
||||
assert.equal(quick.code, 0);
|
||||
assert.deepEqual(s.calls.map((c) => c.verb), ["message.send"]);
|
||||
const slow = await s.run(["talk", "coder", "hello", "--wait", "1", "--business", "demo"]);
|
||||
assert.equal(slow.code, 1);
|
||||
assert.match(slow.error, /no reply from coder within 1 s; .*mosaic inbox/);
|
||||
});
|
||||
|
||||
test("talk, stop and launches off/on refuse inside an agent run; bad arguments are usage errors", async (t) => {
|
||||
const s = setup(t);
|
||||
const agent = () => io({ ...io().env, CLAUDECODE: "1" });
|
||||
for (const argv of [["talk", "coder", "hi", "--business", "demo"], ["stop", "r0123456789ab"], ["launches", "off", "--business", "demo"], ["launches", "on", "--business", "demo"]]) {
|
||||
assert.equal((await s.run(argv, agent())).code, 3, argv.join(" "));
|
||||
}
|
||||
const usage = [
|
||||
["talk", "coder", "--business", "demo"],
|
||||
["talk", "coder", "a", "b", "--business", "demo"],
|
||||
["talk", "coder", "hi", "--wait", "-1", "--business", "demo"],
|
||||
["talk", "coder", "hi", "--wait", "1.5", "--business", "demo"],
|
||||
["stop"],
|
||||
["stop", "a", "b"],
|
||||
["stop", "a", "--business", "demo"],
|
||||
["launches"],
|
||||
["launches", "maybe"],
|
||||
["launches", "off", "x", "--business", "demo"],
|
||||
["launches", "list", "x"],
|
||||
];
|
||||
for (const argv of usage) assert.equal((await s.run(argv)).code, 4, argv.join(" "));
|
||||
assert.equal(s.calls.length, 0);
|
||||
});
|
||||
|
||||
test("launches off and on go to the broker and change the business's launch state", async (t) => {
|
||||
const s = setup(t);
|
||||
const state = () => s.store.get("SELECT state FROM launch_state WHERE business=?", "demo")?.state;
|
||||
const off = await s.run(["launches", "off", "--business", "demo"]);
|
||||
assert.equal(off.code, 0, off.error);
|
||||
assert.match(off.io.out.text, /^launches off for demo: .*running sessions keep running\n$/);
|
||||
assert.equal(state(), "revoked");
|
||||
const on = await s.run(["launches", "on", "--business", "demo"]);
|
||||
assert.equal(on.io.out.text, "launches on for demo\n");
|
||||
assert.notEqual(state(), "revoked");
|
||||
assert.deepEqual(s.calls.map((c) => c.verb), ["launch.revoke", "launch.restore"]);
|
||||
});
|
||||
|
||||
test("launches list reads sessions.json and marks a stale entry; stop reports it and refuses a non-runner", async (t) => {
|
||||
const s = setup(t);
|
||||
assert.equal((await s.run(["launches", "list"])).io.out.text, "no sessions\n");
|
||||
const r = new Registry(s.dataRoot);
|
||||
const base = { business: "demo", harness: "pi", model: "claude-sonnet-5-5", launchedBy: "pm", startedAt: "2026-10-09T00:00:00.000Z", runnerPid: process.pid };
|
||||
r.add({ ...base, instance: "coder", run: "r000000000001", runnerStartTime: startTimeOf(process.pid) });
|
||||
r.add({ ...base, instance: "reviewer", run: "r000000000002", runnerStartTime: "1" });
|
||||
const list = await s.run(["launches", "list"]);
|
||||
assert.equal(
|
||||
list.io.out.text,
|
||||
"r000000000001 demo/coder pi claude-sonnet-5-5 by pm since 2026-10-09T00:00:00.000Z running\n" +
|
||||
"r000000000002 demo/reviewer pi claude-sonnet-5-5 by pm since 2026-10-09T00:00:00.000Z not running (stale)\n",
|
||||
);
|
||||
const json = JSON.parse((await s.run(["launches", "list", "--json"])).io.out.text);
|
||||
assert.deepEqual(json.map((e) => [e.run, e.live]), [["r000000000001", true], ["r000000000002", false]]);
|
||||
|
||||
const stale = await s.run(["stop", "r000000000002"]);
|
||||
assert.equal(stale.code, 0);
|
||||
assert.equal(stale.io.out.text, "run r000000000002 is not running (stale entry)\n");
|
||||
const unknown = await s.run(["stop", "nope"]);
|
||||
assert.equal(unknown.code, 2);
|
||||
assert.match(unknown.error, /no running session has run id nope/);
|
||||
// This test process is live with the recorded start time but is not a runner.
|
||||
const notRunner = await s.run(["stop", "r000000000001"]);
|
||||
assert.equal(notRunner.code, 2);
|
||||
assert.match(notRunner.error, /is not a session runner; refusing to signal it/);
|
||||
assert.equal(s.calls.length, 0, "stop and list never touch the bus");
|
||||
});
|
||||
Reference in New Issue
Block a user