AgentHub/server/lib/remote-sync.js

/**
 * @file remote-sync.js
 * @description Pull Claude Code session history from remote machines over SSH so
 * one dashboard can monitor usage collected elsewhere (e.g. a dev box or cloud
 * VM the user drives over SSH while running AgentHub on their laptop).
 *
 * Design (see repo issue "Live remote/multi-machine data collection"):
 *   1. Authentication ALWAYS defers to the host's own SSH stack — ~/.ssh/config,
 *      ssh-agent, keys, known_hosts. This module stores and handles NO secrets;
 *      a source's `host` is just an ssh destination (user@host or a config alias)
 *      and `identity_file` is at most a path to a key the user already controls.
 *   2. `scp -r` mirrors the remote `<remote_home>/projects` tree into a sandboxed
 *      per-source staging dir. Only OpenSSH is required on the remote (the SFTP
 *      subsystem built into sshd) — no rsync or other packages.
 *   3. The SAME importer the dashboard uses for local history
 *      (scripts/import-history.js `importFromDirectory`) parses that staging dir,
 *      so remote sessions/agents/tokens/costs line up bit-for-bit with local
 *      ones. Imported rows are then tagged with the source id.
 *   4. Codex, per source and opt-in (`sync_codex`): `<remote_codex_home>/sessions`
 *      (default ~/.codex) and its `session_index.jsonl` are mirrored with their
 *      mtimes kept (`scp -p`, or tar from WSL) into `<staging>/codex/`, and read by
 *      the SAME journal reader local Codex history uses (server/lib/codex-usage.js,
 *      `home` = that mirror), which writes the rows with the source id.
 *
 * Security posture: every external command runs via child_process with an
 * ARGUMENT ARRAY and no shell (`shell` is never set), so user-controlled values
 * are never interpolated into a shell line. On top of that, host / path / port /
 * identity-file inputs are validated against strict allowlists BEFORE they reach
 * any command (see validateSourceInput). StrictHostKeyChecking is left at the
 * SSH default — an unknown host key fails the sync rather than being trusted
 * blindly, so the user must have connected once manually (host in known_hosts).
 *
 * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
 */

const fs = require("fs");
const path = require("path");
const os = require("os");
const { spawn, execFileSync } = require("child_process");

const { getDataDir } = require("./claude-home");
const { importFromDirectory, collectJsonlFiles } = require("../../scripts/import-history");

// Per-source sync timeout. A first sync of a large history can take a while;
// keep it generous but bounded so a hung SSH never wedges the poller.
const SYNC_TIMEOUT_MS = parseInt(process.env.DASHBOARD_REMOTE_SYNC_TIMEOUT_MS || "600000", 10);
// Connection-test timeout — short; this is a liveness probe, not a transfer.
const TEST_TIMEOUT_MS = parseInt(process.env.DASHBOARD_REMOTE_TEST_TIMEOUT_MS || "15000", 10);
// A remote session is treated as still-active while its mirrored transcript was
// modified within this window. scp preserves remote mtimes — the same
// "recently touched ⇒ probably running" signal as local import
// (scripts/import-history.js RECENT_THRESHOLD_MS). Once the mirror stops
// advancing (the remote session ended), the session flips to completed on the
// next sync. Configurable for slow links / long-idle turns.
const REMOTE_ACTIVE_WINDOW_MS = parseInt(
  process.env.DASHBOARD_REMOTE_ACTIVE_WINDOW_MS || "600000",
  10
);

// In-flight source ids, so a manual "Sync now" and the background poller can't
// run two syncs against the same source (and the same staging dir) at once.
const inFlight = new Set();

// ── Validation ──────────────────────────────────────────────────────────────

// SSH destination: [user@]host or a ~/.ssh/config alias. Allowlist only chars
// that legitimately appear in those; critically, forbid a leading '-' so the
// value can never be mistaken for an ssh option (OpenSSH has no `--` end-of-
// options terminator), and forbid whitespace / shell metacharacters / ':'
// (which would break scp's host:path parsing), whitespace, and metachars.
const HOST_RE = /^[A-Za-z0-9][A-Za-z0-9._@-]*$/;
// Remote path (the remote CLAUDE_HOME). POSIX absolute, ~-rooted, Windows
// `C:/...`, WSL `wsl:~/.claude` / `wsl:/home/user/.claude`, or a UNC path
// such as `//wsl.localhost/Ubuntu/home/user/.claude` (forward slashes only).
const REMOTE_PATH_RE =
  /^(wsl:(~[A-Za-z0-9._/~-]*|\/[A-Za-z0-9._/~-]*)|~[A-Za-z0-9._/~-]*|\/[A-Za-z0-9._/~-]*|[A-Za-z]:\/[A-Za-z0-9._/~-]*|\/\/[A-Za-z0-9._-]+(?:\/[A-Za-z0-9._/~-]*)*)$/;
// Whitespace + control chars (space, tab, newline, DEL, all C0 controls).
// eslint-disable-next-line no-control-regex -- intentional control-char reject
const CONTROL_OR_SPACE_RE = /[- ]/;

class ValidationError extends Error {
  constructor(code, message) {
    super(message);
    this.code = code;
  }
}

/**
 * Validate + normalize raw source config from the API into the shape the DB and
 * command builders expect. Throws ValidationError (with a stable `code`) on any
 * malformed field so the route can return a structured 400.
 *
 * @param {object} input raw body
 * @param {boolean} [partial] when true (PATCH), only validate provided fields
 * @returns {{label?:string,host?:string,sshPort?:number|null,identityFile?:string|null,remoteHome?:string|null,enabled?:number}}
 */
