Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
156 changes: 156 additions & 0 deletions apps/server/src/process/taskScope.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,156 @@
import * as NodeChildProcess from "node:child_process";
import * as NodeFSP from "node:fs/promises";
import * as NodeOS from "node:os";
import * as NodePath from "node:path";
import * as NodeUtil from "node:util";

import { HostProcessPlatform } from "@t3tools/shared/hostProcess";

import { afterEach, describe, expect, it } from "vite-plus/test";
import { taskScopeCommand } from "./taskScope.ts";

const execFile = NodeUtil.promisify(NodeChildProcess.execFile);
const directories: string[] = [];

afterEach(async () => {
await Promise.all(
directories.splice(0).map((dir) => NodeFSP.rm(dir, { recursive: true, force: true })),
);
});

async function fixture(busExit = 0, startExit = 0) {
const directory = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3-task-scope-"));
directories.push(directory);
const log = NodePath.join(directory, "calls");
for (const [name, body] of Object.entries({
timeout: 'shift; exec "$@"',
busctl: `printf 'busctl %s\\n' "$*" >> "$CALLS"; exit ${busExit}`,
systemctl: `printf 'systemctl %s\\n' "$*" >> "$CALLS"; exit ${startExit}`,
})) {
await NodeFSP.writeFile(NodePath.join(directory, name), `#!/bin/sh\n${body}\n`, {
mode: 0o755,
});
}
return { directory, log, env: { PATH: directory, CALLS: log, VALUE: "inherited value" } };
}

async function launch(f: Awaited<ReturnType<typeof fixture>>, script: string, args: string[] = []) {
const command = taskScopeCommand("/bin/sh", ["-c", script, "workload", ...args], "linux");
return execFile(command.command, command.args, { cwd: f.directory, env: f.env });
}

