"use strict";

// Central SFTP Transfer Manager (Main process). Every file is one job with its own SFTP channel,
// a bounded pipeline of positioned reads/writes (backpressure: at most PIPELINE chunks in flight)
// and a confirmed offset that only advances over contiguous acknowledged chunks. Pause and cancel
// stop issuing requests on the channel itself; resume reopens the partial file, verifies the tail
// against the source and continues from that offset. Completed uploads are published by the shared
// remote helper (verified hash, original metadata, atomic rename, privileged when needed); completed
// downloads are hashed locally and renamed into place. Job state (never secrets) survives restarts.
const crypto = require("node:crypto");
const fs = require("node:fs");
const path = require("node:path");
const { EventEmitter } = require("node:events");
const { atomicWriteJson } = require("../storage/atomicFile");
const { fileError, normalizeRemotePath, safeEntryName } = require("./remoteFileService");

const STATES = Object.freeze(["waiting", "connecting", "transferring", "paused", "interrupted", "reconnecting", "verifying", "finalizing", "completed", "failed", "cancelled"]);
const RUNNING = new Set(["connecting", "transferring", "verifying", "finalizing"]);
const FINAL = new Set(["completed", "failed", "cancelled"]);
const CHUNK = 256 * 1024;
const PIPELINE = 8;
const TAIL_CHECK = 1024 * 1024;
const STALL_MS = 60000;
const STOP_GRACE_MS = 5000;
const EMIT_MS = 250;
const PERSIST_MS = 1000;
const RETRY_DELAYS_MS = [2000, 4000, 8000, 16000, 32000];
const MAX_TOTAL_RETRIES = 50;
const MAX_HISTORY = 500;
const MAX_JOBS = 20000;
const MAX_PLAN_ENTRIES = 10000;
const MAX_PLAN_BYTES = 4 * 1024 * 1024 * 1024 * 1024;
const PLAN_TTL_MS = 10 * 60 * 1000;
const ORPHAN_AGE_MS = 24 * 3600 * 1000;
const HASH_RATE = 10 * 1024 * 1024;
const DEFAULT_SETTINGS = Object.freeze({ maxConcurrent: 2, autoRetry: true, verify: true });
const CONNECTION_CODES = new Set(["SESSION_NOT_CONNECTED", "SFTP_OPEN_FAILED", "TRANSFER_STALLED", "REMOTE_COMMAND_TIMEOUT", "REMOTE_COMMAND_FAILED",
  "FILE_EXEC_FAILED", "ECONNRESET", "ETIMEDOUT", "EPIPE"]);
const NO_HELPER_CODES = new Set(["FILE_DEPENDENCY_UNAVAILABLE", "FEATURE_PLATFORM_UNSUPPORTED"]);
const AUTH_CODES = new Set(["SUDO_PASSWORD_REQUIRED", "SUDO_PASSWORD_INCORRECT", "FILE_PERMISSION_DENIED", "FILE_METADATA_PERMISSION"]);

class StopSignal extends Error {
  constructor(kind) { super(`Transfer ${kind}`); this.stop = kind; }
}

function hex(bytes = 16) { return crypto.randomBytes(bytes).toString("hex"); }
function call(sftp, method, ...args) {
  return new Promise((resolve, reject) => sftp[method](...args, (error, ...results) => error ? reject(error) : resolve(results.length > 1 ? results : results[0])));
}
function errorCode(error) {
  if (error?.code === 6 || error?.code === 7) return "SESSION_NOT_CONNECTED";
  if (typeof error?.code === "string") return error.code;
  if (/no response|not connected|connection (lost|closed)|channel|ECONNRESET|socket/i.test(String(error?.message || ""))) return "SESSION_NOT_CONNECTED";
  if (error?.code === 3) return "FILE_PERMISSION_DENIED";
  if (error?.code === 2) return "FILE_NOT_FOUND";
  return "TRANSFER_FAILED";
}
function localIdentity(stats) { return { size: stats.size, mtimeMs: Math.trunc(stats.mtimeMs), ino: Number(stats.ino) || 0, dev: Number(stats.dev) || 0 }; }
function sameIdentity(left, right) { return Boolean(left && right) && ["size", "mtimeMs", "ino", "dev"].every((key) => left[key] === right[key]); }
function percent(transferred, size) {
  if (!Number.isFinite(size) || size < 0) return null;
  if (size === 0) return null;
  return Math.min(100, Math.floor((Math.min(transferred, size) / size) * 1000) / 10);
}
// A dead channel never answers: cleanup calls must not block the job forever.
function within(promise, ms = 3000) {
  let timer;
  return Promise.race([Promise.resolve(promise).catch(() => {}), new Promise((resolve) => { timer = setTimeout(resolve, ms); timer.unref?.(); })]).finally(() => clearTimeout(timer));
}
function partName(name, id) { return `.${String(name).slice(0, 120)}.chj-part-${id}`; }

// Aggregate view: percentages are byte based; an unknown size yields an indeterminate total.
function summarize(jobs) {
  const counted = jobs.filter((job) => job.state !== "cancelled");
  const sizesKnown = counted.every((job) => Number.isFinite(job.size) && job.size >= 0);
  const totalBytes = counted.reduce((sum, job) => sum + (Number.isFinite(job.size) ? job.size : 0), 0);
  const transferredBytes = counted.reduce((sum, job) => sum + Math.min(job.transferred || 0, Number.isFinite(job.size) ? job.size : job.transferred || 0), 0);
  const completed = counted.filter((job) => job.state === "completed").length;
  const active = jobs.filter((job) => RUNNING.has(job.state));
  const speed = active.reduce((sum, job) => sum + (job.speed || 0), 0);
  const remaining = counted.filter((job) => !FINAL.has(job.state)).reduce((sum, job) => sum + Math.max(0, (job.size || 0) - (job.transferred || 0)), 0);
  const etaReliable = active.length > 0 && active.every((job) => job.state !== "transferring" || job.eta !== null);
  let overall = null;
  if (sizesKnown && totalBytes > 0) overall = Math.floor((transferredBytes / totalBytes) * 1000) / 10;
  else if (sizesKnown && counted.length && completed === counted.length) overall = 100;
  if (overall !== null && overall >= 100 && completed !== counted.length) overall = 99.9;
  return {
    total: jobs.length, completed, active: active.length,
    waiting: jobs.filter((job) => job.state === "waiting" || job.state === "reconnecting").length,
    paused: jobs.filter((job) => job.state === "paused").length,
    interrupted: jobs.filter((job) => job.state === "interrupted").length,
    failed: jobs.filter((job) => job.state === "failed").length,
    cancelled: jobs.length - counted.length,
    totalBytes, transferredBytes, sizesKnown, percent: overall, speed,
    eta: speed > 0 && etaReliable && remaining > 0 ? Math.ceil(remaining / speed) : remaining === 0 && completed === counted.length ? 0 : null
  };
}

class TransferManager extends EventEmitter {
  constructor(options) {
    super();
    this.sessionManager = options.sessionManager;
    this.files = options.remoteFileService;
    this.selectUploadPath = options.selectUploadPath || options.remoteFileService?.selectUploadPath;
    this.selectDownloadPath = options.selectDownloadPath || options.remoteFileService?.selectDownloadPath;
    this.selectDownloadDirectory = options.selectDownloadDirectory || options.remoteFileService?.selectDownloadDirectory;
    this.logger = options.logger;
    this.now = options.now || (() => Date.now());
    this.stopGraceMs = Number.isFinite(options.stopGraceMs) ? options.stopGraceMs : STOP_GRACE_MS;
    this.retryDelays = Array.isArray(options.retryDelaysMs) && options.retryDelaysMs.length ? options.retryDelaysMs : RETRY_DELAYS_MS;
    this.file = options.storageRoot ? path.join(options.storageRoot, "transfers.json") : null;
    this.settings = { ...DEFAULT_SETTINGS };
    this.jobs = new Map();
    this.runtime = new Map();
    this.secrets = new Map(); // jobId -> sudo password, memory only
    this.plans = new Map();
    this.changed = new Set();
    this.cleanedIdentities = new Set();
    this.order = 0;
    this._load();
    this._onState = (status) => this._sessionState(status);
    this.sessionManager?.on?.("state", this._onState);
  }

  // ---- public API (owner = plugin id) ---------------------------------------------------------

  list(owner) {
    const jobs = this._owned(owner).map((job) => this._public(job));
    this._cleanupOrphansLater();
    return { jobs, summary: summarize(jobs), settings: { ...this.settings } };
  }

