remote-transport.js

// see .ai/contexts/session-cache.md ("Remote hosts")
'use strict';

const fs = require('fs');
const path = require('path');
const { isSafeMirrorRelPath } = require('./remote-hosts');

const REMOTE_PROJECTS_REL = '.claude/projects';
const REMOTE_SESSIONS_REL = '.claude/sessions';
const SESSIONS_MARKER = '\u0001SWITCHBOARD-SESSIONS\u0001';
const MAX_SESSION_DESCRIPTORS = 200;
const MAX_SESSION_DESCRIPTOR_BYTES = 8192;
const DEFAULT_CONNECT_TIMEOUT_S = 10;
const DEFAULT_LIST_TIMEOUT_MS = 60_000;
const DEFAULT_FETCH_TIMEOUT_MS = 120_000;
const MAX_LIST_BYTES = 8 * 1024 * 1024;
const DEFAULT_CONCURRENCY = 4;
// see .ai/contexts/session-cache.md ("Remote hosts — incremental fetch")
const MAX_RANGE_FETCH_BYTES = 64 * 1024 * 1024;

const SSH_BASE_OPTS = [
  '-o', 'BatchMode=yes',
  '-o', `ConnectTimeout=${DEFAULT_CONNECT_TIMEOUT_S}`,
];

// STX-prefixed liveness marker printed after each descriptor — see .ai/contexts/session-cache.md ("Remote SSH hosts", liveness)
const ALIVE_MARKER_PREFIX = 'ALIVE:';

// see .ai/contexts/session-cache.md ("Remote SSH hosts (issue #211)" and
// "Remote hosts — meta.json sidecars"). F9: each descriptor is followed by a
// \002ALIVE:0|1 line so parseSessions() can drop dead pids without a 2nd ssh.
const LIST_COMMAND =
  `find ${REMOTE_PROJECTS_REL} -type f \\( -name '*.jsonl' -o -name '*.meta.json' \\) -printf '%T@\\t%s\\t%P\\n' || exit $?; ` +
  `printf '\\001SWITCHBOARD-SESSIONS\\001\\n'; ` +
  `find ${REMOTE_SESSIONS_REL} -maxdepth 1 -type f -name '[0-9]*.json' 2>/dev/null | LC_ALL=C sort | ` +
  `head -n ${MAX_SESSION_DESCRIPTORS} | while IFS= read -r f; do head -c ${MAX_SESSION_DESCRIPTOR_BYTES} "$f"; printf '\\n'; ` +
  `pid=$(basename "$f" .json); printf '\\002ALIVE:%s\\n' "$( [ -d "/proc/$pid" ] && echo 1 || echo 0 )"; done`;

function parseInventory(stdout) {
  const out = [];
  for (const line of stdout.split('\n')) {
    if (!line) continue;
    const parts = line.split('\t');
    if (parts.length < 3) continue;
    const mtime = Number.parseFloat(parts[0]);
    const size = Number.parseInt(parts[1], 10);
    const rel = parts.slice(2).join('\t').replace(/\r$/, '');
    if (!Number.isFinite(mtime) || !Number.isFinite(size)) continue;
    if (!isSafeMirrorRelPath(rel)) continue;
    out.push({ rel, size, mtimeMs: Math.round(mtime * 1000) });
  }
  return out;
}

// see .ai/contexts/session-cache.md ("Remote SSH hosts (issue #211)")
function splitListOutput(stdout) {
  const idx = stdout.indexOf(SESSIONS_MARKER);
  if (idx === -1) return { inventoryBlock: stdout, sessionsBlock: '' };
  const afterIdx = idx + SESSIONS_MARKER.length;
  const validStart = idx === 0 || stdout[idx - 1] === '\n';
  const validEnd = stdout[afterIdx] === '\n';
  if (!validStart || !validEnd) return { inventoryBlock: stdout, sessionsBlock: '' };
  return { inventoryBlock: stdout.slice(0, idx), sessionsBlock: stdout.slice(afterIdx + 1) };
}

