jochen
4295aad88e
Awaited at import, a bucket not yet on the bus — or a grant the bus had not reloaded — answered after the runtime's 10s handshake, and the module was left unserved on its first assignment. Also cites module state as ADR 0201, as hq main numbers it (folds #256).
218 lines
11 KiB
TypeScript
218 lines
11 KiB
TypeScript
// claude-code's bundle (novox/hq design 36, ADR 0183). The node's runtime launches it over stdio, as the
|
|
// operator account (ADR 0193), and is its bus (ADR 0198): it asks tools, emits and consumes through the
|
|
// runtime. It is given its state directory and two files the mesh renders into it (ADR 0192), beside the
|
|
// runtime's own words. **stdout is the MCP channel**: everything this module says, it says on stderr.
|
|
//
|
|
// At start it renders the agent's managed directory, asks the licence manager for this node's token,
|
|
// begins watching the credentials file for a login, takes the manager's licence events, and watches the
|
|
// module's `servers` state — every node's MCP server registrations (novox/hq ADR 0201). node.ts holds the
|
|
// logic.
|
|
|
|
import { readFileSync, watchFile } from "node:fs";
|
|
import { spawnSync } from "node:child_process";
|
|
import { join } from "node:path";
|
|
import { registerModuleTools, type ToolDefinition } from "@novox/mesh-sdk/tools";
|
|
import { broker } from "@novox/mesh-sdk/messaging";
|
|
import { on } from "@novox/mesh-sdk/events";
|
|
import { state } from "@novox/mesh-sdk/state";
|
|
|
|
import {
|
|
MANAGED_DIR, SEAT, ServerView, concerns, keypair, offerLogin, onServerChange, pull, readJson, registerServer,
|
|
registered, renderNow, type Ask, type Paths, type Registration, type ServerChange, type ServerState, type WriteManaged,
|
|
} from "../node.js";
|
|
import { grantOf, holdsLogin, readCredentials } from "../grant.js";
|
|
import { createHash } from "node:crypto";
|
|
|
|
const say = (line: string) => console.error(`[claude-code] ${line}`);
|
|
const fingerprint = (s: string) => "sha256:" + createHash("sha256").update(s).digest("hex").slice(0, 16);
|
|
|
|
function pathsFrom(env: NodeJS.ProcessEnv): Paths | null {
|
|
const state = env.MESH_CLAUDE_CODE_STATE, facts = env.MESH_CLAUDE_CODE_FACTS;
|
|
const settings = env.MESH_CLAUDE_CODE_SETTINGS, home = env.MESH_OPERATOR_HOME, node = env.MESH_NODE;
|
|
if (!state || !facts || !settings || !home || !node) return null;
|
|
return { state, facts, settings, home, node };
|
|
}
|
|
|
|
/** Write one managed file as root, only when its content changed. */
|
|
const writeManaged: WriteManaged = (name, content) => {
|
|
const path = join(MANAGED_DIR, name);
|
|
try {
|
|
if (readFileSync(path, "utf8") === content) return `${name}: unchanged`;
|
|
} catch {
|
|
/* absent */
|
|
}
|
|
const asRoot = process.getuid?.() === 0;
|
|
const cmd = asRoot ? ["install", "-D", "-m", "0644", "/dev/stdin", path] : ["sudo", "-n", "install", "-D", "-m", "0644", "/dev/stdin", path];
|
|
const r = spawnSync(cmd[0], cmd.slice(1), { input: content, encoding: "utf8" });
|
|
if (r.status !== 0) {
|
|
throw new Error(`${name}: could not be written to ${MANAGED_DIR} (${(r.stderr || r.error?.message || "").trim()}); ` +
|
|
`the module writes there through the operator account's passwordless sudo`);
|
|
}
|
|
return `${name}: written`;
|
|
};
|
|
|
|
/** A tool on the bus, through the runtime; its MCP answer read back as JSON where it is JSON. */
|
|
const ask: Ask = async (address, args) => {
|
|
const answer = (await broker().request<Record<string, unknown>, { content?: { text?: string }[]; isError?: boolean }>(address, args)) ?? {};
|
|
const text = answer.content?.map((c) => c.text ?? "").join("") ?? "";
|
|
if (answer.isError) throw new Error(`${address}: ${text}`);
|
|
try {
|
|
return JSON.parse(text);
|
|
} catch {
|
|
return text;
|
|
}
|
|
};
|
|
|
|
/** The nodes claude-code runs on, from the controller's list of modules — for the register tool's question. */
|
|
async function nodesRunningMe(): Promise<string[]> {
|
|
const out = await ask("mesh-controller.modules", {});
|
|
const text = typeof out === "string" ? out : String((out as { output?: string })?.output ?? "");
|
|
const line = text.split("\n").find((l) => /^claude-code\s/.test(l)) ?? "";
|
|
const on = line.split(" on ")[1] ?? "";
|
|
return on.trim() === "nothing" ? [] : on.split(",").map((s) => s.trim()).filter(Boolean);
|
|
}
|
|
|
|
function status(p: Paths): Record<string, unknown> {
|
|
const creds = readCredentials(join(p.home, ".claude", ".credentials.json"));
|
|
const grant = grantOf(creds);
|
|
const managed = ["managed-mcp.json", "managed-settings.json", "CLAUDE.md"].map((f) => {
|
|
try {
|
|
return { file: join(MANAGED_DIR, f), fingerprint: fingerprint(readFileSync(join(MANAGED_DIR, f), "utf8")) };
|
|
} catch {
|
|
return { file: join(MANAGED_DIR, f), fingerprint: null };
|
|
}
|
|
});
|
|
return {
|
|
node: p.node,
|
|
licence: readJson(join(p.state, "licence.json"), null),
|
|
token: grant ? { fingerprint: fingerprint(grant.accessToken), expiresAt: new Date(grant.expiresAt).toISOString(),
|
|
loginWaiting: holdsLogin(creds) } : null,
|
|
managed,
|
|
registered: Object.keys(registered(p)),
|
|
};
|
|
}
|
|
|
|
/** The module's MCP servers on the bus (ADR 0201): its own state, which every node of it watches. */
|
|
const servers = () => state<Record<string, unknown>>("servers") as unknown as ServerState;
|
|
|
|
/** What this node takes from that state, kept from the watch. One per process. */
|
|
let view: ServerView | null = null;
|
|
const viewOf = (p: Paths) => (view ??= new ServerView(p));
|
|
|
|
function tools(p: Paths): ToolDefinition[] {
|
|
const nodesArg = { type: "string", description: 'more nodes: "all" for every node running claude-code, or a comma-separated list; absent is this node only' };
|
|
const nodesOf = (v: unknown): Registration["nodes"] =>
|
|
v === undefined || v === "" ? undefined : v === "all" ? "all" : String(v).split(",").map((s) => s.trim()).filter(Boolean);
|
|
return [
|
|
{
|
|
name: "claude_code_status",
|
|
description: "Claude Code on this machine as the mesh configured it: the licence it holds and when its token expires, the managed files, the MCP servers registered here. Fingerprints only, never a token.",
|
|
input: {},
|
|
run: async () => status(p),
|
|
},
|
|
{
|
|
name: "claude_code_render",
|
|
description: "Write Claude Code's managed directory now, from the mesh's facts, this module's settings and the servers registered here.",
|
|
input: {},
|
|
run: async () => ({ rendered: renderNow(p, writeManaged) }),
|
|
},
|
|
{
|
|
name: "claude_code_pull",
|
|
description: "Ask the licence manager for this node's current token now and apply it, rather than waiting for its next event.",
|
|
input: {},
|
|
run: async () => pull(p, ask, writeManaged),
|
|
},
|
|
{
|
|
name: "claude_code_mcp_list",
|
|
description: "The MCP servers registered through this module: those that apply on this node (beside the console, `mesh`, and those set in the module's settings), and every registration on the mesh, by key — `all.<server>` for every node, `<node>.<server>` for one.",
|
|
input: {},
|
|
run: async () => ({ here: registered(p), everywhere: await servers().keys() }),
|
|
},
|
|
{
|
|
name: "claude_code_mcp_register",
|
|
description: "Register an MCP server with Claude Code on this node, every node, or a list — an http/sse server by url, or a stdio server by command. Kept on the bus, so a node that joins later takes it too. Never put a secret in env or headers: the mesh refuses one.",
|
|
input: {
|
|
name: { type: "string", description: "the server's name: letters, digits, - and _" },
|
|
type: { type: "string", description: "http, sse or stdio (default stdio when a command is given, http when a url is)" },
|
|
url: { type: "string", description: "an http or sse server's url" },
|
|
command: { type: "string", description: "a stdio server's program" },
|
|
args: { type: "array", description: "a stdio server's arguments" },
|
|
env: { type: "object", description: "a stdio server's environment" },
|
|
headers: { type: "object", description: "an http server's headers" },
|
|
nodes: nodesArg,
|
|
},
|
|
run: async (a) => {
|
|
const entry: Record<string, unknown> = { type: a.type ?? (a.url ? "http" : "stdio") };
|
|
for (const k of ["url", "command", "args", "env", "headers"]) if (a[k] !== undefined) entry[k] = a[k];
|
|
return registerServer(p, { name: String(a.name ?? ""), entry, nodes: nodesOf(a.nodes) }, servers(), viewOf(p), writeManaged, nodesRunningMe);
|
|
},
|
|
},
|
|
{
|
|
name: "claude_code_mcp_unregister",
|
|
description: "Remove an MCP server registered through this module, on this node or more.",
|
|
input: { name: { type: "string", description: "the server's name" }, nodes: nodesArg },
|
|
run: async (a) => registerServer(p, { name: String(a.name ?? ""), nodes: nodesOf(a.nodes) }, servers(), viewOf(p), writeManaged, nodesRunningMe),
|
|
},
|
|
];
|
|
}
|
|
|
|
registerModuleTools("claude-code", (env) => {
|
|
const p = pathsFrom(env);
|
|
if (!p) return [];
|
|
try {
|
|
keypair(p);
|
|
for (const line of renderNow(p, writeManaged)) if (!line.endsWith("unchanged")) say(line);
|
|
} catch (err) {
|
|
say(err instanceof Error ? err.message : String(err));
|
|
}
|
|
return tools(p);
|
|
});
|
|
|
|
// Launched by the runtime: the bus is there from the first line (ADR 0198). Outside it — a test, a
|
|
// build — nothing below runs.
|
|
const p = process.env.MESH_SERVED_MODULE ? pathsFrom(process.env) : null;
|
|
if (p) {
|
|
const loud = (what: string) => (err: unknown) => say(`${what}: ${err instanceof Error ? err.message : String(err)}`);
|
|
|
|
void on<{ licence?: string; node?: string }>("claude-licence-manager.licence.*", async (event) => {
|
|
if (!concerns(p, event.type, event.body ?? {})) return;
|
|
say(`${event.type} — asking ${SEAT} for this node's token`);
|
|
say(JSON.stringify(await pull(p, ask, writeManaged).catch((e) => ({ failed: String(e) }))));
|
|
}).catch(loud("the licence events"));
|
|
|
|
// Every node's MCP servers: the whole current set first, then each change (ADR 0201). **Not awaited
|
|
// where the module is imported**: the runtime waits on the handshake, and a bucket that is not on the
|
|
// bus yet — or a grant the bus has not reloaded — answers late; awaited here, that left the bundle
|
|
// unable to answer `initialize` in time and the module unserved (found on its first assignment). So it
|
|
// watches beside the handshake and asks again until the state answers; until then the managed
|
|
// directory holds what the file kept from the last run.
|
|
const watchServers = (attempt = 0): void => {
|
|
state<Record<string, unknown>>("servers").watch((c) => {
|
|
try {
|
|
const done = onServerChange(viewOf(p), c as ServerChange, p, writeManaged);
|
|
if (done) say(done);
|
|
} catch (err) {
|
|
loud(`taking ${c.op} ${c.key}`)(err); // the view took it; the next render writes it
|
|
}
|
|
}).then(
|
|
() => say(`watching the MCP servers${attempt ? ` (after ${attempt} refusal(s))` : ""}`),
|
|
(err) => {
|
|
const wait = [2, 5, 10, 30][attempt] ?? 60;
|
|
say(`the MCP servers cannot be watched yet (${err instanceof Error ? err.message : String(err)}); asking again in ${wait}s`);
|
|
setTimeout(() => watchServers(attempt + 1), wait * 1000);
|
|
});
|
|
};
|
|
watchServers();
|
|
|
|
// Catch up once at start: a node that was off takes its current token now.
|
|
void pull(p, ask, writeManaged).then((r) => say(`at start: ${JSON.stringify(r)}`), loud("asking for this node's token at start"));
|
|
|
|
// A login: a refresh token appears in the credentials file. Polled, because the file is replaced by
|
|
// rename and a watch on the old inode would go quiet.
|
|
const credentials = join(p.home, ".claude", ".credentials.json");
|
|
watchFile(credentials, { interval: 5000 }, () => {
|
|
void offerLogin(p, ask).then((r) => { if (r) say(`a login here was offered to ${SEAT}: ${JSON.stringify(r)}`); },
|
|
loud("offering a login to the licence manager"));
|
|
});
|
|
}
|