  updateSettings(_owner, patch = {}) {
    if (patch.maxConcurrent !== undefined) {
      const value = Number(patch.maxConcurrent);
      if (!Number.isInteger(value) || value < 1 || value > 8) throw fileError("TRANSFER_INVALID_SETTINGS", "Concurrent transfers must be between 1 and 8.");
      this.settings.maxConcurrent = value;
    }
    if (patch.autoRetry !== undefined) this.settings.autoRetry = patch.autoRetry === true;
    if (patch.verify !== undefined) this.settings.verify = patch.verify === true;
    this._persist(true);
    this._schedule();
    return { ...this.settings };
  }

  async planUpload(owner, sessionId, remoteDirectory, options = {}) {
    const directory = normalizeRemotePath(remoteDirectory || "/");
    const identity = this._identity(sessionId);
    if (typeof this.selectUploadPath !== "function") throw fileError("TRANSFER_DIALOG_UNAVAILABLE", "Upload dialog is unavailable.");
    const selected = await this.selectUploadPath();
    const localPaths = selected && !selected.canceled ? [...new Set((selected.paths || (selected.path ? [selected.path] : [])).map(String).filter(Boolean))] : [];
    if (!localPaths.length) return { canceled: true };
    const plan = { id: hex(), owner, direction: "upload", sessionId, identity, directory, files: [], directories: [], skipped: [], bytes: 0, createdAt: this.now() };
    for (const localPath of localPaths) await this._walkLocal(plan, localPath, path.posix.join(directory, safeEntryName(path.basename(localPath))));
    await this.files._withSftp(sessionId, async (sftp) => {
      plan.mode = await this._directoryMode(sftp, sessionId, directory);
      await this._remoteConflicts(sftp, plan);
    });
    if (plan.mode === "privileged") plan.privilege = await this.files.privileges(sessionId);
    if (plan.mode === "privileged" && plan.privilege.mode === "none") {
      throw fileError(plan.privilege.sudoInstalled ? "SUDO_NOT_PERMITTED" : "SUDO_UNAVAILABLE", "The destination is not writable and the SSH account cannot use sudo.");
    }
    return this._storePlan(plan);
  }

  async planDownload(owner, sessionId, remotePaths, options = {}) {
    const identity = this._identity(sessionId);
    const targets = [...new Set((Array.isArray(remotePaths) ? remotePaths : [remotePaths]).filter((value) => value !== undefined && value !== null).map(normalizeRemotePath))];
    if (!targets.length || targets.length > 100) throw fileError("TRANSFER_INVALID_SELECTION", "Select between 1 and 100 remote entries.");
    if (targets.includes("/")) throw fileError("TRANSFER_INVALID_SELECTION", "Downloading the entire remote root is not allowed in one batch.");
    const plan = { id: hex(), owner, direction: "download", sessionId, identity, files: [], directories: [], skipped: [], bytes: 0, createdAt: this.now(),
      mode: options.sudo === true ? "privileged-allowed" : "direct" };
    await this.files._withSftp(sessionId, async (sftp) => {
      const first = targets.length === 1 ? await this._remoteStat(sftp, targets[0], options, sessionId) : null;
      if (first && first.type === "file") {
        if (typeof this.selectDownloadPath !== "function") throw fileError("TRANSFER_DIALOG_UNAVAILABLE", "Download dialog is unavailable.");
        const chosen = await this.selectDownloadPath(path.posix.basename(targets[0]));
        if (!chosen || chosen.canceled || !chosen.path) { plan.canceled = true; return; }
        plan.localDirectory = path.dirname(chosen.path);
        this._addDownload(plan, targets[0], chosen.path, first);
        plan.files[0].overwriteConfirmed = true; // The native save dialog already confirmed replacing.
        return;
      }
      if (typeof this.selectDownloadDirectory !== "function") throw fileError("TRANSFER_DIALOG_UNAVAILABLE", "Download directory dialog is unavailable.");
      const chosen = await this.selectDownloadDirectory();
      if (!chosen || chosen.canceled || !chosen.path) { plan.canceled = true; return; }
      plan.localDirectory = chosen.path;
      for (const target of targets) await this._walkRemote(sftp, sessionId, plan, target, path.join(chosen.path, safeEntryName(path.posix.basename(target))), options, true);
    });
    if (plan.canceled) return { canceled: true };
    for (const file of plan.files) {
      if (!file.overwriteConfirmed && await fs.promises.lstat(file.localPath).then(() => true, () => false)) file.conflict = true;
    }
    return this._storePlan(plan);
  }

  async start(owner, planId, options = {}) {
    const plan = this.plans.get(String(planId || ""));
    if (!plan || plan.owner !== owner || this.now() - plan.createdAt > PLAN_TTL_MS) throw fileError("TRANSFER_PLAN_EXPIRED", "The transfer plan expired; select the files again.");
    const conflict = ["overwrite", "skip"].includes(options.conflict) ? options.conflict : "fail";
    const conflicts = plan.files.filter((file) => file.conflict);
    if (conflicts.length && conflict === "fail") throw fileError("FILE_EXISTS", `${conflicts.length} destination file(s) already exist.`);
    if (this.jobs.size + plan.files.length > MAX_JOBS) throw fileError("TRANSFER_QUEUE_FULL", "Remove finished transfers before starting more.");
    const password = options.sudoPassword ? String(options.sudoPassword) : "";
    const privileged = plan.mode === "privileged" || (plan.direction === "download" && (plan.mode === "privileged-allowed" || options.sudo === true));
    if (plan.mode === "privileged") {
      // Verifies the sudo authorization once before any data moves (and caches it on request).
      await this.files.helper(plan.sessionId, { operation: "stat", path: plan.directory }, { sudo: true, sudoPassword: password, rememberSudo: options.rememberSudo === true });
    }
    await this._createDirectories(plan, privileged, password);
    const created = [];
    for (const file of plan.files) {
      if (file.conflict && conflict === "skip") continue;
      const job = {
        id: hex(), owner, batchId: plan.id, direction: plan.direction, identity: plan.identity, sessionId: plan.sessionId, order: ++this.order,
        name: file.name, localPath: file.localPath, remotePath: file.remotePath, size: file.size, transferred: 0, movedBytes: 0, activeMs: 0,
        state: "waiting", mode: plan.direction === "upload" ? plan.mode : privileged ? "privileged-allowed" : "direct",
        overwrite: conflict === "overwrite" || file.overwriteConfirmed === true, localIdentity: file.localIdentity || null, remoteIdentity: file.remoteIdentity || null, remoteMode: file.mode || null,
        finalize: { state: "pending" }, attempts: 0, totalRetries: 0, error: null, createdAt: this.now(), updatedAt: this.now()
      };
      this.jobs.set(job.id, job);
      if (password && privileged) this.secrets.set(job.id, password);
      created.push(job.id);
      this._mark(job);
    }
    this.plans.delete(plan.id);
    this._trimHistory();
    this._persist(true);
    this.logger?.info("Transfer batch queued.", { direction: plan.direction, files: created.length, bytes: plan.bytes, mode: plan.mode, sessionId: plan.sessionId });
    this._schedule();
    return { batchId: plan.id, jobs: created.length, skipped: plan.files.length - created.length };
  }

  pause(owner, ids) {
    let count = 0;
    for (const job of this._select(owner, ids)) {
      if (job.state === "waiting" || job.state === "reconnecting" || job.state === "interrupted") {
        this._clearRetry(job);
        this._set(job, "paused");
        count += 1;
      } else if (RUNNING.has(job.state) && job.state !== "finalizing") {
        this._stop(job, "pause");
        count += 1;
      }
    }
    this._persist(true);
    return { paused: count };
  }

  resume(owner, ids, options = {}) {
    let count = 0;
    for (const job of this._select(owner, ids)) {
      const resumable = ["paused", "interrupted", "reconnecting"].includes(job.state) || (job.state === "failed" && job.resumable);
      if (!resumable) continue;
      this._authorize(job, options);
      if (options.overwrite === true && ["FILE_EXISTS", "FILE_CONFLICT"].includes(job.error?.code)) {
        job.overwrite = true;
        job.targetBaseline = undefined;
        job.finalize = { state: "pending" };
      }
      this._clearRetry(job);
      job.attempts = 0;
      job.authRequired = false;
      this._set(job, "waiting");
      count += 1;
    }
    this._persist(true);
    this._schedule();
    return { resumed: count };
  }