describe("taskScopeCommand", () => {
it.each(["darwin", "win32", "freebsd"] as const)("leaves %s launches unchanged", (platform) => {
expect(taskScopeCommand("agent", ["--arg", "a b"], platform)).toEqual({
command: "agent",
args: ["--arg", "a b"],
});
});

it.skipIf(HostProcessPlatform.defaultValue() !== "linux")(
"waits for scope startup before exec and preserves PID, cwd, environment and arguments",
async () => {
const f = await fixture();
const args = ["a b", "$(touch should-not-exist)", "$HOME", "", "--flag"];
const result = await launch(
f,
'printf "workload %s\\n" "$$" >> "$CALLS"; printf "%s\\n" "$PWD" "$VALUE" "$@"',
args,
);
expect(result.stdout).toBe([f.directory, "inherited value", ...args, ""].join("\n"));
expect(result.stderr).toBe("");
const calls = (await NodeFSP.readFile(f.log, "utf8")).trim().split("\n");
const pid = calls[2]?.split(" ")[1];
const unit = calls[0]?.match(/t3code-task-[0-9a-f-]+\.scope/)?.[0];
expect(unit).toBeDefined();
expect(calls[0]).toBe(
`busctl --user --timeout=2s call org.freedesktop.systemd1 /org/freedesktop/systemd1 org.freedesktop.systemd1.Manager StartTransientUnit ssa(sv)a(sa(sv)) ${unit} fail 3 PIDs au 1 ${pid} Slice s app.slice CollectMode s inactive-or-failed 0`,
);
expect(calls[1]).toBe(`systemctl --user --no-ask-password start ${unit}`);
expect(await NodeFSP.readdir(f.directory)).not.toContain("should-not-exist");
},
);

it.skipIf(HostProcessPlatform.defaultValue() !== "linux").each([
[1, 0],
[0, 1],
])("falls back when systemd setup fails (%s, %s)", async (busExit, startExit) => {
const f = await fixture(busExit, startExit);
const result = await launch(f, 'printf "ran\\n" >> "$CALLS"; printf ok');
expect(result.stdout).toBe("ok");
expect(result.stderr).toBe("");
const calls = await NodeFSP.readFile(f.log, "utf8");
expect(calls.match(/^ran$/gm)).toHaveLength(1);
if (busExit !== 0) expect(calls).not.toContain("systemctl");
});

it.skipIf(HostProcessPlatform.defaultValue() !== "linux")(
"falls back without systemd tools",
async () => {
const f = await fixture();
await NodeFSP.unlink(NodePath.join(f.directory, "busctl"));
expect((await launch(f, "printf ok")).stdout).toBe("ok");
await NodeFSP.unlink(NodePath.join(f.directory, "timeout"));
expect((await launch(f, "printf ok")).stdout).toBe("ok");
},
);

it.skipIf(HostProcessPlatform.defaultValue() !== "linux")(
"does not retry a failing workload and preserves its exit code and stderr",
async () => {
const f = await fixture();
await expect(
launch(f, 'printf "ran\\n" >> "$CALLS"; printf failure >&2; exit 42'),
).rejects.toMatchObject({ code: 42, stderr: "failure" });
expect((await NodeFSP.readFile(f.log, "utf8")).match(/^ran$/gm)).toHaveLength(1);
},
);

it.skipIf(HostProcessPlatform.defaultValue() !== "linux")("preserves stdin", async () => {
const f = await fixture();
const command = taskScopeCommand("/bin/cat", [], "linux");
const child = NodeChildProcess.spawn(command.command, command.args, { env: f.env });
const output: Buffer[] = [];
child.stdout.on("data", (chunk: Buffer) => output.push(chunk));
const closed = new Promise<number | null>((resolve, reject) => {
child.once("error", reject);
child.once("close", resolve);
});
child.stdin.end("input\n");
expect(await closed).toBe(0);
expect(Buffer.concat(output).toString()).toBe("input\n");
});
it.skipIf(HostProcessPlatform.defaultValue() !== "linux")(
"preserves termination signals after exec",
async () => {
const f = await fixture();
const command = taskScopeCommand("/bin/sh", ["-c", "printf ready; exec /bin/cat"], "linux");
const child = NodeChildProcess.spawn(command.command, command.args, { env: f.env });
const closed = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>(
(resolve, reject) => {
child.once("error", reject);
child.once("close", (code, signal) => resolve({ code, signal }));
},
);
await new Promise<void>((resolve, reject) => {
child.once("error", reject);
child.stdout.once("data", () => resolve());
});
child.kill("SIGTERM");
expect(await closed).toEqual({ code: null, signal: "SIGTERM" });
},
);
it.skipIf(HostProcessPlatform.defaultValue() !== "linux")(
"bounds a hung user-manager setup",
async () => {
const f = await fixture();
await NodeFSP.copyFile("/usr/bin/timeout", NodePath.join(f.directory, "timeout"));
await NodeFSP.writeFile(
NodePath.join(f.directory, "busctl"),
"#!/bin/sh\nexec /bin/sleep 30\n",
);
expect((await launch(f, "printf fallback")).stdout).toBe("fallback");
},
6000,
);
});
52 changes: 52 additions & 0 deletions apps/server/src/process/taskScope.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
// @effect-diagnostics nodeBuiltinImport:off -- Unique scope names are generated at the process boundary.
import * as NodeCrypto from "node:crypto";

import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
import { resolveSpawnCommand } from "@t3tools/shared/shell";
import * as Effect from "effect/Effect";

// Move the launcher before exec, so even the workload's first children inherit
// its scope. A separate helper performs the D-Bus call and waits for the start
// job. Failure (including missing tools/user bus) simply leaves the launcher in
// its original cgroup. Never retry the workload itself: a nonzero exit may have
// already performed side effects. exec preserves the PID, stdio and signals.
const SCOPE_LAUNCHER = `
timeout 3s /bin/sh -c '
busctl --user --timeout=2s call org.freedesktop.systemd1 \\
/org/freedesktop/systemd1 org.freedesktop.systemd1.Manager \\
StartTransientUnit "ssa(sv)a(sa(sv))" "$1" fail 3 \\
PIDs au 1 "$2" Slice s app.slice CollectMode s inactive-or-failed 0 &&
systemctl --user --no-ask-password start "$1"
' t3code-scope "$1" "$$" </dev/null >/dev/null 2>&1
shift
exec "$@"
`;

