diff --git a/apps/server/package.json b/apps/server/package.json index ca74368348ea..a0ebf1005d87 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -16,10 +16,11 @@ "type": "module", "scripts": { "dev": "node --watch src/bin.ts", - "build:bundle": "vp pack && vp pack src/service-launcher.ts --out-dir dist --no-clean", + "build:bundle": "vp pack && vp pack src/service-launcher.ts --out-dir dist --no-clean && node ../../native/process-placement/build.mjs", "start": "node dist/bin.mjs", "typecheck": "tsgo --noEmit", - "test": "vp test run" + "test": "vp test run", + "test:placement-build": "node --test ../../native/process-placement/build.test.mjs" }, "dependencies": { "@anthropic-ai/claude-agent-sdk": "^0.3.170", diff --git a/apps/server/src/cloud/bootService.test.ts b/apps/server/src/cloud/bootService.test.ts index 3ccdecd861e1..fb9c6f64d8ce 100644 --- a/apps/server/src/cloud/bootService.test.ts +++ b/apps/server/src/cloud/bootService.test.ts @@ -719,3 +719,111 @@ it.layer(NodeServices.layer)("boot service install", (it) => { }), ); }); + +it("places the stable launcher before execution and leaves server priority to its child", () => { + const unit = BootService.renderBootServiceUnit({ + nodePath: "/usr/bin/node", + launcherPath: "/home/user/.t3/runtime/service-launcher.mjs", + baseDir: "/home/user/.t3", + logPath: "/home/user/.t3/logs/service.log", + unitPath: "/home/user/.config/systemd/user/t3code.service", + processPlacement: { + version: 1, + helperPath: "/opt/t3/process-placement", + helperSha256: "a".repeat(64), + helperDevice: "1", + helperInode: "10", + control: { path: "/sys/fs/cgroup/t3/control", device: "1", inode: "2" }, + workload: { path: "/sys/fs/cgroup/t3/workload", device: "1", inode: "3" }, + serverPriority: { nice: -15, resetOnFork: true }, + }, + }); + expect(unit).toContain( + `ExecStart=:/opt/t3/process-placement --helper-path /opt/t3/process-placement --helper-sha256 ${"a".repeat(64)} --helper-device 1 --helper-inode 10 --role control --cgroup /sys/fs/cgroup/t3/control --device 1 --inode 2 -- /usr/bin/node /home/user/.t3/runtime/service-launcher.mjs`, + ); + expect(unit).toContain("Environment=T3_PROCESS_PLACEMENT="); + expect(unit.split("\n").find((line) => line.startsWith("ExecStart="))).not.toContain( + "--server-priority", + ); +}); + +it("boots through an explicitly bound external producer before starting the stable launcher", () => { + const unit = BootService.renderBootServiceUnit({ + nodePath: "/usr/bin/node", + launcherPath: "/home/user/.t3/runtime/service-launcher.mjs", + baseDir: "/home/user/.t3", + logPath: "/home/user/.t3/logs/service.log", + unitPath: "/home/user/.config/systemd/user/t3code.service", + processPlacementBootstrap: { + version: 1, + executablePath: "/usr/bin/python3.13", + executableSha256: "a".repeat(64), + scriptPath: "/opt/t3/bootstrap.py", + scriptSha256: "b".repeat(64), + policyPath: "/opt/t3/policy.json", + policySha256: "c".repeat(64), + helperPath: "/opt/t3/process-placement", + helperSha256: "d".repeat(64), + }, + }); + expect(unit).toContain("Delegate=cpu memory pids"); + expect(unit).toContain("Environment=T3_PROCESS_PLACEMENT_BOOTSTRAP="); + expect(unit).toContain( + `ExecStart=:/usr/bin/python3.13 /opt/t3/bootstrap.py --script-sha256 ${"b".repeat(64)} --policy /opt/t3/policy.json --policy-sha256 ${"c".repeat(64)} --helper /opt/t3/process-placement --helper-sha256 ${"d".repeat(64)} -- /usr/bin/node /home/user/.t3/runtime/service-launcher.mjs`, + ); + expect(unit).not.toContain("Environment=T3_PROCESS_PLACEMENT="); +}); + +it("preserves literal systemd dollar and percent characters in placement paths", () => { + const unit = BootService.renderBootServiceUnit({ + nodePath: "/opt/${version}/node", + launcherPath: "/home/user/.t3/launcher.mjs", + baseDir: "/home/user/.t3", + logPath: "/home/user/.t3/logs/service.log", + unitPath: "/home/user/.config/systemd/user/t3code.service", + processPlacement: { + version: 1, + helperPath: "/opt/t3%name/${helper}", + helperSha256: "a".repeat(64), + helperDevice: "1", + helperInode: "10", + control: { path: "/sys/fs/cgroup/t3/control", device: "1", inode: "2" }, + workload: { path: "/sys/fs/cgroup/t3/workload", device: "1", inode: "3" }, + }, + }); + expect(unit).toContain( + "ExecStart=:/opt/t3%%name/${helper} --helper-path /opt/t3%%name/${helper}", + ); + expect(unit).toContain("-- /opt/${version}/node /home/user/.t3/launcher.mjs"); + expect( + JSON.parse(BootService.placementEnvironmentFromUnit(unit).T3_PROCESS_PLACEMENT!).helperPath, + ).toBe("/opt/t3%name/${helper}"); +}); + +it("recovers required placement from its persisted systemd unit without silently dropping it", () => { + const binding = { + version: 1 as const, + helperPath: "/opt/t3/helper%$literal", + helperSha256: "a".repeat(64), + helperDevice: "1", + helperInode: "10", + control: { path: "/sys/fs/cgroup/t3/control", device: "1", inode: "2" }, + workload: { path: "/sys/fs/cgroup/t3/workload", device: "1", inode: "3" }, + }; + const unit = BootService.renderBootServiceUnit({ ...macPlan, processPlacement: binding }); + const recovered = BootService.placementEnvironmentFromUnit(unit); + expect(JSON.parse(recovered.T3_PROCESS_PLACEMENT!)).toEqual(binding); + expect(() => + BootService.placementEnvironmentFromUnit( + `${unit}\nEnvironment=T3_PROCESS_PLACEMENT="duplicate"`, + ), + ).toThrow("duplicate"); + expect(() => + BootService.placementEnvironmentFromUnit('Environment="T3_PROCESS_PLACEMENT=bad"'), + ).toThrow("unsupported"); + expect(() => + BootService.placementEnvironmentFromUnit( + `${unit}\nEnvironment=T3_PROCESS_PLACEMENT_BOOTSTRAP="{}"`, + ), + ).toThrow("ambiguous"); +}); diff --git a/apps/server/src/cloud/bootService.ts b/apps/server/src/cloud/bootService.ts index ff103176a173..a559bba1b7da 100644 --- a/apps/server/src/cloud/bootService.ts +++ b/apps/server/src/cloud/bootService.ts @@ -14,6 +14,21 @@ import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; +import { + placementBootstrapArguments, + readProcessPlacementBootstrap, + validateProcessPlacementBootstrap, + type ProcessPlacementBootstrap, +} from "../process/processPlacementBootstrap.ts"; +import { + PROCESS_PLACEMENT_BOOTSTRAP_ENV, + ProcessPlacementError, + PROCESS_PLACEMENT_ENV, + placementArguments, + readProcessPlacementBinding, + validateProcessPlacementBinding, + type ProcessPlacementBinding, +} from "../process/processPlacement.ts"; import * as ProcessRunner from "../processRunner.ts"; import { ensurePinnedRuntimeInstalled, @@ -51,6 +66,47 @@ function quoteSystemdValue(value: string): string { : escaped; } +export function placementEnvironmentFromUnit(unit: string): NodeJS.ProcessEnv { + const environment: NodeJS.ProcessEnv = {}; + for (const line of unit.split("\n")) { + const keys = [PROCESS_PLACEMENT_ENV, PROCESS_PLACEMENT_BOOTSTRAP_ENV].filter((key) => + line.includes(key), + ); + if (keys.length === 0) continue; + if (keys.length !== 1) + throw new ProcessPlacementError("ambiguous installed placement environment"); + const key = keys[0]!; + const prefix = `Environment=${key}=`; + if (!line.startsWith(prefix) || environment[key] !== undefined) { + throw new ProcessPlacementError("unsupported or duplicate installed placement environment"); + } + const encoded = line.slice(prefix.length); + if (Buffer.byteLength(encoded, "utf8") > 32 * 1024) + throw new ProcessPlacementError("installed placement environment too large"); + let decoded: unknown; + try { + decoded = JSON.parse(encoded); + } catch { + throw new ProcessPlacementError("invalid installed placement environment"); + } + if (typeof decoded !== "string") + throw new ProcessPlacementError("invalid installed placement environment"); + const text = decoded.replaceAll("%%", "%"); + if (quoteSystemdValue(text) !== encoded) + throw new ProcessPlacementError( + "installed placement environment failed roundtrip validation", + ); + environment[key] = text; + } + if ( + environment[PROCESS_PLACEMENT_ENV] !== undefined && + environment[PROCESS_PLACEMENT_BOOTSTRAP_ENV] !== undefined + ) { + throw new ProcessPlacementError("ambiguous installed placement routes"); + } + return environment; +} + export interface BootServicePlan { readonly nodePath: string; readonly launcherPath: string; @@ -59,10 +115,15 @@ export interface BootServicePlan { readonly authorityStateDir?: string; readonly logPath: string; readonly unitPath: string; + readonly processPlacement?: ProcessPlacementBinding; + readonly processPlacementBootstrap?: ProcessPlacementBootstrap; } /** Pure renderer: service units cannot rely on the user's shell or PATH. */ export function renderBootServiceUnit(plan: BootServicePlan): string { + if (plan.processPlacement !== undefined && plan.processPlacementBootstrap !== undefined) { + throw new ProcessPlacementError("choose either a cold bootstrap or a ready binding"); + } // The user manager has no reliable network-online target; server networking retries itself. return [ "[Unit]", @@ -72,6 +133,7 @@ export function renderBootServiceUnit(plan: BootServicePlan): string { "", "[Service]", "Type=simple", + ...(plan.processPlacementBootstrap === undefined ? [] : ["Delegate=cpu memory pids"]), "WorkingDirectory=%h", `Environment=T3CODE_HOME=${quoteSystemdValue(plan.baseDir)}`, ...(plan.authorityStateDir === undefined @@ -79,12 +141,41 @@ export function renderBootServiceUnit(plan: BootServicePlan): string { : [ `Environment=T3CODE_NATIVE_AUTHORITY_STATE_DIR=${quoteSystemdValue(plan.authorityStateDir)}`, ]), + ...(plan.processPlacementBootstrap === undefined + ? [] + : [ + `Environment=${PROCESS_PLACEMENT_BOOTSTRAP_ENV}=${quoteSystemdValue(JSON.stringify(plan.processPlacementBootstrap))}`, + ]), + ...(plan.processPlacement === undefined + ? [] + : [ + `Environment=${PROCESS_PLACEMENT_ENV}=${quoteSystemdValue(JSON.stringify(plan.processPlacement))}`, + ]), `Environment=${BOOT_SERVICE_UNIT_ENV}=${BOOT_SERVICE_UNIT_FILE}`, - `ExecStart=${quoteSystemdValue(plan.nodePath)} ${quoteSystemdValue(plan.launcherPath)}`, + // The colon prefix disables environment expansion in every bound executable and argument. + `ExecStart=${plan.processPlacement !== undefined || plan.processPlacementBootstrap !== undefined ? ":" : ""}${(plan.processPlacementBootstrap !== + undefined + ? [ + plan.processPlacementBootstrap.executablePath, + ...placementBootstrapArguments(plan.processPlacementBootstrap), + plan.nodePath, + plan.launcherPath, + ] + : plan.processPlacement === undefined + ? [plan.nodePath, plan.launcherPath] + : [ + plan.processPlacement.helperPath, + ...placementArguments(plan.processPlacement, "control"), + plan.nodePath, + plan.launcherPath, + ] + ) + .map(quoteSystemdValue) + .join(" ")}`, // Let the launcher mark an explicit stop before it signals the server. // systemd still SIGKILLs the whole cgroup if graceful shutdown times out. "KillMode=mixed", - // Agent tool calls run as children of the server, so they share this cgroup. + // Keep the unit alive if the kernel kills an agent under a delegated child cgroup. // With the systemd default of OOMPolicy=stop, the kernel killing one greedy // child stops the whole unit: the server, every live agent, and the user's // connection. Keep running and let Restart=always cover the main process. @@ -110,6 +201,8 @@ export function renderBootServicePlist( plan: BootServicePlan, options: { readonly homeDir: string; readonly environmentPath: string }, ): string { + if (plan.processPlacement !== undefined || plan.processPlacementBootstrap !== undefined) + throw new Error("Process placement requires Linux cgroup v2."); // KeepAlive + ThrottleInterval mirror Restart=always + RestartSec=5. launchd // has no StartLimitBurst analog; a hard crash loop respawns every 5s forever. // ExitTimeOut 90 matches systemd's default TimeoutStopSec. A plain stop @@ -499,6 +592,7 @@ export class BootService extends Context.Service< { readonly install: (options?: { readonly allowDowngrade?: boolean; + readonly disableProcessPlacement?: true; }) => Effect.Effect; readonly uninstall: Effect.Effect; readonly status: Effect.Effect; @@ -574,7 +668,46 @@ export const make = Effect.fn("cloud.boot_service.make")(function* (input: { yield* (yield* fs.open(directory, { flag: "r" })).sync; }), ).pipe(Effect.mapError((cause) => new BootServiceInstallError({ cause }))); + const existingUnit = + detectedManager?.kind === "systemd" && (yield* fs.exists(unitPath)) + ? yield* fs + .readFileString(unitPath) + .pipe(Effect.mapError((cause) => new BootServiceInstallError({ cause }))) + : undefined; + const placementEnvironment = yield* Effect.try({ + try: () => + process.env[PROCESS_PLACEMENT_ENV] !== undefined || + process.env[PROCESS_PLACEMENT_BOOTSTRAP_ENV] !== undefined + ? process.env + : existingUnit === undefined + ? {} + : placementEnvironmentFromUnit(existingUnit), + catch: (cause) => new BootServiceInstallError({ cause }), + }); + const { processPlacement, processPlacementBootstrap } = yield* Effect.try({ + try: () => { + const processPlacementBootstrap = readProcessPlacementBootstrap(placementEnvironment); + if ( + processPlacementBootstrap !== undefined && + process.env[PROCESS_PLACEMENT_ENV] !== undefined + ) { + throw new ProcessPlacementError("choose either a cold bootstrap or a ready binding"); + } + const processPlacement = + processPlacementBootstrap === undefined + ? readProcessPlacementBinding(placementEnvironment) + : undefined; + if (processPlacement !== undefined) + validateProcessPlacementBinding(processPlacement, platform); + if (processPlacementBootstrap !== undefined) + validateProcessPlacementBootstrap(processPlacementBootstrap, platform); + return { processPlacement, processPlacementBootstrap }; + }, + catch: (cause) => new BootServiceInstallError({ cause }), + }); const plan: BootServicePlan = { + ...(processPlacement === undefined ? {} : { processPlacement }), + ...(processPlacementBootstrap === undefined ? {} : { processPlacementBootstrap }), nodePath: host.execPath, launcherPath, baseDir: input.baseDir, @@ -700,8 +833,32 @@ export const make = Effect.fn("cloud.boot_service.make")(function* (input: { const install = Effect.fn("cloud.boot_service.install")(function* (options?: { readonly allowDowngrade?: boolean; + readonly disableProcessPlacement?: true; }) { const manager = yield* requireManager; + const installationPlan: BootServicePlan = + options?.disableProcessPlacement === true + ? { + nodePath: plan.nodePath, + launcherPath: plan.launcherPath, + baseDir: plan.baseDir, + ...(plan.authorityStateDir === undefined + ? {} + : { authorityStateDir: plan.authorityStateDir }), + logPath: plan.logPath, + unitPath: plan.unitPath, + } + : plan; + if ( + existingUnit !== undefined && + (yield* fs + .readFileString(unitPath) + .pipe(Effect.mapError((cause) => new BootServiceInstallError({ cause })))) !== existingUnit + ) { + return yield* new BootServiceInstallError({ + cause: new ProcessPlacementError("installed unit changed before update preparation"), + }); + } yield* fs .makeDirectory(input.logsDir, { recursive: true }) .pipe(Effect.mapError((cause) => new BootServiceInstallError({ cause }))); @@ -764,6 +921,24 @@ export const make = Effect.fn("cloud.boot_service.make")(function* (input: { .readFileString(launcherSourcePath) .pipe(Effect.mapError((cause) => new BootServiceInstallError({ cause }))); + if ( + processPlacementBootstrap !== undefined && + options?.disableProcessPlacement !== true && + placementEnvironment !== process.env && + existingUnit !== undefined + ) { + const installedLauncher = yield* fs + .readFileString(launcherPath) + .pipe(Effect.mapError((cause) => new BootServiceInstallError({ cause }))); + if (installedLauncher !== launcherSource) { + return yield* new BootServiceInstallError({ + cause: new ProcessPlacementError( + "cold bootstrap requires a renewed policy for the changed launcher source", + ), + }); + } + } + const installed = yield* fs .exists(unitPath) .pipe(Effect.mapError((cause) => new BootServiceInstallError({ cause }))); @@ -809,7 +984,7 @@ export const make = Effect.fn("cloud.boot_service.make")(function* (input: { 2, )}\n`, ); - yield* writeDurably(unitPath, manager.render(plan)); + yield* writeDurably(unitPath, manager.render(installationPlan)); yield* runSteps(manager.activate); }).pipe( @@ -817,7 +992,7 @@ export const make = Effect.fn("cloud.boot_service.make")(function* (input: { installed ? runSteps(manager.restart).pipe(Effect.ignore) : Effect.void, ), ); - return plan; + return installationPlan; }); const uninstall: BootService["Service"]["uninstall"] = Effect.gen(function* () { diff --git a/apps/server/src/process/claudeProcessPlacement.test.ts b/apps/server/src/process/claudeProcessPlacement.test.ts new file mode 100644 index 000000000000..b9ac1473abdd --- /dev/null +++ b/apps/server/src/process/claudeProcessPlacement.test.ts @@ -0,0 +1,30 @@ +import { describe, expect, it } from "vite-plus/test"; +import { claudeProcessPlacementOptions } from "./claudeProcessPlacement.ts"; +import { ProcessPlacementError } from "./processPlacement.ts"; + +describe("Claude SDK placement", () => { + it("preserves the SDK's native spawn implementation when placement is disabled", () => { + const previous = process.env.T3_PROCESS_PLACEMENT; + delete process.env.T3_PROCESS_PLACEMENT; + try { + expect(claudeProcessPlacementOptions()).toEqual({}); + } finally { + if (previous !== undefined) process.env.T3_PROCESS_PLACEMENT = previous; + } + }); + it("rejects malformed required placement before a session or capability probe can call the SDK", () => { + const previous = process.env.T3_PROCESS_PLACEMENT; + process.env.T3_PROCESS_PLACEMENT = "invalid-required-binding"; + let invokedSdk = false; + try { + expect(() => { + claudeProcessPlacementOptions(); + invokedSdk = true; + }).toThrow(ProcessPlacementError); + expect(invokedSdk).toBe(false); + } finally { + if (previous === undefined) delete process.env.T3_PROCESS_PLACEMENT; + else process.env.T3_PROCESS_PLACEMENT = previous; + } + }); +}); diff --git a/apps/server/src/process/claudeProcessPlacement.ts b/apps/server/src/process/claudeProcessPlacement.ts new file mode 100644 index 000000000000..18d38d6341c4 --- /dev/null +++ b/apps/server/src/process/claudeProcessPlacement.ts @@ -0,0 +1,58 @@ +// @effect-diagnostics nodeBuiltinImport:off - the SDK owns this subprocess lifecycle. +import * as NodeChildProcess from "node:child_process"; +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import type { Options } from "@anthropic-ai/claude-agent-sdk"; +import { + placementCommand, + readProcessPlacementBinding, + validateProcessPlacementBinding, +} from "./processPlacement.ts"; + +export function claudeProcessPlacementOptions( + options: Pick = {}, + platform = HostProcessPlatform.defaultValue(), +): Pick { + const binding = readProcessPlacementBinding(); + if (binding === undefined) return {}; + validateProcessPlacementBinding(binding, platform); + return { + spawnClaudeCodeProcess: (input) => { + const placed = placementCommand( + input.command, + input.args, + "workload", + binding, + false, + platform, + ); + const debug = + input.env.DEBUG_CLAUDE_AGENT_SDK === "1" || input.env.DEBUG_CLAUDE_AGENT_SDK === "true"; + const child = NodeChildProcess.spawn(placed.command, [...placed.args], { + cwd: input.cwd, + env: input.env, + signal: input.signal, + windowsHide: true, + stdio: ["pipe", "pipe", debug || options.stderr !== undefined ? "pipe" : "ignore"], + }); + child.stderr?.on("data", (data: Buffer) => { + const text = data.toString(); + if (debug) process.stderr.write(text); + options.stderr?.(text); + }); + return { + stdin: child.stdin!, + stdout: child.stdout!, + get killed() { + return child.killed; + }, + get exitCode() { + return child.exitCode; + }, + kill: child.kill.bind(child), + on: child.on.bind(child), + once: child.once.bind(child), + off: child.off.bind(child), + }; + }, + }; +} diff --git a/apps/server/src/process/processPlacement.test.ts b/apps/server/src/process/processPlacement.test.ts new file mode 100644 index 000000000000..38e1993e4cbf --- /dev/null +++ b/apps/server/src/process/processPlacement.test.ts @@ -0,0 +1,203 @@ +// @effect-diagnostics nodeBuiltinImport:off - native rejection checks spawn only a captured test child. +import * as NodeChildProcess from "node:child_process"; +import * as NodeCrypto from "node:crypto"; +import * as NodeFS from "node:fs"; +import * as NodeURL from "node:url"; +import { HostProcessArchitecture, HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import { describe, expect, it } from "vite-plus/test"; +import { + helperIdentityArguments, + placementArguments, + placementCommand, + readProcessPlacementBinding, + validateProcessPlacementBinding, + ProcessPlacementError, + type ProcessPlacementBinding, +} from "./processPlacement.ts"; + +const binding: ProcessPlacementBinding = { + version: 1, + helperPath: "/opt/t3/process-placement", + helperSha256: "a".repeat(64), + helperDevice: "1", + helperInode: "10", + control: { path: "/sys/fs/cgroup/t3-test/control", device: "29", inode: "101" }, + workload: { path: "/sys/fs/cgroup/t3-test/workload", device: "29", inode: "102" }, +}; +const environment = (value: unknown) => ({ T3_PROCESS_PLACEMENT: JSON.stringify(value) }); + +describe("required process placement", () => { + it.each(["darwin", "win32"] as const)( + "honors the injected %s host and refuses Linux placement", + (platform) => { + expect(() => validateProcessPlacementBinding(binding, platform)).toThrow( + "Linux cgroup v2 is required", + ); + }, + ); + it("leaves disabled commands and arguments intact", () => { + const args = ["argument with spaces", "$HOME", "--"]; + expect(readProcessPlacementBinding({})).toBeUndefined(); + expect(placementCommand("payload", args, "workload", undefined)).toEqual({ + command: "payload", + args, + }); + }); + it("decodes the versioned binding without exposing unrelated environment", () => { + expect(readProcessPlacementBinding(environment(binding))).toEqual(binding); + expect( + readProcessPlacementBinding( + environment({ ...binding, serverPriority: { nice: -15, resetOnFork: true } }), + ), + ).toEqual({ ...binding, serverPriority: { nice: -15, resetOnFork: true } }); + }); + it.each([ + "", + "null", + "{}", + JSON.stringify({ ...binding, version: 2 }), + JSON.stringify({ ...binding, helperPath: "helper" }), + JSON.stringify({ ...binding, extra: true }), + JSON.stringify({ ...binding, helperPath: "/opt/unsafe\nhelper" }), + JSON.stringify({ + ...binding, + workload: { ...binding.workload, device: "18446744073709551616" }, + }), + JSON.stringify({ ...binding, workload: { ...binding.workload, inode: "1".repeat(21) } }), + JSON.stringify({ ...binding, workload: { ...binding.workload, extra: true } }), + JSON.stringify({ ...binding, serverPriority: { nice: -15, resetOnFork: true, extra: true } }), + JSON.stringify({ ...binding, helperSha256: "bad" }), + JSON.stringify({ ...binding, helperDevice: undefined }), + JSON.stringify({ ...binding, helperInode: "0" }), + JSON.stringify({ ...binding, helperDevice: "18446744073709551616" }), + JSON.stringify({ ...binding, workload: binding.control }), + JSON.stringify({ + ...binding, + workload: { ...binding.workload, path: `${binding.control.path}/child` }, + }), + JSON.stringify({ + ...binding, + workload: { ...binding.workload, path: "/sys/fs/cgroup/a/../b" }, + }), + JSON.stringify({ ...binding, serverPriority: { nice: -15, resetOnFork: false } }), + ])("rejects malformed, unsupported, or overlapping required bindings", (text) => { + expect(() => readProcessPlacementBinding({ T3_PROCESS_PLACEMENT: text })).toThrow( + ProcessPlacementError, + ); + }); + it("limits server priority to the actual server, not the launcher or telemetry", () => { + const priority = { ...binding, serverPriority: { nice: -15, resetOnFork: true } } as const; + expect(placementArguments(priority, "control")).not.toContain("--server-priority"); + expect(placementArguments(priority, "control", true)).toEqual([ + ...helperIdentityArguments(binding), + "--role", + "control", + "--cgroup", + binding.control.path, + "--device", + "29", + "--inode", + "101", + "--server-priority", + "-15", + "--", + ]); + expect(placementArguments(priority, "workload")).not.toContain("--server-priority"); + }); + it("rejects unavailable placement before handing any executable to a spawner", () => { + let executed = false; + expect(() => { + const command = placementCommand("sentinel", [], "workload", binding); + executed = command.command === "sentinel"; + }).toThrow(ProcessPlacementError); + expect(executed).toBe(false); + }); +}); + +const helperPath = NodeURL.fileURLToPath( + new URL( + `../../dist/native/process-placement-linux-${HostProcessArchitecture.defaultValue()}`, + import.meta.url, + ), +); +describe.skipIf(HostProcessPlatform.defaultValue() !== "linux" || !NodeFS.existsSync(helperPath))( + "native placement gate", + () => { + const helperIdentity = () => { + const stat = NodeFS.statSync(helperPath, { bigint: true }); + return helperIdentityArguments({ + ...binding, + helperPath, + helperDevice: String(stat.dev), + helperInode: String(stat.ino), + helperSha256: NodeCrypto.createHash("sha256") + .update(NodeFS.readFileSync(helperPath)) + .digest("hex"), + }); + }; + it.each(["control", "workload"])("never executes %s payload when placement fails", (role) => { + const result = NodeChildProcess.spawnSync( + helperPath, + [ + ...helperIdentity(), + "--role", + role, + "--cgroup", + "/sys/fs/cgroup/t3-placement-test-missing", + "--device", + "0", + "--inode", + "1", + "--", + process.execPath, + "-e", + "process.stdout.write('PAYLOAD_EXECUTED')", + ], + { encoding: "utf8" }, + ); + expect(result.error).toBeUndefined(); + expect(result.status).toBe(125); + expect(result.stdout).toBe(""); + expect(result.stderr).toContain("cgroup unavailable or symlinked"); + }); + it.each(["path", "sha256", "device", "inode"])( + "rejects stale executed helper %s identity", + (field) => { + const args = [...helperIdentity()]; + const index = args.indexOf(`--helper-${field}`) + 1; + args[index] = + field === "path" ? `${helperPath}.other` : field === "sha256" ? "0".repeat(64) : "0"; + const result = NodeChildProcess.spawnSync( + helperPath, + [...args, "--verify-server-priority", String(process.pid)], + { encoding: "utf8" }, + ); + expect(result.status).toBe(125); + expect(result.stderr).toMatch(/helper (identity|hash) changed/); + }, + ); + it("rejects ordinary filesystem directories before payload execution", () => { + const result = NodeChildProcess.spawnSync( + helperPath, + [ + ...helperIdentity(), + "--role", + "workload", + "--cgroup", + "/tmp", + "--device", + "0", + "--inode", + "1", + "--", + process.execPath, + "-e", + "process.stdout.write('PAYLOAD_EXECUTED')", + ], + { encoding: "utf8" }, + ); + expect(result.status).toBe(125); + expect(result.stdout).toBe(""); + }); + }, +); diff --git a/apps/server/src/process/processPlacement.ts b/apps/server/src/process/processPlacement.ts new file mode 100644 index 000000000000..9ff8bfe6260e --- /dev/null +++ b/apps/server/src/process/processPlacement.ts @@ -0,0 +1,282 @@ +// @effect-diagnostics nodeBuiltinImport:off - shared with the standalone Node-only service launcher. +import * as NodeChildProcess from "node:child_process"; +import * as NodeCrypto from "node:crypto"; +import * as NodeFS from "node:fs"; +import * as NodePath from "node:path"; +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; + +export const PROCESS_PLACEMENT_ENV = "T3_PROCESS_PLACEMENT"; +export const PROCESS_PLACEMENT_BOOTSTRAP_ENV = "T3_PROCESS_PLACEMENT_BOOTSTRAP"; +export type ProcessRole = "control" | "workload"; +export interface CgroupBinding { + readonly path: string; + readonly device: string; + readonly inode: string; +} +export interface ProcessPlacementBinding { + readonly version: 1; + readonly helperPath: string; + readonly helperSha256: string; + readonly helperDevice: string; + readonly helperInode: string; + readonly control: CgroupBinding; + readonly workload: CgroupBinding; + readonly serverPriority?: { readonly nice: -15; readonly resetOnFork: true }; +} + +export class ProcessPlacementError extends Error { + constructor(message: string) { + super(`Process placement required: ${message}`); + this.name = "ProcessPlacementError"; + } +} + +const validIdentity = (value: unknown, positive: boolean): value is string => + typeof value === "string" && + (positive ? /^[1-9][0-9]{0,19}$/ : /^(0|[1-9][0-9]{0,19})$/).test(value) && + BigInt(value) <= 18446744073709551615n; + +const fail = (message: string): never => { + throw new ProcessPlacementError(message); +}; + +// Parsing remains synchronous for the standalone launcher. +export function readProcessPlacementBinding( + environment: NodeJS.ProcessEnv = process.env, +): ProcessPlacementBinding | undefined { + const text = environment[PROCESS_PLACEMENT_ENV]; + if (text === undefined) { + if (environment[PROCESS_PLACEMENT_BOOTSTRAP_ENV] !== undefined) + return fail("bootstrap must establish a ready binding before T3 executes"); + return undefined; + } + if (Buffer.byteLength(text, "utf8") > 16 * 1024) return fail("binding exceeds 16 KiB"); + let value: unknown; + try { + value = JSON.parse(text); + } catch { + return fail("invalid binding JSON"); + } + if (typeof value !== "object" || value === null || Array.isArray(value)) + return fail("invalid binding"); + const record = value as Record; + if ( + Object.keys(record).some( + (key) => + ![ + "version", + "helperPath", + "helperSha256", + "helperDevice", + "helperInode", + "control", + "workload", + "serverPriority", + ].includes(key), + ) + ) + return fail("unknown binding field"); + if ( + record.version !== 1 || + typeof record.helperPath !== "string" || + !NodePath.isAbsolute(record.helperPath) || + NodePath.normalize(record.helperPath) !== record.helperPath || + /[\u0000-\u001f\u007f]/.test(record.helperPath) || + typeof record.helperSha256 !== "string" || + !/^[0-9a-f]{64}$/.test(record.helperSha256) || + !validIdentity(record.helperDevice, false) || + !validIdentity(record.helperInode, true) + ) { + return fail("unsupported binding or helper identity"); + } + const group = (input: unknown): CgroupBinding => { + if (typeof input !== "object" || input === null || Array.isArray(input)) + return fail("invalid cgroup binding"); + const item = input as Record; + if (Object.keys(item).some((key) => !["path", "device", "inode"].includes(key))) + return fail("unknown cgroup field"); + if ( + typeof item.path !== "string" || + !item.path.startsWith("/sys/fs/cgroup/") || + /[\u0000-\u001f\u007f]/.test(item.path) || + NodePath.normalize(item.path) !== item.path || + typeof item.device !== "string" || + !/^(0|[1-9][0-9]{0,19})$/.test(item.device) || + BigInt(item.device) > 18446744073709551615n || + typeof item.inode !== "string" || + !/^[1-9][0-9]{0,19}$/.test(item.inode) || + BigInt(item.inode) > 18446744073709551615n + ) { + return fail("invalid cgroup identity"); + } + return { path: item.path, device: item.device, inode: item.inode }; + }; + const control = group(record.control); + const workload = group(record.workload); + if (NodePath.dirname(control.path) !== NodePath.dirname(workload.path)) + return fail("roles must share one delegated parent"); + if ( + control.path === workload.path || + control.path.startsWith(`${workload.path}/`) || + workload.path.startsWith(`${control.path}/`) + ) + return fail("roles require separate cgroup branches"); + if (record.serverPriority !== undefined) { + const priority = record.serverPriority as Record | null; + if ( + typeof priority !== "object" || + priority === null || + Object.keys(priority).some((key) => !["nice", "resetOnFork"].includes(key)) || + priority.nice !== -15 || + priority.resetOnFork !== true + ) { + return fail("unsupported server scheduling policy"); + } + } + return { + version: 1, + helperPath: record.helperPath, + helperSha256: record.helperSha256, + helperDevice: record.helperDevice as string, + helperInode: record.helperInode as string, + control, + workload, + ...(record.serverPriority === undefined + ? {} + : { serverPriority: { nice: -15, resetOnFork: true } }), + }; +} + +export function validateProcessPlacementBinding( + binding: ProcessPlacementBinding, + platform = HostProcessPlatform.defaultValue(), +): void { + if (platform !== "linux") fail("Linux cgroup v2 is required"); + try { + if (NodeFS.realpathSync(binding.helperPath) !== binding.helperPath) + fail("helper symlink rejected"); + const helper = NodeFS.lstatSync(binding.helperPath, { bigint: true }); + if ( + !helper.isFile() || + helper.isSymbolicLink() || + (helper.mode & 0o111n) === 0n || + (helper.mode & 0o022n) !== 0n || + helper.uid !== BigInt(process.getuid?.() ?? -1) || + String(helper.dev) !== binding.helperDevice || + String(helper.ino) !== binding.helperInode + ) + fail("unsafe helper executable"); + if ( + NodeCrypto.createHash("sha256") + .update(NodeFS.readFileSync(binding.helperPath)) + .digest("hex") !== binding.helperSha256 + ) { + fail("helper identity changed"); + } + for (const role of ["control", "workload"] as const) { + const target = binding[role]; + if (NodeFS.realpathSync(target.path) !== target.path) fail("cgroup symlink rejected"); + const stat = NodeFS.statSync(target.path, { bigint: true }); + if ( + !stat.isDirectory() || + String(stat.dev) !== target.device || + String(stat.ino) !== target.inode + ) { + fail("cgroup identity changed"); + } + NodeFS.accessSync(NodePath.join(target.path, "cgroup.procs"), NodeFS.constants.W_OK); + NodeFS.accessSync(NodePath.join(target.path, "cpu.max"), NodeFS.constants.R_OK); + } + } catch (cause) { + if (cause instanceof ProcessPlacementError) throw cause; + fail("helper or delegated CPU cgroup unavailable"); + } +} + +export function placementCommand( + command: string, + args: ReadonlyArray, + role: ProcessRole = "workload", + binding = readProcessPlacementBinding(), + server = false, + platform = HostProcessPlatform.defaultValue(), +): { readonly command: string; readonly args: ReadonlyArray } { + if (binding === undefined) return { command, args }; + if (server && role !== "control") fail("server priority requires control role"); + validateProcessPlacementBinding(binding, platform); + return { + command: binding.helperPath, + args: [...placementArguments(binding, role, server), command, ...args], + }; +} + +export function assertControlProcessPlacement( + binding = readProcessPlacementBinding(), + platform = HostProcessPlatform.defaultValue(), +): void { + if (binding === undefined) return; + validateProcessPlacementBinding(binding, platform); + let membership: string; + try { + membership = NodeFS.readFileSync("/proc/self/cgroup", "utf8"); + } catch { + return fail("cannot verify server membership"); + } + const cgroup = membership + .split("\n") + .find((line) => line.startsWith("0::")) + ?.slice(3); + if (cgroup === undefined || NodePath.join("/sys/fs/cgroup", cgroup) !== binding.control.path) { + fail("server must start in its control cgroup before execution"); + } +} + +export function assertServerProcessPlacement( + binding = readProcessPlacementBinding(), + platform = HostProcessPlatform.defaultValue(), +): void { + assertControlProcessPlacement(binding, platform); + if (binding?.serverPriority !== undefined) { + const result = NodeChildProcess.spawnSync( + binding.helperPath, + [...helperIdentityArguments(binding), "--verify-server-priority", String(process.pid)], + { stdio: "ignore" }, + ); + if (result.error !== undefined || result.status !== 0) + fail("server priority or reset-on-fork not established at startup"); + } +} + +export function helperIdentityArguments(binding: ProcessPlacementBinding): ReadonlyArray { + return [ + "--helper-path", + binding.helperPath, + "--helper-sha256", + binding.helperSha256, + "--helper-device", + binding.helperDevice, + "--helper-inode", + binding.helperInode, + ]; +} + +export function placementArguments( + binding: ProcessPlacementBinding, + role: ProcessRole, + server = false, +): ReadonlyArray { + const group = binding[role]; + return [ + ...helperIdentityArguments(binding), + "--role", + role, + "--cgroup", + group.path, + "--device", + group.device, + "--inode", + group.inode, + ...(server && binding.serverPriority !== undefined ? ["--server-priority", "-15"] : []), + "--", + ]; +} diff --git a/apps/server/src/process/processPlacementBootstrap.test.ts b/apps/server/src/process/processPlacementBootstrap.test.ts new file mode 100644 index 000000000000..156b85cb07c4 --- /dev/null +++ b/apps/server/src/process/processPlacementBootstrap.test.ts @@ -0,0 +1,53 @@ +import { expect, it } from "vite-plus/test"; +import { + placementBootstrapArguments, + readProcessPlacementBootstrap, + validateProcessPlacementBootstrap, +} from "./processPlacementBootstrap.ts"; +import { readProcessPlacementBinding } from "./processPlacement.ts"; + +const binding = { + version: 1 as const, + executablePath: "/usr/bin/python3.13", + executableSha256: "a".repeat(64), + scriptPath: "/opt/t3/bootstrap.py", + scriptSha256: "b".repeat(64), + policyPath: "/opt/t3/policy.json", + policySha256: "c".repeat(64), + helperPath: "/opt/t3/process-placement", + helperSha256: "d".repeat(64), +}; + +it("keeps an absent external bootstrap disabled", () => { + expect(readProcessPlacementBootstrap({})).toBeUndefined(); +}); +it("decodes only the explicit versioned external artifact identities", () => { + expect( + readProcessPlacementBootstrap({ T3_PROCESS_PLACEMENT_BOOTSTRAP: JSON.stringify(binding) }), + ).toEqual(binding); + expect(placementBootstrapArguments(binding).at(-1)).toBe("--"); +}); +it.each([ + "", + "null", + "{}", + JSON.stringify({ ...binding, version: 2 }), + JSON.stringify({ ...binding, extra: true }), + JSON.stringify({ ...binding, executablePath: "python3" }), + JSON.stringify({ ...binding, policySha256: "bad" }), +])("rejects invalid required external bootstrap bindings", (text) => { + expect(() => readProcessPlacementBootstrap({ T3_PROCESS_PLACEMENT_BOOTSTRAP: text })).toThrow( + "Process placement required", + ); +}); +it("blocks direct serve while a cold bootstrap has not established its ready binding", () => { + expect(() => + readProcessPlacementBinding({ T3_PROCESS_PLACEMENT_BOOTSTRAP: JSON.stringify(binding) }), + ).toThrow("bootstrap must establish a ready binding"); +}); + +it("honors the injected host when validating cold bootstrap artifacts", () => { + expect(() => validateProcessPlacementBootstrap(binding, "darwin")).toThrow( + "bootstrap requires Linux cgroup v2", + ); +}); diff --git a/apps/server/src/process/processPlacementBootstrap.ts b/apps/server/src/process/processPlacementBootstrap.ts new file mode 100644 index 000000000000..52f7f5bc2368 --- /dev/null +++ b/apps/server/src/process/processPlacementBootstrap.ts @@ -0,0 +1,98 @@ +// @effect-diagnostics nodeBuiltinImport:off - explicitly bound external bootstrap artifacts. +import * as NodeCrypto from "node:crypto"; +import * as NodeFS from "node:fs"; +import * as NodePath from "node:path"; +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import { PROCESS_PLACEMENT_BOOTSTRAP_ENV, ProcessPlacementError } from "./processPlacement.ts"; + +export interface ProcessPlacementBootstrap { + readonly version: 1; + readonly executablePath: string; + readonly executableSha256: string; + readonly scriptPath: string; + readonly scriptSha256: string; + readonly policyPath: string; + readonly policySha256: string; + readonly helperPath: string; + readonly helperSha256: string; +} +const fields = ["executable", "script", "policy", "helper"] as const; +export function readProcessPlacementBootstrap( + environment: NodeJS.ProcessEnv = process.env, +): ProcessPlacementBootstrap | undefined { + const text = environment[PROCESS_PLACEMENT_BOOTSTRAP_ENV]; + if (text === undefined) return undefined; + if (Buffer.byteLength(text, "utf8") > 16 * 1024) + throw new ProcessPlacementError("bootstrap binding exceeds 16 KiB"); + let value: unknown; + try { + value = JSON.parse(text); + } catch { + throw new ProcessPlacementError("invalid bootstrap JSON"); + } + if (typeof value !== "object" || value === null || Array.isArray(value)) + throw new ProcessPlacementError("invalid bootstrap binding"); + const record = value as Record; + const keys = ["version", ...fields.flatMap((field) => [`${field}Path`, `${field}Sha256`])]; + if (record.version !== 1 || Object.keys(record).some((key) => !keys.includes(key))) + throw new ProcessPlacementError("unsupported bootstrap binding"); + for (const field of fields) { + const path = record[`${field}Path`]; + const sha = record[`${field}Sha256`]; + if ( + typeof path !== "string" || + !NodePath.isAbsolute(path) || + NodePath.normalize(path) !== path || + /[\u0000-\u001f\u007f]/.test(path) || + typeof sha !== "string" || + !/^[0-9a-f]{64}$/.test(sha) + ) + throw new ProcessPlacementError("invalid bootstrap artifact identity"); + } + return record as unknown as ProcessPlacementBootstrap; +} +export function validateProcessPlacementBootstrap( + binding: ProcessPlacementBootstrap, + platform = HostProcessPlatform.defaultValue(), +): void { + if (platform !== "linux") throw new ProcessPlacementError("bootstrap requires Linux cgroup v2"); + try { + for (const field of fields) { + const path = binding[`${field}Path`]; + const stat = NodeFS.lstatSync(path); + if ( + NodeFS.realpathSync(path) !== path || + !stat.isFile() || + (stat.mode & 0o022) !== 0 || + (stat.uid !== 0 && stat.uid !== process.getuid?.()) + ) + throw new ProcessPlacementError("unsafe bootstrap artifact"); + if ((field === "executable" || field === "helper") && (stat.mode & 0o111) === 0) + throw new ProcessPlacementError("bootstrap executable unavailable"); + const sha = NodeCrypto.createHash("sha256").update(NodeFS.readFileSync(path)).digest("hex"); + if (sha !== binding[`${field}Sha256`]) + throw new ProcessPlacementError("bootstrap artifact identity changed"); + } + } catch (cause) { + if (cause instanceof ProcessPlacementError) throw cause; + throw new ProcessPlacementError("bootstrap artifact unavailable"); + } +} +export function placementBootstrapArguments( + binding: ProcessPlacementBootstrap, +): ReadonlyArray { + return [ + binding.scriptPath, + "--script-sha256", + binding.scriptSha256, + "--policy", + binding.policyPath, + "--policy-sha256", + binding.policySha256, + "--helper", + binding.helperPath, + "--helper-sha256", + binding.helperSha256, + "--", + ]; +} diff --git a/apps/server/src/process/processPlacementSpawner.test.ts b/apps/server/src/process/processPlacementSpawner.test.ts new file mode 100644 index 000000000000..3ed842f70af6 --- /dev/null +++ b/apps/server/src/process/processPlacementSpawner.test.ts @@ -0,0 +1,99 @@ +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Layer from "effect/Layer"; +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; +import { expect, it } from "@effect/vitest"; +import { vi } from "vite-plus/test"; +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import { controlCommand, layer, placeCommand } from "./processPlacementSpawner.ts"; +import { placementCommand, type ProcessPlacementBinding } from "./processPlacement.ts"; + +vi.mock("./processPlacement.ts", async (importOriginal) => { + const original = await importOriginal(); + return { ...original, placementCommand: vi.fn(original.placementCommand) }; +}); + +it("preserves disabled command identity, pipeline options and streams", () => { + const command = ChildProcess.pipeTo( + ChildProcess.make("one", ["argument"], { + cwd: "/workspace", + env: { KEY: "value" }, + shell: "/bin/sh", + }), + ChildProcess.make("two"), + { from: "stderr" }, + ); + expect(placeCommand(command, undefined)).toBe(command); +}); + +it.effect("fails required placement before the platform spawner executes a provider payload", () => + Effect.gen(function* () { + let spawned = false; + const underlying = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => + Effect.sync(() => { + spawned = true; + throw new Error("payload must not execute"); + }), + ), + ); + const binding: ProcessPlacementBinding = { + version: 1, + helperPath: "/missing/t3-placement-helper", + helperSha256: "a".repeat(64), + helperDevice: "1", + helperInode: "10", + control: { path: "/sys/fs/cgroup/missing/control", device: "0", inode: "1" }, + workload: { path: "/sys/fs/cgroup/missing/workload", device: "0", inode: "2" }, + }; + const exit = yield* Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + return yield* spawner.spawn(ChildProcess.make("provider-sentinel")); + }).pipe( + Effect.scoped, + Effect.provide(layer(binding).pipe(Layer.provide(underlying))), + Effect.provideService(HostProcessPlatform, "linux"), + Effect.exit, + ); + expect(Exit.isFailure(exit)).toBe(true); + expect(spawned).toBe(false); + }), +); + +it("places both pipeline payloads and preserves options with an explicit telemetry control role", () => { + const binding: ProcessPlacementBinding = { + version: 1, + helperPath: "/bound/helper", + helperSha256: "a".repeat(64), + helperDevice: "1", + helperInode: "10", + control: { path: "/sys/fs/cgroup/bound/control", device: "1", inode: "2" }, + workload: { path: "/sys/fs/cgroup/bound/workload", device: "1", inode: "3" }, + }; + const fakePlacement = vi.mocked(placementCommand).mockImplementation((command, args, role) => ({ + command: binding.helperPath, + args: [role!, command, ...args], + })); + try { + const options = { cwd: "/workspace", env: { KEY: "value" }, shell: "/bin/sh" } as const; + const pipeline = ChildProcess.pipeTo( + ChildProcess.make("git", ["status"], options), + controlCommand(ChildProcess.make("telemetry", ["--bounded"])), + { from: "stderr" }, + ); + const placed = placeCommand(pipeline, binding); + expect(placed._tag).toBe("PipedCommand"); + if (placed._tag !== "PipedCommand") throw new Error("expected pipeline"); + expect(placed.options).toEqual(pipeline.options); + expect(placed.left._tag).toBe("StandardCommand"); + if (placed.left._tag !== "StandardCommand" || placed.right._tag !== "StandardCommand") + throw new Error("expected payload commands"); + expect(placed.left.command).toBe(binding.helperPath); + expect(placed.left.args).toEqual(["workload", "/bin/sh", "-c", "git status"]); + expect(placed.left.options).toEqual({ ...options, shell: false }); + expect(placed.right.args).toEqual(["control", "telemetry", "--bounded"]); + } finally { + fakePlacement.mockRestore(); + } +}); diff --git a/apps/server/src/process/processPlacementSpawner.ts b/apps/server/src/process/processPlacementSpawner.ts new file mode 100644 index 000000000000..c8ec8ae334a4 --- /dev/null +++ b/apps/server/src/process/processPlacementSpawner.ts @@ -0,0 +1,69 @@ +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as PlatformError from "effect/PlatformError"; +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; +import { + placementCommand, + type ProcessPlacementBinding, + type ProcessRole, +} from "./processPlacement.ts"; + +const roles = new WeakMap(); +export function controlCommand(command: T): T { + roles.set(command, "control"); + return command; +} + +export function placeCommand( + command: ChildProcess.Command, + binding: ProcessPlacementBinding | undefined, + platform = HostProcessPlatform.defaultValue(), +): ChildProcess.Command { + if (binding === undefined) return command; + if (command._tag === "PipedCommand") { + return ChildProcess.pipeTo( + placeCommand(command.left, binding, platform), + placeCommand(command.right, binding, platform), + command.options, + ); + } + const shell = command.options.shell; + const payload = shell + ? { + command: typeof shell === "string" ? shell : "/bin/sh", + args: ["-c", [command.command, ...command.args].join(" ")], + } + : { command: command.command, args: command.args }; + const placed = placementCommand( + payload.command, + payload.args, + roles.get(command) ?? "workload", + binding, + false, + platform, + ); + return ChildProcess.make(placed.command, placed.args, { ...command.options, shell: false }); +} + +export const layer = (binding: ProcessPlacementBinding | undefined) => + Layer.effect( + ChildProcessSpawner.ChildProcessSpawner, + Effect.gen(function* () { + const platform = yield* HostProcessPlatform; + const underlying = yield* ChildProcessSpawner.ChildProcessSpawner; + return ChildProcessSpawner.make((command) => + Effect.try({ + try: () => placeCommand(command, binding, platform), + catch: (cause) => + PlatformError.systemError({ + _tag: "Unknown", + module: "ProcessPlacement", + method: "spawn", + description: "Required process placement unavailable", + cause, + }), + }).pipe(Effect.flatMap((placed) => underlying.spawn(placed))), + ); + }), + ); diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index c08a5cbace19..ceeba43a09a5 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -305,6 +305,32 @@ const RESUME_THREAD_ID = ThreadId.make("thread-claude-resume"); const SYNTHETIC_SUBAGENT_MODEL = "claude-synthetic-subagent[expanded]"; describe("ClaudeAdapterLive", () => { + it.effect("blocks required placement before creating a Claude SDK session", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const previous = process.env.T3_PROCESS_PLACEMENT; + process.env.T3_PROCESS_PLACEMENT = "invalid-required-binding"; + try { + const result = yield* adapter + .startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }) + .pipe(Effect.result); + assert.equal(result._tag, "Failure"); + assert.isUndefined(harness.getLastCreateQueryInput()); + } finally { + if (previous === undefined) delete process.env.T3_PROCESS_PLACEMENT; + else process.env.T3_PROCESS_PLACEMENT = previous; + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("returns validation error for non-claude provider on startSession", () => { const harness = makeHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 5848bcc457de..e8a928132ef2 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -19,6 +19,8 @@ import { type SDKUserMessage, type ModelUsage, } from "@anthropic-ai/claude-agent-sdk"; +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import { claudeProcessPlacementOptions } from "../../process/claudeProcessPlacement.ts"; import { parseCliArgs } from "@t3tools/shared/cliArgs"; import { isWorkspaceImagePreviewPath } from "@t3tools/shared/filePreview"; import { type ClaudeScopedLimitNames, claudeRateLimitEventToUpdate } from "./claudeUsageLimits.ts"; @@ -1731,6 +1733,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( claudeSettings: ClaudeSettings, options?: ClaudeAdapterLiveOptions, ) { + const placementPlatform = yield* HostProcessPlatform; const boundInstanceId = options?.instanceId ?? ProviderInstanceId.make("claudeAgent"); const modelCatalogEffect = ( options?.modelCatalog ?? Effect.succeed(BUNDLED_CLAUDE_MODEL_CATALOG) @@ -4443,7 +4446,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( try: () => createQuery({ prompt, - options: queryOptions, + options: { + ...queryOptions, + ...claudeProcessPlacementOptions(queryOptions, placementPlatform), + }, }), catch: (cause) => new ProviderAdapterProcessError({ diff --git a/apps/server/src/provider/Layers/ClaudeCapabilitiesProbe.test.ts b/apps/server/src/provider/Layers/ClaudeCapabilitiesProbe.test.ts index 253118299820..a8d36d602c4c 100644 --- a/apps/server/src/provider/Layers/ClaudeCapabilitiesProbe.test.ts +++ b/apps/server/src/provider/Layers/ClaudeCapabilitiesProbe.test.ts @@ -162,3 +162,23 @@ it.layer(NodeServices.layer)("Claude capability probe SDK boundary", (it) => { }).pipe(Effect.scoped), ); }); + +it("blocks a required-placement capability probe before the SDK can execute its provider", () => { + const previous = process.env.T3_PROCESS_PLACEMENT; + process.env.T3_PROCESS_PLACEMENT = "invalid-required-binding"; + try { + assert.throws( + () => + buildClaudeCapabilitiesProbeQueryOptions({ + executablePath: "provider-sentinel", + abortController: new AbortController(), + environment: {}, + cwd: "/workspace", + }), + "Process placement required", + ); + } finally { + if (previous === undefined) delete process.env.T3_PROCESS_PLACEMENT; + else process.env.T3_PROCESS_PLACEMENT = previous; + } +}); diff --git a/apps/server/src/provider/Layers/ClaudeProvider.ts b/apps/server/src/provider/Layers/ClaudeProvider.ts index e4ec8c522da7..f4029a3c63bb 100644 --- a/apps/server/src/provider/Layers/ClaudeProvider.ts +++ b/apps/server/src/provider/Layers/ClaudeProvider.ts @@ -1,3 +1,5 @@ +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import { claudeProcessPlacementOptions } from "../../process/claudeProcessPlacement.ts"; import { type ClaudeSettings, type ModelCapabilities, @@ -186,8 +188,10 @@ export function buildClaudeCapabilitiesProbeQueryOptions(input: { readonly abortController: AbortController; readonly environment: NodeJS.ProcessEnv; readonly cwd: string | undefined; + readonly platform?: NodeJS.Platform; }): ClaudeQueryOptions { return { + ...claudeProcessPlacementOptions({}, input.platform), persistSession: false, pathToClaudeCodeExecutable: input.executablePath, abortController: input.abortController, @@ -334,6 +338,7 @@ const probeClaudeCapabilities = ( ) => { const abort = new AbortController(); return Effect.gen(function* () { + const platform = yield* HostProcessPlatform; const claudeEnvironment = yield* makeClaudeEnvironment(claudeSettings, environment); const executablePath = yield* resolveClaudeSdkExecutablePath( claudeSettings.binaryPath, @@ -348,6 +353,7 @@ const probeClaudeCapabilities = ( await waitForAbortSignal(abort.signal); })(), options: buildClaudeCapabilitiesProbeQueryOptions({ + platform, executablePath, abortController: abort, environment: claudeEnvironment, diff --git a/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts b/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts index 232079d9dc9b..722bcf985936 100644 --- a/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts +++ b/apps/server/src/resourceTelemetry/NativeTelemetryClient.ts @@ -19,6 +19,7 @@ import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; +import { controlCommand } from "../process/processPlacementSpawner.ts"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; @@ -553,7 +554,7 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu }); const handle = yield* Effect.acquireRelease( spawner - .spawn(command) + .spawn(controlCommand(command)) .pipe( Effect.mapError( (cause) => new NativeTelemetrySpawnFailed({ path: executablePath, cause }), diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 177466b8bec1..452e095442dc 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -1,3 +1,4 @@ +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; import { EnvironmentHttpApi, ProviderDriverKind } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Duration from "effect/Duration"; @@ -24,6 +25,11 @@ import { import { guardHttpResponseWriteErrors } from "./httpResponseErrorGuard.ts"; import { fixPath } from "./os-jank.ts"; import { websocketRpcRouteLayer } from "./ws.ts"; +import { + assertServerProcessPlacement, + readProcessPlacementBinding, +} from "./process/processPlacement.ts"; +import * as ProcessPlacementSpawner from "./process/processPlacementSpawner.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; import { pullRequestHttpApiLayer } from "./pullRequest/http.ts"; import * as PullRequestProviderRegistry from "./pullRequest/PullRequestProviderRegistry.ts"; @@ -264,17 +270,18 @@ const HttpServerLive = Layer.unwrap( }), ); -const PlatformServicesLive = Layer.unwrap( - Effect.gen(function* () { - if (typeof Bun !== "undefined") { - const { layer } = yield* Effect.promise(() => import("@effect/platform-bun/BunServices")); - return layer; - } else { - const { layer } = yield* Effect.promise(() => import("@effect/platform-node/NodeServices")); - return layer; - } - }), -); +const platformServicesLive = (binding: ReturnType) => + Layer.unwrap( + Effect.gen(function* () { + if (typeof Bun !== "undefined") { + const { layer } = yield* Effect.promise(() => import("@effect/platform-bun/BunServices")); + return ProcessPlacementSpawner.layer(binding).pipe(Layer.provideMerge(layer)); + } else { + const { layer } = yield* Effect.promise(() => import("@effect/platform-node/NodeServices")); + return ProcessPlacementSpawner.layer(binding).pipe(Layer.provideMerge(layer)); + } + }), + ); const ReactorLayerLive = Layer.empty.pipe( Layer.provideMerge(OrchestrationReactorLive), @@ -569,6 +576,9 @@ export const makeRoutesLayer = Layer.mergeAll( export const makeServerLayer = Layer.unwrap( Effect.gen(function* () { const config = yield* ServerConfig.ServerConfig; + const platform = yield* HostProcessPlatform; + const placement = yield* Effect.sync(() => readProcessPlacementBinding()); + yield* Effect.sync(() => assertServerProcessPlacement(placement, platform)); const activation = yield* Deferred.make(); const awaitActivation = Deferred.await(activation); const activationLayer = Layer.succeed(ServerActivation, awaitActivation); @@ -774,7 +784,7 @@ export const makeServerLayer = Layer.unwrap( Layer.provideMerge(FetchHttpClient.layer), // PR reads, Git operations, and WebSocket discovery share one process limiter. Layer.provide(VcsProcess.layer), - Layer.provideMerge(PlatformServicesLive), + Layer.provideMerge(platformServicesLive(placement)), ); }), ); diff --git a/apps/server/src/serviceLauncher.test.ts b/apps/server/src/serviceLauncher.test.ts index 5cf6dd858420..fe0ee1d6a34e 100644 --- a/apps/server/src/serviceLauncher.test.ts +++ b/apps/server/src/serviceLauncher.test.ts @@ -3,7 +3,7 @@ import * as NodeFS from "node:fs"; import * as NodePath from "node:path"; import * as NodeSqlite from "node:sqlite"; import * as NodeServices from "@effect/platform-node/NodeServices"; -import { assert, it } from "@effect/vitest"; +import { assert, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; @@ -596,3 +596,18 @@ if (context.update?.status === "pending") { }), ); }); + +it("rejects required placement before launcher recovery or any server generation executes", async () => { + const previous = process.env.T3_PROCESS_PLACEMENT; + process.env.T3_PROCESS_PLACEMENT = "invalid-required-binding"; + try { + const launcher = new Launcher("/not-an-application-state-directory", { + protocol: SERVICE_LAUNCHER_PROTOCOL, + activeVersion: "1.0.0", + }); + await expect(launcher.run()).rejects.toThrow("Process placement required"); + } finally { + if (previous === undefined) delete process.env.T3_PROCESS_PLACEMENT; + else process.env.T3_PROCESS_PLACEMENT = previous; + } +}); diff --git a/apps/server/src/serviceLauncher.ts b/apps/server/src/serviceLauncher.ts index a314ea1a8e83..68d1029da703 100644 --- a/apps/server/src/serviceLauncher.ts +++ b/apps/server/src/serviceLauncher.ts @@ -32,6 +32,11 @@ import { fenceNativeStoreAuthorityForBaseDir, nativeStoreAuthorityStateDirForBaseDir, } from "./environment/nativeStoreAuthorityPersistence.ts"; +import { + assertControlProcessPlacement, + placementCommand, + readProcessPlacementBinding, +} from "./process/processPlacement.ts"; import { isEntrypoint } from "./entrypoint.ts"; const HANDOFF_DELAY_MS = 2_000; @@ -395,6 +400,7 @@ export class Launcher { } async run(): Promise { + assertControlProcessPlacement(); const onSigterm = () => void this.stop("SIGTERM"); const onSigint = () => void this.stop("SIGINT"); process.once("SIGTERM", onSigterm); @@ -551,7 +557,14 @@ export class Launcher { childVersion: version, ...(update === undefined ? {} : { update }), }; - const child = NodeChildProcess.spawn(process.execPath, [paths.entryPath, "serve"], { + const placed = placementCommand( + process.execPath, + [paths.entryPath, "serve"], + "control", + readProcessPlacementBinding(), + true, + ); + const child = NodeChildProcess.spawn(placed.command, [...placed.args], { env: { ...process.env, [SERVICE_LAUNCHER_CONTEXT_ENV]: JSON.stringify(context) }, stdio: ["inherit", "inherit", "inherit", "ipc"], }); diff --git a/apps/server/src/terminal/BunPtyAdapter.test.ts b/apps/server/src/terminal/BunPtyAdapter.test.ts index e04a54e6d333..6b03c67f6e54 100644 --- a/apps/server/src/terminal/BunPtyAdapter.test.ts +++ b/apps/server/src/terminal/BunPtyAdapter.test.ts @@ -3,8 +3,12 @@ import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as Schema from "effect/Schema"; import * as BunPtyAdapter from "./BunPtyAdapter.ts"; +import * as PtyAdapter from "./PtyAdapter.ts"; +import { ProcessPlacementError } from "../process/processPlacement.ts"; +import { vi } from "vite-plus/test"; it("describes unavailable Bun PTY operations structurally", () => { const error = new BunPtyAdapter.BunPtyOperationUnavailableError({ @@ -42,3 +46,34 @@ it.effect("reports unsupported platforms with a structured startup defect", () = } }), ); + +it.effect("rejects required placement before Bun executes the terminal shell", () => + Effect.gen(function* () { + const previous = process.env.T3_PROCESS_PLACEMENT; + process.env.T3_PROCESS_PLACEMENT = "invalid-required-binding"; + const backend = vi.fn(() => { + throw new Error("Bun backend must not execute"); + }); + vi.stubGlobal("Bun", { spawn: backend }); + try { + const adapter = yield* BunPtyAdapter.make().pipe( + Effect.provideService(HostProcessPlatform, "linux"), + ); + const result = yield* adapter + .spawn({ shell: "shell-sentinel", cwd: "/workspace", cols: 80, rows: 24, env: {} }) + .pipe(Effect.exit); + expect(Exit.isFailure(result)).toBe(true); + if (Exit.isFailure(result)) { + const error = Cause.squash(result.cause); + expect(error).toBeInstanceOf(PtyAdapter.PtySpawnError); + if (!Schema.is(PtyAdapter.PtySpawnError)(error)) throw error; + expect(error.cause).toBeInstanceOf(ProcessPlacementError); + } + expect(backend).not.toHaveBeenCalled(); + } finally { + vi.unstubAllGlobals(); + if (previous === undefined) delete process.env.T3_PROCESS_PLACEMENT; + else process.env.T3_PROCESS_PLACEMENT = previous; + } + }), +); diff --git a/apps/server/src/terminal/BunPtyAdapter.ts b/apps/server/src/terminal/BunPtyAdapter.ts index 88b68940de11..eaf6e8dfaf1f 100644 --- a/apps/server/src/terminal/BunPtyAdapter.ts +++ b/apps/server/src/terminal/BunPtyAdapter.ts @@ -5,6 +5,7 @@ import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import { placementCommand, readProcessPlacementBinding } from "../process/processPlacement.ts"; import * as PtyAdapter from "./PtyAdapter.ts"; export class BunPtyUnsupportedPlatformError extends Schema.TaggedErrorClass()( @@ -127,7 +128,15 @@ export const make = Effect.fn("BunPtyAdapter.make")(function* () { Effect.try({ try: () => { let processHandle: BunPtyProcess | null = null; - const command = [input.shell, ...(input.args ?? [])]; + const placed = placementCommand( + input.shell, + input.args ?? [], + "workload", + readProcessPlacementBinding(), + false, + platform, + ); + const command = [placed.command, ...placed.args]; const subprocess = Bun.spawn(command, { cwd: input.cwd, env: input.env, diff --git a/apps/server/src/terminal/NodePtyAdapter.test.ts b/apps/server/src/terminal/NodePtyAdapter.test.ts index 7cf6a167ecfa..b1be8723c7c1 100644 --- a/apps/server/src/terminal/NodePtyAdapter.test.ts +++ b/apps/server/src/terminal/NodePtyAdapter.test.ts @@ -114,3 +114,22 @@ it.effect("reports native module load failures as structured startup defects", ( ), ), ); + +it.effect("rejects required placement before executing the terminal shell", () => + Effect.gen(function* () { + spawn.mockClear(); + const previous = process.env.T3_PROCESS_PLACEMENT; + process.env.T3_PROCESS_PLACEMENT = "invalid-required-binding"; + try { + const adapter = yield* PtyAdapter.PtyAdapter; + const result = yield* adapter + .spawn({ shell: "shell-sentinel", cwd: "/workspace", cols: 80, rows: 24, env: {} }) + .pipe(Effect.exit); + assert.isTrue(Exit.isFailure(result)); + assert.equal(spawn.mock.calls.length, 0); + } finally { + if (previous === undefined) delete process.env.T3_PROCESS_PLACEMENT; + else process.env.T3_PROCESS_PLACEMENT = previous; + } + }).pipe(Effect.provide(testLayer)), +); diff --git a/apps/server/src/terminal/NodePtyAdapter.ts b/apps/server/src/terminal/NodePtyAdapter.ts index e9c462ab2c20..0c118c39e6c7 100644 --- a/apps/server/src/terminal/NodePtyAdapter.ts +++ b/apps/server/src/terminal/NodePtyAdapter.ts @@ -7,6 +7,7 @@ import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import { HostProcessArchitecture, HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import { placementCommand, readProcessPlacementBinding } from "../process/processPlacement.ts"; import * as PtyAdapter from "./PtyAdapter.ts"; export class NodePtyModuleLoadError extends Schema.TaggedErrorClass()( @@ -149,14 +150,23 @@ export const make = Effect.fn("NodePtyAdapter.make")(function* ( ? { ...input.env, TERM: "xterm-256color" } : input.env; const ptyProcess = yield* Effect.try({ - try: () => - nodePty.spawn(input.shell, input.args ?? [], { + try: () => { + const placed = placementCommand( + input.shell, + input.args ?? [], + "workload", + readProcessPlacementBinding(), + false, + platform, + ); + return nodePty.spawn(placed.command, [...placed.args], { cwd: input.cwd, cols: input.cols, rows: input.rows, env, name: "xterm-256color", - }), + }); + }, catch: (cause) => new PtyAdapter.PtySpawnError({ adapter: "node-pty", diff --git a/docs/internals/process-placement.md b/docs/internals/process-placement.md new file mode 100644 index 000000000000..6126b543906a --- /dev/null +++ b/docs/internals/process-placement.md @@ -0,0 +1,125 @@ +# Linux process placement + +Placement is opt-in. With neither `T3_PROCESS_PLACEMENT` nor +`T3_PROCESS_PLACEMENT_BOOTSTRAP` configured, process launch behavior is unchanged. +A configured route is required: malformed, unsupported, missing, symlinked, or +stale artifacts fail before the intended payload executes. There is no fallback +to launching an unplaced payload. + +## Ready binding + +`T3_PROCESS_PLACEMENT` is a JSON object with these fields: + +```json +{ + "version": 1, + "helperPath": "/absolute/immutable/process-placement-linux-x64", + "helperSha256": "64 lowercase hexadecimal characters", + "helperDevice": "29", + "helperInode": "100", + "control": { + "path": "/sys/fs/cgroup/delegated-t3/control", + "device": "29", + "inode": "101" + }, + "workload": { + "path": "/sys/fs/cgroup/delegated-t3/workload", + "device": "29", + "inode": "102" + }, + "serverPriority": { "nice": -15, "resetOnFork": true } +} +``` + +The role paths must be distinct siblings. The owner of the delegated hierarchy +must establish and verify the CPU controller and policy before supplying this +binding. T3 checks helper path, SHA-256 and device/inode, and cgroup +device/inode identity on each launch. The executed helper verifies those same +helper identities against `/proc/self/exe` before touching a cgroup; replacement +between T3 validation and exec therefore fails closed. The native helper opens every directory component without following +symlinks, checks cgroup v2 identity and the CPU controller, writes only its own +PID to `cgroup.procs`, verifies its membership, then replaces itself with the +payload using `execvp`. It does not create cgroups or move existing processes. + +The helper keeps the same PID, cwd, environment, process group, open standard +streams and IPC descriptors. Existing cancellation, exit, signal and PTY handles +continue to refer to that PID. Workloads use `SCHED_OTHER` and nice 0. The helper +drops its Linux capabilities before payload execution. + +The optional `serverPriority` policy is applied only when the launcher starts +an actual server generation. It requests `SCHED_OTHER`, nice -15 and Linux +`SCHED_FLAG_RESET_ON_FORK`, and verifies all three before executing Node. An +unavailable permission or kernel API fails the server launch. This does not +configure capabilities or grant permission to install them. The launcher and +telemetry process receive no negative nice request. Linux resets scheduling for +new child tasks, including runtime threads; this policy gives the server main +thread the requested nice value rather than promising every Node thread has it. +The server checks its own control membership and, when requested, its scheduler +policy before building its runtime layers. + +## Cold bootstrap + +For a fresh delegated unit, `T3_PROCESS_PLACEMENT_BOOTSTRAP` supplies a version-1 +JSON object containing `executablePath`, `executableSha256`, `scriptPath`, +`scriptSha256`, `policyPath`, `policySha256`, `helperPath` and `helperSha256`. +All paths are explicit canonical absolute paths and hashes are lowercase +SHA-256. T3 validates the artifacts before installing the unit. The systemd unit +requests `Delegate=cpu memory pids`, persists this binding, and invokes: + +```text +EXECUTABLE SCRIPT --script-sha256 HASH --policy POLICY --policy-sha256 HASH --helper HELPER --helper-sha256 HASH -- NODE LAUNCHER +``` + +The external bootstrap owns its source/interpreter/policy/helper validation, +creation of a fresh delegated hierarchy, CPU policy readback, and failure +handling. It must remove the cold bootstrap variable, emit the verified ready +binding, and replace itself with the launcher only after that work succeeds. +Configured placement `ExecStart` commands use systemd's `:` executable prefix +to disable environment substitution for the whole command line, preserving +literal `$` and `${...}` in helper paths and nested payload arguments. Percent +specifiers are escaped separately. This follows the upstream +[systemd.service command-line contract](https://raw.githubusercontent.com/systemd/systemd/main/man/systemd.service.xml). +T3 has no dependency on a particular external policy producer. Configuring both +routes for service installation is rejected. Direct `serve` with a cold binding +and no ready binding is rejected. A ready binding also requires the invoking +server or launcher to already be in its control cgroup. + +## Process boundaries + +| Route | Placement | +| ------------------------------------------------------- | ----------------------------------------------------------------------- | +| Stable launcher | Control, before execution through the unit helper or external bootstrap | +| Active, update trial and rollback server generations | Control, native helper before execution | +| Codex sessions and CLI probes | Workload, shared Effect spawner | +| Claude SDK sessions and SDK capability probes | Workload, explicit SDK spawn hook | +| Cursor, Grok and Antigravity ACP sessions/probes | Workload, shared Effect spawner | +| OpenCode server sessions/probes | Workload, shared Effect spawner | +| Ephemeral title, branch, commit and PR text generation | Workload, shared Effect spawner | +| Git, setup scripts and ordinary process-runner commands | Workload, shared Effect spawner | +| Node and Bun PTYs | Workload, explicit adapter boundary | +| Native resource telemetry | Control, explicitly marked Effect command | +| Provider tools and MCP children | Inherit their provider's workload cgroup | + +Independent shells, Codex, OMP or other agent roots outside this T3 process tree +need their own explicitly bound startup wrapper. T3 placement does not place +those processes, move existing sessions, or establish a host-wide CPU cap. + +## Build and evidence + +The Linux server bundle builds the helper with `cc` into +`dist/native/process-placement-linux-ARCH`. Its adjacent JSON manifest records +the captured native source SHA-256 and compiled helper SHA-256. Each compile +uses a unique staging directory under that output directory, routes compiler +temporary files there, waits for compiler exit on ordinary cancellation, and +removes only its captured staging root in `finally`. Failure preserves the prior +helper; cleanup failure preserves the original compiler failure and is reported. +The standalone build lifecycle tests use workspace-local fixtures and clean each +captured fixture root. Other platforms keep the +disabled behavior and reject a required Linux placement route. + +Focused tests exercise disabled behavior, strict binding validation, rejection +before backend/SDK/PTY execution, service unit commands, and native rejection +with an unexecuted payload sentinel. Successful placement into a delegated Linux +cgroup and optional negative nice/reset-on-fork capability require separate +host evidence. Compilation, mocked tests, and source bundle tests do not prove +installation, live delegation, or deployment provenance. diff --git a/native/process-placement/build.mjs b/native/process-placement/build.mjs new file mode 100644 index 000000000000..2ee11fbbd86f --- /dev/null +++ b/native/process-placement/build.mjs @@ -0,0 +1,106 @@ +import * as NodeCrypto from "node:crypto"; +import * as NodeChildProcess from "node:child_process"; +import * as NodeFS from "node:fs"; +import * as NodeURL from "node:url"; +import * as NodePath from "node:path"; +import * as NodeOS from "node:os"; + +const hash = (bytes) => NodeCrypto.createHash("sha256").update(bytes).digest("hex"); +export async function buildPlacement({ + // oxlint-disable-next-line t3code/no-global-process-runtime -- Plain MJS build boundary supports Node 22 without TS loading; tests inject this input. + platform = NodeOS.platform(), + // oxlint-disable-next-line t3code/no-global-process-runtime -- Native compiler architecture default stays in the plain MJS boundary; tests inject it. + architecture = NodeOS.arch(), + source = NodeURL.fileURLToPath(new URL("./placement.c", import.meta.url)), + outputDir = NodeURL.fileURLToPath(new URL("../../apps/server/dist/native", import.meta.url)), + compiler = "cc", + compilerArgs = [], + signal, + removeScratch = NodeFS.rmSync, +} = {}) { + if (platform !== "linux") return; + const sourceBytes = NodeFS.readFileSync(source); + NodeFS.mkdirSync(outputDir, { recursive: true }); + const staging = NodeFS.mkdtempSync(NodePath.join(outputDir, ".placement-build-")); + let primaryFailure; + try { + const sourceCopy = NodePath.join(staging, "placement.c"); + const name = `process-placement-linux-${architecture}`; + const candidate = NodePath.join(staging, name); + NodeFS.writeFileSync(sourceCopy, sourceBytes); + const status = await new Promise((accept, reject) => { + const child = NodeChildProcess.spawn( + compiler, + [ + ...compilerArgs, + "-O2", + "-std=c11", + "-Wall", + "-Wextra", + "-Werror", + sourceCopy, + "-o", + candidate, + ], + { + stdio: "inherit", + env: { ...process.env, TMPDIR: staging }, + }, + ); + const abort = () => child.kill("SIGTERM"); + signal?.addEventListener("abort", abort, { once: true }); + if (signal?.aborted) abort(); + child.once("error", reject); + child.once("close", (code, terminated) => { + signal?.removeEventListener("abort", abort); + accept(signal?.aborted ? 143 : (code ?? (terminated ? 143 : 1))); + }); + }); + if (status !== 0) + throw Object.assign(new Error(`Placement compiler failed (${status})`), { exitCode: status }); + NodeFS.chmodSync(candidate, 0o755); + const manifest = `${JSON.stringify( + { + version: 1, + architecture, + sourceSha256: hash(sourceBytes), + helperSha256: hash(NodeFS.readFileSync(candidate)), + }, + null, + 2, + )}\n`; + NodeFS.writeFileSync(`${candidate}.json`, manifest); + NodeFS.renameSync(candidate, NodePath.join(outputDir, name)); + NodeFS.renameSync(`${candidate}.json`, NodePath.join(outputDir, `${name}.json`)); + } catch (cause) { + primaryFailure = cause; + throw cause; + } finally { + try { + removeScratch(staging, { recursive: true, force: true }); + } catch (cleanupError) { + if (primaryFailure === undefined) throw cleanupError; + primaryFailure.cleanupError = cleanupError; + console.error(`Placement scratch cleanup failed: ${cleanupError.message}`); + } + } +} + +if ( + process.argv[1] !== undefined && + NodePath.resolve(process.argv[1]) === NodeURL.fileURLToPath(import.meta.url) +) { + const cancellation = new AbortController(); + const stop = () => cancellation.abort(); + process.once("SIGTERM", stop); + process.once("SIGINT", stop); + try { + await buildPlacement({ signal: cancellation.signal }); + } catch (cause) { + console.error(cause.message); + process.exitCode = cause.exitCode ?? 1; + } finally { + process.removeListener("SIGTERM", stop); + process.removeListener("SIGINT", stop); + } +} diff --git a/native/process-placement/build.test.mjs b/native/process-placement/build.test.mjs new file mode 100644 index 000000000000..54a20b6a6f21 --- /dev/null +++ b/native/process-placement/build.test.mjs @@ -0,0 +1,114 @@ +import * as NodeTest from "node:test"; +import * as NodeAssert from "node:assert/strict"; +import * as NodeFS from "node:fs"; +import * as NodePath from "node:path"; +import * as NodeURL from "node:url"; +import * as NodeCrypto from "node:crypto"; +import { buildPlacement } from "./build.mjs"; + +async function fixture(action) { + const root = NodeFS.mkdtempSync( + NodePath.join(NodePath.dirname(NodeURL.fileURLToPath(import.meta.url)), ".build-test-"), + ); + try { + const source = NodePath.join(root, "input.c"); + const compiler = NodePath.join(root, "compiler.mjs"); + NodeFS.writeFileSync(source, "captured native source\n"); + NodeFS.writeFileSync( + compiler, + `import {writeFileSync} from 'node:fs'; +const args=process.argv.slice(2); const mode=args.shift(); +if(mode==='fail') process.exit(23); +if(mode==='wait') { process.on('SIGTERM',()=>process.exit(143)); setInterval(()=>{},1000); } +else writeFileSync(args[args.indexOf('-o')+1], 'compiled fixture');\n`, + ); + const build = (outputDir, mode = "ok", signal, removeScratch) => + buildPlacement({ + platform: "linux", + architecture: "fixture", + source, + outputDir, + compiler: process.execPath, + compilerArgs: [compiler, mode], + signal, + removeScratch, + }); + await action({ root, source, build }); + } finally { + NodeFS.rmSync(root, { recursive: true, force: true }); + } + NodeAssert.equal(NodeFS.existsSync(root), false); +} +const sha = (value) => NodeCrypto.createHash("sha256").update(value).digest("hex"); + +NodeTest.test("success retains only helper and hash manifest and removes owned build scratch", () => + fixture(async ({ root, source, build }) => { + const output = NodePath.join(root, "output"); + await build(output); + NodeAssert.deepEqual(NodeFS.readdirSync(output).sort(), [ + "process-placement-linux-fixture", + "process-placement-linux-fixture.json", + ]); + const manifest = JSON.parse( + NodeFS.readFileSync(NodePath.join(output, "process-placement-linux-fixture.json"), "utf8"), + ); + NodeAssert.equal(manifest.sourceSha256, sha(NodeFS.readFileSync(source))); + NodeAssert.equal( + manifest.helperSha256, + sha(NodeFS.readFileSync(NodePath.join(output, "process-placement-linux-fixture"))), + ); + }), +); +NodeTest.test( + "compiler failure preserves status and prior output while removing build scratch", + () => + fixture(async ({ root, build }) => { + const output = NodePath.join(root, "output"); + await build(output); + await NodeAssert.rejects(build(output, "fail"), (cause) => cause.exitCode === 23); + NodeAssert.equal(NodeFS.readdirSync(output).length, 2); + NodeAssert.equal( + NodeFS.readFileSync(NodePath.join(output, "process-placement-linux-fixture"), "utf8"), + "compiled fixture", + ); + }), +); +NodeTest.test("normal cancellation reaps the compiler before removing its captured scratch", () => + fixture(async ({ root, build }) => { + const output = NodePath.join(root, "output"); + const controller = new AbortController(); + const pending = build(output, "wait", controller.signal); + controller.abort(); + await NodeAssert.rejects(pending, (cause) => cause.exitCode === 143); + NodeAssert.deepEqual(NodeFS.readdirSync(output), []); + }), +); +NodeTest.test("concurrent build invocations keep separate captured scratch roots", () => + fixture(async ({ root, build }) => { + const one = NodePath.join(root, "one"), + two = NodePath.join(root, "two"); + await Promise.all([ + build(one), + build(two, "fail").catch((cause) => NodeAssert.equal(cause.exitCode, 23)), + ]); + NodeAssert.equal(NodeFS.readdirSync(one).length, 2); + NodeAssert.deepEqual(NodeFS.readdirSync(two), []); + }), +); + +NodeTest.test("cleanup failure fails success and does not hide an original compiler failure", () => + fixture(async ({ root, build }) => { + const cleanupFailure = new Error("fixture cleanup failure"); + const removeScratch = () => { + throw cleanupFailure; + }; + await NodeAssert.rejects( + build(NodePath.join(root, "success"), "ok", undefined, removeScratch), + (cause) => cause === cleanupFailure, + ); + await NodeAssert.rejects( + build(NodePath.join(root, "failed"), "fail", undefined, removeScratch), + (cause) => cause.exitCode === 23 && cause.cleanupError === cleanupFailure, + ); + }), +); diff --git a/native/process-placement/placement.c b/native/process-placement/placement.c new file mode 100644 index 000000000000..a53d5bf4ae41 --- /dev/null +++ b/native/process-placement/placement.c @@ -0,0 +1,187 @@ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +static void fail(const char *message) { + fprintf(stderr, "t3 process placement: %s\n", message); + exit(125); +} +struct sha256 { uint32_t state[8]; uint64_t bytes; unsigned char block[64]; size_t used; }; +static uint32_t rotate(uint32_t value, unsigned count) { return (value >> count) | (value << (32 - count)); } +static void sha_block(struct sha256 *hash) { + static const uint32_t k[64] = { + 0x428a2f98,0x71374491,0xb5c0fbcf,0xe9b5dba5,0x3956c25b,0x59f111f1,0x923f82a4,0xab1c5ed5, + 0xd807aa98,0x12835b01,0x243185be,0x550c7dc3,0x72be5d74,0x80deb1fe,0x9bdc06a7,0xc19bf174, + 0xe49b69c1,0xefbe4786,0x0fc19dc6,0x240ca1cc,0x2de92c6f,0x4a7484aa,0x5cb0a9dc,0x76f988da, + 0x983e5152,0xa831c66d,0xb00327c8,0xbf597fc7,0xc6e00bf3,0xd5a79147,0x06ca6351,0x14292967, + 0x27b70a85,0x2e1b2138,0x4d2c6dfc,0x53380d13,0x650a7354,0x766a0abb,0x81c2c92e,0x92722c85, + 0xa2bfe8a1,0xa81a664b,0xc24b8b70,0xc76c51a3,0xd192e819,0xd6990624,0xf40e3585,0x106aa070, + 0x19a4c116,0x1e376c08,0x2748774c,0x34b0bcb5,0x391c0cb3,0x4ed8aa4a,0x5b9cca4f,0x682e6ff3, + 0x748f82ee,0x78a5636f,0x84c87814,0x8cc70208,0x90befffa,0xa4506ceb,0xbef9a3f7,0xc67178f2 + }; + uint32_t w[64]; + for (int i = 0; i < 16; i++) w[i] = (uint32_t)hash->block[4*i] << 24 | (uint32_t)hash->block[4*i+1] << 16 | (uint32_t)hash->block[4*i+2] << 8 | hash->block[4*i+3]; + for (int i = 16; i < 64; i++) { + uint32_t a = w[i-15], b = w[i-2]; + w[i] = w[i-16] + (rotate(a,7) ^ rotate(a,18) ^ (a >> 3)) + w[i-7] + (rotate(b,17) ^ rotate(b,19) ^ (b >> 10)); + } + uint32_t a=hash->state[0], b=hash->state[1], c=hash->state[2], d=hash->state[3], e=hash->state[4], f=hash->state[5], g=hash->state[6], h=hash->state[7]; + for (int i = 0; i < 64; i++) { + uint32_t first=h+(rotate(e,6)^rotate(e,11)^rotate(e,25))+((e&f)^(~e&g))+k[i]+w[i]; + uint32_t second=(rotate(a,2)^rotate(a,13)^rotate(a,22))+((a&b)^(a&c)^(b&c)); + h=g;g=f;f=e;e=d+first;d=c;c=b;b=a;a=first+second; + } + hash->state[0]+=a;hash->state[1]+=b;hash->state[2]+=c;hash->state[3]+=d;hash->state[4]+=e;hash->state[5]+=f;hash->state[6]+=g;hash->state[7]+=h; +} +static void sha_byte(struct sha256 *hash, unsigned char value) { + hash->block[hash->used++] = value; + if (hash->used == 64) { sha_block(hash); hash->used = 0; } +} +static void verify_helper(const char *path, const char *expected, uintmax_t device, uintmax_t inode) { + char actual[PATH_MAX]; + ssize_t size = readlink("/proc/self/exe", actual, sizeof(actual)-1); + if (size < 0 || size >= (ssize_t)sizeof(actual)-1) fail("helper path unavailable"); + actual[size] = 0; + if (strcmp(actual, path) || strlen(expected) != 64) fail("helper identity changed"); + int fd = open("/proc/self/exe", O_RDONLY | O_CLOEXEC); + struct stat stat; + if (fd < 0 || fstat(fd, &stat) || !S_ISREG(stat.st_mode) || (uintmax_t)stat.st_dev != device || + (uintmax_t)stat.st_ino != inode || (stat.st_mode & 0022) || stat.st_uid != getuid()) fail("helper identity changed"); + struct sha256 hash = { .state = {0x6a09e667,0xbb67ae85,0x3c6ef372,0xa54ff53a,0x510e527f,0x9b05688c,0x1f83d9ab,0x5be0cd19} }; + unsigned char buffer[4096]; ssize_t count; + while ((count = read(fd, buffer, sizeof(buffer))) > 0) { + hash.bytes += (uint64_t)count; + for (ssize_t i = 0; i < count; i++) sha_byte(&hash, buffer[i]); + } + if (count < 0) fail("helper hash unavailable"); + close(fd); + uint64_t bits = hash.bytes * 8; + sha_byte(&hash, 0x80); + while (hash.used != 56) sha_byte(&hash, 0); + for (int i = 7; i >= 0; i--) sha_byte(&hash, (unsigned char)(bits >> (8*i))); + char digest[65]; + for (int i = 0; i < 8; i++) snprintf(digest+8*i, 9, "%08" PRIx32, hash.state[i]); + if (strcmp(digest, expected)) fail("helper hash changed"); +} +static uintmax_t number(const char *input) { + char *end; + errno = 0; + if (!*input || strlen(input) > 20) fail("invalid identity"); + for (const char *digit = input; *digit; digit++) if (*digit < '0' || *digit > '9') fail("invalid identity"); + uintmax_t value = strtoumax(input, &end, 10); + if (errno || *end) fail("invalid identity"); + return value; +} +static int open_directory(const char *path) { + if (path[0] != '/' || strlen(path) >= PATH_MAX) fail("invalid cgroup path"); + int fd = open("/", O_PATH | O_DIRECTORY | O_CLOEXEC); + if (fd < 0) fail("cannot open filesystem root"); + char copy[PATH_MAX]; + strcpy(copy, path + 1); + char *save = NULL; + for (char *part = strtok_r(copy, "/", &save); part; part = strtok_r(NULL, "/", &save)) { + if (!strcmp(part, ".") || !strcmp(part, "..")) fail("invalid cgroup path"); + int next = openat(fd, part, O_PATH | O_NOFOLLOW | O_DIRECTORY | O_CLOEXEC); + close(fd); + if (next < 0) fail("cgroup unavailable or symlinked"); + fd = next; + } + return fd; +} +static void identity(int fd, uintmax_t device, uintmax_t inode) { + struct stat stat; + struct statfs fs; + if (fstat(fd, &stat) || fstatfs(fd, &fs) || fs.f_type != CGROUP2_SUPER_MAGIC || + (uintmax_t)stat.st_dev != device || (uintmax_t)stat.st_ino != inode) fail("stale cgroup identity"); +} +struct placement_sched_attr { + unsigned int size; + unsigned int policy; + unsigned long long flags; + int nice; + unsigned int priority; + unsigned long long runtime, deadline, period; +}; +static void verify_server_priority(pid_t pid) { + struct placement_sched_attr attr = { .size = sizeof(attr) }; + if (syscall(SYS_sched_getattr, pid, &attr, sizeof(attr), 0) || attr.policy != SCHED_OTHER || + attr.nice != -15 || !(attr.flags & 1ULL)) fail("server priority or reset-on-fork unavailable"); +} +int main(int argc, char **argv) { + if (argc < 11 || strcmp(argv[1], "--helper-path") || strcmp(argv[3], "--helper-sha256") || + strcmp(argv[5], "--helper-device") || strcmp(argv[7], "--helper-inode")) fail("invalid helper invocation"); + verify_helper(argv[2], argv[4], number(argv[6]), number(argv[8])); + argc -= 8; argv += 8; + if (argc == 3 && !strcmp(argv[1], "--verify-server-priority")) { + uintmax_t pid = number(argv[2]); + if (pid == 0 || pid > INT_MAX || (pid_t)pid != getppid()) fail("only the invoking parent may be verified"); + verify_server_priority((pid_t)pid); + return 0; + } + int server_priority = argc >= 13 && !strcmp(argv[9], "--server-priority") && !strcmp(argv[10], "-15"); + int delimiter = server_priority ? 11 : 9; + if (argc < 11 || strcmp(argv[1], "--role") || strcmp(argv[3], "--cgroup") || + strcmp(argv[5], "--device") || strcmp(argv[7], "--inode") || strcmp(argv[delimiter], "--") || + (strcmp(argv[2], "control") && strcmp(argv[2], "workload"))) fail("invalid invocation"); + if (server_priority && strcmp(argv[2], "control")) fail("server priority requires control role"); + const char *path = argv[4]; + if (strncmp(path, "/sys/fs/cgroup/", 15)) fail("cgroup must be beneath /sys/fs/cgroup"); + uintmax_t device = number(argv[6]), inode = number(argv[8]); + int directory = open_directory(path); + identity(directory, device, inode); + int cpu = openat(directory, "cpu.max", O_RDONLY | O_NOFOLLOW | O_CLOEXEC); + if (cpu < 0) fail("CPU controller unavailable"); + close(cpu); + int procs = openat(directory, "cgroup.procs", O_WRONLY | O_NOFOLLOW | O_CLOEXEC); + if (procs < 0) fail("delegation unavailable"); + char pid[32]; + int size = snprintf(pid, sizeof(pid), "%ld\n", (long)getpid()); + if (write(procs, pid, (size_t)size) != size) fail("placement rejected"); + close(procs); + FILE *membership = fopen("/proc/self/cgroup", "re"); + if (!membership) fail("cannot verify membership"); + char *line = NULL; size_t capacity = 0; int verified = 0; + while (getline(&line, &capacity, membership) >= 0) { + if (strncmp(line, "0::", 3)) continue; + line[strcspn(line, "\n")] = 0; + char current[PATH_MAX]; + if (snprintf(current, sizeof(current), "/sys/fs/cgroup%s", line + 3) >= (int)sizeof(current)) fail("membership too long"); + int actual = open_directory(current); + identity(actual, device, inode); + close(actual); verified = 1; + } + free(line); fclose(membership); + if (!verified) fail("membership mismatch"); + identity(directory, device, inode); + close(directory); + if (!strcmp(argv[2], "workload")) { + struct sched_param params = { .sched_priority = 0 }; + if (sched_setscheduler(0, SCHED_OTHER, ¶ms) || setpriority(PRIO_PROCESS, 0, 0)) fail("cannot reset workload scheduling"); + } + if (server_priority) { + struct placement_sched_attr attr = { .size = sizeof(attr), .policy = SCHED_OTHER, .flags = 1ULL, .nice = -15 }; + if (syscall(SYS_sched_setattr, 0, &attr, 0)) fail("server priority capability unavailable"); + verify_server_priority(0); + } + struct __user_cap_header_struct capability_header = { .version = _LINUX_CAPABILITY_VERSION_3, .pid = 0 }; + struct __user_cap_data_struct capability_data[2] = { { 0 }, { 0 } }; + if (prctl(PR_CAP_AMBIENT, PR_CAP_AMBIENT_CLEAR_ALL, 0, 0, 0) || + syscall(SYS_capset, &capability_header, capability_data)) fail("cannot drop helper capabilities"); + execvp(argv[delimiter + 1], &argv[delimiter + 1]); + fail("payload execution failed"); +}