  // Restart from offset zero (used when the partial data cannot be safely continued).
  async retry(owner, ids, options = {}) {
    let count = 0;
    for (const job of this._select(owner, ids)) {
      if (!["failed", "interrupted", "paused", "cancelled"].includes(job.state)) continue;
      // The deterministic partial file (same job id) is truncated and rewritten from offset zero.
      if (job.localPart) await fs.promises.unlink(job.localPart).catch(() => {});
      this._authorize(job, options);
      Object.assign(job, { transferred: 0, staged: null, targetBaseline: undefined, hash: null, verification: null, finalize: { state: "pending" },
        attempts: 0, authRequired: false, error: null, resumable: false, finalizeWithSudo: false });
      if (job.direction === "upload") {
        const stats = await fs.promises.stat(job.localPath).catch(() => null);
        if (stats) { job.localIdentity = localIdentity(stats); job.size = stats.size; }
      } else job.remoteIdentity = null;
      if (options.overwrite === true) job.overwrite = true;
      this._set(job, "waiting");
      count += 1;
    }
    this._persist(true);
    this._schedule();
    return { retried: count };
  }

  async cancel(owner, ids) {
    let count = 0;
    for (const job of this._select(owner, ids)) {
      if (FINAL.has(job.state)) continue;
      count += 1;
      if (RUNNING.has(job.state)) {
        if (job.state === "finalizing") continue; // The remote replacement must finish or fail on its own.
        this._stop(job, "cancel");
        continue;
      }
      this._clearRetry(job);
      await this._finishCancel(job);
    }
    this._persist(true);
    return { cancelled: count };
  }

  remove(owner, ids) {
    let count = 0;
    for (const job of this._select(owner, ids)) {
      if (!FINAL.has(job.state)) continue;
      if (job.cleanupPending) continue; // Kept until its temporary data is removed.
      this.jobs.delete(job.id);
      this.secrets.delete(job.id);
      this.runtime.delete(job.id);
      this.changed.add(job.id);
      count += 1;
    }
    this._persist(true);
    this._emitSoon(owner, true);
    return { removed: count };
  }

  shutdown() {
    if (this.closed) return;
    clearTimeout(this.emitTimer);
    this.sessionManager?.removeListener?.("state", this._onState);
    for (const job of this.jobs.values()) {
      this._clearRetry(job);
      if (RUNNING.has(job.state)) this._stop(job, "shutdown");
    }
    this._persist(true, true);
    this.closed = true; // Late completions of stopped jobs must not overwrite the saved state.
    this.secrets.clear();
  }

  // ---- planning helpers -----------------------------------------------------------------------

  _storePlan(plan) {
    if (plan.files.length + plan.directories.length > MAX_PLAN_ENTRIES) throw fileError("TRANSFER_TOO_MANY_ENTRIES", `A transfer can contain at most ${MAX_PLAN_ENTRIES} entries.`);
    if (plan.bytes > MAX_PLAN_BYTES) throw fileError("TRANSFER_TOO_LARGE", "The selection exceeds the transfer size limit.");
    for (const [id, other] of this.plans) if (this.now() - other.createdAt > PLAN_TTL_MS) this.plans.delete(id);
    this.plans.set(plan.id, plan);
    const conflicts = plan.files.filter((file) => file.conflict);
    return {
      canceled: false, planId: plan.id, direction: plan.direction, files: plan.files.length, directories: plan.directories.length, bytes: plan.bytes,
      mode: plan.mode, requiresSudo: plan.mode === "privileged" && plan.privilege?.mode !== "root",
      sudoPasswordRequired: plan.mode === "privileged" && plan.privilege?.mode === "sudo-password" && !plan.privilege?.sudoCached,
      conflicts: conflicts.slice(0, 50).map((file) => file.direction === "upload" ? file.remotePath : file.localPath), conflictCount: conflicts.length,
      skipped: plan.skipped.slice(0, 50), skippedCount: plan.skipped.length, destination: plan.direction === "upload" ? plan.directory : plan.localDirectory
    };
  }

  async _walkLocal(plan, localPath, remotePath, depth = 0) {
    if (plan.files.length + plan.directories.length > MAX_PLAN_ENTRIES) throw fileError("TRANSFER_TOO_MANY_ENTRIES", `A transfer can contain at most ${MAX_PLAN_ENTRIES} entries.`);
    const stats = await fs.promises.lstat(localPath);
    if (stats.isSymbolicLink()) { plan.skipped.push({ path: localPath, reason: "symlink" }); return; }
    if (stats.isDirectory()) {
      plan.directories.push(remotePath);
      if (depth > 64) throw fileError("TRANSFER_TOO_DEEP", "The selected directory tree is too deep.");
      for (const entry of await fs.promises.readdir(localPath, { withFileTypes: true })) {
        let name;
        try { name = safeEntryName(entry.name); } catch { plan.skipped.push({ path: path.join(localPath, entry.name), reason: "name" }); continue; }
        await this._walkLocal(plan, path.join(localPath, name), path.posix.join(remotePath, name), depth + 1);
      }
      return;
    }
    if (!stats.isFile()) { plan.skipped.push({ path: localPath, reason: "type" }); return; }
    plan.bytes += stats.size;
    plan.files.push({ direction: "upload", name: path.basename(localPath), localPath, remotePath, size: stats.size, localIdentity: localIdentity(stats) });
  }

  // Writable by the SSH account: direct SFTP; otherwise the privileged staging route.
  async _directoryMode(sftp, sessionId, directory) {
    const attrs = await call(sftp, "stat", directory).catch((error) => { throw fileError(errorCode(error) === "FILE_NOT_FOUND" ? "FILE_NOT_FOUND" : errorCode(error), `${directory} is not accessible.`); });
    if (typeof attrs?.isDirectory === "function" && !attrs.isDirectory()) throw fileError("FILE_NOT_DIRECTORY", `${directory} is not a directory.`);
    const probe = path.posix.join(directory, `.chj-probe-${hex(6)}`);
    try {
      const handle = await call(sftp, "open", probe, "wx", { mode: 0o600 });
      await call(sftp, "close", handle).catch(() => {});
      await call(sftp, "unlink", probe).catch(() => {});
      return "direct";
    } catch (error) {
      if (errorCode(error) === "FILE_PERMISSION_DENIED") return "privileged";
      throw fileError(errorCode(error) === "TRANSFER_FAILED" ? "FILE_REMOTE_IO_FAILED" : errorCode(error), `${directory} is not writable.`);
    }
  }

  async _remoteConflicts(sftp, plan) {
    const queue = [...plan.files];
    const worker = async () => {
      for (let file = queue.shift(); file; file = queue.shift()) {
        const attrs = await call(sftp, "lstat", file.remotePath).catch(() => null);
        if (!attrs) continue;
        if (typeof attrs.isDirectory === "function" && attrs.isDirectory()) { file.blocked = true; plan.skipped.push({ path: file.remotePath, reason: "directory-exists" }); }
        else file.conflict = true;
      }
    };
    await Promise.all(Array.from({ length: 16 }, worker));
    plan.files = plan.files.filter((file) => !file.blocked);
  }

  async _remoteStat(sftp, target, options, sessionId) {
    try {
      const attrs = await call(sftp, "stat", target);
      return { type: typeof attrs.isDirectory === "function" && attrs.isDirectory() ? "directory" : "file", size: Number(attrs.size) || 0, mtime: Number(attrs.mtime) || 0, mode: Number(attrs.mode) || 0 };
    } catch (error) {
      if (errorCode(error) !== "FILE_PERMISSION_DENIED" || options.sudo !== true) throw fileError(errorCode(error), `${target} is not accessible.`);
      const value = await this.files.helper(sessionId, { operation: "stat", path: target }, { ...options, sudo: true });
      if (!value.exists) throw fileError("FILE_NOT_FOUND", `${target} does not exist.`);
      return { type: value.type, size: value.baseline.size, mtime: Math.trunc(Number(value.baseline.mtimeNs) / 1e9), mode: value.baseline.mode };
    }
  }

  _addDownload(plan, remotePath, localPath, info) {
    plan.bytes += info.size;
    plan.files.push({ direction: "download", name: path.posix.basename(remotePath), remotePath, localPath, size: info.size,
      remoteIdentity: { size: info.size, mtime: info.mtime }, mode: info.mode & 0o777 });
  }