/** Only long-lived local workload roots belong here, not utility commands. */
export function taskScopeCommand(
command: string,
args: ReadonlyArray<string>,
platform: NodeJS.Platform,
) {
return platform === "linux"
? {
command: "/bin/sh",
args: [
"-c",
SCOPE_LAUNCHER,
"t3code-task",
`t3code-task-${NodeCrypto.randomUUID()}.scope`,
command,
...args,
],
}
: { command, args };
}

export const resolveTaskSpawnCommand = Effect.fn("resolveTaskSpawnCommand")(function* (
...input: Parameters<typeof resolveSpawnCommand>
) {
const resolved = yield* resolveSpawnCommand(...input);
const platform = yield* HostProcessPlatform;
return { ...resolved, ...taskScopeCommand(resolved.command, resolved.args, platform) };
});
29 changes: 27 additions & 2 deletions apps/server/src/provider/Layers/ClaudeAdapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
*/

import * as NodeUtil from "node:util";
import * as NodeChildProcess from "node:child_process";
import { taskScopeCommand } from "../../process/taskScope.ts";
import {
type CanUseTool,
query,
Expand Down Expand Up @@ -70,7 +72,7 @@ import {
CLAUDE_RESUME_COMPACTION_NEVER_ANSWER,
formatClaudeResumeCompactionQuestion,
} from "@t3tools/shared/claudeCompaction";
import { HostProcessIsExecutable } from "@t3tools/shared/hostProcess";
import { HostProcessIsExecutable, HostProcessPlatform } from "@t3tools/shared/hostProcess";
import * as Cause from "effect/Cause";
import * as Crypto from "effect/Crypto";
import * as DateTime from "effect/DateTime";
Expand Down Expand Up @@ -2074,6 +2076,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
claudeSettings: ClaudeSettings,
options?: ClaudeAdapterLiveOptions,
) {
const platform = yield* HostProcessPlatform;
const boundInstanceId = options?.instanceId ?? ProviderInstanceId.make("claudeAgent");
const modelCatalogEffect = (
options?.modelCatalog ?? Effect.succeed(BUNDLED_CLAUDE_MODEL_CATALOG)
Expand Down Expand Up @@ -2108,7 +2111,29 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
}) =>
query({
prompt: input.prompt,
options: input.options,
options: {
...input.options,
...(platform === "linux"
? {
spawnClaudeCodeProcess: (spawnOptions) => {
const command = taskScopeCommand(
spawnOptions.command,
spawnOptions.args,
platform,
);
const child = NodeChildProcess.spawn(command.command, [...command.args], {
cwd: spawnOptions.cwd,
env: spawnOptions.env,
signal: spawnOptions.signal,
stdio: ["pipe", "pipe", "pipe"],
});
child.stderr.setEncoding("utf8");
child.stderr.on("data", (chunk: string) => input.options.stderr?.(chunk));
return child;
},
}
: {}),
},
}) as ClaudeQueryRuntime);

