fix(pi-ssh): harden remote execution and search

pi-ssh/src/agent-connection.ts

@@ -1,3 +1,5 @@
+import type { PiSshConfig, SshHostConfig } from "./config.ts";
+
 export interface HostSelection {
   hostId: string;
   remotePath?: string;
@@ -6,7 +8,7 @@ export interface HostSelection {
 export const SSH_CONNECT_TOOL_METADATA = {
   name: "ssh_connect",
   label: "ssh_connect",
-  description: "Establish a persistent SSH2 connection to an explicitly imported host. Use this when the user names a remote server as part of a concrete task; the connection request is reviewed before any network connection is opened.",
+  description: "Establish a persistent SSH2 connection to an explicitly imported host as a sequential state transition. Use this as a separate step when the user names a remote server as part of a concrete task, and wait for success before calling dependent ssh_* tools; the connection request is reviewed before any network connection is opened.",
   parameters: {
     type: "object",
     properties: {
@@ -16,6 +18,7 @@ export const SSH_CONNECT_TOOL_METADATA = {
     required: ["hostId"],
     additionalProperties: false,
   },
+  executionMode: "sequential",
 } as const;
 
 export function parseConnectInput(input: Record<string, unknown>): HostSelection {
@@ -31,3 +34,16 @@ export function parseConnectInput(input: Record<string, unknown>): HostSelection
   }
   return { hostId, remotePath };
 }
+
+export function getConfiguredHost(config: PiSshConfig, hostId: string): SshHostConfig {
+  const host = config.hosts[hostId];
+  if (host) return host;
+  const hostIds = Object.keys(config.hosts).sort();
+  if (hostIds.length === 0) {
+    throw new Error(`unknown pi-ssh host '${hostId}'; no hosts are imported; run ssh_config.sh import <alias>`);
+  }
+  const shown = hostIds.slice(0, 10);
+  const remaining = hostIds.length - shown.length;
+  const available = `${shown.join(", ")}${remaining > 0 ? `, … (+${remaining} more)` : ""}`;
+  throw new Error(`unknown pi-ssh host '${hostId}'; available imported host IDs: ${available}`);
+}

pi-ssh/src/remote-bash.ts

@@ -0,0 +1,27 @@
+import type { BashOperations } from "@earendil-works/pi-coding-agent";
+import type { RemoteTransport } from "./ssh2-transport.ts";
+
+export interface RemoteBashConnection {
+  remoteCwd: string;
+}
+
+/**
+ * Adapt Pi's Bash output machinery to the active remote workspace.
+ *
+ * The cwd supplied by Pi's Bash factory is deliberately ignored: it belongs
+ * to the factory's filesystem namespace and must never leak into SSH command
+ * execution. The connection's mutable remoteCwd is the sole shell base.
+ */
+export function createRemoteBashOps(
+  connection: RemoteBashConnection,
+  transport: RemoteTransport,
+): BashOperations {
+  return {
+    exec: (command, _factoryCwd, { onData, signal, timeout }) => {
+      if (!connection.remoteCwd.startsWith("/")) {
+        throw new Error(`ssh_bash requires an absolute remote cwd, received '${connection.remoteCwd}'`);
+      }
+      return transport.exec(command, connection.remoteCwd, { onData, signal, timeout });
+    },
+  };
+}

pi-ssh/src/remote-cwd.ts

@@ -0,0 +1,59 @@
+import { posix as posixPath } from "node:path";
+
+export const SSH_CD_EXECUTION_MODE = "sequential" as const;
+
+export interface RemoteWorkspaceConnection {
+  remoteCwd: string;
+  remoteHome: string;
+}
+
+export interface RemotePathMappingConnection extends RemoteWorkspaceConnection {
+  localCwd: string;
+  localHome: string;
+}
+
+export interface RemoteCwdChange {
+  previousCwd: string;
+  remoteCwd: string;
+}
+
+export function resolveRemoteCwd(path: string, currentCwd: string, remoteHome: string): string {
+  if (typeof path !== "string" || path.length === 0) {
+    throw new Error("path must be a non-empty string");
+  }
+  if (/[\0\r\n]/u.test(path)) {
+    throw new Error("path must not contain NUL or newline characters");
+  }
+  if (!currentCwd.startsWith("/") || !remoteHome.startsWith("/")) {
+    throw new Error("the active SSH connection has an invalid remote workspace");
+  }
+  if (path === "~") return posixPath.normalize(remoteHome);
+  if (path.startsWith("~/")) return posixPath.normalize(posixPath.join(remoteHome, path.slice(2)));
+  if (path.startsWith("~")) throw new Error("path supports only '~' or '~/' home expansion");
+  if (path.startsWith("/")) return posixPath.normalize(path);
+  return posixPath.normalize(posixPath.join(currentCwd, path));
+}
+
+export async function changeRemoteCwd(
+  connection: RemoteWorkspaceConnection,
+  path: string,
+  verifyDirectory: (requestedCwd: string) => Promise<string>,
+): Promise<RemoteCwdChange> {
+  const previousCwd = connection.remoteCwd;
+  const requestedCwd = resolveRemoteCwd(path, previousCwd, connection.remoteHome);
+  const remoteCwd = await verifyDirectory(requestedCwd);
+  connection.remoteCwd = remoteCwd;
+  return { previousCwd, remoteCwd };
+}
+
+export function mapLocalPathToRemote(path: string, connection: RemotePathMappingConnection): string {
+  if (path === connection.localCwd) return connection.remoteCwd;
+  if (path.startsWith(`${connection.localCwd}/`)) {
+    return `${connection.remoteCwd}${path.slice(connection.localCwd.length)}`;
+  }
+  if (path === connection.localHome) return connection.remoteHome;
+  if (path.startsWith(`${connection.localHome}/`)) {
+    return `${connection.remoteHome}${path.slice(connection.localHome.length)}`;
+  }
+  return path;
+}

pi-ssh/src/remote-probe.ts

@@ -0,0 +1,70 @@
+import { randomBytes } from "node:crypto";
+import { posix as posixPath } from "node:path";
+import type { RemoteTransport } from "./ssh2-transport.ts";
+
+const PROBE_TIMEOUT_SECONDS = 20;
+const MAX_PROBE_OUTPUT_BYTES = 64 * 1024;
+
+export type RemotePathProbe = "home" | "cwd";
+
+function probeCommand(kind: RemotePathProbe, token: string): { command: string; start: string; end: string } {
+  const start = `__PI_SSH_PROBE_${token}_START__`;
+  const end = `__PI_SSH_PROBE_${token}_END__`;
+  const assign = kind === "home"
+    ? "pi_ssh_probe_value=$HOME"
+    : "pi_ssh_probe_value=$(pwd -P) || exit $?";
+  return {
+    command: `${assign}\nprintf '%s%s%s' '${start}' "$pi_ssh_probe_value" '${end}'`,
+    start,
+    end,
+  };
+}
+
+export function parseRemotePathProbe(output: Buffer, start: string, end: string, kind: RemotePathProbe): string {
+  if (output.length > MAX_PROBE_OUTPUT_BYTES) {
+    throw new Error(`remote ${kind} probe output exceeded ${MAX_PROBE_OUTPUT_BYTES} bytes`);
+  }
+  const text = output.toString("utf8");
+  const startIndex = text.indexOf(start);
+  const endIndex = startIndex < 0 ? -1 : text.indexOf(end, startIndex + start.length);
+  if (startIndex < 0 || endIndex < 0 || text.indexOf(start, startIndex + start.length) >= 0) {
+    throw new Error(`remote ${kind} probe returned an invalid framed response`);
+  }
+  const value = text.slice(startIndex + start.length, endIndex);
+  if (!value.startsWith("/") || /[\0\r\n]/u.test(value)) {
+    throw new Error(`remote ${kind} probe did not return one absolute POSIX path`);
+  }
+  return posixPath.normalize(value);
+}
+
+export async function probeRemotePath(
+  transport: RemoteTransport,
+  kind: RemotePathProbe,
+  cwd = ".",
+  signal?: AbortSignal,
+): Promise<string> {
+  const token = randomBytes(12).toString("hex");
+  const probe = probeCommand(kind, token);
+  const chunks: Buffer[] = [];
+  let captured = 0;
+  let overflow = false;
+  const result = await transport.exec(probe.command, cwd, {
+    signal,
+    timeout: PROBE_TIMEOUT_SECONDS,
+    onData(data) {
+      if (captured + data.length > MAX_PROBE_OUTPUT_BYTES) {
+        overflow = true;
+        return;
+      }
+      chunks.push(data);
+      captured += data.length;
+    },
+  });
+  const output = Buffer.concat(chunks);
+  if (overflow) throw new Error(`remote ${kind} probe output exceeded ${MAX_PROBE_OUTPUT_BYTES} bytes`);
+  if (result.exitCode !== 0) {
+    const detail = output.toString("utf8").replace(/\s+/gu, " ").trim().slice(0, 300);
+    throw new Error(`remote ${kind} probe failed${result.exitCode === null ? "" : ` with exit code ${result.exitCode}`}${detail ? `: ${detail}` : ""}`);
+  }
+  return parseRemotePathProbe(output, probe.start, probe.end, kind);
+}

pi-ssh/src/remote-search.ts

@@ -15,6 +15,7 @@ export interface RemoteGrepInput {
   literal?: boolean;
   caseSensitive?: boolean;
   include?: string;
+  includeHidden?: boolean;
   limit?: number;
 }
 
@@ -30,6 +31,8 @@ const DEFAULT_LIMIT = 50;
 const MAX_LIMIT = 200;
 const MAX_LINE_CHARS = 800;
 const MAX_CAPTURE_CHARS = 512_000;
+const SEARCH_TIMEOUT_SECONDS = 30;
+const MAX_DIAGNOSTIC_CHARS = 16_000;
 
 function shellQuote(value: string): string {
   return `'${value.replace(/'/g, `'"'"'`)}'`;
@@ -55,6 +58,7 @@ export function resolveRemoteSearchPath(path: string | undefined, remoteCwd: str
   validateField(value, "path");
   if (value === "~") return remoteHome;
   if (value.startsWith("~/")) return posixPath.normalize(posixPath.join(remoteHome, value.slice(2)));
+  if (value.startsWith("~")) throw new Error("path supports only ~ or ~/... remote HOME expansion");
   if (value.startsWith("/")) return posixPath.normalize(value);
   return posixPath.normalize(posixPath.join(remoteCwd, value));
 }
@@ -70,6 +74,27 @@ const STATUS_HELPER = [
   `}`,
 ].join("\n");
 
+const GREP_RECORD_HELPER = [
+  `pi_ssh_limit_colon_records() {`,
+  `  local path match line content count=0`,
+  `  while IFS= read -r -d '' path && IFS= read -r match; do`,
+  '    line="${match%%:*}"',
+  '    content="${match#*:}"',
+  '    printf \'%s\\0%s\\0%s\\0\' "${path:0:' + MAX_LINE_CHARS + '}" "$line" "${content:0:' + MAX_LINE_CHARS + '}"',
+  `    count=$((count + 1))`,
+  `    if [ "$count" -ge "$PI_SSH_TAKE" ]; then return 0; fi`,
+  `  done`,
+  `}`,
+  `pi_ssh_limit_git_records() {`,
+  `  local path line content count=0`,
+  `  while IFS= read -r -d '' path && IFS= read -r -d '' line && IFS= read -r content; do`,
+  '    printf \'%s\\0%s\\0%s\\0\' "${path:0:' + MAX_LINE_CHARS + '}" "$line" "${content:0:' + MAX_LINE_CHARS + '}"',
+  `    count=$((count + 1))`,
+  `    if [ "$count" -ge "$PI_SSH_TAKE" ]; then return 0; fi`,
+  `  done`,
+  `}`,
+] .join("\n");
+
 function findPipeline(input: RemoteFindInput, root: string, limit: number): string {
   const pattern = validateField(input.pattern, "pattern") as string;
   const fdCase = input.caseSensitive ? "--case-sensitive" : "--ignore-case";
@@ -81,24 +106,24 @@ function findPipeline(input: RemoteFindInput, root: string, limit: number): stri
     STATUS_HELPER,
     `if command -v fd >/dev/null 2>&1; then`,
     `  printf '${MARKER}fd\\n'`,
-    `  fd --type f --color never --fixed-strings ${fdCase} ${fdHidden} --exclude .git --exclude node_modules -- ${shellQuote(pattern)} ${shellQuote(root)} 2>&1 | head -n ${take} | cut -c 1-${MAX_LINE_CHARS}`,
+    `  fd --type f --color never --fixed-strings ${fdCase} ${fdHidden} --exclude .git --exclude node_modules -- ${shellQuote(pattern)} ${shellQuote(root)} | head -n ${take} | cut -c 1-${MAX_LINE_CHARS}`,
     '  statuses=("${PIPESTATUS[@]}")',
     '  pi_ssh_accept_status "${statuses[0]}" 0 141 || exit $?',
     `elif command -v fdfind >/dev/null 2>&1; then`,
     `  printf '${MARKER}fdfind\\n'`,
-    `  fdfind --type f --color never --fixed-strings ${fdCase} ${fdHidden} --exclude .git --exclude node_modules -- ${shellQuote(pattern)} ${shellQuote(root)} 2>&1 | head -n ${take} | cut -c 1-${MAX_LINE_CHARS}`,
+    `  fdfind --type f --color never --fixed-strings ${fdCase} ${fdHidden} --exclude .git --exclude node_modules -- ${shellQuote(pattern)} ${shellQuote(root)} | head -n ${take} | cut -c 1-${MAX_LINE_CHARS}`,
     '  statuses=("${PIPESTATUS[@]}")',
     '  pi_ssh_accept_status "${statuses[0]}" 0 141 || exit $?',
     `elif command -v git >/dev/null 2>&1 && git -C ${shellQuote(root)} rev-parse --is-inside-work-tree >/dev/null 2>&1; then`,
     `  printf '${MARKER}git-ls-files\\n'`,
-    `  git -C ${shellQuote(root)} ls-files -co --exclude-standard 2>&1 | ${hiddenFilter} | grep -F ${grepCase} -- ${shellQuote(pattern)} | head -n ${take} | cut -c 1-${MAX_LINE_CHARS}`,
+    `  git -C ${shellQuote(root)} ls-files -co --exclude-standard | ${hiddenFilter} | grep -F ${grepCase} -- ${shellQuote(pattern)} | head -n ${take} | cut -c 1-${MAX_LINE_CHARS}`,
     '  statuses=("${PIPESTATUS[@]}")',
     '  pi_ssh_accept_status "${statuses[0]}" 0 141 || exit $?',
     '  pi_ssh_accept_status "${statuses[1]}" 0 1 141 || exit $?',
     '  pi_ssh_accept_status "${statuses[2]}" 0 1 141 || exit $?',
     `else`,
     `  printf '${MARKER}find\\n'`,
-    `  find ${shellQuote(root)} -type f ! -path '*/.git/*' ! -path '*/node_modules/*' 2>&1 | ${hiddenFilter} | grep -F ${grepCase} -- ${shellQuote(pattern)} | head -n ${take} | cut -c 1-${MAX_LINE_CHARS}`,
+    `  find ${shellQuote(root)} -type f ! -path '*/.git/*' ! -path '*/node_modules/*' | ${hiddenFilter} | grep -F ${grepCase} -- ${shellQuote(pattern)} | head -n ${take} | cut -c 1-${MAX_LINE_CHARS}`,
     '  statuses=("${PIPESTATUS[@]}")',
     '  pi_ssh_accept_status "${statuses[0]}" 0 141 || exit $?',
     '  pi_ssh_accept_status "${statuses[1]}" 0 1 141 || exit $?',
@@ -110,29 +135,64 @@ function findPipeline(input: RemoteFindInput, root: string, limit: number): stri
 function grepPipeline(input: RemoteGrepInput, root: string, limit: number): string {
   const pattern = validateField(input.pattern, "pattern") as string;
   const include = validateField(input.include, "include", true);
-  const fixed = input.literal === false ? "" : "-F";
+  if (include?.includes("/")) throw new Error("include must be a basename glob without /");
+  const rgMode = input.literal === false ? "" : "-F";
+  const grepMode = input.literal === false ? "-E" : "-F";
   const insensitive = input.caseSensitive ? "" : "-i";
+  const rgHidden = input.includeHidden
+    ? "--hidden"
+    : "--glob '!.*' --glob '!**/.*' --glob '!**/.*/**'";
   const rgGlob = include ? `-g ${shellQuote(include)}` : "";
-  const gitPath = include ? `-- ${shellQuote(include)}` : "";
+  const gitPathspecs = [
+    ...(include ? [`:(glob)**/${include}`] : []),
+    ...(input.includeHidden ? [] : [":(exclude,glob)**/.*", ":(exclude,glob)**/.*/**"]),
+  ];
+  const gitPath = gitPathspecs.length > 0 ? `-- ${gitPathspecs.map(shellQuote).join(" ")}` : "";
   const findName = include ? `-name ${shellQuote(include)}` : "";
+  const includeHidden = input.includeHidden ? 1 : 0;
   const take = limit + 1;
   return [
     STATUS_HELPER,
+    GREP_RECORD_HELPER,
+    `PI_SSH_TAKE=${take}`,
+    `PI_SSH_ROOT=${shellQuote(root)}`,
+    `PI_SSH_INCLUDE_HIDDEN=${includeHidden}`,
     `if command -v rg >/dev/null 2>&1; then`,
     `  printf '${MARKER}ripgrep\\n'`,
-    `  rg --line-number --no-heading --color never --with-filename --max-columns 500 --max-columns-preview ${fixed} ${insensitive} ${rgGlob} --glob '!.git/**' --glob '!node_modules/**' -- ${shellQuote(pattern)} ${shellQuote(root)} 2>&1 | head -n ${take} | cut -c 1-${MAX_LINE_CHARS}`,
+    `  rg --null --line-number --no-heading --color never --with-filename --max-columns ${MAX_LINE_CHARS} --max-columns-preview ${rgMode} ${insensitive} ${rgHidden} ${rgGlob} --glob '!.git/**' --glob '!node_modules/**' -- ${shellQuote(pattern)} ${shellQuote(root)} | pi_ssh_limit_colon_records`,
     '  statuses=("${PIPESTATUS[@]}")',
     '  pi_ssh_accept_status "${statuses[0]}" 0 1 141 || exit $?',
+    '  pi_ssh_accept_status "${statuses[1]}" 0 || exit $?',
     `elif command -v git >/dev/null 2>&1 && git -C ${shellQuote(root)} rev-parse --is-inside-work-tree >/dev/null 2>&1; then`,
     `  printf '${MARKER}git-grep\\n'`,
-    `  git -C ${shellQuote(root)} grep --untracked --exclude-standard -n -I ${fixed} ${insensitive} -e ${shellQuote(pattern)} ${gitPath} 2>&1 | head -n ${take} | cut -c 1-${MAX_LINE_CHARS}`,
+    `  git -C ${shellQuote(root)} grep --untracked --exclude-standard -z -n -I ${grepMode} ${insensitive} -e ${shellQuote(pattern)} ${gitPath} | pi_ssh_limit_git_records`,
     '  statuses=("${PIPESTATUS[@]}")',
     '  pi_ssh_accept_status "${statuses[0]}" 0 1 141 || exit $?',
+    '  pi_ssh_accept_status "${statuses[1]}" 0 || exit $?',
     `else`,
     `  printf '${MARKER}grep\\n'`,
-    `  find ${shellQuote(root)} -type f ! -path '*/.git/*' ! -path '*/node_modules/*' ${findName} -exec grep -nH -I ${fixed} ${insensitive} -- ${shellQuote(pattern)} {} + 2>&1 | head -n ${take} | cut -c 1-${MAX_LINE_CHARS}`,
+    `  find ${shellQuote(root)} -type f ! -path '*/.git/*' ! -path '*/node_modules/*' ${findName} -print0 | (`,
+    `    while IFS= read -r -d '' file; do`,
+    `      if [ "$PI_SSH_INCLUDE_HIDDEN" -eq 0 ]; then`,
+    `        if [ -d "$PI_SSH_ROOT" ]; then`,
+    '          relative=${file#"$PI_SSH_ROOT"/}',
+    `        else`,
+    '          relative=${file##*/}',
+    `        fi`,
+    '        case "$relative" in .*|*/.*) continue ;; esac',
+    `      fi`,
+    `      grep -n -I ${grepMode} ${insensitive} -- ${shellQuote(pattern)} "$file" | while IFS= read -r match; do`,
+    `        printf '%s\\0%s\\n' "$file" "$match"`,
+    `      done`,
+    '      grep_statuses=("${PIPESTATUS[@]}")',
+    '      pi_ssh_accept_status "${grep_statuses[0]}" 0 1 141 || exit $?',
+    '      pi_ssh_accept_status "${grep_statuses[1]}" 0 1 141 || exit $?',
+    `    done`,
+    `  ) | pi_ssh_limit_colon_records`,
     '  statuses=("${PIPESTATUS[@]}")',
     '  pi_ssh_accept_status "${statuses[0]}" 0 141 || exit $?',
+    '  pi_ssh_accept_status "${statuses[1]}" 0 141 || exit $?',
+    '  pi_ssh_accept_status "${statuses[2]}" 0 || exit $?',
     `fi`,
   ].join("\n");
 }
@@ -149,30 +209,54 @@ export function buildRemoteGrepCommand(input: RemoteGrepInput, root: string): {
   return { command: grepPipeline(input, root, limit), limit };
 }
 
-function prefixGitPath(line: string, root: string, grep: boolean): string {
-  if (line.startsWith("/") || line.startsWith("../")) return line;
-  if (!grep) return posixPath.join(root, line);
-  const separator = line.indexOf(":");
-  if (separator < 1) return line;
-  return `${posixPath.join(root, line.slice(0, separator))}${line.slice(separator)}`;
+function prefixGitPath(path: string, root: string): string {
+  return path.startsWith("/") || path.startsWith("../") ? path : posixPath.join(root, path);
+}
+
+function visibleField(value: string): string {
+  return value.replace(/\r/gu, "\\r").replace(/\n/gu, "\\n");
+}
+
+function normalizeDiagnostic(raw: Buffer): string {
+  return raw
+    .toString("utf8")
+    .replace(/\0/gu, " ")
+    .replace(new RegExp(`${MARKER}[^\\n]*`, "gu"), "")
+    .replace(/\s+/gu, " ")
+    .trim()
+    .slice(0, 300);
 }
 
 export function formatRemoteSearchOutput(
-  raw: string,
+  raw: string | Buffer,
   root: string,
   limit: number,
   kind: "find" | "grep",
 ): RemoteSearchResult {
-  const lines = raw.replace(/\r\n?/gu, "\n").split("\n");
-  const markerIndex = lines.findIndex((line) => line.startsWith(MARKER));
-  if (markerIndex < 0) throw new Error(`remote ${kind} did not report a search backend`);
-  const backend = lines[markerIndex].slice(MARKER.length).trim() || "unknown";
-  const sourceRows = lines.slice(markerIndex + 1).filter((line) => line.length > 0);
+  const output = Buffer.isBuffer(raw) ? raw.toString("utf8") : raw;
+  const markerIndex = output.indexOf(MARKER);
+  const markerEnd = markerIndex < 0 ? -1 : output.indexOf("\n", markerIndex);
+  if (markerIndex < 0 || markerEnd < 0) throw new Error(`remote ${kind} did not report a search backend`);
+  const backend = output.slice(markerIndex + MARKER.length, markerEnd).trim() || "unknown";
+  const payload = output.slice(markerEnd + 1);
+  let sourceRows: string[];
+  if (kind === "grep") {
+    const fields = payload.length === 0 ? [] : payload.split("\0");
+    if (fields.at(-1) === "") fields.pop();
+    if (fields.length % 3 !== 0) throw new Error("remote grep returned a malformed or truncated NUL-delimited result");
+    sourceRows = [];
+    for (let index = 0; index < fields.length; index += 3) {
+      const path = backend.startsWith("git-") ? prefixGitPath(fields[index], root) : fields[index];
+      sourceRows.push(`${visibleField(path)}:${visibleField(fields[index + 1])}:${visibleField(fields[index + 2])}`);
+    }
+  } else {
+    sourceRows = payload.replace(/\r\n?/gu, "\n").split("\n").filter((line) => line.length > 0);
+    if (backend.startsWith("git-")) sourceRows = sourceRows.map((line) => prefixGitPath(line, root));
+  }
   const truncated = sourceRows.length > limit;
-  const rows = sourceRows.slice(0, limit).map((line) => {
-    const normalized = backend.startsWith("git-") ? prefixGitPath(line, root, kind === "grep") : line;
-    return normalized.length > MAX_LINE_CHARS ? `${normalized.slice(0, MAX_LINE_CHARS - 1)}…` : normalized;
-  });
+  const rows = sourceRows.slice(0, limit).map((line) =>
+    line.length > MAX_LINE_CHARS ? `${line.slice(0, MAX_LINE_CHARS - 1)}…` : line,
+  );
   const header = `Remote ${kind}: ${rows.length} result${rows.length === 1 ? "" : "s"} (backend: ${backend}, root: ${root}, truncated: ${truncated ? "yes" : "no"})`;
   return {
     text: rows.length > 0 ? `${header}\n\n${rows.join("\n")}` : `${header}\n\nNo matches found.`,
@@ -185,37 +269,50 @@ export function formatRemoteSearchOutput(
 async function runRemoteSearch(
   transport: RemoteTransport,
   command: string,
+  executionCwd: string,
   root: string,
   limit: number,
   kind: "find" | "grep",
   signal?: AbortSignal,
 ): Promise<RemoteSearchResult> {
   const chunks: Buffer[] = [];
+  const stderrChunks: Buffer[] = [];
   let captured = 0;
-  const result = await transport.exec(command, root, {
-    signal,
-    timeout: 30,
-    onData(data) {
-      if (captured >= MAX_CAPTURE_CHARS) return;
-      const remaining = MAX_CAPTURE_CHARS - captured;
-      const chunk = data.length > remaining ? data.subarray(0, remaining) : data;
-      chunks.push(chunk);
-      captured += chunk.length;
-    },
-  });
-  const output = Buffer.concat(chunks).toString("utf8");
+  let stderrCaptured = 0;
+  const capture = (target: Buffer[], data: Buffer, stderr = false) => {
+    const current = stderr ? stderrCaptured : captured;
+    const maximum = stderr ? MAX_DIAGNOSTIC_CHARS : MAX_CAPTURE_CHARS;
+    if (current >= maximum) return;
+    const remaining = maximum - current;
+    const chunk = data.length > remaining ? data.subarray(0, remaining) : data;
+    target.push(chunk);
+    if (stderr) stderrCaptured += chunk.length;
+    else captured += chunk.length;
+  };
+  let result: { exitCode: number | null };
+  try {
+    result = await transport.exec(command, executionCwd, {
+      signal,
+      timeout: SEARCH_TIMEOUT_SECONDS,
+      onData(data) { capture(chunks, data); },
+      onStderr(data) { capture(stderrChunks, data, true); },
+    });
+  } catch (error) {
+    if (!signal?.aborted && /timed out/iu.test(error instanceof Error ? error.message : String(error))) {
+      throw new Error(`remote ${kind} timed out after ${SEARCH_TIMEOUT_SECONDS}s (root: ${root}); narrow the remote path with ssh_find before retrying`);
+    }
+    throw error;
+  }
+  const output = Buffer.concat(chunks);
+  const stderr = Buffer.concat(stderrChunks);
   if (result.exitCode !== 0) {
-    const detail = output
-      .replace(/\r\n?/gu, "\n")
-      .split("\n")
-      .filter((line) => !line.startsWith(MARKER))
-      .join(" ")
-      .replace(/\s+/gu, " ")
-      .trim()
-      .slice(0, 300);
+    const detail = normalizeDiagnostic(stderr.length > 0 ? stderr : output);
     throw new Error(`remote ${kind} failed${result.exitCode === null ? "" : ` with exit code ${result.exitCode}`}${detail ? `: ${detail}` : ""}`);
   }
-  return formatRemoteSearchOutput(output, root, limit, kind);
+  const formatted = formatRemoteSearchOutput(output, root, limit, kind);
+  const warning = normalizeDiagnostic(stderr);
+  if (warning) formatted.text += `\n\nRemote warning: ${warning}`;
+  return formatted;
 }
 
 export function runRemoteFind(
@@ -227,7 +324,7 @@ export function runRemoteFind(
 ): Promise<RemoteSearchResult> {
   const root = resolveRemoteSearchPath(input.path, remoteCwd, remoteHome);
   const built = buildRemoteFindCommand(input, root);
-  return runRemoteSearch(transport, built.command, root, built.limit, "find", signal);
+  return runRemoteSearch(transport, built.command, remoteCwd, root, built.limit, "find", signal);
 }
 
 export function runRemoteGrep(
@@ -239,5 +336,5 @@ export function runRemoteGrep(
 ): Promise<RemoteSearchResult> {
   const root = resolveRemoteSearchPath(input.path, remoteCwd, remoteHome);
   const built = buildRemoteGrepCommand(input, root);
-  return runRemoteSearch(transport, built.command, root, built.limit, "grep", signal);
+  return runRemoteSearch(transport, built.command, remoteCwd, root, built.limit, "grep", signal);
 }

pi-ssh/src/ssh2-transport.ts

@@ -7,12 +7,13 @@ import { expandUserPath, type SshHostConfig } from "./config.ts";
 
 export interface RemoteExecOptions {
   onData: (data: Buffer) => void;
+  onStderr?: (data: Buffer) => void;
   signal?: AbortSignal;
   timeout?: number;
 }
 
 export interface RemoteTransport {
-  connect(): Promise<void>;
+  connect(signal?: AbortSignal): Promise<void>;
   dispose(): Promise<void>;
   exec(command: string, cwd: string, options: RemoteExecOptions): Promise<{ exitCode: number | null }>;
   capture(command: string, cwd?: string, timeout?: number): Promise<{ exitCode: number | null; output: Buffer }>;
@@ -36,6 +37,66 @@ class CommandQueue {
   }
 }
 
+interface SemaphoreWaiter {
+  resolve: (release: () => void) => void;
+  reject: (error: Error) => void;
+  signal?: AbortSignal;
+  onAbort?: () => void;
+}
+
+export class AsyncSemaphore {
+  private active = 0;
+  private readonly waiters: SemaphoreWaiter[] = [];
+  private readonly limit: number;
+
+  constructor(limit: number) {
+    if (!Number.isInteger(limit) || limit < 1) throw new Error("semaphore limit must be a positive integer");
+    this.limit = limit;
+  }
+
+  acquire(signal?: AbortSignal): Promise<() => void> {
+    if (signal?.aborted) return Promise.reject(new Error("SSH command aborted"));
+    if (this.active < this.limit) {
+      this.active += 1;
+      return Promise.resolve(this.createRelease());
+    }
+    return new Promise((resolve, reject) => {
+      const waiter: SemaphoreWaiter = { resolve, reject, signal };
+      waiter.onAbort = () => {
+        const index = this.waiters.indexOf(waiter);
+        if (index >= 0) this.waiters.splice(index, 1);
+        reject(new Error("SSH command aborted"));
+      };
+      signal?.addEventListener("abort", waiter.onAbort, { once: true });
+      this.waiters.push(waiter);
+    });
+  }
+
+  private createRelease(): () => void {
+    let released = false;
+    return () => {
+      if (released) return;
+      released = true;
+      this.active -= 1;
+      this.dispatch();
+    };
+  }
+
+  private dispatch(): void {
+    while (this.active < this.limit) {
+      const waiter = this.waiters.shift();
+      if (!waiter) return;
+      if (waiter.onAbort) waiter.signal?.removeEventListener("abort", waiter.onAbort);
+      if (waiter.signal?.aborted) {
+        waiter.reject(new Error("SSH command aborted"));
+        continue;
+      }
+      this.active += 1;
+      waiter.resolve(this.createRelease());
+    }
+  }
+}
+
 function shellQuote(value: string): string {
   return `'${value.replace(/'/g, `'"'"'`)}'`;
 }
@@ -80,7 +141,8 @@ function errorMessage(error: unknown): string {
 
 export class Ssh2Transport implements RemoteTransport {
   private readonly client: Client;
-  private readonly queue = new CommandQueue();
+  private readonly sftpQueue = new CommandQueue();
+  private readonly execSemaphore = new AsyncSemaphore(4);
   private connected = false;
   private disposed = false;
   private disconnectError: Error | null = null;
@@ -92,14 +154,17 @@ export class Ssh2Transport implements RemoteTransport {
     this.client = client;
   }
 
-  async connect(): Promise<void> {
+  async connect(signal?: AbortSignal): Promise<void> {
     if (this.connected) return;
     if (this.disposed) throw new Error("SSH2 transport is disposed");
+    if (signal?.aborted) throw new Error("SSH connection aborted");
     await new Promise<void>((resolve, reject) => {
       let settled = false;
+      const cleanup = () => signal?.removeEventListener("abort", onAbort);
       const succeed = () => {
         if (settled) return;
         settled = true;
+        cleanup();
         this.connected = true;
         resolve();
       };
@@ -108,8 +173,17 @@ export class Ssh2Transport implements RemoteTransport {
         this.disconnectError = normalized;
         if (settled) return;
         settled = true;
+        cleanup();
         reject(normalized);
       };
+      const onAbort = () => {
+        try {
+          this.client.destroy();
+        } catch {
+          // client may already be closed
+        }
+        fail(new Error("SSH connection aborted"));
+      };
       this.client.once("ready", succeed);
       this.client.on("error", fail);
       this.client.on("close", () => {
@@ -122,6 +196,7 @@ export class Ssh2Transport implements RemoteTransport {
           finish(prompts.map(() => this.host.auth.type === "password" ? this.host.auth.password : ""));
         });
       }
+      signal?.addEventListener("abort", onAbort, { once: true });
       try {
         this.client.connect(buildConnectConfig(this.host));
       } catch (error) {
@@ -141,8 +216,13 @@ export class Ssh2Transport implements RemoteTransport {
     if (!this.connected) throw this.disconnectError ?? new Error("SSH2 connection is not active");
   }
 
-  exec(command: string, cwd: string, options: RemoteExecOptions): Promise<{ exitCode: number | null }> {
-    return this.queue.enqueue(() => this.execUnqueued(command, cwd, options));
+  async exec(command: string, cwd: string, options: RemoteExecOptions): Promise<{ exitCode: number | null }> {
+    const release = await this.execSemaphore.acquire(options.signal);
+    try {
+      return await this.execUnqueued(command, cwd, options);
+    } finally {
+      release();
+    }
   }
 
   private async execUnqueued(command: string, cwd: string, options: RemoteExecOptions): Promise<{ exitCode: number | null }> {
@@ -197,7 +277,10 @@ export class Ssh2Transport implements RemoteTransport {
         }
         channel = stream;
         stream.on("data", (data: Buffer | string) => options.onData(Buffer.isBuffer(data) ? data : Buffer.from(data)));
-        stream.stderr.on("data", (data: Buffer | string) => options.onData(Buffer.isBuffer(data) ? data : Buffer.from(data)));
+        stream.stderr.on("data", (data: Buffer | string) => {
+          const chunk = Buffer.isBuffer(data) ? data : Buffer.from(data);
+          (options.onStderr ?? options.onData)(chunk);
+        });
         stream.once("error", fail);
         stream.once("close", (code: number | undefined) => {
           if (settled) return;
@@ -225,7 +308,7 @@ export class Ssh2Transport implements RemoteTransport {
   }
 
   async readFile(remotePath: string): Promise<Buffer> {
-    return this.queue.enqueue(async () => {
+    return this.sftpQueue.enqueue(async () => {
       const sftp = await this.sftp();
       return new Promise<Buffer>((resolve, reject) => {
         sftp.readFile(remotePath, (error, data) => error ? reject(error) : resolve(data));
@@ -234,7 +317,7 @@ export class Ssh2Transport implements RemoteTransport {
   }
 
   private async ensureOpen(remotePath: string, flags: "r" | "r+"): Promise<void> {
-    return this.queue.enqueue(async () => {
+    return this.sftpQueue.enqueue(async () => {
       const sftp = await this.sftp();
       await new Promise<void>((resolve, reject) => {
         sftp.open(remotePath, flags, (error, handle) => {
@@ -267,7 +350,7 @@ export class Ssh2Transport implements RemoteTransport {
   }
 
   async mkdir(remoteDir: string): Promise<void> {
-    return this.queue.enqueue(() => this.mkdirUnqueued(remoteDir));
+    return this.sftpQueue.enqueue(() => this.mkdirUnqueued(remoteDir));
   }
 
   private async mkdirUnqueued(remoteDir: string): Promise<void> {
@@ -290,7 +373,7 @@ export class Ssh2Transport implements RemoteTransport {
   }
 
   async writeFile(remotePath: string, content: Buffer): Promise<void> {
-    return this.queue.enqueue(async () => {
+    return this.sftpQueue.enqueue(async () => {
       const sftp = await this.sftp();
       await this.mkdirUnqueued(posixPath.dirname(remotePath));
       const temporary = `${remotePath}.pi-ssh-${randomBytes(8).toString("hex")}.tmp`;