fix: stream runner tran ssh output

2026-05-25 11:41:56 +00:00
parent 230c13efe4
commit 4a03052747
12 changed files with 381 additions and 152 deletions
@@ -162,6 +162,7 @@ export function sshHelp(): unknown {
      "bun scripts/cli.ts ssh D601:k3s script <<'SCRIPT'",
      "bun scripts/cli.ts ssh D601:k3s:hwlab-dev:hwlab-cloud-api/app pwd",
      "bun scripts/cli.ts ssh D601:k3s:hwlab-dev:hwlab-cloud-api apply-patch <<'PATCH'",
+      "tar -C /path/to/files -cf - . | bun scripts/cli.ts ssh D601:k3s:unidesk:code-queue/root/unidesk exec --stdin -- tar -xf - -C /root/unidesk",
      "bun scripts/cli.ts ssh D601:k3s:hwlab-dev:hwlab-cloud-api node -e 'console.log(process.version)'",
      "bun scripts/cli.ts ssh D601:k3s:hwlab-dev:hwlab-cloud-api script <<'SCRIPT'",
      "bun scripts/cli.ts ssh D601:k3s:hwlab-dev:hwlab-cloud-api logs --tail 80",
@@ -169,6 +170,7 @@ export function sshHelp(): unknown {
    notes: [
      "ssh --help and ssh <route> --help print this JSON help and never open an interactive session.",
      "For non-interactive remote commands, prefer argv for a single process and script/stdin for shell logic.",
+      "For arbitrary stdin streams into a workload command, use a workload route plus `exec --stdin -- <command> ...`; this keeps the route as location-only and avoids heredoc/base64/tar shell wrapping.",
      "apply-patch rejects low-context update hunks by default, reports the matched file:line for each hunk on stderr, and only accepts --allow-loose when the caller has manually reviewed an intentionally ambiguous insertion.",
      "script defaults to target /bin/sh and inherits provider proxy variables such as HTTP_PROXY/HTTPS_PROXY/ALL_PROXY/NO_PROXY; use --shell bash only for bash syntax such as pipefail, arrays, or [[ ... ]], not as a proxy workaround.",
      "Route syntax is `{provider}:{plane}[:{scope...}] {operation} [operation-args...]`: the first argv token locates a distributed target only, and every following token belongs to the operation parser. Host workspace routes use `<provider>:/absolute/workspace`; native k3s providers such as D601 and G14 use <provider>:k3s for the control plane, <provider>:k3s:<namespace>:<workload> for a workload, and <provider>:k3s:<namespace>:<workload>/<pod-workspace> for a pod workspace.",
@@ -6,7 +6,7 @@ import { type UniDeskConfig } from "./config";
 import { type DebugDispatchCommand, isDebugDispatchCommand } from "./debug";
 import { summarizeMicroserviceHealthResponse, summarizeMicroserviceObservation, summarizeMicroserviceProxyResponse } from "./microservices";
 import { parseNetworkPerfOptions, runNetworkPerf } from "./network-perf";
-import { formatSshFailureHint, isSshSkillDiscoveryArgs, parseSshInvocation, sshFailureHint } from "./ssh";
+import { formatSshFailureHint, parseSshInvocation, sshFailureHint, wrapSshRemoteCommand } from "./ssh";
 import { codexJudgeQueryAsync, codexOutputQueryAsync, codexPrPreflightQueryAsync, codexQueuesQueryAsync, codexTaskQueryAsync, codexTasksQueryAsync, codexUnreadTriageAsync } from "./code-queue";
 import { runDecisionCenterCommandAsync } from "./decision-center";
 import {
@@ -536,95 +536,6 @@ function jsonOption(args: string[], name: string): Record<string, unknown> | und
  return parsed as Record<string, unknown>;
 }

-const compactSkillDiscoverPython = String.raw`import os,json,socket,platform,getpass
-from pathlib import Path as P
-S=os.getenv('S','all');L=int(os.getenv('L','0'));D=int(os.getenv('D','4'));skip={'node_modules','.git','.state','logs','references','__pycache__'}
-def isw():
- try:r=P('/proc/sys/kernel/osrelease').read_text(errors='ignore').lower()
- except Exception:r=''
- return 'microsoft' in r or 'wsl' in r or 'WSL_INTEROP' in os.environ
-def wp(p):
- s=str(p)
- return s[5].upper()+':\\'+s[7:].replace('/','\\') if s.startswith('/mnt/') and len(s)>6 and s[5].isalpha() and s[6]=='/' else None
-def md(f):
- n=f.parent.name;d=''
- try:ls=f.read_text(errors='replace')[:8192].splitlines()
- except Exception:ls=[]
- if ls and ls[0].strip()=='---':
-  for l in ls[1:]:
-   if l.strip()=='---':break
-   if ':' in l:
-    k,v=l.split(':',1);k=k.strip().lower();v=v.strip().strip('"\' ')
-    if k=='name' and v:n=v
-    if k=='description' and v:d=v
- if not d:
-  for l in ls:
-   x=l.strip()
-   if x and not x.startswith('---') and not x.startswith('#'):
-    d=x;break
- return n,d
-def scan(sc,root):
- rec={'scope':sc,'path':str(root),'windowsPath':wp(root),'exists':False,'skillCount':0,'error':None};out=[]
- try:rec['exists']=root.exists()
- except Exception as e:rec['error']=str(e)
- if not rec['exists']:return rec,out
- try:
-  for f in root.rglob('SKILL.md'):
-   rel=f.relative_to(root).parts[:-1]
-   if not rel or len(rel)>D or any(x in skip for x in rel):continue
-   n,d=md(f);out.append({'scope':sc,'name':n,'description':d,'path':str(f.parent),'skillMd':str(f),'windowsPath':wp(f.parent),'root':str(root)})
- except Exception as e:rec['error']=str(e)
- rec['skillCount']=len(out);return rec,out
-roots=[];h=P.home()
-if S!='windows':roots += [('wsl',h/'.agents/skills'),('wsl',h/'.codex/skills'),('wsl',P('/root/.agents/skills')),('wsl',P('/root/.codex/skills'))]
-if S!='wsl' and isw():
- try:users=list(P('/mnt/c/Users').iterdir())
- except Exception:users=[]
- for u in users:
-  if u.is_dir() and u.name.lower() not in {'all users','default','default user','public'}:roots += [('windows',u/'.agents/skills'),('windows',u/'.codex/skills')]
-seen=set();rr=[];ss=[]
-for sc,r in roots:
- k=(sc,str(r))
- if k in seen:continue
- seen.add(k);a,b=scan(sc,r);rr.append(a);ss+=b
-ss.sort(key=lambda x:(0 if x['scope']=='wsl' else 1,x['name'].lower(),x['path']));tb=len(ss)
-if L>0:ss=ss[:L]
-c={'total':len(ss),'totalBeforeLimit':tb,'wsl':sum(1 for x in ss if x['scope']=='wsl'),'windows':sum(1 for x in ss if x['scope']=='windows')}
-print(json.dumps({'ok':True,'command':'unidesk ssh skills','node':{'hostname':socket.gethostname(),'user':getpass.getuser(),'home':str(P.home()),'platform':platform.platform(),'isWsl':isw()},'counts':c,'roots':rr,'skills':ss},ensure_ascii=False,indent=2))`;
-
-function remoteFrontendSkillDiscoverCommand(args: string[]): string {
-  let scope = "all";
-  let limit = 0;
-  let maxDepth = 4;
-  const start = args[0] === "skill" ? 2 : 1;
-  for (let index = start; index < args.length; index += 1) {
-    const arg = args[index] ?? "";
-    const next = args[index + 1];
-    if (arg === "--scope") {
-      if (next !== "all" && next !== "wsl" && next !== "windows") throw new Error("ssh skills --scope must be one of: all, wsl, windows");
-      scope = next;
-      index += 1;
-      continue;
-    }
-    if (arg === "--limit") {
-      const value = Number(next);
-      if (!Number.isInteger(value) || value < 0) throw new Error("ssh skills --limit must be >= 0");
-      limit = value;
-      index += 1;
-      continue;
-    }
-    if (arg === "--max-depth") {
-      const value = Number(next);
-      if (!Number.isInteger(value) || value <= 0) throw new Error("ssh skills --max-depth must be positive");
-      maxDepth = value;
-      index += 1;
-      continue;
-    }
-    throw new Error(`remote frontend ssh skills does not support option: ${arg}`);
-  }
-  return `S=${shellQuote(scope)} L=${shellQuote(String(limit))} D=${shellQuote(String(maxDepth))} python3 -c ${shellQuote(compactSkillDiscoverPython)}`;
-}
-
 function dispatchPayload(args: string[], command: DebugDispatchCommand): Record<string, unknown> {
  const explicit = jsonOption(args, "--payload-json") ?? {};
  if (command === "provider.upgrade") {
@@ -951,12 +862,178 @@ async function remoteNetworkPerf(options: RemoteCliOptions, config: UniDeskConfi
  };
 }

+function frontendSshWebSocketUrl(session: FrontendSession): string {
+  const url = new URL("/ws/ssh", session.baseUrl);
+  url.protocol = url.protocol === "https:" ? "wss:" : "ws:";
+  return url.toString();
+}
+
+function webSocketDataText(data: unknown): string {
+  if (typeof data === "string") return data;
+  if (data instanceof ArrayBuffer) return Buffer.from(data).toString("utf8");
+  if (ArrayBuffer.isView(data)) return Buffer.from(data.buffer, data.byteOffset, data.byteLength).toString("utf8");
+  return String(data);
+}
+
+function openFrontendSshWebSocket(session: FrontendSession): WebSocket {
+  const WebSocketWithHeaders = WebSocket as unknown as new (
+    url: string,
+    options?: { headers?: Record<string, string> },
+  ) => WebSocket;
+  return new WebSocketWithHeaders(frontendSshWebSocketUrl(session), { headers: { cookie: session.cookie } });
+}
+
+async function runRemoteSshWebSocket(
+  session: FrontendSession,
+  invocation: ReturnType<typeof parseSshInvocation>,
+): Promise<number> {
+  const parsed = invocation.parsed;
+  const size = {
+    cols: Number(process.stdout.columns) > 0 ? Number(process.stdout.columns) : 100,
+    rows: Number(process.stdout.rows) > 0 ? Number(process.stdout.rows) : 30,
+  };
+  const openTimeoutMs = Math.max(15000, Number(process.env.UNIDESK_SSH_OPEN_TIMEOUT_MS || 60000));
+  const payload = {
+    providerId: invocation.providerId,
+    command: wrapSshRemoteCommand(parsed.remoteCommand),
+    cwd: invocation.route.plane === "host" ? invocation.route.workspace ?? undefined : undefined,
+    tty: parsed.remoteCommand === null,
+    stdinEotOnEnd: parsed.remoteCommand !== null,
+    openTimeoutMs,
+    cols: size.cols,
+    rows: size.rows,
+  };
+  const ws = openFrontendSshWebSocket(session);
+  let exitCode = 255;
+  let settled = false;
+  let canSend = false;
+  let sessionReady = false;
+  const pending: string[] = [];
+  const pendingSessionMessages: string[] = [];
+
+  const send = (value: unknown): void => {
+    const text = JSON.stringify(value);
+    if (!canSend || ws.readyState !== WebSocket.OPEN) {
+      pending.push(text);
+      return;
+    }
+    ws.send(text);
+  };
+  const flush = (): void => {
+    while (pending.length > 0 && ws.readyState === WebSocket.OPEN) ws.send(pending.shift()!);
+  };
+  const sendInput = (value: Buffer | string): void => {
+    sendWhenSessionReady({ type: "ssh.input", data: Buffer.from(value).toString("base64"), encoding: "base64" });
+  };
+  const sendWhenSessionReady = (value: unknown): void => {
+    const text = JSON.stringify(value);
+    if (!sessionReady || ws.readyState !== WebSocket.OPEN) {
+      pendingSessionMessages.push(text);
+      return;
+    }
+    ws.send(text);
+  };
+  const flushSessionMessages = (): void => {
+    if (!sessionReady || ws.readyState !== WebSocket.OPEN) return;
+    while (pendingSessionMessages.length > 0) ws.send(pendingSessionMessages.shift()!);
+  };
+
+  return await new Promise<number>((resolve) => {
+    const rawMode = parsed.remoteCommand === null && process.stdin.isTTY && typeof process.stdin.setRawMode === "function";
+    const openTimer = setTimeout(() => {
+      if (sessionReady || settled) return;
+      process.stderr.write("unidesk remote frontend ssh bridge timed out waiting for provider session\n");
+      exitCode = 255;
+      try {
+        ws.close();
+      } catch {
+        // Ignore close failures while resolving the timeout path.
+      }
+    }, openTimeoutMs);
+
+    const restore = (): void => {
+      clearTimeout(openTimer);
+      process.stdin.off("data", onStdinData);
+      process.stdin.off("end", onStdinEnd);
+      if (rawMode) process.stdin.setRawMode(false);
+    };
+    const finish = (code: number): void => {
+      if (settled) return;
+      settled = true;
+      restore();
+      const hint = sshFailureHint(invocation.providerId, parsed, code, "");
+      if (hint !== null) process.stderr.write(formatSshFailureHint(hint));
+      resolve(code);
+    };
+    const onStdinData = (chunk: Buffer): void => {
+      sendInput(chunk);
+    };
+    const onStdinEnd = (): void => {
+      if (parsed.stdinSuffix) sendInput(parsed.stdinSuffix);
+      if (payload.stdinEotOnEnd === true) sendInput(Buffer.from([4]));
+      sendWhenSessionReady({ type: "ssh.eof" });
+    };
+
+    ws.addEventListener("open", () => {
+      canSend = true;
+      send({ type: "ssh.open", ...payload });
+      flush();
+    });
+    ws.addEventListener("message", (event) => {
+      const text = webSocketDataText(event.data);
+      let message: Record<string, unknown>;
+      try {
+        message = JSON.parse(text) as Record<string, unknown>;
+      } catch {
+        process.stderr.write(`${text}\n`);
+        return;
+      }
+      if (message.type === "ssh.dispatched") return;
+      if (message.type === "ssh.opened") {
+        sessionReady = true;
+        clearTimeout(openTimer);
+        flushSessionMessages();
+        return;
+      }
+      if (message.type === "ssh.data") {
+        const chunk = Buffer.from(String(message.data ?? ""), message.encoding === "base64" ? "base64" : "utf8");
+        if (message.stream === "stderr") process.stderr.write(chunk);
+        else process.stdout.write(chunk);
+        return;
+      }
+      if (message.type === "ssh.error") {
+        process.stderr.write(`${String(message.message || "ssh bridge error")}\n`);
+        exitCode = 255;
+        ws.close();
+        return;
+      }
+      if (message.type === "ssh.exit") {
+        exitCode = Number.isInteger(message.exitCode) ? Number(message.exitCode) : 255;
+        ws.close();
+      }
+    });
+    ws.addEventListener("close", () => finish(exitCode));
+    ws.addEventListener("error", () => {
+      process.stderr.write("unidesk remote frontend ssh bridge websocket error\n");
+      finish(255);
+    });
+
+    if (rawMode) process.stdin.setRawMode(true);
+    process.stdin.resume();
+    if (parsed.stdinPrefix) sendInput(parsed.stdinPrefix);
+    process.stdin.on("data", onStdinData);
+    process.stdin.on("end", onStdinEnd);
+  });
+}
+
 export function remoteSshFrontendPlanForTest(target: string, args: string[]): Record<string, unknown> {
  const invocation = parseSshInvocation(target, args);
  return {
+    transport: "frontend-websocket",
    providerId: invocation.providerId,
    route: invocation.route,
    remoteCommand: invocation.parsed.remoteCommand,
+    wrappedRemoteCommand: wrapSshRemoteCommand(invocation.parsed.remoteCommand),
    requiresStdin: invocation.parsed.requiresStdin,
    invocationKind: invocation.parsed.invocationKind,
    payloadCwd: invocation.route.plane === "host" ? invocation.route.workspace : null,
@@ -966,54 +1043,7 @@ export function remoteSshFrontendPlanForTest(target: string, args: string[]): Re
 async function runRemoteSshOverFrontend(session: FrontendSession, target: string | undefined, args: string[]): Promise<number> {
  if (!target) throw new Error("remote ssh requires a route, for example: bun scripts/cli.ts --main-server-ip 74.48.78.17 ssh D601 hostname");
  const invocation = parseSshInvocation(target, args);
-  const parsed = invocation.parsed;
-  if (parsed.requiresStdin) {
-    process.stderr.write("remote frontend transport does not stream stdin for ssh helper subcommands such as script, apply-patch or py; run the command on the main server, use --main-server-transport ssh, or use an argv/pod-route operation that does not need stdin\n");
-    return 255;
-  }
-  if (parsed.remoteCommand === null) {
-    process.stderr.write("remote frontend transport supports ssh remote commands only; pass a command such as: ssh D601 hostname\n");
-    return 255;
-  }
-  if (args[0] === "glob") {
-    process.stderr.write("remote frontend transport does not support the ssh glob helper because host.ssh exec has a short command-length limit; run it on the main server CLI instead\n");
-    return 255;
-  }
-  const remoteCommand = isSshSkillDiscoveryArgs(args) ? remoteFrontendSkillDiscoverCommand(args) : parsed.remoteCommand;
-  const dispatch = await frontendJson(session, "/api/dispatch", {
-    method: "POST",
-    body: JSON.stringify({
-      providerId: invocation.providerId,
-      command: "host.ssh",
-      payload: {
-        source: "cli-remote-ssh",
-        mode: "exec",
-        command: remoteCommand,
-        ...(invocation.route.plane === "host" && invocation.route.workspace !== null ? { cwd: invocation.route.workspace } : {}),
-        timeoutMs: isSshSkillDiscoveryArgs(args) ? 30000 : 15000,
-      },
-    }),
-  });
-  const taskId = (dispatch as { body?: { taskId?: string } }).body?.taskId ?? "";
-  if (!dispatch.ok || taskId.length === 0) {
-    process.stderr.write(`${JSON.stringify(dispatch, null, 2)}\n`);
-    return 255;
-  }
-  const wait = await waitForFrontendTask(session, taskId, 20_000);
-  const task = (wait as { task?: { status?: string; result?: Record<string, unknown>; message?: string } }).task;
-  const result = task?.result ?? {};
-  const stdout = typeof result.stdout === "string" ? result.stdout : "";
-  const stderr = typeof result.stderr === "string" ? result.stderr : "";
-  if (stdout.length > 0) process.stdout.write(stdout);
-  if (stderr.length > 0) process.stderr.write(stderr);
-  if (task?.status !== "succeeded") {
-    if (stdout.length === 0 && stderr.length === 0) process.stderr.write(`${JSON.stringify({ taskId, task }, null, 2)}\n`);
-    const exitCode = typeof result.exitCode === "number" ? result.exitCode : 255;
-    const hint = sshFailureHint(invocation.providerId, parsed, exitCode, stderr.length > 0 ? stderr : String(task?.message ?? ""));
-    if (hint !== null) process.stderr.write(formatSshFailureHint(hint));
-    return exitCode;
-  }
-  return typeof result.exitCode === "number" ? result.exitCode : 0;
+  return runRemoteSshWebSocket(session, invocation);
 }

 async function runRemoteCliOverFrontend(options: RemoteCliOptions, config: UniDeskConfig): Promise<number> {
@@ -39,7 +39,7 @@ export interface SshFailureHint {
  note: string;
 }

-const argvQuotedSshSubcommands = new Set(["rg", "grep", "sed", "nl", "stat", "du", "ls", "cat", "head", "tail", "wc", "pwd"]);
+const argvQuotedSshSubcommands = new Set(["git", "rg", "grep", "sed", "nl", "stat", "du", "ls", "cat", "head", "tail", "wc", "pwd"]);
 const nativeK3sKubeconfig = "/etc/rancher/k3s/k3s.yaml";
 const k3sResourceKindAliases = new Set(["pod", "po", "pods", "deployment", "deploy", "deployments", "statefulset", "sts", "daemonset", "ds", "job", "jobs"]);
 const legacyK3sOperationRouteSegments = new Set([
@@ -1018,7 +1018,12 @@ function parseK3sTargetOperation(route: ParsedSshRoute, args: string[]): ParsedS
  }
  if (operation === "kubectl") throw new Error(`ssh k3s kubectl is a control-plane operation; use ssh ${route.providerId}:k3s kubectl ...`);
  if (operation === "exec") {
-    return { remoteCommand: buildK3sExecCommand([...targetArgs, ...k3sRouteCommandArgs(operationArgs)]), requiresStdin: false, invocationKind: "helper" };
+    const execArgs = k3sRouteExecOperationArgs(operationArgs);
+    return {
+      remoteCommand: buildK3sExecCommand([...targetArgs, ...execArgs]),
+      requiresStdin: execArgs.includes("--stdin") || execArgs.includes("-i"),
+      invocationKind: "helper",
+    };
  }
  return { remoteCommand: buildK3sExecCommand([...targetArgs, ...k3sRouteCommandArgs(args)]), requiresStdin: false, invocationKind: "helper" };
 }
@@ -1045,6 +1050,31 @@ function k3sRouteCommandArgs(args: string[]): string[] {
  return args[0] === "--" ? args : ["--", ...args];
 }

+function k3sRouteExecOperationArgs(args: string[]): string[] {
+  if (args.length === 0) throw new Error("ssh k3s target exec operation requires a command to exec");
+  const result: string[] = [];
+  for (let index = 0; index < args.length; index += 1) {
+    const arg = args[index] ?? "";
+    if (arg === "--") {
+      if (index === args.length - 1) throw new Error("ssh k3s target exec operation requires a command after --");
+      return [...result, ...args.slice(index)];
+    }
+    if (arg === "--stdin" || arg === "-i" || arg === "--tty" || arg === "-t") {
+      result.push(arg);
+      continue;
+    }
+    if (arg === "--container" || arg === "-c" || arg === "--workdir" || arg === "--cwd") {
+      const value = args[index + 1];
+      if (value === undefined || value.length === 0) throw new Error(`ssh k3s target exec ${arg} requires a value`);
+      result.push(arg, value);
+      index += 1;
+      continue;
+    }
+    return [...result, "--", ...args.slice(index)];
+  }
+  throw new Error("ssh k3s target exec operation requires a command to exec");
+}
+
 function buildK3sCommand(providerId: string, args: string[]): string {
  const action = args[0] ?? "";
  if (action.length === 0 || action === "--help" || action === "-h" || action === "help") {