remote-index.js

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

const fs = require('fs');
const {
  enabledHosts,
  normalizeRefreshMs,
  joinFolderKey,
  parseFolderKey,
  mirrorProjectsDirFor,
  manifestPathFor,
} = require('./remote-hosts');
const { syncMirror } = require('./remote-mirror');
const { encodeProjectPath } = require('./encode-project-path');
const { computeHostProfile } = require('./remote-host-profile');

const NOOP_LOG = { info() {}, warn() {}, error() {} };

// see .ai/contexts/session-cache.md ("Remote hosts backoff")
const MAX_BACKOFF_MS = 30 * 60 * 1000;

// see .ai/contexts/session-cache.md ("Remote hosts — capability tiers", the probe)
const PROBE_INTERVAL_MS = 6 * 60 * 60 * 1000;
const PROBE_RETRY_MS = 30 * 60 * 1000;

// see .ai/contexts/session-cache.md ("Remote hosts backoff")
function backoffDelayMs(failures, intervalMs) {
  if (failures <= 0) return 0;
  return Math.min(intervalMs * Math.pow(2, failures - 1), MAX_BACKOFF_MS);
}

// see .ai/contexts/session-cache.md ("Remote hosts — descriptor-only sessions")
function placeholderTitle(cwd) {
  if (typeof cwd !== 'string' || !cwd) return null;
  const trimmed = cwd.replace(/[\\/]+$/, '');
  const parts = trimmed.split(/[\\/]/);
  return parts[parts.length - 1] || cwd;
}

const MAX_WAITING_FOR_LENGTH = 64;

// see .ai/contexts/cli-session-state.md ("waiting")
function sanitizeWaitingFor(value) {
  if (typeof value !== 'string') return null;
  const text = value.trim();
  if (!text || text.length > MAX_WAITING_FOR_LENGTH) return null;
  for (let k = 0; k < text.length; k++) {
    const code = text.charCodeAt(k);
    if (code < 32 || code === 127) return null;
    if (code >= 0x200b && code <= 0x200f) return null;
    if (code >= 0x202a && code <= 0x202e) return null;
    if (code >= 0x2066 && code <= 0x2069) return null;
    if (code >= 0x80 && code <= 0x9f) return null;
    if (code === 0xad || code === 0x61c || code === 0x180e || code === 0xfeff) return null;
    if (code === 0x2028 || code === 0x2029) return null;
    if (code >= 0x2060 && code <= 0x2064) return null;
  }
  return text;
}

// see .ai/contexts/session-cache.md ("Remote hosts — descriptor-only sessions")
function buildPlaceholderSession(alias, descriptor) {
  const id = (typeof descriptor.sessionId === 'string' && descriptor.sessionId)
    ? descriptor.sessionId
    : `pid:${descriptor.pid}`;
  return {
    sessionId: id,
    remoteAlias: alias,
    projectPath: descriptor.cwd,
    folder: encodeProjectPath(descriptor.cwd),
    remoteDescriptorSeen: true,
    status: descriptor.status || null,
    statusUpdatedAt: descriptor.statusUpdatedAt || null,
    waitingFor: sanitizeWaitingFor(descriptor.waitingFor),
    modified: descriptor.statusUpdatedAt || descriptor.startedAt || null,
    messageCount: 0,
    summary: placeholderTitle(descriptor.cwd),
    placeholder: true,
  };
}

/**
 * Periodic mirror + index of every declared SSH host.
 * see .ai/contexts/session-cache.md ("Remote SSH hosts")
 *
 * ctx (everything reaching the outside world is injected):
 *   getHosts()                       -> raw host array from settings
 *   getRefreshMs()                   -> configured interval (floored at 60 s)
 *   dataDir                          -> where <dataDir>/remote/<alias>/ lives
 *   transport                        -> see remote-mirror.js
 *   scanFolders({projectsDir, folderPrefix, folders}) -> Promise
 *   listIndexedFolderKeys()          -> every folder key already in the cache
 *   dropFolder(folderKey)            -> remove a folder from cache + search
 *   setRemoteRoots(Map<alias,dir>)   -> tell the cache where each mirror lives
 *   notify()                         -> push a sidebar refresh
 *   timers                           -> { setInterval, clearInterval } (test seam)
 *   sync                             -> syncMirror override (test seam)
 *   now()                            -> current epoch ms (test seam, defaults to Date.now)
 */