  async _walkRemote(sftp, sessionId, plan, remotePath, localPath, options, top = false, depth = 0) {
    if (plan.files.length + plan.directories.length > MAX_PLAN_ENTRIES) throw fileError("TRANSFER_TOO_MANY_ENTRIES", `A transfer can contain at most ${MAX_PLAN_ENTRIES} entries.`);
    let info;
    if (top) info = await this._remoteStat(sftp, remotePath, options, sessionId);
    else {
      const attrs = await call(sftp, "lstat", remotePath).catch((error) => { throw fileError(errorCode(error), `${remotePath} is not accessible.`); });
      if (typeof attrs.isSymbolicLink === "function" && attrs.isSymbolicLink()) { plan.skipped.push({ path: remotePath, reason: "symlink" }); return; }
      info = { type: typeof attrs.isDirectory === "function" && attrs.isDirectory() ? "directory" : typeof attrs.isFile === "function" && !attrs.isFile() ? "other" : "file",
        size: Number(attrs.size) || 0, mtime: Number(attrs.mtime) || 0, mode: Number(attrs.mode) || 0 };
    }
    if (info.type === "other") { plan.skipped.push({ path: remotePath, reason: "type" }); return; }
    if (info.type !== "directory") { this._addDownload(plan, remotePath, localPath, info); return; }
    if (depth > 64) throw fileError("TRANSFER_TOO_DEEP", "The selected directory tree is too deep.");
    plan.directories.push(localPath);
    let names;
    try { names = (await call(sftp, "readdir", remotePath) || []).map((entry) => String(entry.filename)); }
    catch (error) {
      if (errorCode(error) !== "FILE_PERMISSION_DENIED" || options.sudo !== true) throw fileError(errorCode(error), `Directory ${remotePath} cannot be listed.`);
      const listed = await this.files.helper(sessionId, { operation: "listdir", path: remotePath }, { ...options, sudo: true });
      for (const entry of listed.entries || []) {
        let name;
        try { name = safeEntryName(entry.name); } catch { plan.skipped.push({ path: path.posix.join(remotePath, String(entry.name)), reason: "name" }); continue; }
        const child = path.posix.join(remotePath, name);
        if (entry.type === "symlink" || entry.type === "other") plan.skipped.push({ path: child, reason: entry.type });
        else if (entry.type === "directory") await this._walkRemote(sftp, sessionId, plan, child, path.join(localPath, name), options, true, depth + 1);
        else this._addDownload(plan, child, path.join(localPath, name), { size: entry.size, mtime: entry.mtime, mode: entry.mode });
      }
      return;
    }
    for (const raw of names) {
      if (raw === "." || raw === "..") continue;
      let name;
      try { name = safeEntryName(raw); } catch { plan.skipped.push({ path: path.posix.join(remotePath, raw), reason: "name" }); continue; }
      await this._walkRemote(sftp, sessionId, plan, path.posix.join(remotePath, name), path.join(localPath, name), options, false, depth + 1);
    }
  }

  async _createDirectories(plan, privileged, password) {
    if (plan.direction === "download") {
      for (const directory of plan.directories) await fs.promises.mkdir(directory, { recursive: true, mode: 0o700 });
      return;
    }
    if (!plan.directories.length) return;
    if (privileged) {
      for (const directory of plan.directories) {
        await this.files.helper(plan.sessionId, { operation: "mkdir", path: directory, parents: true }, { sudo: true, sudoPassword: password })
          .catch((error) => { if (error.code !== "FILE_EXISTS") throw error; });
      }
      return;
    }
    await this.files._withSftp(plan.sessionId, async (sftp) => {
      for (const directory of plan.directories) {
        try { await call(sftp, "mkdir", directory, { mode: 0o755 }); }
        catch (error) {
          const existing = await call(sftp, "stat", directory).catch(() => null);
          if (!existing || typeof existing.isDirectory !== "function" || !existing.isDirectory()) throw fileError(errorCode(error), `Directory ${directory} could not be created.`);
        }
      }
    });
  }

  // ---- scheduling ----------------------------------------------------------------------------

  _schedule() {
    if (this.scheduling || this.closed) return;
    this.scheduling = true;
    setImmediate(() => {
      this.scheduling = false;
      let running = [...this.runtime.values()].filter((rt) => rt.running).length;
      const waiting = [...this.jobs.values()].filter((job) => job.state === "waiting").sort((a, b) => a.order - b.order);
      for (const job of waiting) {
        if (running >= this.settings.maxConcurrent) break;
        running += 1;
        void this._run(job);
      }
    });
  }

  _rt(job) {
    let rt = this.runtime.get(job.id);
    if (!rt) { rt = { running: false, stop: null, speed: 0, samples: 0, lastBytes: job.transferred, lastTime: this.now() }; this.runtime.set(job.id, rt); }
    return rt;
  }

  async _run(job) {
    const rt = this._rt(job);
    if (rt.running) return;
    Object.assign(rt, { running: true, stop: null, speed: 0, samples: 0, lastBytes: job.transferred, lastTime: this.now(), abort: null, progressSinceResume: 0,
      generation: (rt.generation || 0) + 1 });
    job.error = null;
    job.resumable = false;
    try {
      this._set(job, "connecting");
      const sessionId = this._sessionFor(job);
      job.sessionId = sessionId;
      if (job.direction === "upload") await this._runUpload(job, rt, sessionId);
      else await this._runDownload(job, rt, sessionId);
    } catch (error) {
      await this._failure(job, rt, error);
    } finally {
      rt.running = false;
      rt.abort = null;
      clearTimeout(rt.stopTimer);
      rt.stopTimer = null;
      this._persist(true);
      this._schedule();
    }
  }

  async _failure(job, rt, error) {
    if (error instanceof StopSignal || rt.stop) {
      const kind = error.stop || rt.stop;
      if (kind === "cancel") await this._finishCancel(job);
      else if (kind === "pause") this._set(job, "paused");
      else this._set(job, "interrupted", { code: "APP_RESTARTED", message: "The application was closed during the transfer." });
      return;
    }
    const code = errorCode(error);
    const message = String(error?.message || code).replace(/^[A-Z_]+: /, "").slice(0, 400);
    this.logger?.warn("Transfer interrupted.", { jobId: job.id, direction: job.direction, state: job.state, code, transferred: job.transferred, size: job.size });
    if (job.state === "finalizing" && (code === "FILE_RESULT_UNKNOWN" || CONNECTION_CODES.has(code))) {
      job.finalize = { ...job.finalize, state: "unknown" };
    }
    if (AUTH_CODES.has(code)) {
      job.authRequired = true;
      this._set(job, "interrupted", { code, message });
      return;
    }
    if (CONNECTION_CODES.has(code) || code === "FILE_RESULT_UNKNOWN") {
      this._set(job, "interrupted", { code, message });
      if (this.settings.autoRetry && job.totalRetries < MAX_TOTAL_RETRIES) this._scheduleRetry(job);
      return;
    }
    job.resumable = ["FILE_NO_SPACE", "FILE_EXISTS", "FILE_CONFLICT", "LOCAL_NO_SPACE", "ENOSPC", "FILE_READ_ONLY_FILESYSTEM", "FILE_BUSY"].includes(code);
    this._set(job, "failed", { code: code === "ENOSPC" ? "LOCAL_NO_SPACE" : code, message });
  }

  _scheduleRetry(job) {
    this._clearRetry(job);
    if (job.attempts >= this.retryDelays.length) return; // Manual resume from here on: no endless loop.
    const delay = this.retryDelays[job.attempts];
    job.attempts += 1;
    job.totalRetries += 1;
    job.nextRetryAt = this.now() + delay;
    this._set(job, "reconnecting", job.error);
    const timer = setTimeout(() => {
      this._rt(job).retryTimer = null;
      if (job.state !== "reconnecting") return;
      if (this._findSession(job)) { this._set(job, "waiting"); this._schedule(); }
      else if (job.attempts < this.retryDelays.length) this._scheduleRetry(job);
      else this._set(job, "interrupted", job.error);
    }, delay);
    timer.unref?.();
    this._rt(job).retryTimer = timer;
  }

  _clearRetry(job) {
    const rt = this.runtime.get(job.id);
    if (rt?.retryTimer) { clearTimeout(rt.retryTimer); rt.retryTimer = null; }
    job.nextRetryAt = null;
  }