function validateSourceInput(input, partial = false) {
  const out = {};
  const has = (k) => input[k] !== undefined && input[k] !== null;

  if (!partial || has("label")) {
    const label = typeof input.label === "string" ? input.label.trim() : "";
    if (!label) throw new ValidationError("INVALID_LABEL", "`label` is required");
    if (label.length > 100)
      throw new ValidationError("INVALID_LABEL", "`label` must be 100 characters or fewer");
    out.label = label;
  }

  if (!partial || has("host")) {
    const host = typeof input.host === "string" ? input.host.trim() : "";
    if (!host) throw new ValidationError("INVALID_HOST", "`host` is required");
    if (host.length > 255 || !HOST_RE.test(host)) {
      throw new ValidationError(
        "INVALID_HOST",
        "`host` must be an ssh destination (user@host or a ~/.ssh/config alias); no spaces or shell characters, cannot start with '-'"
      );
    }
    out.host = host;
  }

  if (has("ssh_port")) {
    const port = Number(input.ssh_port);
    if (!Number.isInteger(port) || port < 1 || port > 65535) {
      throw new ValidationError("INVALID_PORT", "`ssh_port` must be an integer 1-65535");
    }
    out.sshPort = port;
  } else if (!partial) {
    out.sshPort = null;
  }

  if (has("identity_file")) {
    const raw = String(input.identity_file).trim();
    if (!raw) {
      out.identityFile = null;
    } else {
      // Expand ~ and normalize slashes for the local OS (Windows accepts C:/ or C:\).
      let expanded = raw.startsWith("~") ? path.join(os.homedir(), raw.slice(1)) : raw;
      if (process.platform === "win32") expanded = expanded.replace(/\//g, "\\");
      if (CONTROL_OR_SPACE_RE.test(expanded)) {
        throw new ValidationError(
          "INVALID_IDENTITY_FILE",
          "`identity_file` must be a path with no spaces or control characters"
        );
      }
      if (!path.isAbsolute(expanded)) {
        throw new ValidationError(
          "INVALID_IDENTITY_FILE",
          "`identity_file` must be an absolute path"
        );
      }
      out.identityFile = expanded;
    }
  } else if (!partial) {
    out.identityFile = null;
  }

  if (has("remote_home")) {
    const rh = String(input.remote_home).trim();
    if (rh.includes("..") || !REMOTE_PATH_RE.test(rh)) {
      throw new ValidationError(
        "INVALID_REMOTE_HOME",
        "`remote_home` must be a POSIX path (`~/.claude`, `/opt/cc`), a Windows path (`C:/Users/you/.claude`), a WSL path (`wsl:~/.claude`), or a UNC path (`//wsl.localhost/Ubuntu/home/you/.claude`); no spaces or `..`"
      );
    }
    out.remoteHome = rh;
  } else if (!partial) {
    out.remoteHome = null;
  }

  if (has("remote_codex_home")) {
    const ch = String(input.remote_codex_home).trim();
    if (!ch) {
      out.remoteCodexHome = null;
    } else if (ch.includes("..") || !REMOTE_PATH_RE.test(ch)) {
      throw new ValidationError(
        "INVALID_REMOTE_CODEX_HOME",
        "`remote_codex_home` must be a path like `remote_home` (`~/.codex`, `/opt/codex`, `C:/Users/you/.codex`, `wsl:~/.codex`); no spaces or `..`"
      );
    } else {
      out.remoteCodexHome = ch;
    }
  } else if (!partial) {
    out.remoteCodexHome = null;
  }

  if (has("sync_codex")) {
    out.syncCodex = input.sync_codex ? 1 : 0;
  } else if (!partial) {
    out.syncCodex = 0;
  }

  if (has("enabled")) {
    out.enabled = input.enabled ? 1 : 0;
  }

  return out;
}

// ── Command building ──────────────────────────────────────────────────────────

/** Strip ANSI color / style escapes from command stderr (common on Windows shells). */
function stripAnsi(text) {
  return String(text || "").replace(/\u001b\[[0-9;]*m/g, "");
}

/**
 * Best-effort `SSH_AUTH_SOCK` for GUI/desktop launches that omit it. Never
 * guesses Secretive — that comes only from the user's `ssh -G` / `~/.ssh/config`.
 */
function discoverSshAuthSock() {
  if (process.env.SSH_AUTH_SOCK && fs.existsSync(process.env.SSH_AUTH_SOCK)) {
    return process.env.SSH_AUTH_SOCK;
  }
  if (process.platform === "darwin") {
    try {
      const sock = execFileSync("launchctl", ["getenv", "SSH_AUTH_SOCK"], {
        encoding: "utf8",
        stdio: ["ignore", "pipe", "ignore"],
      }).trim();
      if (sock && fs.existsSync(sock)) return sock;
    } catch {
      /* no launchd agent — file-based keys or IdentityAgent from ssh -G */
    }
  }
  // Windows OpenSSH agent service (optional — file keys still work without it).
  if (process.platform === "win32") {
    return "\\\\.\\pipe\\openssh-ssh-agent";
  }
  return null;
}

/**
 * Environment passed to every `ssh`/`scp` child. GUI/desktop launches often omit
 * `HOME` and `SSH_AUTH_SOCK`; file-based keys need `HOME`, ssh-agent users need
 * the socket, and Secretive/1Password users get `IdentityAgent` from `ssh -G`.
 */
function buildSshChildEnv() {
  const env = { ...process.env };
  const home = os.homedir();
  if (!env.HOME) env.HOME = home;
  if (!env.USERPROFILE) env.USERPROFILE = home;
  if (process.platform === "win32") {
    if (!env.SYSTEMROOT && process.env.SystemRoot) env.SYSTEMROOT = process.env.SystemRoot;
    if (!env.USER && process.env.USERNAME) env.USER = process.env.USERNAME;
  }
  if (!env.SSH_AUTH_SOCK) {
    const sock = discoverSshAuthSock();
    if (sock) env.SSH_AUTH_SOCK = sock;
  }
  return env;
}

/** Expand `~` in paths from `ssh -G` (IdentityAgent, etc.). */
function expandSshConfigPath(p) {
  const raw = String(p || "")
    .trim()
    .replace(/^"|"$/g, "");
  if (!raw || raw === "none" || raw === "SSH_AUTH_SOCK") return raw;
  if (raw.startsWith("~/")) return path.join(os.homedir(), raw.slice(2));
  if (raw === "~") return os.homedir();
  return raw;
}

/** Explicit user config so ssh/scp match the interactive shell even when env is sparse. */
function sshConfigFileArgs() {
  const cfg = path.join(os.homedir(), ".ssh", "config");
  return fs.existsSync(cfg) ? ["-F", cfg] : [];
}

function parseSshGOutput(stdout) {
  const out = {};
  for (const line of String(stdout || "").split(/\r?\n/)) {
    const idx = line.indexOf(" ");
    if (idx === -1) continue;
    out[line.slice(0, idx)] = line.slice(idx + 1);
  }
  return out;
}

/** Cache `ssh -G` identity-agent resolution per host. */
const sshExtraOptionsCache = new Map();

/**
 * Build `-o IdentityAgent=…` only when `ssh -G` names a concrete socket path
 * (Secretive, 1Password, custom agent). `SSH_AUTH_SOCK` / `none` → rely on env
 * or file-based keys — never force Secretive when the user's config does not.
 */
function identityAgentArgsFromConfig(identityagent) {
  const agent = expandSshConfigPath(identityagent);
  if (agent && agent !== "none" && agent !== "SSH_AUTH_SOCK") {
    return ["-o", `IdentityAgent=${agent}`];
  }
  return [];
}

/**
 * Match the interactive `ssh` stack: read effective config via `ssh -G` and
 * forward `IdentityAgent` only when config names a concrete agent socket.
 */
async function resolveExtraSshOptions(source) {
  const cacheKey = `${source.host}\0${source.ssh_port ?? ""}\0${source.identity_file ?? ""}`;
  if (sshExtraOptionsCache.has(cacheKey)) return sshExtraOptionsCache.get(cacheKey);

  const probeArgs = [
    "-G",
    ...sshConfigFileArgs(),
    "-o",
    "BatchMode=yes",
    "-o",
    "ConnectTimeout=10",
  ];
  if (source.ssh_port) probeArgs.push("-p", String(source.ssh_port));
  if (source.identity_file) {
    probeArgs.push("-i", source.identity_file, "-o", "IdentitiesOnly=yes");
  }
  probeArgs.push(source.host);

  let extra = [];
  try {
    const { code, stdout } = await runCommand(resolveSshBinary("ssh"), probeArgs, {
      timeoutMs: 10000,
      env: buildSshChildEnv(),
    });
    if (code === 0) {
      extra = identityAgentArgsFromConfig(parseSshGOutput(stdout).identityagent);
    }
  } catch {
    /* no extra options — OpenSSH will use file keys or SSH_AUTH_SOCK from env */
  }

  sshExtraOptionsCache.set(cacheKey, extra);
  return extra;
}

/**
 * Resolve OpenSSH client binaries (`ssh`, `scp`). Prefer PATH so we use the same
 * build as the user's shell (Git for Windows vs System32). On Windows, fall back
 * to the built-in OpenSSH install when PATH has no match.
 */
function resolveSshBinary(tool) {
  const exe = process.platform === "win32" ? `${tool}.exe` : tool;
  const pathDirs = (process.env.PATH || "").split(path.delimiter).filter(Boolean);
  for (const dir of pathDirs) {
    const full = path.join(dir, exe);
    try {
      if (fs.existsSync(full)) return full;
    } catch {
      /* try next PATH entry */
    }
  }
  const winDir = process.env.WINDIR || process.env.SystemRoot;
  if (process.platform === "win32" && winDir) {
    const system32 = path.join(winDir, "System32", "OpenSSH", exe);
    if (fs.existsSync(system32)) return system32;
  }
  return exe;
}

/**
 * SSH options shared by the connection test and the scp transport. BatchMode
 * fails fast instead of hanging on a password prompt; ConnectTimeout bounds a
 * dead host. Host-key policy is left at the SSH default on purpose (see file
 * header). Returned as a flat arg array — never a shell string.
 */
async function sshOptionArgs(source) {
  const args = [...sshConfigFileArgs(), "-o", "BatchMode=yes", "-o", "ConnectTimeout=10"];
  if (source.ssh_port) args.push("-p", String(source.ssh_port));
  if (source.identity_file) args.push("-i", source.identity_file, "-o", "IdentitiesOnly=yes");
  args.push(...(await resolveExtraSshOptions(source)));
  return args;
}

/** `scp` shares most SSH options but uses `-P` (capital) for the port. */
async function scpOptionArgs(source) {
  const args = [...sshConfigFileArgs(), "-o", "BatchMode=yes", "-o", "ConnectTimeout=10"];
  if (source.ssh_port) args.push("-P", String(source.ssh_port));
  if (source.identity_file) args.push("-i", source.identity_file, "-o", "IdentitiesOnly=yes");
  args.push(...(await resolveExtraSshOptions(source)));
  return args;
}

/** Effective remote_home from a source row (default `~/.claude`). */
function normalizedRemoteHome(source) {
  return source.remote_home && source.remote_home.trim() ? source.remote_home.trim() : "~/.claude";
}

/** `<claude_home>/projects` inside a WSL distro (POSIX path, no `wsl:` prefix). */
function wslProjectsPathFromHome(wslHome) {
  return `${wslHome.replace(/\/+$/, "")}/projects`;
}

/** True when `remote_home` targets Claude Code inside WSL on a Windows SSH host. */
function isWslRemoteHome(home) {
  return typeof home === "string" && home.startsWith("wsl:");
}

/** True when the user may rely on automatic WSL fallback (blank or default home). */
function allowsWslAutoFallback(source) {
  const home = normalizedRemoteHome(source);
  return home === "~/.claude" && !isWslRemoteHome(home);
}

/** Remote projects directory to mirror (default ~/.claude/projects). */
function remoteProjectsPath(source) {
  const home = normalizedRemoteHome(source);
  if (isWslRemoteHome(home)) {
    return `wsl:${wslProjectsPathFromHome(home.slice(4))}`;
  }
  const projects = `${home.replace(/\/+$/, "")}/projects`;
  // scp host:path uses forward slashes even for Windows remotes.
  return projects.replace(/\\/g, "/");
}

/** `host:path/.` spec for recursive scp (validated path — no shell interpolation). */
function scpRemoteSpec(source) {
  return `${source.host}:${remoteProjectsPath(source)}/.`;
}

/**
 * Remote shell probes for testConnection. Order: POSIX `sh` (Linux/macOS +
 * Git-Bash-on-Windows), PowerShell (Windows OpenSSH + ~ home), WSL when the
 * default home is used, then cmd.exe for explicit `C:/` or UNC homes. Paths are
 * validation-sanitized before embedding.
 */
function connectionProbeCommands(source) {
  const home = normalizedRemoteHome(source);
  const probes = [];

  if (isWslRemoteHome(home)) {
    const projects = wslProjectsPathFromHome(home.slice(4));
    probes.push(
      `wsl.exe -e sh -c 'test -d ${projects} && echo AGENTHUB_OK || echo AGENTHUB_NO_DIR'`
    );
    return probes;
  }

  const remoteProjects = remoteProjectsPath(source);

  if (/^\/\//.test(home)) {
    const winPath = remoteProjects.replace(/\//g, "\\");
    probes.push(`cmd /c "if exist ${winPath} (echo AGENTHUB_OK) else (echo AGENTHUB_NO_DIR)"`);
    probes.push(
      `powershell.exe -NoProfile -Command "if (Test-Path -LiteralPath '${remoteProjects}') { 'AGENTHUB_OK' } else { 'AGENTHUB_NO_DIR' }"`
    );
    return probes;
  }

  if (/^[A-Za-z]:\//.test(home)) {
    const winPath = remoteProjects.replace(/\//g, "\\");
    probes.push(`cmd /c "if exist ${winPath} (echo AGENTHUB_OK) else (echo AGENTHUB_NO_DIR)"`);
    return probes;
  }

  const shProbe = `sh -c 'test -d ${remoteProjects} && echo AGENTHUB_OK || echo AGENTHUB_NO_DIR'`;
  probes.push(shProbe);

  if (home.startsWith("~")) {
    const tail = home.replace(/^~/, "").replace(/\/+$/, "").replace(/\//g, "\\");
    const rel = tail ? `${tail}\\projects` : ".claude\\projects";
    probes.push(
      `powershell.exe -NoProfile -Command "if (Test-Path -LiteralPath (Join-Path $env:USERPROFILE '${rel}')) { 'AGENTHUB_OK' } else { 'AGENTHUB_NO_DIR' }"`
    );
    // Claude Code on Windows is often installed in WSL while SSH lands in Windows.
    probes.push(
      `wsl.exe -e sh -c 'test -d ${remoteProjects} && echo AGENTHUB_OK || echo AGENTHUB_NO_DIR'`
    );
  }
  return probes;
}

/** `ssh` remote command that streams a tar of a WSL home's `sub` dir to stdout. */
function wslTarRemoteCmd(wslHome, sub = "projects") {
  const dir = `${wslHome.replace(/\/+$/, "")}/${sub}`;
  return `wsl.exe -e sh -c 'tar -cC ${dir} .'`;
}

/** Whether Claude's projects dir exists inside the default (or given) WSL home. */
async function wslPathExists(source, wslHome, timeoutMs) {
  const projects = wslProjectsPathFromHome(wslHome);
  const { code, stdout } = await runSsh(
    source,
    `wsl.exe -e sh -c 'test -d ${projects} && echo AGENTHUB_OK || echo AGENTHUB_NO_DIR'`,
    timeoutMs
  );
  return code === 0 && stripAnsi(stdout).includes("AGENTHUB_OK");
}

function shouldTryWslFallback(source, err) {
  if (!allowsWslAutoFallback(source)) return false;
  const msg = String((err && err.message) || err || "").toLowerCase();
  return (
    msg.includes("no such file") ||
    msg.includes("not found") ||
    msg.includes("missing") ||
    msg.includes("does not exist")
  );
}

/** Sandboxed local staging dir for a source's mirrored transcripts. */
function stagingDir(sourceId) {
  return path.join(getDataDir(), "remote-sources", sourceId, "projects");
}

/**
 * Run a command with an argument array and no shell. Resolves with
 * {code, stdout, stderr}; rejects on spawn error or timeout. Never interpolates
 * arguments into a shell line.
 */
function runCommand(cmd, args, { timeoutMs, env } = {}) {
  return new Promise((resolve, reject) => {
    let child;
    try {
      child = spawn(cmd, args, { shell: false, env: env || process.env });
    } catch (err) {
      reject(err);
      return;
    }
    let stdout = "";
    let stderr = "";
    let timedOut = false;
    const timer =
      timeoutMs && timeoutMs > 0
        ? setTimeout(() => {
            timedOut = true;
            child.kill("SIGKILL");
          }, timeoutMs)
        : null;
    // Cap captured output so a chatty remote can't balloon memory.
    child.stdout.on("data", (d) => {
      if (stdout.length < 65536) stdout += d.toString();
    });
    child.stderr.on("data", (d) => {
      if (stderr.length < 65536) stderr += d.toString();
    });
    child.on("error", (err) => {
      if (timer) clearTimeout(timer);
      if (err && err.code === "ENOENT") {
        const tool = path.basename(cmd);
        reject(
          new Error(
            `${tool} not found — install the OpenSSH client (macOS/Linux: openssh-client; Windows: Optional Features → OpenSSH Client)`
          )
        );
        return;
      }
      reject(err);
    });
    child.on("close", (code) => {
      if (timer) clearTimeout(timer);
      if (timedOut) {
        reject(new Error(`${path.basename(cmd)} timed out after ${timeoutMs}ms`));
        return;
      }
      resolve({ code, stdout, stderr });
    });
  });
}

function formatTransferError(tool, { code, stderr }) {
  const detail = stripAnsi((stderr || "").trim() || `exit ${code}`).slice(0, 400);
  return new Error(`${tool} failed (exit ${code}): ${detail}`);
}

async function runSsh(source, remoteCmd, timeoutMs) {
  const args = [...(await sshOptionArgs(source)), source.host, remoteCmd];
  return runCommand(resolveSshBinary("ssh"), args, {
    timeoutMs,
    env: buildSshChildEnv(),
  });
}

async function runScpMirror(
  source,
  dest,
  timeoutMs,
  { legacy = false, remoteSpec = scpRemoteSpec(source), preserve = false } = {}
) {
  fs.rmSync(dest, { recursive: true, force: true });
  fs.mkdirSync(dest, { recursive: true });
  const args = ["-r"];
  // -p keeps the remote mtimes, which the Codex reader takes as activity.
  if (preserve) args.push("-p");
  if (legacy) args.push("-O");
  args.push(...(await scpOptionArgs(source)), remoteSpec, dest);
  return runCommand(resolveSshBinary("scp"), args, {
    timeoutMs,
    env: buildSshChildEnv(),
  });
}

function isLegacyScpProtocolError(stderr) {
  const s = stripAnsi(stderr || "").toLowerCase();
  return (
    s.includes("sftp") ||
    s.includes("subsystem") ||
    s.includes("protocol error") ||
    s.includes("remote side closed")
  );
}

/**
 * Mirror the remote projects tree into `dest` via recursive scp over SSH.
 * Requires only OpenSSH client locally and OpenSSH server (sftp/scp) remotely.
 */
async function mirrorViaScp(source, dest, timeoutMs) {
  let scpResult = await runScpMirror(source, dest, timeoutMs);
  if (scpResult.code !== 0 && isLegacyScpProtocolError(scpResult.stderr)) {
    scpResult = await runScpMirror(source, dest, timeoutMs, { legacy: true });
  }
  if (scpResult.code !== 0) {
    const detail = stripAnsi((scpResult.stderr || "").trim()).toLowerCase();
    if (detail.includes("no such file") || detail.includes("not found")) {
      throw new Error(
        `Remote projects directory missing (${remoteProjectsPath(source)}). Install Claude Code on the remote or set remote_home.`
      );
    }
    throw formatTransferError("scp", scpResult);
  }
}

/**
 * Pull a WSL-hosted projects tree over SSH by streaming `tar` from `wsl.exe`.
 * Used when Claude Code runs inside WSL but SSH lands in Windows.
 */
function mirrorViaWslTar(source, wslHome, dest, timeoutMs, sub = "projects") {
  fs.rmSync(dest, { recursive: true, force: true });
  fs.mkdirSync(dest, { recursive: true });

  return new Promise((resolve, reject) => {
    let sshChild;
    let tarChild;
    let timedOut = false;
    let stderr = "";
    const remoteCmd = wslTarRemoteCmd(wslHome, sub);

    const timer =
      timeoutMs && timeoutMs > 0
        ? setTimeout(() => {
            timedOut = true;
            try {
              sshChild?.kill("SIGKILL");
            } catch {
              /* ignore */
            }
            try {
              tarChild?.kill("SIGKILL");
            } catch {
              /* ignore */
            }
          }, timeoutMs)
        : null;

    const fail = (err) => {
      if (timer) clearTimeout(timer);
      reject(err);
    };

    try {
      tarChild = spawn("tar", ["-xf", "-", "-C", dest], {
        shell: false,
        stdio: ["pipe", "pipe", "pipe"],
      });
    } catch {
      fail(
        new Error(
          "tar not found — required to pull Claude Code history from WSL over SSH (install tar or use a UNC remote_home path)"
        )
      );
      return;
    }

    sshOptionArgs(source)
      .then((opts) => {
        const args = [...opts, source.host, remoteCmd];
        try {
          sshChild = spawn(resolveSshBinary("ssh"), args, {
            shell: false,
            env: buildSshChildEnv(),
            stdio: ["ignore", "pipe", "pipe"],
          });
        } catch (err) {
          try {
            tarChild.kill();
          } catch {
            /* ignore */
          }
          fail(err);
          return;
        }

        const onStderr = (d) => {
          if (stderr.length < 65536) stderr += d.toString();
        };
        sshChild.stderr.on("data", onStderr);
        tarChild.stderr.on("data", onStderr);
        sshChild.stdout.pipe(tarChild.stdin);

        let sshCode;
        let tarCode;
        let pending = 2;
        const done = () => {
          if (--pending > 0) return;
          if (timer) clearTimeout(timer);
          if (timedOut) {
            fail(new Error(`WSL transfer timed out after ${timeoutMs}ms`));
            return;
          }
          if (sshCode !== 0) {
            const detail = stripAnsi(stderr.trim()).toLowerCase();
            if (detail.includes("wsl") && detail.includes("not recognized")) {
              fail(
                new Error(
                  "WSL is not available on the remote Windows host — install WSL, SSH directly into the Linux distro, or set remote_home to a UNC path such as //wsl.localhost/Ubuntu/home/you/.claude"
                )
              );
              return;
            }
            fail(formatTransferError("ssh", { code: sshCode, stderr }));
            return;
          }
          if (tarCode !== 0) {
            fail(
              new Error(
                `tar extract failed (exit ${tarCode}): ${stripAnsi(stderr).trim().slice(0, 400)}`
              )
            );
            return;
          }
          resolve();
        };

        sshChild.on("close", (code) => {
          sshCode = code;
          done();
        });
        tarChild.on("close", (code) => {
          tarCode = code;
          done();
        });
        sshChild.on("error", fail);
        tarChild.on("error", fail);
      })
      .catch(fail);
  });
}

/**
 * Mirror the remote projects tree into `dest`. Uses scp for native paths; when
 * `remote_home` is `wsl:…` or the default home is missing on Windows but present
 * in WSL, streams a tar archive from `wsl.exe` instead.
 */
async function mirrorRemoteTree(source, dest, timeoutMs) {
  const home = normalizedRemoteHome(source);
  if (isWslRemoteHome(home)) {
    await mirrorViaWslTar(source, home.slice(4), dest, timeoutMs);
    return;
  }

  try {
    await mirrorViaScp(source, dest, timeoutMs);
  } catch (err) {
    if (
      shouldTryWslFallback(source, err) &&
      (await wslPathExists(source, "~/.claude", TEST_TIMEOUT_MS))
    ) {
      await mirrorViaWslTar(source, "~/.claude", dest, timeoutMs);
      return;
    }
    throw err;
  }
}

// ── Codex ─────────────────────────────────────────────────────────────────────

/** The remote CODEX_HOME: `remote_codex_home`, else ~/.codex (in WSL when Claude's home is). */
function normalizedCodexHome(source) {
  const own = source.remote_codex_home && source.remote_codex_home.trim();
  if (own) return own;
  return isWslRemoteHome(normalizedRemoteHome(source)) ? "wsl:~/.codex" : "~/.codex";
}

/** Sandboxed staging dir for a source's Codex mirror (`sessions/` + title index). */
function codexStagingDir(sourceId) {
  return path.join(getDataDir(), "remote-sources", sourceId, "codex");
}

/**
 * Mirror the remote `<codex home>/sessions` tree into `<dest>/sessions`, mtimes
 * kept, plus `session_index.jsonl` (Codex's own thread names; holds no
 * credentials, and is absent until a session is renamed, so never fatal). From
 * WSL only `sessions/` comes over; names then fall back to the first prompt.
 */
async function mirrorCodex(source, dest, timeoutMs) {
  const home = normalizedCodexHome(source);
  const sessionsDest = path.join(dest, "sessions");
  fs.mkdirSync(dest, { recursive: true });
  if (isWslRemoteHome(home)) {
    await mirrorViaWslTar(source, home.slice(4), sessionsDest, timeoutMs, "sessions");
    return;
  }
  const base = home.replace(/\\/g, "/").replace(/\/+$/, "");
  const remoteSpec = `${source.host}:${base}/sessions/.`;
  let r = await runScpMirror(source, sessionsDest, timeoutMs, { remoteSpec, preserve: true });
  if (r.code !== 0 && isLegacyScpProtocolError(r.stderr)) {
    r = await runScpMirror(source, sessionsDest, timeoutMs, {
      remoteSpec,
      preserve: true,
      legacy: true,
    });
  }
  if (r.code !== 0) {
    const detail = stripAnsi((r.stderr || "").trim()).toLowerCase();
    if (detail.includes("no such file") || detail.includes("not found")) {
      throw new Error(
        `Remote Codex sessions directory missing (${base}/sessions). Install Codex on the remote or set remote_codex_home.`
      );
    }
    throw formatTransferError("scp", r);
  }
  const index = path.join(dest, "session_index.jsonl");
  fs.rmSync(index, { force: true });
  await runCommand(
    resolveSshBinary("scp"),
    ["-p", ...(await scpOptionArgs(source)), `${source.host}:${base}/session_index.jsonl`, index],
    { timeoutMs, env: buildSshChildEnv() }
  ).catch(() => {});
}

/**
 * Pull and read one source's Codex history: sessions land as `tool = 'codex'`
 * rows tagged with the source id, with their tokens, through codex-usage.js.
 * The activity window is REMOTE_ACTIVE_WINDOW_MS, as for Claude transcripts.
 */
async function syncCodex(source, broadcast) {
  const dest = codexStagingDir(source.id);
  await mirrorCodex(source, dest, SYNC_TIMEOUT_MS);
  const { ingestCodexUsage, settleActive } = require("./codex-usage");
  const opts = { home: dest, source: source.id, activeMs: REMOTE_ACTIVE_WINDOW_MS };
  const send = broadcast || (() => {});
  const r = await ingestCodexUsage({ broadcast: send, ...opts });
  settleActive(send, Date.now(), opts);
  // A model only the remote used has no rate row yet: price it now, as the
  // local Codex poll does.
  const pricingSync = require("./pricing-sync");
  if (r.sessions > 0 && pricingSync.syncEnabled()) {
    void pricingSync.syncPricing(require("../db").db).catch(() => {});
  }
  return r;
}

// ── Public operations ─────────────────────────────────────────────────────────

/**
 * User-facing copy after a successful projects-dir probe.
 * Distinguishes explicit `wsl:` homes from auto-detected WSL fallback.
 */
function connectionSuccessMessage(source, remoteCmd) {
  const home = normalizedRemoteHome(source);
  if (isWslRemoteHome(home)) {
    return `Connected. Claude Code history found in WSL (${home}).`;
  }
  if (remoteCmd.includes("wsl.exe")) {
    return "Connected. Claude Code history found in WSL (auto-detected). Set remote home to wsl:~/.claude to sync WSL instead of Windows.";
  }
  return "Connected. Remote Claude Code history found.";
}

/**
 * Best-effort "last activity" for a mirrored transcript. Prefer the newest
 * event timestamp inside the JSONL over filesystem mtime so a touched-but-stale
 * seed file (or scp preserving an old tree) does not look still-running.
 */
function lastTranscriptActivityMs(filePath) {
  let mtimeMs = null;
  try {
    mtimeMs = fs.statSync(filePath).mtimeMs;
  } catch {
    return null;
  }

  let contentMs = null;
  try {
    const text = fs.readFileSync(filePath, "utf8");
    const lines = text.trim().split(/\n/).filter(Boolean);
    for (let i = lines.length - 1; i >= 0; i--) {
      try {
        const row = JSON.parse(lines[i]);
        const raw = row.timestamp || row.ts;
        if (!raw) continue;
        const ms = new Date(raw).getTime();
        if (!Number.isNaN(ms)) {
          contentMs = ms;
          break;
        }
      } catch {
        /* skip malformed tail line */
      }
    }
  } catch {
    /* unreadable — fall back to mtime */
  }

  if (contentMs != null) return contentMs;
  return mtimeMs;
}

/**
 * Probe a source: can we SSH in, and does the remote projects dir exist?
 * Returns {ok, message, remoteProjects}. Never throws for an ordinary failure
 * (bad host, unreachable) — those come back as {ok:false, message}.
 */
async function testConnection(source) {
  const remoteProjects = remoteProjectsPath(source);

  try {
    const auth = await runSsh(source, "echo AGENTHUB_AUTH", TEST_TIMEOUT_MS);
    const authOut = stripAnsi(auth.stdout);
    if (auth.code !== 0 || !authOut.includes("AGENTHUB_AUTH")) {
      const msg = stripAnsi(auth.stderr.trim() || `ssh exited with code ${auth.code}`).slice(
        0,
        500
      );
      return {
        ok: false,
        message: msg || "SSH authentication failed.",
        remoteProjects,
      };
    }

    const probes = connectionProbeCommands(source);
    let lastMessage = "";

    for (const remoteCmd of probes) {
      const { code, stdout, stderr } = await runSsh(source, remoteCmd, TEST_TIMEOUT_MS);
      const out = stripAnsi(stdout);
      if (out.includes("AGENTHUB_OK")) {
        return {
          ok: true,
          message: connectionSuccessMessage(source, remoteCmd),
          remoteProjects,
        };
      }
      if (out.includes("AGENTHUB_NO_DIR")) {
        return {
          ok: false,
          message: `Connected, but ${remoteProjects} does not exist on the remote. Is Claude Code installed there? Set a custom remote home if it lives elsewhere.`,
          remoteProjects,
        };
      }
      lastMessage = stripAnsi(stderr.trim() || `ssh exited with code ${code}`).slice(0, 500);
    }
    return {
      ok: false,
      message:
        lastMessage || "SSH connected but the remote projects directory could not be verified.",
      remoteProjects,
    };
  } catch (err) {
    return { ok: false, message: stripAnsi(err.message).slice(0, 500), remoteProjects };
  }
}

/**
 * Session ids present in a mirrored staging tree. A Claude Code session file is
 * `<projects>/<encoded-cwd>/<sessionId>.jsonl`; subagent transcripts live under
 * `.../<sessionId>/subagents/agent-*.jsonl`. So a top-level session file is any
 * `.jsonl` whose basename is not `agent-*` and whose parent dir is not
 * `subagents` — those are exactly the ids to tag with this source.
 */
function stagedSessionIds(dir) {
  const ids = [];
  for (const f of collectJsonlFiles(dir)) {
    const base = path.basename(f, ".jsonl");
    if (base.startsWith("agent-")) continue;
    if (path.basename(path.dirname(f)) === "subagents") continue;
    ids.push(base);
  }
  return ids;
}

/**
 * Reconcile the live status of a source's sessions from its mirrored transcripts.
 *
 * Remote sessions get NO live hooks and are excluded from every local liveness /
 * staleness heuristic (the process probe, the watchdog transcript scan, the
 * startup + periodic stale sweeps — all gated on `source = 'local'`). The scp
 * mirror is therefore the single source of truth for whether a remote session is
 * still running: a transcript file touched within REMOTE_ACTIVE_WINDOW_MS means
 * the remote CLI is still writing to it (⇒ active); once it stops advancing the
 * session has ended (⇒ completed). The shared importer only sets status when it
 * FIRST inserts a session, so this pass is what keeps an already-imported remote
 * session's status correct on every subsequent sync (and self-heals any session
 * a pre-fix build wrongly completed).
 *
 * @param {object} dbModule require("../db")
 * @param {object} source remote_sources row (needs `id`)
 * @param {string} dest staging dir mirrored for this source
 * @param {Function} [broadcast] websocket broadcaster
 */
function reconcileRemoteSessionStatus(dbModule, source, dest, broadcast) {
  const { stmts } = dbModule;
  const now = Date.now();

  // Newest transcript activity per top-level session id (content timestamp preferred
  // over mirror mtime — see lastTranscriptActivityMs).
  const mtimeById = new Map();
  for (const f of collectJsonlFiles(dest)) {
    const base = path.basename(f, ".jsonl");
    if (base.startsWith("agent-")) continue;
    if (path.basename(path.dirname(f)) === "subagents") continue;
    const activityMs = lastTranscriptActivityMs(f);
    if (activityMs == null) continue;
    if (!mtimeById.has(base) || activityMs > mtimeById.get(base)) mtimeById.set(base, activityMs);
  }

  for (const [id, mtimeMs] of mtimeById) {
    const sess = stmts.getSession.get(id);
    // Only touch sessions this source actually owns — never reinterpret a local
    // (or other-source) row that happens to share an id.
    if (!sess || sess.source !== source.id) continue;
    const active = now - mtimeMs < REMOTE_ACTIVE_WINDOW_MS;

    if (active && sess.status !== "active") {
      // Heal a session the mirror shows is still running. Mirror the importer's
      // fresh-import shape: session active, main agent back to waiting (we can't
      // know from a mirror whether it's mid-turn; waiting matches import).
      stmts.reactivateSession.run(id);
      const mainId = `${id}-main`;
      const main = stmts.getAgent.get(mainId);
      if (main && main.status !== "working" && main.status !== "waiting") {
        stmts.updateAgent.run(null, "waiting", null, null, null, null, mainId);
      }
      if (broadcast) {
        broadcast("session_updated", stmts.getSession.get(id));
        const refreshed = stmts.getAgent.get(mainId);
        if (refreshed) broadcast("agent_updated", refreshed);
      }
    } else if (!active && sess.status === "active") {
      // Mirror stopped advancing ⇒ the remote session ended. Land it in the same
      // terminal state a real SessionEnd produces (agents completed, ended_at
      // stamped). Leave error/completed/abandoned rows untouched.
      const ts = new Date().toISOString();
      stmts.updateSession.run(null, "completed", ts, null, id);
      const agents = stmts.listAgentsBySession.all(id);
      for (const a of agents) {
        if (a.status !== "completed" && a.status !== "error") {
          stmts.updateAgent.run(null, "completed", null, null, ts, null, a.id);
        }
      }
      if (broadcast) {
        broadcast("session_updated", stmts.getSession.get(id));
        for (const a of agents) {
          const refreshed = stmts.getAgent.get(a.id);
          if (refreshed) broadcast("agent_updated", refreshed);
        }
      }
    }
  }
}

/**
 * Sync one source: mirror its remote history, import it, tag the imported
 * sessions with the source id, and record status. Returns import counters.
 * Guarded against concurrent runs of the same source.
 *
 * @param {object} dbModule require("../db")
 * @param {object} source remote_sources row
 * @param {{onProgress?:Function, broadcast?:Function}} [opts]
 */
async function syncSource(dbModule, source, opts = {}) {
  const { stmts } = dbModule;
  const { onProgress, broadcast } = opts;
  if (inFlight.has(source.id)) {
    return { skipped: true, reason: "already-syncing" };
  }
  inFlight.add(source.id);

  const emitStatus = (status, error) => {
    try {
      stmts.setRemoteSourceStatus.run(status, error || null, source.id);
    } catch {
      /* non-fatal */
    }
    if (broadcast)
      broadcast("remote_source.status", { id: source.id, status, error: error || null });
  };

  const dest = stagingDir(source.id);
  const preExisting = new Set();
  try {
    for (const id of stagedSessionIds(dest)) {
      if (stmts.getSession.get(id)) preExisting.add(id);
    }
  } catch {
    /* staging dir may not exist yet */
  }

  try {
    emitStatus("syncing", null);
    fs.mkdirSync(dest, { recursive: true });

    // With Codex opted in, a machine may have only one of the two CLIs: one
    // half failing is reported, both failing fails the sync.
    let counters = {};
    let ids = [];
    let claudeError = null;
    try {
      await mirrorRemoteTree(source, dest, SYNC_TIMEOUT_MS);

      // Parse the mirror through the SAME importer local history uses.
      counters = await importFromDirectory(dbModule, dest, { onProgress });

      // Tag every session that came from this mirror with the source id.
      ids = stagedSessionIds(dest);
      const tag = dbModule.db.transaction((sessionIds) => {
        for (const id of sessionIds) stmts.setSessionSource.run(source.id, id);
      });
      tag(ids);

      // Now that this source's sessions are tagged, reconcile their live status
      // from the freshly mirrored transcripts. This is the ONLY thing that keeps a
      // remote session's active/completed state correct after its first import,
      // since every local liveness/staleness sweep skips remote sessions.
      try {
        reconcileRemoteSessionStatus(dbModule, source, dest, broadcast);
      } catch {
        /* non-fatal — status reconciliation must never fail a sync */
      }
    } catch (err) {
      if (!source.sync_codex) throw err;
      claudeError = err;
    }

    let codex = null;
    let partialError = claudeError ? `Claude Code: ${claudeError.message}` : null;
    if (source.sync_codex) {
      try {
        codex = await syncCodex(source, broadcast);
      } catch (err) {
        if (claudeError) throw new Error(`${claudeError.message} — Codex: ${err.message}`);
        partialError = `Codex: ${err.message}`;
      }
    }

    const result = {
      imported: counters.imported || 0,
      skipped: counters.skipped || 0,
      backfilled: counters.backfilled || 0,
      errors: counters.errors || 0,
      sessions_seen: counters.sessionsSeen || 0,
      sessions_tagged: ids.length,
      ...(codex && { codex_files: codex.files, codex_sessions: codex.sessions }),
    };
    const now = new Date().toISOString();
    try {
      stmts.setRemoteSourceSyncResult.run(
        "ok",
        partialError ? partialError.slice(0, 500) : null,
        now,
        JSON.stringify(result),
        source.id
      );
    } catch {
      /* non-fatal */
    }
    if (broadcast) {
      broadcast("remote_source.status", { id: source.id, status: "ok", last_sync_at: now });
      // Mirror local session sync: push per-session frames so Kanban/Sessions
      // update immediately without waiting for a full refetch.
      for (const id of ids) {
        const row = stmts.getSession.get(id);
        if (!row) continue;
        broadcast(preExisting.has(id) ? "session_updated" : "session_created", row);
        try {
          const mainAgent = dbModule.db
            .prepare("SELECT * FROM agents WHERE session_id = ? AND type = 'main' LIMIT 1")
            .get(id);
          if (mainAgent) {
            broadcast(preExisting.has(id) ? "agent_updated" : "agent_created", mainAgent);
          }
        } catch {
          /* best-effort */
        }
      }
      // Nudge open pages (Sessions/Dashboard) to refetch — remote sessions were
      // written without going through the live hook broadcast path.
      broadcast("import.progress", {
        importId: `remote-${source.id}`,
        phase: "complete",
        source: "remote",
        counters: result,
      });
      broadcast("remote_data.updated", {
        sourceId: source.id,
        source: source.id,
        label: source.label,
        counters: result,
        last_sync_at: now,
      });
    }
    return result;
  } catch (err) {
    const msg = (err && err.message ? err.message : String(err)).slice(0, 500);
    try {
      stmts.setRemoteSourceStatus.run("error", msg, source.id);
    } catch {
      /* non-fatal */
    }
    if (broadcast)
      broadcast("remote_source.status", { id: source.id, status: "error", error: msg });
    throw err;
  } finally {
    inFlight.delete(source.id);
  }
}

/**
 * Sync every enabled source sequentially (one SSH connection at a time — avoids
 * a connection storm on the local box and on any shared bastion). Per-source
 * failures are isolated: one bad source never aborts the rest. Returns a
 * per-source summary array.
 */
async function syncAllEnabled(dbModule, opts = {}) {
  const sources = dbModule.stmts.listEnabledRemoteSources.all();
  const results = [];
  for (const source of sources) {
    try {
      const r = await syncSource(dbModule, source, opts);
      results.push({ id: source.id, ok: true, ...r });
    } catch (err) {
      results.push({ id: source.id, ok: false, error: err.message });
    }
  }
  return results;
}

module.exports = {
  validateSourceInput,
  ValidationError,
  testConnection,
  syncSource,
  syncAllEnabled,
  normalizedRemoteHome,
  wslProjectsPathFromHome,
  isWslRemoteHome,
  allowsWslAutoFallback,
  wslTarRemoteCmd,
  wslPathExists,
  mirrorViaWslTar,
  mirrorViaScp,
  stagingDir,
  codexStagingDir,
  normalizedCodexHome,
  syncCodex,
  remoteProjectsPath,
  stagedSessionIds,
  reconcileRemoteSessionStatus,
  // exported for tests
  sshOptionArgs,
  scpOptionArgs,
  connectionProbeCommands,
  connectionSuccessMessage,
  lastTranscriptActivityMs,
  stripAnsi,
  resolveSshBinary,
  buildSshChildEnv,
  expandSshConfigPath,
  identityAgentArgsFromConfig,
  sshConfigFileArgs,
  scpRemoteSpec,
  parseSshGOutput,
  isLegacyScpProtocolError,
  HOST_RE,
  REMOTE_PATH_RE,
};