// see .ai/contexts/session-cache.md ("Remote hosts — descriptor-only sessions")
function transcriptSessionIds(files) {
  const ids = new Set();
  for (const f of files) {
    if (!f || typeof f.rel !== 'string' || !f.rel.endsWith('.jsonl')) continue;
    const base = f.rel.slice(f.rel.lastIndexOf('/') + 1, -'.jsonl'.length);
    if (base) ids.add(base);
  }
  return ids;
}

// see .ai/contexts/session-cache.md ("Remote SSH hosts (issue #211)", liveness marker)
function parseSessions(block) {
  const lines = block.split('\n');
  const sessions = [];
  const warnings = [];
  let dropped = 0;
  for (let i = 0; i < lines.length; i++) {
    const line = lines[i].replace(/\r$/, '');
    if (!line) continue;

    const next = i + 1 < lines.length ? lines[i + 1].replace(/\r$/, '') : undefined;
    const hasMarker = next === `${ALIVE_MARKER_PREFIX}0` || next === `${ALIVE_MARKER_PREFIX}1`;
    const alive = hasMarker ? next === `${ALIVE_MARKER_PREFIX}1` : null;
    if (hasMarker) i++; // consume the marker line unconditionally, valid JSON or not

    let parsed;
    try {
      parsed = JSON.parse(line);
    } catch {
      warnings.push('skipped a session descriptor: invalid JSON');
      continue;
    }
    if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) {
      warnings.push('skipped a session descriptor: not a JSON object');
      continue;
    }
    if (!Number.isInteger(parsed.pid) || parsed.pid <= 0) {
      warnings.push('skipped a session descriptor: missing/invalid pid');
      continue;
    }
    if (typeof parsed.sessionId !== 'string' || !parsed.sessionId) {
      warnings.push('skipped a session descriptor: missing/invalid sessionId');
      continue;
    }
    if (alive === false) {
      dropped++;
      continue;
    }
    sessions.push(parsed);
  }
  return { sessions, warnings, dropped };
}

/**
 * ssh/scp transport. `spawn` is injected so the process bounding, the argv and
 * the parsing are all testable without a network or an ssh binary.
 */