  _sessionState(status) {
    if (status && status.state === "disconnected") {
      // A dropped SSH connection does not answer outstanding SFTP requests: end those jobs now.
      for (const job of this.jobs.values()) {
        const rt = this.runtime.get(job.id);
        if (!rt?.running || job.sessionId !== status.sessionId) continue;
        rt.abort?.();
        rt.kill?.(fileError("SESSION_NOT_CONNECTED", "The SSH connection was lost."));
      }
      return;
    }
    if (!status || status.state !== "connected") return;
    const identity = JSON.stringify([status.profileId, status.host, status.port, status.username]);
    for (const job of this.jobs.values()) {
      if (JSON.stringify([job.identity?.profileId, job.identity?.host, job.identity?.port, job.identity?.username]) !== identity) continue;
      if (job.state === "reconnecting" || (job.state === "interrupted" && this.settings.autoRetry && CONNECTION_CODES.has(job.error?.code) && job.totalRetries < MAX_TOTAL_RETRIES)) {
        this._clearRetry(job);
        this._set(job, "waiting");
      }
    }
    this.cleanedIdentities.delete(identity);
    this._cleanupOrphansLater();
    this._schedule();
  }

  _stop(job, kind) {
    const rt = this._rt(job);
    rt.stop = kind;
    rt.wake?.();
    // Requests already on the wire normally settle quickly; a dead link is cut after a grace period.
    // The timer belongs to this run only: a resumed run must never be stopped by an old pause.
    const generation = rt.generation;
    clearTimeout(rt.stopTimer);
    rt.stopTimer = setTimeout(() => {
      if (rt.generation !== generation || !rt.running) return;
      rt.abort?.();
      rt.kill?.(new StopSignal(kind));
    }, kind === "shutdown" ? 0 : this.stopGraceMs);
    rt.stopTimer.unref?.();
  }

  // ---- transfer engine -----------------------------------------------------------------------

  async _openChannel(job, rt, sessionId) {
    const sftp = await this.sessionManager.openSftp(sessionId);
    rt.abort = () => { try { sftp.end?.(); } catch {} };
    if (rt.stop) { rt.abort(); throw new StopSignal(rt.stop); }
    return sftp;
  }

  _helperAuth(job, internal = {}) {
    const privileged = job.mode === "privileged";
    return privileged ? [{ sudo: true, sudoPassword: this.secrets.get(job.id) || "" }, { useCachedSudo: true, ...internal }] : [{}, internal];
  }

  async _runUpload(job, rt, sessionId) {
    const stats = await fs.promises.lstat(job.localPath).catch((error) => { throw fileError(error.code === "ENOENT" ? "LOCAL_NOT_FOUND" : "LOCAL_READ_FAILED", `${job.localPath} cannot be read.`); });
    if (!stats.isFile()) throw fileError("LOCAL_NOT_FILE", `${job.localPath} is no longer a regular file.`);
    if (!sameIdentity(localIdentity(stats), job.localIdentity)) {
      if (job.transferred > 0) throw fileError("FILE_SOURCE_CHANGED", "The local file changed since the transfer started; restart it from the beginning.");
      job.localIdentity = localIdentity(stats);
      job.size = stats.size;
    }
    if (job.finalize?.state === "unknown" || (job.hash && job.transferred === job.size && job.partPath)) return this._finalizeUpload(job, rt, sessionId);
    await this._prepareUploadTarget(job, sessionId);
    const sftp = await this._openChannel(job, rt, sessionId);
    let local;
    let handle;
    try {
      local = await fs.promises.open(job.localPath, "r");
      handle = await this._openRemotePart(sftp, job);
      if (job.transferred > 0) {
        const remote = await call(sftp, "fstat", handle);
        job.transferred = Math.min(job.transferred, Number(remote.size) || 0);
        await this._verifyTail(job, (buffer, position) => local.read(buffer, 0, buffer.length, position).then((r) => r.bytesRead),
          (buffer, position) => this._remoteRead(sftp, handle, buffer, position));
      }
      this._set(job, "transferring");
      await this._pump(job, rt, {
        produce: async (position, length) => {
          const buffer = Buffer.allocUnsafe(length);
          const { bytesRead } = await local.read(buffer, 0, length, position);
          if (bytesRead !== length) throw fileError("FILE_SOURCE_CHANGED", "The local file shrank during the transfer.");
          return buffer;
        },
        consume: (buffer, position) => call(sftp, "write", handle, buffer, 0, buffer.length, position)
      });
      const remote = await call(sftp, "fstat", handle);
      if ((Number(remote.size) || 0) > job.size) await call(sftp, "fsetstat", handle, { size: job.size });
      if (typeof sftp.ext_openssh_fsync === "function") await call(sftp, "ext_openssh_fsync", handle).catch(() => {});
      await call(sftp, "close", handle).catch(() => {});
      handle = null;
      if (!sameIdentity(localIdentity(await local.stat()), job.localIdentity)) throw fileError("FILE_SOURCE_CHANGED", "The local file changed during the transfer.");
      this._set(job, "verifying");
      job.hash = await this._localHash(job.localPath, rt);
      job.verification = { method: "sha256", local: job.hash };
    } finally {
      if (handle) await within(call(sftp, "close", handle));
      await local?.close().catch(() => {});
      try { sftp.end?.(); } catch {}
      rt.abort = null;
    }
    return this._finalizeUpload(job, rt, sessionId);
  }

  async _prepareUploadTarget(job, sessionId) {
    if (job.resolvedPath && job.partPath && job.targetBaseline !== undefined) return;
    const [auth, internal] = this._helperAuth(job);
    let stat;
    try {
      stat = await this.files.helper(sessionId, { operation: "stat", path: job.remotePath }, auth, internal);
      job.helper = true;
    } catch (error) {
      if (!NO_HELPER_CODES.has(error.code) || job.mode === "privileged") throw error;
      // No python3 / Windows server: plain SFTP is enough for new files; existing ones need the
      // metadata-preserving Linux helper.
      job.helper = false;
      const exists = await this.files._withSftp(sessionId, (sftp) => call(sftp, "lstat", job.remotePath).then(() => true, () => false));
      if (exists) throw fileError(error.code, error.code === "FEATURE_PLATFORM_UNSUPPORTED"
        ? "Replacing an existing file on this server is not supported (no metadata-preserving helper); remove it first or choose another name."
        : "Replacing an existing file requires python3 on the server to preserve its owner, permissions and ACLs.");
      stat = { exists: false, resolvedPath: job.remotePath };
    }
    if (stat.exists && stat.type !== "file") throw fileError("FILE_UNSAFE_TYPE_OR_LINK", `${job.remotePath} is not a regular file.`);
    if (stat.exists && stat.baseline.nlink > 1) throw fileError("FILE_HARDLINK_UNSUPPORTED", `${job.remotePath} has several hard links.`);
    if (stat.exists && !job.overwrite) throw fileError("FILE_EXISTS", `${job.remotePath} already exists.`);
    job.resolvedPath = stat.resolvedPath || job.remotePath;
    job.targetBaseline = stat.exists ? stat.baseline : null;
    if (job.mode === "privileged") {
      const workspace = await this.files.helper(sessionId, { operation: "transfer-workspace" });
      job.ownerUid = workspace.uid;
      job.partPath = path.posix.join(workspace.path, `${job.id}.part`);
    } else {
      if (job.helper) job.ownerUid = (await this.files.helper(sessionId, { operation: "transfer-workspace" })).uid;
      job.partPath = path.posix.join(path.posix.dirname(job.resolvedPath), partName(path.posix.basename(job.resolvedPath), job.id));
    }
    this._persist(true);
  }

  async _openRemotePart(sftp, job) {
    if (job.transferred > 0) {
      const attrs = await call(sftp, "lstat", job.partPath).catch(() => null);
      if (!attrs || (typeof attrs.isFile === "function" && !attrs.isFile())) {
        job.transferred = 0; // The partial file disappeared; start this file again.
      } else {
        return call(sftp, "open", job.partPath, "r+");
      }
    }
    try { return await call(sftp, "open", job.partPath, "wx", { mode: 0o600 }); }
    catch (error) {
      // Only our own leftover (same job id) can exist here; remove the entry itself, never follow it.
      if (errorCode(error) === "FILE_PERMISSION_DENIED") throw fileError("FILE_PERMISSION_DENIED", `${path.posix.dirname(job.partPath)} is not writable.`);
      await call(sftp, "unlink", job.partPath).catch(() => {});
      return call(sftp, "open", job.partPath, "wx", { mode: 0o600 });
    }
  }