const sessions = new Map<ThreadId, ClaudeSessionContext>();
Expand Down
4 changes: 2 additions & 2 deletions apps/server/src/provider/Layers/CodexSessionRuntime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ import {
ThreadId,
TurnId,
} from "@t3tools/contracts";
import { resolveSpawnCommand } from "@t3tools/shared/shell";
import { resolveTaskSpawnCommand } from "../../process/taskScope.ts";
import { normalizeModelSlug } from "@t3tools/shared/model";
import * as Crypto from "effect/Crypto";
import * as DateTime from "effect/DateTime";
Expand Down Expand Up @@ -1332,7 +1332,7 @@ export const makeCodexSessionRuntime = (
};
const extendEnv = options.environment === undefined;
const appServerArgs = codexSessionAppServerArgs(options.appServerArgs, options.launchArgs);
const spawnCommand = yield* resolveSpawnCommand(options.binaryPath, appServerArgs, {
const spawnCommand = yield* resolveTaskSpawnCommand(options.binaryPath, appServerArgs, {
env,
extendEnv,
});
Expand Down
4 changes: 2 additions & 2 deletions apps/server/src/provider/acp/AcpSessionRuntime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ import * as EffectAcpClient from "effect-acp/client";
import * as EffectAcpErrors from "effect-acp/errors";
import type * as EffectAcpSchema from "effect-acp/schema";
import type * as EffectAcpProtocol from "effect-acp/protocol";
import { resolveSpawnCommand } from "@t3tools/shared/shell";
import { resolveTaskSpawnCommand } from "../../process/taskScope.ts";

import { appendAcpStderrTail, sanitizeAcpStderrExcerpt } from "./AcpStderr.ts";
import {
Expand Down Expand Up @@ -460,7 +460,7 @@ export const make = (
),
);

const spawnCommand = yield* resolveSpawnCommand(options.spawn.command, options.spawn.args, {
const spawnCommand = yield* resolveTaskSpawnCommand(options.spawn.command, options.spawn.args, {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High acp/AcpSessionRuntime.ts:463

With extendEnv: false and a restricted PATH, a valid absolute ACP command falls back to the unwrapped process when resolveTaskSpawnCommand cannot find timeout, busctl, or systemctl, leaving the workload in T3's cgroup and unprotected from the intended OOM isolation. Keep the ACP-specific resolveSpawnCommand path, or otherwise resolve these wrapper dependencies independently of the agent-supplied PATH.

🤖 Copy this AI Prompt to have your agent fix this:
In file @apps/server/src/provider/acp/AcpSessionRuntime.ts around line 463:

With `extendEnv: false` and a restricted `PATH`, a valid absolute ACP command falls back to the unwrapped process when `resolveTaskSpawnCommand` cannot find `timeout`, `busctl`, or `systemctl`, leaving the workload in T3's cgroup and unprotected from the intended OOM isolation. Keep the ACP-specific `resolveSpawnCommand` path, or otherwise resolve these wrapper dependencies independently of the agent-supplied `PATH`.

...(options.spawn.env ? { env: options.spawn.env } : {}),
extendEnv: options.spawn.extendEnv ?? true,
});
Expand Down
7 changes: 6 additions & 1 deletion apps/server/src/provider/opencodeRuntime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ import * as NetService from "@t3tools/shared/Net";
import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
import { compareSemverVersions, parseSemver } from "@t3tools/shared/semver";
import { resolveSpawnCommand } from "@t3tools/shared/shell";
import { resolveTaskSpawnCommand } from "../process/taskScope.ts";
const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonString(Schema.Unknown));
const OPENCODE_EMPTY_CONFIG_CONTENT = "{}";

Expand Down Expand Up @@ -682,7 +683,11 @@ const makeOpenCodeRuntime = Effect.gen(function* () {
));
const timeoutMs = input.timeoutMs ?? DEFAULT_OPENCODE_SERVER_TIMEOUT_MS;
const args = ["serve", `--hostname=${hostname}`, `--port=${port}`];
const spawnCommand = yield* resolveCommand(input.binaryPath, args, input.environment);
const spawnCommand = yield* resolveTaskSpawnCommand(
input.binaryPath,
args,
input.environment ? { env: input.environment } : {},
);
const serverPassword = resolveOpenCodeServerPassword({
external: false,
...(input.serverPassword !== undefined ? { serverPassword: input.serverPassword } : {}),
Expand Down
4 changes: 3 additions & 1 deletion apps/server/src/terminal/NodePtyAdapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import * as Schema from "effect/Schema";
import { HostProcessArchitecture, HostProcessPlatform } from "@t3tools/shared/hostProcess";

import * as PtyAdapter from "./PtyAdapter.ts";
import { taskScopeCommand } from "../process/taskScope.ts";

export class NodePtyModuleLoadError extends Schema.TaggedError<NodePtyModuleLoadError>()(
"NodePtyModuleLoadError",
Expand Down Expand Up @@ -271,7 +272,8 @@ export const make = Effect.fn("NodePtyAdapter.make")(function* () {
: input.env;
const ptyProcess = yield* Effect.try({
try: () => {
const nativeProcess = nodePty.spawn(input.shell, input.args ?? [], {
const command = taskScopeCommand(input.shell, input.args ?? [], platform);
const nativeProcess = nodePty.spawn(command.command, [...command.args], {
cwd: input.cwd,
cols: input.cols,
rows: input.rows,
Expand Down
Loading