function createSshTransport(opts = {}) {
  const spawn = opts.spawn || require('child_process').spawn;
  const log = opts.log || { info() {}, warn() {}, error() {} };
  const listTimeoutMs = opts.listTimeoutMs || DEFAULT_LIST_TIMEOUT_MS;
  const fetchTimeoutMs = opts.fetchTimeoutMs || DEFAULT_FETCH_TIMEOUT_MS;
  const concurrency = Math.max(1, Math.min(8, opts.concurrency || DEFAULT_CONCURRENCY));

  // Every child stays registered until it settles; dispose() kills the set.
  const live = new Set();
  let disposed = false;

  // `binary: true` captures stdout as a raw Buffer. see .ai/contexts/session-cache.md ("Remote hosts — incremental fetch")
  function run(command, args, { timeoutMs, maxBytes, binary }) {
    return new Promise((resolve) => {
      if (disposed) {
        resolve({ code: -1, stdout: '', stdoutBuffer: Buffer.alloc(0), stderr: 'transport disposed', timedOut: false });
        return;
      }
      let child;
      try {
        child = spawn(command, args, { windowsHide: true, stdio: ['ignore', 'pipe', 'pipe'] });
      } catch (err) {
        resolve({ code: -1, stdout: '', stdoutBuffer: Buffer.alloc(0), stderr: err.message, timedOut: false });
        return;
      }
      live.add(child);

      let stdout = '';
      const chunks = [];
      let stdoutBytes = 0;
      let stderr = '';
      let truncated = false;
      let timedOut = false;
      let settled = false;

      const kill = () => {
        try { child.kill('SIGKILL'); } catch {}
      };
      // Never unref'd: this timer IS the bound on the child. Unref'd, an
      // otherwise-idle loop exits before it fires and the promise never
      // settles. see .ai/contexts/session-cache.md ("Remote SSH hosts")
      const timer = setTimeout(() => { timedOut = true; kill(); }, timeoutMs);

      const finish = (code) => {
        if (settled) return;
        settled = true;
        clearTimeout(timer);
        live.delete(child);
        resolve({
          code,
          stdout,
          stdoutBuffer: binary ? Buffer.concat(chunks) : undefined,
          stderr: stderr.slice(0, 4096),
          timedOut,
          truncated,
        });
      };

      if (child.stdout) {
        if (binary) {
          child.stdout.on('data', (chunk) => {
            stdoutBytes += chunk.length;
            if (stdoutBytes > (maxBytes || MAX_RANGE_FETCH_BYTES)) {
              truncated = true;
              kill();
              return;
            }
            chunks.push(chunk);
          });
        } else {
          child.stdout.setEncoding('utf8');
          child.stdout.on('data', (chunk) => {
            if (stdout.length + chunk.length > (maxBytes || MAX_LIST_BYTES)) {
              truncated = true;
              kill();
              return;
            }
            stdout += chunk;
          });
        }
      }
      if (child.stderr) {
        child.stderr.setEncoding('utf8');
        child.stderr.on('data', (chunk) => { if (stderr.length < 4096) stderr += chunk; });
      }
      child.on('error', (err) => { stderr += err.message; finish(-1); });
      child.on('close', (code) => finish(code == null ? -1 : code));
    });
  }

  async function listFiles(alias) {
    const res = await run('ssh', [...SSH_BASE_OPTS, '-n', alias, LIST_COMMAND], {
      timeoutMs: listTimeoutMs,
      maxBytes: MAX_LIST_BYTES,
    });
    if (res.timedOut) throw new Error(`ssh inventory timed out after ${listTimeoutMs} ms`);
    if (res.truncated) throw new Error('ssh inventory output exceeded the size cap');
    if (res.code !== 0) throw new Error(`ssh inventory failed (exit ${res.code}): ${res.stderr.trim() || 'no stderr'}`);
    const { inventoryBlock, sessionsBlock } = splitListOutput(res.stdout);
    const files = parseInventory(inventoryBlock);
    const { sessions: rawSessions, warnings, dropped } = parseSessions(sessionsBlock);
    for (const w of warnings) log.warn(`[remote:${alias}] ${w}`);
    if (dropped) log.warn(`[remote:${alias}] dropped ${dropped} dead session descriptor(s)`);
    const transcriptIds = transcriptSessionIds(files);
    const sessions = rawSessions.map(s => ({ ...s, descriptorOnly: !transcriptIds.has(s.sessionId) }));
    return { files, sessions };
  }

  async function fetchOne(alias, rel, destRoot) {
    const destPath = path.join(destRoot, rel);
    fs.mkdirSync(path.dirname(destPath), { recursive: true });
    const tmpPath = destPath + '.part';
    // Deliberately unquoted; isSafeMirrorRelPath is the guard. see .ai/contexts/session-cache.md ("Remote SSH hosts")
    const remote = `${alias}:${REMOTE_PROJECTS_REL}/${rel}`;
    const res = await run('scp', [...SSH_BASE_OPTS, '-p', '-q', remote, tmpPath], {
      timeoutMs: fetchTimeoutMs,
    });
    if (res.code !== 0 || res.timedOut) {
      try { fs.rmSync(tmpPath, { force: true }); } catch {}
      const why = res.timedOut ? 'timed out' : `exit ${res.code}: ${res.stderr.trim()}`;
      log.warn(`[remote:${alias}] scp ${rel} failed — ${why}`);
      return false;
    }
    try {
      fs.renameSync(tmpPath, destPath);
    } catch (err) {
      try { fs.rmSync(tmpPath, { force: true }); } catch {}
      log.warn(`[remote:${alias}] could not place ${rel}: ${err.message}`);
      return false;
    }
    return true;
  }

  async function fetchFiles(alias, rels, destRoot) {
    const fetched = [];
    const failed = [];
    const queue = rels.filter(isSafeMirrorRelPath);
    let cursor = 0;

    const workers = Array.from({ length: Math.min(concurrency, queue.length) }, async () => {
      while (!disposed) {
        const i = cursor++;
        if (i >= queue.length) return;
        const rel = queue[i];
        if (await fetchOne(alias, rel, destRoot)) fetched.push(rel);
        else failed.push(rel);
      }
    });
    await Promise.all(workers);
    // A dispose mid-run leaves the rest unfetched — report them as failures.
    for (let i = cursor; i < queue.length; i++) failed.push(queue[i]);
    return { fetched, failed };
  }

  // offset comes from the local mirror's byte count — see .ai/contexts/session-cache.md ("Remote hosts — incremental fetch")
  async function fetchOneIncremental(alias, rel, offset, destRoot) {
    if (!Number.isInteger(offset) || offset < 0) return false;
    const destPath = path.join(destRoot, rel);
    if (!fs.existsSync(destPath)) {
      log.warn(`[remote:${alias}] range fetch ${rel} skipped — no local file to append to`);
      return false;
    }
    const tmpPath = destPath + '.part';
    const remoteRel = `${REMOTE_PROJECTS_REL}/${rel}`;
    // tail -c is 1-indexed: offset+1 is the first new byte.
    const rangeCommand = `tail -c +${offset + 1} '${remoteRel}'`;
    const res = await run('ssh', [...SSH_BASE_OPTS, '-n', alias, rangeCommand], {
      timeoutMs: fetchTimeoutMs,
      binary: true,
    });
    if (res.code !== 0 || res.timedOut || res.truncated) {
      const why = res.timedOut ? 'timed out' : res.truncated ? 'range exceeded the size cap' : `exit ${res.code}: ${res.stderr.trim()}`;
      log.warn(`[remote:${alias}] range fetch ${rel} failed — ${why}`);
      return false;
    }
    try {
      // Copy-then-append-then-rename. see .ai/contexts/session-cache.md ("Remote hosts — incremental fetch")
      fs.copyFileSync(destPath, tmpPath);
      fs.appendFileSync(tmpPath, res.stdoutBuffer);
      fs.renameSync(tmpPath, destPath);
    } catch (err) {
      try { fs.rmSync(tmpPath, { force: true }); } catch {}
      log.warn(`[remote:${alias}] could not append ${rel}: ${err.message}`);
      return false;
    }
    return true;
  }

  // `requests` is [{ rel, offset }]. Same fetched/failed shape as fetchFiles.
  async function fetchIncremental(alias, requests, destRoot) {
    const fetched = [];
    const failed = [];
    const queue = (Array.isArray(requests) ? requests : [])
      .filter(r => r && isSafeMirrorRelPath(r.rel) && Number.isInteger(r.offset) && r.offset >= 0);
    let cursor = 0;

    const workers = Array.from({ length: Math.min(concurrency, queue.length) }, async () => {
      while (!disposed) {
        const i = cursor++;
        if (i >= queue.length) return;
        const { rel, offset } = queue[i];
        if (await fetchOneIncremental(alias, rel, offset, destRoot)) fetched.push(rel);
        else failed.push(rel);
      }
    });
    await Promise.all(workers);
    for (let i = cursor; i < queue.length; i++) failed.push(queue[i].rel);
    return { fetched, failed };
  }

  // Kill what is running without ending the transport -- see
  // .ai/contexts/session-cache.md, "Remote hosts".
  function cancelInFlight() {
    for (const child of live) {
      try { child.kill('SIGKILL'); } catch {}
    }
    live.clear();
  }

  function dispose() {
    disposed = true;
    cancelInFlight();
  }

  return { listFiles, fetchFiles, fetchIncremental, cancelInFlight, dispose, liveCount: () => live.size };
}

module.exports = {
  createSshTransport,
  parseInventory,
  parseSessions,
  splitListOutput,
  transcriptSessionIds,
  LIST_COMMAND,
  ALIVE_MARKER_PREFIX,
  REMOTE_PROJECTS_REL,
  REMOTE_SESSIONS_REL,
  SESSIONS_MARKER,
  MAX_SESSION_DESCRIPTORS,
  MAX_SESSION_DESCRIPTOR_BYTES,
  MAX_RANGE_FETCH_BYTES,
};