  async _finalizeUpload(job, rt, sessionId) {
    if (rt.stop) throw new StopSignal(rt.stop);
    this._set(job, "finalizing");
    const retry = job.finalize?.state === "unknown" || job.finalize?.state === "running";
    job.finalize = { state: "running", startedAt: this.now() };
    this._persist(true);
    if (job.helper === false) return this._publishWithoutHelper(job, sessionId);
    const [auth, internal] = this._helperAuth(job.finalizeWithSudo ? { ...job, mode: "privileged" } : job, { timeoutMs: 120000 + Math.ceil((job.size / HASH_RATE) * 1000) });
    const result = await this.files.helper(sessionId, {
      operation: "publish", source: job.partPath, uid: job.ownerUid, path: job.remotePath, hash: job.hash, size: job.size,
      baseline: job.targetBaseline || null, overwrite: job.overwrite === true, strategy: job.mode === "privileged" ? "copy" : "rename",
      parents: true, retry, mode: 0o644
    }, auth, internal);
    job.finalize = { state: "done", resolvedPath: result.resolvedPath, created: result.created === true, metadataPreserved: result.metadataPreserved === true,
      alreadyPublished: result.alreadyPublished === true, finishedAt: this.now() };
    job.verification = { ...(job.verification || {}), remote: result.hash, method: "sha256" };
    if (result.alreadyPublished && job.mode === "privileged") await this.files.helper(sessionId, { operation: "transfer-remove", id: job.id }).catch(() => {});
    this.secrets.delete(job.id);
    this._set(job, "completed");
    this.logger?.info("Transfer completed.", { jobId: job.id, direction: "upload", bytes: job.size, mode: job.mode, path: job.remotePath });
  }

  async _publishWithoutHelper(job, sessionId) {
    const remoteHash = await this.sessionManager.remoteChecksum?.(sessionId, job.partPath).catch(() => null);
    if (remoteHash && remoteHash !== job.hash) {
      await this._discardPartial(job).catch(() => {});
      job.transferred = 0;
      throw fileError("FILE_VERIFY_FAILED", "The uploaded data does not match the local file.");
    }
    job.verification = { ...(job.verification || {}), remote: remoteHash || null, method: remoteHash ? "sha256" : "size" };
    await this.files._withSftp(sessionId, async (sftp) => {
      const attrs = await call(sftp, "stat", job.partPath);
      if (Number(attrs.size) !== job.size) throw fileError("FILE_VERIFY_FAILED", "The uploaded size does not match the local file.");
      if (await call(sftp, "lstat", job.remotePath).then(() => true, () => false)) throw fileError("FILE_EXISTS", `${job.remotePath} appeared during the transfer.`);
      await call(sftp, "rename", job.partPath, job.remotePath);
    });
    job.finalize = { state: "done", resolvedPath: job.remotePath, created: true, metadataPreserved: false, finishedAt: this.now() };
    this._set(job, "completed");
  }

  async _runDownload(job, rt, sessionId) {
    const sftp = await this._openChannel(job, rt, sessionId);
    let handle;
    let local;
    try {
      let source = job.remotePath;
      if (job.mode === "privileged") source = await this._ensureStaged(job, sftp, sessionId);
      else {
        const attrs = await call(sftp, "stat", job.remotePath).catch((error) => { throw fileError(errorCode(error), `${job.remotePath} is not accessible.`); });
        const current = { size: Number(attrs.size) || 0, mtime: Number(attrs.mtime) || 0 };
        if (job.remoteIdentity && (current.size !== job.remoteIdentity.size || current.mtime !== job.remoteIdentity.mtime)) {
          if (job.transferred > 0) throw fileError("FILE_SOURCE_CHANGED", "The remote file changed since the transfer started; restart it from the beginning.");
        }
        job.remoteIdentity = current;
        job.size = current.size;
        job.remoteMode = (Number(attrs.mode) || 0) & 0o777;
      }
      try { handle = await call(sftp, "open", source, "r"); }
      catch (error) {
        if (errorCode(error) === "FILE_PERMISSION_DENIED" && job.mode === "privileged-allowed") {
          job.mode = "privileged";
          source = await this._ensureStaged(job, sftp, sessionId);
          handle = await call(sftp, "open", source, "r");
        } else throw fileError(errorCode(error), `${job.remotePath} cannot be read.`);
      }
      job.localPart ||= path.join(path.dirname(job.localPath), partName(path.basename(job.localPath), job.id));
      local = await this._openLocalPart(job);
      if (job.transferred > 0) {
        await this._verifyTail(job, (buffer, position) => local.read(buffer, 0, buffer.length, position).then((r) => r.bytesRead),
          (buffer, position) => this._remoteRead(sftp, handle, buffer, position));
      }
      this._set(job, "transferring");
      await this._pump(job, rt, {
        produce: async (position, length) => {
          const buffer = Buffer.allocUnsafe(length);
          const read = await this._remoteRead(sftp, handle, buffer, position);
          if (read !== length) throw fileError("FILE_SOURCE_CHANGED", "The remote file shrank during the transfer.");
          return buffer;
        },
        consume: (buffer, position) => local.write(buffer, 0, buffer.length, position)
      });
      await local.truncate(job.size);
      await local.sync();
      await local.close();
      local = null;
      await call(sftp, "close", handle).catch(() => {});
      handle = null;
      this._set(job, "verifying");
      job.hash = await this._localHash(job.localPart, rt);
      job.verification = { method: "size", local: job.hash };
      if (job.mode === "privileged") job.verification = { method: "sha256", local: job.hash, remote: job.staged.hash };
      else {
        const attrs = await call(sftp, "stat", job.remotePath);
        if (Number(attrs.size) !== job.remoteIdentity.size || Number(attrs.mtime) !== job.remoteIdentity.mtime) {
          throw fileError("FILE_SOURCE_CHANGED", "The remote file changed during the transfer; restart it from the beginning.");
        }
        if (this.settings.verify) {
          const remote = await this.files.helper(sessionId, { operation: "hash", path: job.remotePath }, {}, { timeoutMs: 120000 + Math.ceil((job.size / HASH_RATE) * 1000) })
            .then((value) => value.hash, () => this.sessionManager.remoteChecksum?.(sessionId, job.remotePath).catch(() => null));
          if (remote) job.verification = { method: "sha256", local: job.hash, remote };
        }
      }
      if (job.verification.remote && job.verification.remote !== job.hash) {
        await this._discardPartial(job).catch(() => {});
        job.transferred = 0;
        throw fileError("FILE_VERIFY_FAILED", "The downloaded data does not match the remote file.");
      }
    } finally {
      if (handle) await within(call(sftp, "close", handle));
      await local?.close().catch(() => {});
      try { sftp.end?.(); } catch {}
      rt.abort = null;
    }
    if (rt.stop) throw new StopSignal(rt.stop);
    this._set(job, "finalizing");
    const exists = await fs.promises.lstat(job.localPath).then(() => true, () => false);
    if (exists && !job.overwrite) throw fileError("FILE_EXISTS", `${job.localPath} already exists.`);
    await fs.promises.chmod(job.localPart, job.mode === "privileged" ? 0o600 : ((job.remoteMode ?? 0o644) & 0o755) | 0o600).catch(() => {});
    await fs.promises.rename(job.localPart, job.localPath);
    if (job.mode === "privileged") await this.files.helper(sessionId, { operation: "transfer-remove", id: job.id }).catch(() => { job.cleanupPending = true; });
    job.finalize = { state: "done", finishedAt: this.now() };
    this.secrets.delete(job.id);
    this._set(job, "completed");
    this.logger?.info("Transfer completed.", { jobId: job.id, direction: "download", bytes: job.size, mode: job.mode });
  }

  async _ensureStaged(job, sftp, sessionId) {
    if (job.staged) {
      const attrs = await call(sftp, "stat", job.staged.path).catch(() => null);
      if (attrs && Number(attrs.size) === job.staged.size) return job.staged.path;
      job.transferred = 0;
    }
    this._set(job, "connecting");
    const workspace = await this.files.helper(sessionId, { operation: "transfer-workspace" });
    const [auth, internal] = this._helperAuth({ ...job, mode: "privileged" }, { timeoutMs: 120000 + Math.ceil((job.size / HASH_RATE) * 1000) });
    const staged = await this.files.helper(sessionId, { operation: "stage-out", path: job.remotePath, id: job.id, uid: workspace.uid }, auth, internal);
    job.staged = { path: staged.path, size: staged.size, hash: staged.hash, mtimeNs: staged.mtimeNs };
    job.size = staged.size;
    job.transferred = 0;
    this._persist(true);
    return staged.path;
  }