function createRemoteIndexer(ctx) {
  const log = ctx.log || NOOP_LOG;
  const timers = ctx.timers || { setInterval, clearInterval };
  const sync = ctx.sync || syncMirror;
  const now = ctx.now || Date.now;

  let timer = null;
  let inFlight = false;
  let stopped = false;
  const remoteSessions = new Map(); // alias -> sessions array, from the same ssh cycle as the inventory
  const remoteSessionsAt = new Map(); // alias -> epoch ms of the last cycle that did not throw
  const hostBackoff = new Map(); // alias -> { failures, lastError, nextAttemptAt }
  const hostInFlight = new Set();
  const hostTools = new Map(); // alias -> { tools: { tmux, inotifywait } | null, nextProbeAt }

  function backoffState(alias) {
    let s = hostBackoff.get(alias);
    if (!s) {
      s = { failures: 0, lastError: null, lastErrorTransport: false, attempts: 0, nextAttemptAt: 0 };
      hostBackoff.set(alias, s);
    }
    return s;
  }

  function onHostSuccess(alias) {
    const state = backoffState(alias);
    if (state.failures > 0) {
      log.info(`[remote:${alias}] refresh recovered after ${state.failures} consecutive failure(s)`);
    }
    const errorChanged = state.lastError !== null;
    state.failures = 0;
    state.attempts += 1;
    state.lastError = null;
    state.lastErrorTransport = false;
    state.nextAttemptAt = 0;
    return errorChanged;
  }

  function onHostFailure(alias, thrown, intervalMs) {
    const err = thrown instanceof Error ? thrown : new Error(String(thrown));
    const state = backoffState(alias);
    const prevDelay = backoffDelayMs(state.failures, intervalMs);
    state.failures += 1;
    state.attempts += 1;
    const errorChanged = state.lastError !== err.message;
    state.lastError = err.message;
    state.lastErrorTransport = err.remoteTransport === true;
    const delay = backoffDelayMs(state.failures, intervalMs);
    state.nextAttemptAt = now() + delay;
    if (delay !== prevDelay) {
      log.warn(`[remote:${alias}] refresh failed (${state.failures}x consecutive): ${err.message}; ` +
        `retrying in ${Math.round(delay / 1000)}s`);
    }
    return errorChanged;
  }

  function getRemoteHostState(alias) {
    const s = hostBackoff.get(alias);
    if (!s) return { consecutiveFailures: 0, lastError: null, lastErrorTransport: false, attempts: 0, nextAttemptAt: 0 };
    return { consecutiveFailures: s.failures, lastError: s.lastError, lastErrorTransport: s.lastErrorTransport, attempts: s.attempts, nextAttemptAt: s.nextAttemptAt };
  }

  function hosts() {
    return enabledHosts(ctx.getHosts ? ctx.getHosts() : []);
  }

  function publishRoots(list) {
    if (!ctx.setRemoteRoots) return;
    const roots = new Map();
    for (const h of list) roots.set(h.alias, mirrorProjectsDirFor(ctx.dataDir, h.alias));
    ctx.setRemoteRoots(roots);
  }

  // Nothing else ever revisits a folder whose alias is no longer declared.
  async function pruneUnknownAliases(list) {
    if (!ctx.listIndexedFolderKeys || !ctx.dropFolder) return 0;
    const known = new Set(list.map(h => h.alias));
    let dropped = 0;
    for (const key of ctx.listIndexedFolderKeys()) {
      const { alias } = parseFolderKey(key);
      if (alias === null || known.has(alias)) continue;
      await ctx.dropFolder(key);
      dropped++;
    }
    for (const alias of [...remoteSessions.keys()]) {
      if (!known.has(alias)) remoteSessions.delete(alias);
    }
    for (const alias of [...remoteSessionsAt.keys()]) {
      if (!known.has(alias)) remoteSessionsAt.delete(alias);
    }
    for (const alias of [...hostTools.keys()]) {
      if (!known.has(alias)) hostTools.delete(alias);
    }
    for (const alias of [...hostBackoff.keys()]) {
      if (!known.has(alias)) hostBackoff.delete(alias);
    }
    return dropped;
  }

  function mirrorFolders(projectsDir) {
    try {
      return fs.readdirSync(projectsDir, { withFileTypes: true })
        .filter(d => d.isDirectory() && d.name !== '.git')
        .map(d => d.name);
    } catch {
      return [];
    }
  }

  function rearmProbe(alias) {
    const entry = hostTools.get(alias);
    if (entry) entry.nextProbeAt = 0;
  }

  async function probeHostTools(alias) {
    if (!ctx.transport || typeof ctx.transport.probeTools !== 'function') return false;
    const prev = hostTools.get(alias);
    if (prev && now() < prev.nextProbeAt) return false;
    let tools = null;
    try {
      tools = await ctx.transport.probeTools(alias);
    } catch (err) {
      log.warn(`[remote:${alias}] tool probe failed: ${err.message}`);
    }
    const kept = tools || (prev ? prev.tools : null);
    hostTools.set(alias, { tools: kept, nextProbeAt: now() + (kept && kept.tmux && kept.inotifywait && tools ? PROBE_INTERVAL_MS : PROBE_RETRY_MS) });
    return !!tools && (!prev || !prev.tools || prev.tools.tmux !== tools.tmux || prev.tools.inotifywait !== tools.inotifywait);
  }

  async function refreshHost(host) {
    const projectsDir = mirrorProjectsDirFor(ctx.dataDir, host.alias);
    const manifestPath = manifestPathFor(ctx.dataDir, host.alias);
    fs.mkdirSync(projectsDir, { recursive: true });

    const result = await sync({
      alias: host.alias,
      transport: ctx.transport,
      projectsDir,
      manifestPath,
      log,
    });

    remoteSessions.set(host.alias, Array.isArray(result.sessions)
      ? result.sessions.map(s => (s && 'waitingFor' in s ? { ...s, waitingFor: sanitizeWaitingFor(s.waitingFor) } : s))
      : []);

    const folderPrefix = host.alias;
    const toScan = new Set(result.changedFolders);
    const changedFilesByFolder = result.changedFilesByFolder instanceof Map
      ? result.changedFilesByFolder : new Map();

    // A mirror on disk but absent from the cache reports no change; index it once.
    const indexed = new Set();
    if (ctx.listIndexedFolderKeys) {
      for (const key of ctx.listIndexedFolderKeys()) {
        const parsed = parseFolderKey(key);
        if (parsed.alias === host.alias) indexed.add(parsed.folder);
      }
      const present = mirrorFolders(projectsDir);
      for (const folder of present) {
        if (!indexed.has(folder)) toScan.add(folder);
      }
      // And a folder the cache still knows about but the mirror no longer has.
      if (ctx.dropFolder) {
        const presentSet = new Set(present);
        for (const folder of indexed) {
            if (!presentSet.has(folder)) await ctx.dropFolder(joinFolderKey(host.alias, folder));
        }
      }
    }

    if (toScan.size > 0 && ctx.scanFolders) {
      // see .ai/contexts/session-cache.md ("Remote hosts file-level rescan")
      const fileSubsets = new Map();
      for (const folder of toScan) {
        if (!indexed.has(folder)) continue;
        const files = changedFilesByFolder.get(folder);
        if (files && files.size > 0) fileSubsets.set(folder, files);
      }
      await ctx.scanFolders({
        projectsDir,
        folderPrefix,
        folders: [...toScan],
        ...(fileSubsets.size > 0 ? { fileSubsets } : {}),
      });
    }

    log.info(`[remote:${host.alias}] ${result.fetched} fetched, ${result.unchanged} unchanged, ` +
      `${result.removed} removed, ${result.failed} failed, ${toScan.size} folders indexed`);

    const toolsChanged = await probeHostTools(host.alias);
    return toScan.size > 0 || toolsChanged;
  }

  // see .ai/contexts/session-cache.md ("Remote hosts backoff" — manual reconnect, issue #252)
  async function refreshNow({ force = false } = {}) {
    if (stopped || inFlight) return { skipped: true };
    const list = hosts();
    publishRoots(list);
    if (list.length === 0) return { skipped: true, hosts: 0 };

    inFlight = true;
    let changed = false;
    const errors = [];
    const intervalMs = normalizeRefreshMs(ctx.getRefreshMs ? ctx.getRefreshMs() : undefined);
    try {
      changed = await pruneUnknownAliases(list) > 0;
      for (const host of list) {
        if (stopped) break;
        if (hostInFlight.has(host.alias)) continue;
        const state = backoffState(host.alias);
        if (force) {
          state.failures = 0;
          state.nextAttemptAt = 0;
          rearmProbe(host.alias);
        } else if (now() < state.nextAttemptAt) {
          continue; // still backing off: no attempt, no log, no ssh
        }
        try {
          if (await refreshHost(host)) changed = true;
          if (onHostSuccess(host.alias)) changed = true;
          remoteSessionsAt.set(host.alias, now());
        } catch (err) {
          // A failed cycle keeps the last known descriptors — see
          // .ai/contexts/session-cache.md ("Remote hosts — freshness contract").
          errors.push({ alias: host.alias, error: err instanceof Error ? err.message : String(err) });
          if (onHostFailure(host.alias, err, intervalMs)) changed = true;
        }
      }
    } finally {
      inFlight = false;
    }

    if (changed && ctx.notify) ctx.notify();
    return { skipped: false, hosts: list.length, changed, errors };
  }

  // see .ai/contexts/session-cache.md ("Remote hosts — watch channel" and,
  // for `force`, "Remote hosts backoff" — manual reconnect, issue #252)
  async function refreshHostNow(alias, { force = false } = {}) {
    if (stopped) return { skipped: true, reason: 'stopped' };
    if (inFlight || hostInFlight.has(alias)) return { skipped: true, reason: 'in-flight' };
    const host = hosts().find(h => h.alias === alias);
    if (!host) return { skipped: true, reason: 'unknown-host' };
    const state = backoffState(alias);
    if (force) {
      state.failures = 0;
      state.nextAttemptAt = 0;
      rearmProbe(alias);
    } else if (now() < state.nextAttemptAt) {
      return { skipped: true, reason: 'backoff' };
    }
    const intervalMs = normalizeRefreshMs(ctx.getRefreshMs ? ctx.getRefreshMs() : undefined);
    hostInFlight.add(alias);
    let changed = false;
    let error = null;
    try {
      changed = await refreshHost(host);
      if (onHostSuccess(alias)) changed = true;
      remoteSessionsAt.set(alias, now());
    } catch (err) {
      // A failed cycle keeps the last known descriptors — see
      // .ai/contexts/session-cache.md ("Remote hosts — freshness contract").
      if (onHostFailure(alias, err, intervalMs)) changed = true;
      error = err instanceof Error ? err.message : String(err);
    } finally {
      hostInFlight.delete(alias);
    }
    if (changed && ctx.notify) ctx.notify();
    return { skipped: false, changed, error };
  }

  function start() {
    stopped = false;
    const list = hosts();
    publishRoots(list);
    // No host declared: no timer, and the transport is never touched.
    if (list.length === 0) return false;
    pruneUnknownAliases(list).catch(err => log.warn('[remote] alias prune failed:', err.message));
    const intervalMs = normalizeRefreshMs(ctx.getRefreshMs ? ctx.getRefreshMs() : undefined);
    timer = timers.setInterval(() => { refreshNow().catch(() => {}); }, intervalMs);
    if (timer && typeof timer.unref === 'function') timer.unref();
    refreshNow().catch(() => {});
    return true;
  }

  function stop() {
    stopped = true;
    if (timer) {
      timers.clearInterval(timer);
      timer = null;
    }
    if (ctx.transport && typeof ctx.transport.cancelInFlight === 'function') ctx.transport.cancelInFlight();
  }

  // Terminal: for application shutdown only, never for restart().
  function dispose() {
    stop();
    if (ctx.transport && typeof ctx.transport.dispose === 'function') ctx.transport.dispose();
  }

  function restart() {
    stop();
    return start();
  }

  // Freshness contract (issue #212) — see .ai/contexts/session-cache.md ("Remote hosts — freshness contract")
  function getRemoteSessions(alias) {
    const backoff = hostBackoff.get(alias);
    return {
      sessions: remoteSessions.get(alias) || [],
      at: remoteSessionsAt.get(alias) || null,
      error: backoff ? backoff.lastError : null,
    };
  }

  // see .ai/contexts/session-cache.md ("Remote hosts — capability tiers")
  function getRemoteHostProfile(alias) {
    const { sessions, at, error } = getRemoteSessions(alias);
    const known = hostTools.get(alias);
    return computeHostProfile({ at, error, descriptors: sessions, tools: known ? known.tools : null });
  }

  // see .ai/contexts/session-cache.md ("Remote hosts — descriptor-only sessions")
  function getPlaceholderSessions(alias) {
    const list = remoteSessions.get(alias) || [];
    const out = [];
    for (const descriptor of list) {
      if (!descriptor || !descriptor.descriptorOnly) continue;
      if (typeof descriptor.cwd !== 'string' || !descriptor.cwd) continue;
      out.push(buildPlaceholderSession(alias, descriptor));
    }
    return out;
  }

  function getAllPlaceholderSessions() {
    const out = [];
    for (const alias of remoteSessions.keys()) out.push(...getPlaceholderSessions(alias));
    return out;
  }

  // see .ai/contexts/session-cache.md ("Remote hosts — descriptor-only sessions")
  function findSessionAlias(sessionId) {
    for (const [alias, list] of remoteSessions) {
      if (list.some(s => s && s.sessionId === sessionId)) return alias;
    }
    return null;
  }

  function findSessionAliases(sessionId, isEnabled) {
    const aliases = [];
    for (const [alias, list] of remoteSessions) {
      if (isEnabled(alias) && list.some(s => s && s.sessionId === sessionId)) aliases.push(alias);
    }
    return aliases;
  }

  // see .ai/contexts/session-state.md ("The two lifecycle verbs: detach and stop")
  function dropRemoteSession(alias, sessionId) {
    const list = remoteSessions.get(alias);
    if (!list || list.length === 0) return false;
    const next = list.filter(s => s.sessionId !== sessionId);
    if (next.length === list.length) return false;
    remoteSessions.set(alias, next);
    return true;
  }

  return {
    start, stop, dispose, restart, refreshNow, refreshHostNow,
    isRunning: () => timer !== null,
    getRemoteSessions,
    getRemoteHostProfile,
    getPlaceholderSessions,
    getAllPlaceholderSessions,
    findSessionAlias,
    findSessionAliases,
    dropRemoteSession,
    getRemoteHostState,
  };
}

module.exports = { createRemoteIndexer, PROBE_INTERVAL_MS, PROBE_RETRY_MS, backoffDelayMs, buildPlaceholderSession, placeholderTitle, sanitizeWaitingFor };