let5see/clawdbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: remove bash pty mode

Browse Source
This commit is contained in:
Peter Steinberger
2026-01-03 20:15:02 +00:00
parent a15cffb7de
commit 16e3535ac0
16 changed files with 94 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/agents/bash-process-registry.test.ts
View File

@@ -20,7 +20,6 @@ describe("bash process registry", () => {
id: "sess",
command: "echo test",
child: { pid: 123 } as ChildProcessWithoutNullStreams,
stdinMode: "pipe",
startedAt: Date.now(),
cwd: "/tmp",
maxOutputChars: 10,
@@ -49,7 +48,6 @@ describe("bash process registry", () => {
id: "sess",
command: "echo test",
child: { pid: 123 } as ChildProcessWithoutNullStreams,
stdinMode: "pipe",
startedAt: Date.now(),
cwd: "/tmp",
maxOutputChars: 100,

5
src/agents/bash-process-registry.ts
View File

@@ -1,5 +1,4 @@
import type { ChildProcessWithoutNullStreams } from "node:child_process";
import type { IPty } from "node-pty";
const DEFAULT_JOB_TTL_MS = 30 * 60 * 1000; // 30 minutes
const MIN_JOB_TTL_MS = 60 * 1000; // 1 minute
@@ -16,15 +15,11 @@ let jobTtlMs = clampTtl(
export type ProcessStatus = "running" | "completed" | "failed" | "killed";
export type ProcessStdinMode = "pipe" | "pty";
export interface ProcessSession {
id: string;
command: string;
child?: ChildProcessWithoutNullStreams;
pty?: IPty;
pid?: number;
stdinMode: ProcessStdinMode;
startedAt: number;
cwd?: string;
maxOutputChars: number;

68
src/agents/bash-tools.pty.test.ts
View File

@@ -1,68 +0,0 @@
import { describe, expect, it, vi } from "vitest";
describe("bash tool pty mode", () => {
it("falls back to pipe with warning when node-pty fails to load", async () => {
vi.resetModules();
vi.doMock("node-pty", () => {
throw new Error("boom");
});
const { createBashTool } = await import("./bash-tools.js");
const tool = createBashTool({ backgroundMs: 10, timeoutSec: 1 });
const result = await tool.execute("call", {
command: "echo test",
stdinMode: "pty",
});
const text = result.content.find((c) => c.type === "text")?.text ?? "";
expect(text).toContain("Warning: node-pty failed to load");
expect(text).toContain("falling back to pipe mode.");
vi.doUnmock("node-pty");
});
it("uses node-pty when available", async () => {
vi.resetModules();
const spawn = vi.fn(() => {
let onData: ((data: string) => void) | undefined;
let onExit:
| ((event: { exitCode: number | null; signal?: number | null }) => void)
| undefined;
const pty = {
pid: 4321,
onData: (cb: (data: string) => void) => {
onData = cb;
},
onExit: (
cb: (event: {
exitCode: number | null;
signal?: number | null;
}) => void,
) => {
onExit = cb;
},
write: vi.fn(),
kill: vi.fn(),
};
setTimeout(() => {
onData?.("hello\n");
onExit?.({ exitCode: 0, signal: null });
}, 10);
return pty;
});
vi.doMock("node-pty", () => ({ spawn }));
const { createBashTool } = await import("./bash-tools.js");
const tool = createBashTool({ backgroundMs: 10, timeoutSec: 1 });
const result = await tool.execute("call", {
command: "ignored",
stdinMode: "pty",
});
const text = result.content.find((c) => c.type === "text")?.text ?? "";
expect(text).toContain("hello");
expect(text).not.toContain("Warning:");
vi.doUnmock("node-pty");
});
});

281
src/agents/bash-tools.ts
View File

@@ -2,10 +2,8 @@ import { type ChildProcessWithoutNullStreams, spawn } from "node:child_process";
import { randomUUID } from "node:crypto";
import { existsSync, statSync } from "node:fs";
import { homedir } from "node:os";
import path from "node:path";
import type { AgentTool, AgentToolResult } from "@mariozechner/pi-agent-core";
import { Type } from "@sinclair/typebox";
import type { IPty } from "node-pty";
import {
addSession,
@@ -18,7 +16,6 @@ import {
listRunningSessions,
markBackgrounded,
markExited,
type ProcessStdinMode,
setJobTtlMs,
} from "./bash-process-registry.js";
import {
@@ -34,23 +31,6 @@ const DEFAULT_MAX_OUTPUT = clampNumber(
1_000,
150_000,
);
const DEFAULT_SHELL_PATH = "/bin/sh";
const DEFAULT_PATH =
"/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin";
const DEFAULT_PTY_NAME = "xterm-256color";
type PtyModule = typeof import("node-pty");
type PtyLoadResult = { module: PtyModule | null; error?: unknown };
let ptyModulePromise: Promise<PtyLoadResult> | null = null;
async function loadPtyModule(): Promise<PtyLoadResult> {
if (!ptyModulePromise) {
ptyModulePromise = import("node-pty")
.then((mod) => ({ module: mod }))
.catch((error) => ({ module: null, error }));
}
return ptyModulePromise;
}
const stringEnum = (
values: readonly string[],
@@ -81,7 +61,7 @@ const bashSchema = Type.Object({
env: Type.Optional(Type.Record(Type.String(), Type.String())),
yieldMs: Type.Optional(
Type.Number({
description: "Milliseconds to wait before backgrounding (default 20000)",
description: "Milliseconds to wait before backgrounding (default 10000)",
}),
),
background: Type.Optional(
@@ -92,11 +72,6 @@ const bashSchema = Type.Object({
description: "Timeout in seconds (optional, kills process on expiry)",
}),
),
stdinMode: Type.Optional(
stringEnum(["pipe", "pty"] as const, {
description: "stdin mode (pipe or pty when node-pty is available)",
}),
),
});
export type BashToolDetails =
@@ -120,7 +95,7 @@ export function createBashTool(
): AgentTool<any, BashToolDetails> {
const defaultBackgroundMs = clampNumber(
defaults?.backgroundMs ?? readEnvInt("PI_BASH_YIELD_MS"),
20_000,
10_000,
10,
120_000,
);
@@ -133,7 +108,7 @@ export function createBashTool(
name: "bash",
label: "bash",
description:
"Execute bash with background continuation. Use yieldMs/background to continue later via process tool.",
"Execute bash with background continuation. Use yieldMs/background to continue later via process tool. For real TTY mode, use the tmux skill.",
parameters: bashSchema,
execute: async (_toolCallId, args, signal, onUpdate) => {
const params = args as {
@@ -143,7 +118,6 @@ export function createBashTool(
yieldMs?: number;
background?: boolean;
timeout?: number;
stdinMode?: "pipe" | "pty";
};
if (!params.command) {
@@ -169,84 +143,18 @@ export function createBashTool(
const { shell, args: shellArgs } = getShellConfig();
const env = params.env ? { ...process.env, ...params.env } : process.env;
const requestedStdinMode = params.stdinMode === "pty" ? "pty" : "pipe";
let stdinMode: ProcessStdinMode = requestedStdinMode;
let warning: string | null = null;
let child: ChildProcessWithoutNullStreams | undefined;
let pty: IPty | undefined;
if (stdinMode === "pty") {
const { module: ptyModule, error: ptyError } = await loadPtyModule();
if (!ptyModule) {
warning =
`Warning: node-pty failed to load${formatPtyError(ptyError)}; ` +
"falling back to pipe mode.";
stdinMode = "pipe";
} else {
const ptyEnv = ensurePath({
...env,
TERM: env.TERM ?? DEFAULT_PTY_NAME,
} as Record<string, string>);
const ptyShell = resolveShellPath(shell, ptyEnv);
try {
pty = ptyModule.spawn(ptyShell, [...shellArgs, params.command], {
cwd: workdir,
env: ptyEnv,
name: ptyEnv.TERM || DEFAULT_PTY_NAME,
cols: 120,
rows: 30,
});
} catch (error) {
if (
ptyShell !== DEFAULT_SHELL_PATH &&
existsSync(DEFAULT_SHELL_PATH)
) {
try {
pty = ptyModule.spawn(
DEFAULT_SHELL_PATH,
[...shellArgs, params.command],
{
cwd: workdir,
env: ptyEnv,
name: ptyEnv.TERM || DEFAULT_PTY_NAME,
cols: 120,
rows: 30,
},
);
} catch (fallbackError) {
warning =
`Warning: node-pty failed to start${formatPtyError(fallbackError)}; ` +
"falling back to pipe mode.";
stdinMode = "pipe";
}
} else {
warning =
`Warning: node-pty failed to start${formatPtyError(error)}; ` +
"falling back to pipe mode.";
stdinMode = "pipe";
}
}
}
}
if (stdinMode === "pipe") {
child = spawn(shell, [...shellArgs, params.command], {
cwd: workdir,
env,
detached: true,
stdio: ["pipe", "pipe", "pipe"],
});
}
if (warning) warnings.push(warning);
const child = spawn(shell, [...shellArgs, params.command], {
cwd: workdir,
env,
detached: true,
stdio: ["pipe", "pipe", "pipe"],
});
const session = {
id: sessionId,
command: params.command,
child,
pty,
pid: child?.pid ?? pty?.pid,
stdinMode,
pid: child?.pid,
startedAt,
cwd: workdir,
maxOutputChars: maxOutput,
@@ -309,33 +217,21 @@ export function createBashTool(
});
};
if (child) {
child.stdout.on("data", (data) => {
const str = sanitizeBinaryOutput(data.toString());
for (const chunk of chunkString(str)) {
appendOutput(session, "stdout", chunk);
emitUpdate();
}
});
child.stdout.on("data", (data) => {
const str = sanitizeBinaryOutput(data.toString());
for (const chunk of chunkString(str)) {
appendOutput(session, "stdout", chunk);
emitUpdate();
}
});
child.stderr.on("data", (data) => {
const str = sanitizeBinaryOutput(data.toString());
for (const chunk of chunkString(str)) {
appendOutput(session, "stderr", chunk);
emitUpdate();
}
});
}
if (pty) {
pty.onData((data) => {
const str = sanitizeBinaryOutput(data);
for (const chunk of chunkString(str)) {
appendOutput(session, "stdout", chunk);
emitUpdate();
}
});
}
child.stderr.on("data", (data) => {
const str = sanitizeBinaryOutput(data.toString());
for (const chunk of chunkString(str)) {
appendOutput(session, "stderr", chunk);
emitUpdate();
}
});
return new Promise<AgentToolResult<BashToolDetails>>(
(resolve, reject) => {
@@ -434,24 +330,16 @@ export function createBashTool(
);
};
if (child) {
child.once("exit", (code, exitSignal) => {
handleExit(code, exitSignal);
});
child.once("exit", (code, exitSignal) => {
handleExit(code, exitSignal);
});
child.once("error", (err) => {
if (yieldTimer) clearTimeout(yieldTimer);
if (timeoutTimer) clearTimeout(timeoutTimer);
markExited(session, null, null, "failed");
settle(() => reject(err));
});
}
if (pty) {
pty.onExit(({ exitCode, signal }) => {
handleExit(exitCode ?? null, signal ?? null);
});
}
child.once("error", (err) => {
if (yieldTimer) clearTimeout(yieldTimer);
if (timeoutTimer) clearTimeout(timeoutTimer);
markExited(session, null, null, "failed");
settle(() => reject(err));
});
},
);
},
@@ -747,43 +635,25 @@ export function createProcessTool(
details: { status: "failed" },
};
}
if (session.stdinMode === "pty") {
if (!session.pty || session.exited) {
return {
content: [
{
type: "text",
text: `Session ${params.sessionId} stdin is not writable.`,
},
],
details: { status: "failed" },
};
}
session.pty.write(params.data ?? "");
if (params.eof) {
session.pty.write("\x04");
}
} else {
if (!session.child?.stdin || session.child.stdin.destroyed) {
return {
content: [
{
type: "text",
text: `Session ${params.sessionId} stdin is not writable.`,
},
],
details: { status: "failed" },
};
}
await new Promise<void>((resolve, reject) => {
session.child?.stdin.write(params.data ?? "", (err) => {
if (err) reject(err);
else resolve();
});
if (!session.child?.stdin || session.child.stdin.destroyed) {
return {
content: [
{
type: "text",
text: `Session ${params.sessionId} stdin is not writable.`,
},
],
details: { status: "failed" },
};
}
await new Promise<void>((resolve, reject) => {
session.child?.stdin.write(params.data ?? "", (err) => {
if (err) reject(err);
else resolve();
});
if (params.eof) {
session.child.stdin.end();
}
});
if (params.eof) {
session.child.stdin.end();
}
return {
content: [
@@ -908,21 +778,12 @@ export const processTool = createProcessTool();
function killSession(session: {
pid?: number;
stdinMode: ProcessStdinMode;
pty?: IPty;
child?: ChildProcessWithoutNullStreams;
}) {
const pid = session.pid ?? session.child?.pid ?? session.pty?.pid;
const pid = session.pid ?? session.child?.pid;
if (pid) {
killProcessTree(pid);
}
if (session.stdinMode === "pty") {
try {
session.pty?.kill();
} catch {
// ignore kill failures
}
}
}
function resolveWorkdir(workdir: string, warnings: string[]) {
@@ -949,44 +810,6 @@ function safeCwd() {
}
}
function ensurePath(env: Record<string, string>) {
if (!env.PATH?.trim()) {
env.PATH = DEFAULT_PATH;
}
return env;
}
function resolveShellPath(shell: string, env: Record<string, string>) {
if (process.platform === "win32") return shell;
if (shell.includes("/") && existsSync(shell)) {
return shell;
}
const searchPath = env.PATH ?? "";
for (const segment of searchPath.split(path.delimiter)) {
if (!segment) continue;
const candidate = path.join(segment, shell);
if (existsSync(candidate)) return candidate;
}
if (existsSync(DEFAULT_SHELL_PATH)) {
return DEFAULT_SHELL_PATH;
}
return shell;
}
function formatPtyError(error: unknown) {
if (!error) return "";
if (typeof error === "string") return ` (${error})`;
if (error instanceof Error) {
const firstLine = error.message.split(/\r?\n/)[0]?.trim();
return firstLine ? ` (${firstLine})` : "";
}
try {
return ` (${JSON.stringify(error)})`;
} catch {
return "";
}
}
function clampNumber(
value: number | undefined,
defaultValue: number,

2
src/agents/pi-embedded-runner.ts
View File

@@ -494,7 +494,7 @@ export async function runEmbeddedPiAgent(params: {
};
const queueHandle: EmbeddedPiQueueHandle = {
queueMessage: async (text: string) => {
await session.queueMessage(text);
await session.steer(text);
},
isStreaming: () => session.isStreaming,
abort: abortRun,

3
src/discord/monitor.ts
View File

@@ -367,8 +367,7 @@ export async function monitorDiscordProvider(opts: MonitorDiscordOpts = {}) {
message.content?.trim() ??
media?.placeholder ??
message.embeds[0]?.description ??
(forwardedSnapshot ? "<forwarded message>" : "") ??
"";
(forwardedSnapshot ? "<forwarded message>" : "");
if (!text) {
logVerbose(`discord: drop message ${message.id} (empty content)`);
return;

4
src/gateway/server.providers.test.ts
View File

@@ -1,5 +1,4 @@
import { describe, expect, test } from "vitest";
import { readConfigFileSnapshot, writeConfigFile } from "../config/config.js";
import {
connectOk,
installGatewayTestHooks,
@@ -64,6 +63,9 @@ describe("gateway server providers", () => {
test("telegram.logout clears bot token from config", async () => {
const prevToken = process.env.TELEGRAM_BOT_TOKEN;
delete process.env.TELEGRAM_BOT_TOKEN;
const { readConfigFileSnapshot, writeConfigFile } = await import(
"../config/config.js"
);
await writeConfigFile({
telegram: {
botToken: "123:abc",

42
src/gateway/server.ts
View File

@@ -86,6 +86,11 @@ import {
authorizeGatewayConnect,
type ResolvedGatewayAuth,
} from "./auth.js";
import {
type GatewayReloadPlan,
type ProviderKind,
startGatewayConfigReloader,
} from "./config-reload.js";
import { normalizeControlUiBasePath } from "./control-ui.js";
import { resolveHooksConfig } from "./hooks.js";
import {
@@ -94,13 +99,12 @@ import {
resolveGatewayBindHost,
} from "./net.js";
import { createBridgeHandlers } from "./server-bridge.js";
import { createBridgeSubscriptionManager } from "./server-bridge-subscriptions.js";
import { startBrowserControlServerIfEnabled } from "./server-browser.js";
import {
startGatewayConfigReloader,
type GatewayReloadPlan,
type ProviderKind,
} from "./config-reload.js";
type BridgeListConnectedFn,
type BridgeSendEventFn,
createBridgeSubscriptionManager,
} from "./server-bridge-subscriptions.js";
import { startBrowserControlServerIfEnabled } from "./server-browser.js";
import { createAgentEventHandler, createChatRunState } from "./server-chat.js";
import {
DEDUPE_MAX,
@@ -862,6 +866,11 @@ export async function startGatewayServer(
const bridgeSubscribe = bridgeSubscriptions.subscribe;
const bridgeUnsubscribe = bridgeSubscriptions.unsubscribe;
const bridgeUnsubscribeAll = bridgeSubscriptions.unsubscribeAll;
const bridgeSendEvent: BridgeSendEventFn = (opts) => {
bridge?.sendEvent(opts);
};
const bridgeListConnected: BridgeListConnectedFn = () =>
bridge?.listConnected() ?? [];
const bridgeSendToSession = (
sessionKey: string,
event: string,
@@ -871,20 +880,16 @@ export async function startGatewayServer(
sessionKey,
event,
payload,
bridge ? (opts) => bridge.sendEvent(opts) : undefined,
bridgeSendEvent,
);
const bridgeSendToAllSubscribed = (event: string, payload: unknown) =>
bridgeSubscriptions.sendToAllSubscribed(
event,
payload,
bridge ? (opts) => bridge.sendEvent(opts) : undefined,
);
bridgeSubscriptions.sendToAllSubscribed(event, payload, bridgeSendEvent);
const bridgeSendToAllConnected = (event: string, payload: unknown) =>
bridgeSubscriptions.sendToAllConnected(
event,
payload,
bridge ? () => bridge.listConnected() : undefined,
bridge ? (opts) => bridge.sendEvent(opts) : undefined,
bridgeListConnected,
bridgeSendEvent,
);
const broadcastVoiceWakeChanged = (triggers: string[]) => {
@@ -1663,7 +1668,9 @@ export async function startGatewayServer(
if (plan.restartProviders.size > 0) {
if (process.env.CLAWDIS_SKIP_PROVIDERS === "1") {
logProviders.info("skipping provider reload (CLAWDIS_SKIP_PROVIDERS=1)");
logProviders.info(
"skipping provider reload (CLAWDIS_SKIP_PROVIDERS=1)",
);
} else {
const restartProvider = async (
name: ProviderKind,
@@ -1712,10 +1719,7 @@ export async function startGatewayServer(
}
}
setCommandLaneConcurrency(
"cron",
nextConfig.cron?.maxConcurrentRuns ?? 1,
);
setCommandLaneConcurrency("cron", nextConfig.cron?.maxConcurrentRuns ?? 1);
setCommandLaneConcurrency("main", nextConfig.agent?.maxConcurrent ?? 1);
if (plan.hotReasons.length > 0) {

7
src/gateway/test-helpers.ts
View File

@@ -4,7 +4,6 @@ import os from "node:os";
import path from "node:path";
import { afterEach, beforeEach, expect, vi } from "vitest";
import { WebSocket } from "ws";
import { agentCommand } from "../commands/agent.js";
import { resetAgentRunContextForTest } from "../infra/agent-events.js";
import { drainSystemEvents, peekSystemEvents } from "../infra/system-events.js";
import { rawDataToString } from "../infra/ws.js";
@@ -64,6 +63,7 @@ const hoisted = vi.hoisted(() => ({
}>,
},
cronIsolatedRun: vi.fn(async () => ({ status: "ok", summary: "ok" })),
agentCommand: vi.fn().mockResolvedValue(undefined),
testIsNixMode: { value: false },
sessionStoreSaveDelayMs: { value: 0 },
}));
@@ -75,6 +75,7 @@ export const bridgeSendEvent = hoisted.bridgeSendEvent;
export const testTailnetIPv4 = hoisted.testTailnetIPv4;
export const piSdkMock = hoisted.piSdkMock;
export const cronIsolatedRun = hoisted.cronIsolatedRun;
export const agentCommand = hoisted.agentCommand;
export const testState = {
sessionStorePath: undefined as string | undefined,
@@ -290,7 +291,7 @@ vi.mock("../web/outbound.js", () => ({
.mockResolvedValue({ messageId: "msg-1", toJid: "jid-1" }),
}));
vi.mock("../commands/agent.js", () => ({
agentCommand: vi.fn().mockResolvedValue(undefined),
agentCommand,
}));
process.env.CLAWDIS_SKIP_PROVIDERS = "1";
@@ -509,5 +510,3 @@ export async function waitForSystemEvent(timeoutMs = 2000) {
}
throw new Error("timeout waiting for system event");
}
export { agentCommand };