  async _openLocalPart(job) {
    if (job.transferred > 0) {
      const stats = await fs.promises.lstat(job.localPart).catch(() => null);
      if (stats?.isFile()) {
        job.transferred = Math.min(job.transferred, stats.size);
        return fs.promises.open(job.localPart, "r+");
      }
      job.transferred = 0;
    }
    await fs.promises.unlink(job.localPart).catch(() => {});
    return fs.promises.open(job.localPart, "wx", 0o600);
  }

  async _remoteRead(sftp, handle, buffer, position) {
    let total = 0;
    while (total < buffer.length) {
      const result = await call(sftp, "read", handle, buffer, total, buffer.length - total, position + total);
      const read = Array.isArray(result) ? result[0] : result;
      if (!read) break;
      total += read;
    }
    return total;
  }

  // Resume only when the last confirmed bytes of the partial file equal the source.
  async _verifyTail(job, readLocal, readRemote) {
    const window = Math.min(TAIL_CHECK, job.transferred);
    if (!window) return;
    const position = job.transferred - window;
    const left = Buffer.alloc(window);
    const right = Buffer.alloc(window);
    const [leftRead, rightRead] = await Promise.all([readLocal(left, position), readRemote(right, position)]);
    if (leftRead !== window || rightRead !== window || !left.equals(right)) {
      throw fileError("FILE_RESUME_MISMATCH", "The partial data does not match the source; restart the transfer from the beginning.");
    }
  }

  _pump(job, rt, { produce, consume }) {
    return new Promise((resolve, reject) => {
      let next = job.transferred;
      let confirmed = job.transferred;
      const completed = new Map();
      let inflight = 0;
      let failure = null;
      let done = false;
      let lastProgress = this.now();
      const finish = (callback) => {
        if (done) return;
        done = true;
        clearInterval(stall);
        rt.wake = null;
        rt.kill = null;
        callback();
      };
      // Ends the run even while requests are still outstanding on a dead channel; late replies are ignored.
      rt.kill = (error) => { if (!failure) failure = error; finish(() => reject(failure)); };
      const stall = setInterval(() => {
        if (this.now() - lastProgress > STALL_MS && !failure) { rt.abort?.(); rt.kill(fileError("TRANSFER_STALLED", "No data was confirmed for 60 seconds.")); }
      }, 5000);
      stall.unref?.();
      const settle = () => {
        if (inflight || done) return;
        if (failure) finish(() => reject(failure));
        else if (rt.stop) finish(() => reject(new StopSignal(rt.stop)));
        else if (confirmed >= job.size) finish(resolve);
      };
      const issue = () => {
        while (!done && !failure && !rt.stop && inflight < PIPELINE && next < job.size) {
          const position = next;
          const length = Math.min(CHUNK, job.size - position);
          next += length;
          inflight += 1;
          Promise.resolve().then(() => produce(position, length)).then((buffer) => consume(buffer, position)).then(() => {
            inflight -= 1;
            if (done) return;
            completed.set(position, length);
            while (completed.has(confirmed)) {
              const size = completed.get(confirmed);
              completed.delete(confirmed);
              confirmed += size;
            }
            const moved = confirmed - job.transferred;
            job.transferred = confirmed;
            job.movedBytes = (job.movedBytes || 0) + moved;
            rt.progressSinceResume += moved;
            if (rt.progressSinceResume > 1024 * 1024) job.attempts = 0;
            lastProgress = this.now();
            this._progress(job, rt);
            issue();
            settle();
          }, (error) => {
            inflight -= 1;
            if (done) return;
            if (!failure) failure = error;
            settle();
          });
        }
        settle();
      };
      rt.wake = settle;
      issue();
    });
  }

  async _localHash(file, rt) {
    const digest = crypto.createHash("sha256");
    const stream = fs.createReadStream(file, { highWaterMark: 1024 * 1024 });
    rt.abort = () => stream.destroy(new StopSignal(rt.stop || "pause"));
    for await (const chunk of stream) {
      if (rt.stop) { stream.destroy(); throw new StopSignal(rt.stop); }
      digest.update(chunk);
    }
    return digest.digest("hex");
  }

  // ---- cancel / cleanup ------------------------------------------------------------------------

  async _finishCancel(job) {
    this.secrets.delete(job.id);
    try { await this._discardPartial(job); job.cleanupPending = false; }
    catch { job.cleanupPending = true; }
    this._set(job, "cancelled");
  }

  // Removes only temporary data this job created; existing destination files are never touched.
  async _discardPartial(job) {
    if (job.direction === "download") {
      if (job.localPart) await fs.promises.unlink(job.localPart).catch((error) => { if (error.code !== "ENOENT") throw error; });
      if (job.staged) {
        const sessionId = this._findSession(job);
        if (!sessionId) throw fileError("SESSION_NOT_CONNECTED", "Staged data is removed after reconnecting.");
        await this.files.helper(sessionId, { operation: "transfer-remove", id: job.id });
        job.staged = null;
      }
      return;
    }
    if (!job.partPath) return;
    const sessionId = this._findSession(job);
    if (!sessionId) throw fileError("SESSION_NOT_CONNECTED", "Partial data is removed after reconnecting.");
    if (job.mode === "privileged") await this.files.helper(sessionId, { operation: "transfer-remove", id: job.id });
    else {
      await this.files._withSftp(sessionId, async (sftp) => {
        const attrs = await call(sftp, "lstat", job.partPath).catch(() => null);
        if (attrs && !(typeof attrs.isDirectory === "function" && attrs.isDirectory())) await call(sftp, "unlink", job.partPath);
      });
    }
    job.partPath = null;
  }

  _cleanupOrphansLater() {
    if (this.cleanupScheduled) return;
    this.cleanupScheduled = true;
    setImmediate(() => { this.cleanupScheduled = false; void this._cleanupOrphans().catch(() => {}); });
  }

  // After a restart: delete temporary transfer data that no resumable job owns any more.
  async _cleanupOrphans() {
    for (const session of this.sessionManager?.list?.() || []) {
      if (session.state !== "connected") continue;
      const identity = JSON.stringify([session.profileId, session.host, session.port, session.username]);
      if (this.cleanedIdentities.has(identity)) continue;
      this.cleanedIdentities.add(identity);
      const mine = [...this.jobs.values()].filter((job) => JSON.stringify([job.identity?.profileId, job.identity?.host, job.identity?.port, job.identity?.username]) === identity);
      for (const job of mine.filter((entry) => entry.cleanupPending)) {
        await this._discardPartial(job).then(() => { job.cleanupPending = false; this._mark(job); }, () => {});
      }
      const keep = new Set(mine.filter((job) => !FINAL.has(job.state)).map((job) => job.id));
      let listed;
      try { listed = await this.files.helper(session.sessionId, { operation: "transfer-list" }); }
      catch { continue; }
      for (const item of listed.items || []) {
        const known = this.jobs.get(item.id);
        const abandoned = known ? FINAL.has(known.state) : this.now() - Number(item.mtime) * 1000 > ORPHAN_AGE_MS;
        if (keep.has(item.id) || !abandoned) continue;
        await this.files.helper(session.sessionId, { operation: "transfer-remove", id: item.id }).catch(() => {});
        this.logger?.info("Abandoned transfer data removed.", { sessionId: session.sessionId, id: item.id, bytes: item.size });
      }
      this._persist();
    }
  }

  // ---- state, events, persistence ----------------------------------------------------------------

  _authorize(job, options) {
    if (options.sudo === true || options.sudoPassword) {
      // Data is already next to the target: only the metadata-preserving replacement needs root.
      if (job.direction === "upload" && job.mode === "direct" && job.transferred > 0 && job.transferred === job.size) job.finalizeWithSudo = true;
      if (job.direction === "download" && job.mode === "direct") job.mode = "privileged-allowed";
      if (job.direction === "upload" && job.mode === "direct" && job.transferred === 0 && AUTH_CODES.has(job.error?.code)) {
        job.mode = "privileged";
        job.partPath = null;
        job.targetBaseline = undefined;
      }
    }
    if (options.sudoPassword) this.secrets.set(job.id, String(options.sudoPassword));
  }

  _set(job, state, error = null) {
    if (!STATES.includes(state)) throw new Error(`Unknown transfer state ${state}`);
    const rt = this.runtime.get(job.id);
    const now = this.now();
    if (job.state === "transferring" && rt?.activeSince) { job.activeMs = (job.activeMs || 0) + (now - rt.activeSince); rt.activeSince = null; }
    if (state === "transferring" && rt) { rt.activeSince = now; rt.lastTime = now; rt.lastBytes = job.transferred; }
    job.state = state;
    job.error = error;
    job.updatedAt = now;
    if (state === "transferring" && !job.startedAt) job.startedAt = now;
    if (FINAL.has(state)) job.finishedAt = now;
    this._mark(job);
    this._persist(RUNNING.has(state) ? false : true);
  }

  _progress(job, rt) {
    const now = this.now();
    const elapsed = now - rt.lastTime;
    if (elapsed >= 500) {
      const instant = ((job.transferred - rt.lastBytes) * 1000) / elapsed;
      rt.speed = rt.samples ? 0.3 * instant + 0.7 * rt.speed : instant;
      rt.samples += 1;
      rt.lastTime = now;
      rt.lastBytes = job.transferred;
    }
    job.updatedAt = now;
    this._mark(job);
    this._persist();
  }

  _public(job) {
    const rt = this.runtime.get(job.id);
    const now = this.now();
    let speed = 0;
    let eta = null;
    if (job.state === "transferring" && rt) {
      const idle = now - rt.lastTime;
      speed = idle > 2000 ? (rt.speed * 2000) / idle : rt.speed;
      if (rt.samples >= 3 && speed > 1 && Number.isFinite(job.size)) eta = Math.ceil(Math.max(0, job.size - job.transferred) / speed);
    }
    const activeMs = (job.activeMs || 0) + (job.state === "transferring" && rt?.activeSince ? now - rt.activeSince : 0);
    const state = job.state;
    return {
      id: job.id, batchId: job.batchId, direction: job.direction, name: job.name, state,
      source: job.direction === "upload" ? job.localPath : job.remotePath, target: job.direction === "upload" ? job.remotePath : job.localPath,
      resolvedTarget: job.finalize?.resolvedPath || job.resolvedPath || null, mode: job.mode === "privileged" ? "privileged" : "direct",
      size: Number.isFinite(job.size) ? job.size : null, transferred: job.transferred || 0, percent: percent(job.transferred || 0, job.size),
      speed: Math.round(speed), averageSpeed: activeMs > 1000 ? Math.round(((job.movedBytes || 0) * 1000) / activeMs) : null, eta,
      finalize: { state: job.finalize?.state || "pending", metadataPreserved: job.finalize?.metadataPreserved === true, created: job.finalize?.created === true },
      verification: job.verification ? { method: job.verification.method, matched: Boolean(job.verification.remote && job.verification.remote === job.verification.local) } : null,
      error: job.error ? { code: job.error.code, message: job.error.message } : null, authRequired: job.authRequired === true,
      attempts: job.attempts || 0, nextRetryAt: job.nextRetryAt || null, cleanupPending: job.cleanupPending === true,
      actions: {
        pause: ["waiting", "reconnecting", "interrupted", "connecting", "transferring", "verifying"].includes(state),
        resume: ["paused", "interrupted", "reconnecting"].includes(state) || (state === "failed" && job.resumable === true),
        retry: ["failed", "interrupted", "paused", "cancelled"].includes(state),
        cancel: !FINAL.has(state) && state !== "finalizing",
        overwrite: ["FILE_EXISTS", "FILE_CONFLICT"].includes(job.error?.code) && ["failed", "interrupted"].includes(state),
        remove: FINAL.has(state) && !job.cleanupPending
      },
      createdAt: job.createdAt, updatedAt: job.updatedAt, finishedAt: job.finishedAt || null
    };
  }

  _mark(job) {
    this.changed.add(job.id);
    this._emitSoon(job.owner);
  }

  _emitSoon(owner, full = false) {
    this.pendingOwners ||= new Set();
    this.pendingOwners.add(owner);
    if (full) this.fullOwners = new Set([...(this.fullOwners || []), owner]);
    if (this.emitTimer) return;
    this.emitTimer = setTimeout(() => {
      this.emitTimer = null;
      const owners = this.pendingOwners;
      const full = this.fullOwners || new Set();
      this.pendingOwners = new Set();
      this.fullOwners = new Set();
      const changed = this.changed;
      this.changed = new Set();
      for (const owner of owners) {
        const all = this._owned(owner).map((job) => this._public(job));
        const jobs = full.has(owner) ? all : all.filter((job) => changed.has(job.id) || RUNNING.has(job.state));
        const removed = [...changed].filter((id) => !this.jobs.has(id));
        this.emit("update", { owner, jobs, removed, full: full.has(owner), summary: summarize(all), settings: { ...this.settings } });
      }
    }, EMIT_MS);
    this.emitTimer.unref?.();
  }

  _owned(owner) {
    return [...this.jobs.values()].filter((job) => job.owner === owner).sort((a, b) => a.order - b.order);
  }

  _select(owner, ids) {
    const owned = this._owned(owner);
    if (ids === "all") return owned;
    const wanted = new Set((Array.isArray(ids) ? ids : [ids]).map(String));
    return owned.filter((job) => wanted.has(job.id));
  }

  _identity(sessionId) {
    const session = this.sessionManager.list().find((entry) => entry.sessionId === sessionId && entry.state === "connected");
    if (!session) throw fileError("SESSION_NOT_CONNECTED", "The SSH session is not connected.");
    return { profileId: session.profileId, host: session.host, port: session.port, username: session.username };
  }

  _findSession(job) {
    const sessions = (this.sessionManager?.list?.() || []).filter((entry) => entry.state === "connected" && entry.profileId === job.identity?.profileId
      && entry.host === job.identity?.host && entry.port === job.identity?.port && entry.username === job.identity?.username);
    return (sessions.find((entry) => entry.sessionId === job.sessionId) || sessions[0])?.sessionId || null;
  }

  _sessionFor(job) {
    const sessionId = this._findSession(job);
    if (!sessionId) throw fileError("SESSION_NOT_CONNECTED", `Connect to ${job.identity?.username}@${job.identity?.host} to continue.`);
    return sessionId;
  }

  _trimHistory() {
    const finished = [...this.jobs.values()].filter((job) => FINAL.has(job.state) && !job.cleanupPending).sort((a, b) => (a.finishedAt || 0) - (b.finishedAt || 0));
    while (finished.length > MAX_HISTORY) {
      const job = finished.shift();
      this.jobs.delete(job.id);
      this.changed.add(job.id);
    }
  }

  _persist(soon = false, sync = false) {
    if (!this.file || this.closed) return;
    const write = () => {
      this.persistTimer = null;
      this._trimHistory();
      try {
        // Persisted state never contains passwords, keys or runtime handles.
        const jobs = [...this.jobs.values()].map(({ ...job }) => job);
        atomicWriteJson(this.file, { version: 1, settings: this.settings, order: this.order, jobs });
      } catch (error) {
        this.logger?.warn("Transfer state could not be saved.", { message: error.message });
      }
    };
    if (sync) { clearTimeout(this.persistTimer); write(); return; }
    if (this.persistTimer && !soon) return;
    clearTimeout(this.persistTimer);
    this.persistTimer = setTimeout(write, soon ? 50 : PERSIST_MS);
    this.persistTimer.unref?.();
  }

  _load() {
    if (!this.file) return;
    let data;
    try { data = JSON.parse(fs.readFileSync(this.file, "utf8")); }
    catch { return; }
    if (!data || data.version !== 1 || !Array.isArray(data.jobs)) return;
    for (const key of Object.keys(DEFAULT_SETTINGS)) if (data.settings?.[key] !== undefined) this.settings[key] = data.settings[key];
    if (!(Number.isInteger(this.settings.maxConcurrent) && this.settings.maxConcurrent >= 1 && this.settings.maxConcurrent <= 8)) this.settings.maxConcurrent = DEFAULT_SETTINGS.maxConcurrent;
    this.order = Number(data.order) || 0;
    for (const job of data.jobs.slice(-MAX_JOBS)) {
      if (!job || !/^[a-f0-9]{32}$/.test(String(job.id)) || !STATES.includes(job.state)) continue;
      // Nothing restarts on its own after the application was closed: the user is offered to resume.
      if (RUNNING.has(job.state) || job.state === "waiting" || job.state === "reconnecting") {
        if (job.state === "finalizing" || job.finalize?.state === "running") job.finalize = { ...job.finalize, state: "unknown" };
        job.state = "interrupted";
        job.error = { code: "APP_RESTARTED", message: "The application was closed during the transfer." };
      }
      job.nextRetryAt = null;
      job.attempts = 0;
      this.jobs.set(job.id, job);
      this.order = Math.max(this.order, Number(job.order) || 0);
    }
  }
}

module.exports = { CHUNK, PIPELINE, STATES, TransferManager, summarize };