/**
 * SSH tools: SshExec / SshUpload / SshDownload.
 *
 * Implements the 12-step flow from the design plan:
 *
 *   1. Resolve userId (ToolContext) + isAdmin/orgIds (subsystem.getUserAccess).
 *   2. Read pieceName + allowedSshConnections from ToolContext.
 *   3. Piece membership: connectionId ∈ allowedSshConnections (or wildcard).
 *   4. resolveConnection(id) — raw row.
 *   5. resolveAccess({ connection, userId, isAdmin, pieceName, orgIds }).
 *   6. State checks: enabled, !disabled_by_admin, !abuseRepo.isLocked,
 *      hostKeyVerified (for Upload/Download — sshExec runs even on first contact
 *      so the verifier can observe + report).
 *   7. Tool-specific policy: deny-list (Exec), path policy (Upload/Download).
 *   8. auditRepo.begin() — pending row inserted + committed.
 *   9. try {decrypt → sshExec/Upload/Download} catch {abuse.checkAndRecordFailure
 *      + audit.complete(failed)} finally {clearBuffer pem/passphrase}.
 *  10. abuse.recordSuccess(connectionId).
 *  11. audit.complete(id, 'success', { exit_code, ... }).
 *  12. Return wrapOutput JSON envelope (Exec) or success summary (Upload/Download).
 *
 * Steps 3/4/5/6/7 each emit a denied audit row on rejection (beginAndComplete).
 * Step 8 commits BEFORE step 9 so a crash leaves a pending row that the recovery
 * sweep promotes to `aborted` on next boot.
 *
 * Dependency injection: this module relies on a single process-wide subsystem,
 * set by bridge/server.ts at boot. Tests use a stub via setSshSubsystem(...).
 * When the subsystem is null (SSH disabled), every call short-circuits with an
 * informative error — never throws.
 *
 * Plan: docs/superpowers/plans/2026-05-12-ssh-tool-integration.md (Phase 7).
 */
import { createHash } from 'node:crypto';
import * as fs from 'node:fs';
import * as path from 'node:path';
import type { ToolDef } from '../../llm/openai-compat.js';
import { logger } from '../../logger.js';
import type { ToolContext, ToolResult } from './core.js';

import type { SshConnectionRepo, SshConnection } from '../../ssh/connection-repo.js';
import type { SshAuditRepo, BeginAuditInput } from '../../ssh/audit-repo.js';
import type { SshAbuseRepo } from '../../ssh/abuse-repo.js';
import type { SshAccessResolver } from '../../ssh/access.js';
import type { MaintenanceController } from '../../ssh/maintenance.js';
import type { SshRuntimeConfig } from '../../ssh/config.js';
import type {
  ExecArgs,
  UploadArgs,
  DownloadArgs,
  ExecResult,
  TransferResult,
  ResolvedConnection,
  SessionHooks,
  HostKeyObservation,
  openShellChannel as openShellChannelFn,
} from '../../ssh/session.js';
import { SshSessionError } from '../../ssh/session.js';
import type { SessionRegistry } from '../../ssh/console-registry.js';
import { checkCommand, validateCustomPatterns } from '../../ssh/deny-list.js';
import { validateRemotePath, validateLocalPath } from '../../ssh/path-policy.js';
import { clearBuffer, sanitizeError } from '../../ssh/crypto.js';

// ──────────────────────────────────────────────────────────────────────
// Subsystem (dependency injection)
// ──────────────────────────────────────────────────────────────────────

export interface SshSubsystem {
  connectionRepo: SshConnectionRepo;
  auditRepo: SshAuditRepo;
  abuseRepo: SshAbuseRepo;
  accessResolver: SshAccessResolver;
  decryptKeyMaterial(ownerId: string | null, blob: Buffer): Buffer;
  decryptPassphrase(ownerId: string | null, blob: Buffer | null): Buffer | null;
  /** Resolve admin role + org membership for an authenticated user id. */
  getUserAccess(userId: string): { isAdmin: boolean; orgIds: string[] };
  sshExec(args: ExecArgs, hooks: SessionHooks): Promise<ExecResult>;
  sshUpload(args: UploadArgs, hooks: SessionHooks): Promise<TransferResult>;
  sshDownload(args: DownloadArgs, hooks: SessionHooks): Promise<TransferResult>;
  maintenance: MaintenanceController;
  config: SshRuntimeConfig;
  /** Phase 3 (SSH Console): in-memory live-session registry. */
  sessionRegistry: SessionRegistry;
  /** Phase 3 (SSH Console): open a PTY-backed shell channel for a verified connection. */
  openShellChannel: typeof openShellChannelFn;
}

let _subsystem: SshSubsystem | null = null;

export function setSshSubsystem(sub: SshSubsystem | null): void {
  _subsystem = sub;
  logger.debug(`[tools/ssh] subsystem ${sub ? 'set' : 'cleared'}`);
}

export function getSshSubsystem(): SshSubsystem | null {
  return _subsystem;
}

// ──────────────────────────────────────────────────────────────────────
// Tool definitions
// ──────────────────────────────────────────────────────────────────────

export const TOOL_DEFS: Record<string, ToolDef> = {
  SshExec: {
    type: 'function',
    function: {
      name: 'SshExec',
      description:
        'SSH接続先でコマンドを実行します。connection_id (UUID) と command を指定。詳細は ReadToolDoc({ name: "SshExec" })。',
      parameters: {
        type: 'object',
        properties: {
          connection_id: {
            type: 'string',
            description: '使用する SSH 接続の UUID。piece の allowed_ssh_connections に含まれている必要があります。',
          },
          command: {
            type: 'string',
            description: '実行するコマンド (shell 単一行)。組み込み deny-list と接続側 deny/allow パターンでフィルタされます。',
          },
          timeout_ms: {
            type: 'number',
            description: 'ハンドシェイク後の wall-clock 上限 (ms)。未指定なら config.yaml の ssh.call_timeout_seconds。',
          },
        },
        required: ['connection_id', 'command'],
      },
    },
  },
  SshUpload: {
    type: 'function',
    function: {
      name: 'SshUpload',
      description:
        'SSH接続先にローカルファイルをアップロードします。詳細は ReadToolDoc({ name: "SshUpload" })。',
      parameters: {
        type: 'object',
        properties: {
          connection_id: { type: 'string', description: '使用する SSH 接続の UUID。' },
          local_path: {
            type: 'string',
            description: 'workspace 内の送信元ファイル (workspace 外参照は reject)。',
          },
          remote_path: {
            type: 'string',
            description: '接続先の絶対パス。接続の remote_path_prefix 配下である必要があります。',
          },
          timeout_ms: { type: 'number', description: '転送 wall-clock 上限 (ms)。' },
        },
        required: ['connection_id', 'local_path', 'remote_path'],
      },
    },
  },
  SshDownload: {
    type: 'function',
    function: {
      name: 'SshDownload',
      description:
        'SSH接続先からファイルをダウンロードします。詳細は ReadToolDoc({ name: "SshDownload" })。',
      parameters: {
        type: 'object',
        properties: {
          connection_id: { type: 'string', description: '使用する SSH 接続の UUID。' },
          remote_path: {
            type: 'string',
            description: '接続先の絶対パス。接続の remote_path_prefix 配下である必要があります。',
          },
          local_path: {
            type: 'string',
            description: 'workspace 内の保存先パス (既存ファイルへの上書きは reject)。',
          },
          timeout_ms: { type: 'number', description: '転送 wall-clock 上限 (ms)。' },
        },
        required: ['connection_id', 'remote_path', 'local_path'],
      },
    },
  },
  SshListConnections: {
    type: 'function',
    function: {
      name: 'SshListConnections',
      description:
        'この movement で利用可能な SSH 接続 (UUID + label + host) の一覧を返します。詳細は ReadToolDoc({ name: "SshListConnections" })。',
      parameters: {
        type: 'object',
        properties: {},
        required: [],
      },
    },
  },
};

const SSH_TOOL_NAMES = new Set(['SshExec', 'SshUpload', 'SshDownload', 'SshListConnections']);

// ──────────────────────────────────────────────────────────────────────
// Helpers
// ──────────────────────────────────────────────────────────────────────

/** SHA-256 hex prefix (16 chars). Used in audit detail to identify the command without
 * storing it verbatim. Full commands can be PII / contain secrets; the hash is enough
 * to correlate retries of the same command and to spot patterns across audit logs.
 */
function commandHash(command: string): string {
  return createHash('sha256').update(command).digest('hex').slice(0, 16);
}

function err(output: string, extras?: Partial<ToolResult>): ToolResult {
  return { output, isError: true, ...extras };
}

function ok(output: string, extras?: Partial<ToolResult>): ToolResult {
  return { output, isError: false, ...extras };
}

/** Parse a user-supplied newline-delimited regex blob into compiled RegExps.
 * Returns an empty array on parse error — empty list means "no extra rules".
 * Validation already happened at the API layer (validateCustomPatterns), so
 * a parse failure here means corruption / manual DB edit; we fail closed. */
function compilePatterns(blob: string | null): RegExp[] {
  if (!blob || blob.trim().length === 0) return [];
  const sources = blob.split('\n').map((s) => s.trim()).filter((s) => s.length > 0);
  const result = validateCustomPatterns(sources);
  return result.ok && result.compiled ? result.compiled : [];
}

/** Build a ResolvedConnection from a stored row + decrypted key/passphrase. */
function buildResolved(
  conn: SshConnection,
  pem: Buffer,
  passphrase: Buffer | null,
  globalAllowPrivate: boolean,
): ResolvedConnection {
  return {
    id: conn.id,
    ownerId: conn.ownerId,
    host: conn.host,
    port: conn.port,
    username: conn.username,
    privateKeyPem: pem,
    passphrase: passphrase ?? undefined,
    hostKeyB64: conn.hostKeyB64,
    hostKeyVerified: conn.hostKeyVerifiedAt !== null,
    allowPrivate: globalAllowPrivate || conn.allowPrivateAddresses,
  };
}

/** Build SessionHooks that wire host-key observations into connectionRepo + audit log.
 * On first_observe / mismatch, a fresh pending token is persisted; we return it so
 * the verifying user can call /verify-host-key with token + fingerprint. */
function buildHooks(
  sub: SshSubsystem,
  conn: SshConnection,
  actingUserId: string,
  pieceName: string | undefined,
  jobId: string | undefined,
): SessionHooks {
  return {
    onFirstObserve: async (obs: HostKeyObservation) => {
      const ret = sub.connectionRepo.setHostKeyPendingWithToken(
        obs.connectionId,
        obs.b64,
        obs.fingerprint,
        'tofu_record',
      );
      sub.auditRepo.beginAndComplete(
        {
          action: 'ssh.connection.host_key.first_observe',
          connectionId: conn.id,
          ownerId: conn.ownerId,
          actingUserId,
          pieceName,
          jobId,
          detail: { fingerprint: obs.fingerprint },
        },
        'success',
      );
      return ret;
    },
    onMismatch: async (obs: HostKeyObservation) => {
      const ret = sub.connectionRepo.setHostKeyPendingWithToken(
        obs.connectionId,
        obs.b64,
        obs.fingerprint,
        'mismatch',
      );
      sub.auditRepo.beginAndComplete(
        {
          action: 'ssh.connection.host_key.mismatch',
          connectionId: conn.id,
          ownerId: conn.ownerId,
          actingUserId,
          pieceName,
          jobId,
          detail: { fingerprint: obs.fingerprint, prev: conn.hostKeyFingerprint },
        },
        'success',
      );
      return ret;
    },
  };
}

/** Format an SshSessionError into LLM-actionable text. */
function formatSessionError(e: SshSessionError, connectionId: string): string {
  switch (e.code) {
    case 'host_key_first_observe':
      return [
        `Host key first-observe on connection ${connectionId} (fingerprint ${e.observedFingerprint}).`,
        `Verify via UI (SshConnections panel) before retrying. Pending token: ${e.pendingToken ?? '(none)'}`,
      ].join(' ');
    case 'host_key_mismatch':
      return [
        `WARN: Host key MISMATCH on connection ${connectionId} (now ${e.observedFingerprint}).`,
        'Likely possibilities: server rebuild, key rotation, or MITM.',
        `Verify carefully via UI and supply a reason. Pending token: ${e.pendingToken ?? '(none)'}`,
      ].join(' ');
    case 'host_key_not_verified':
      return `Host key for connection ${connectionId} has been recorded but not user-verified. Open SshConnections panel and confirm before retrying.`;
    case 'host_key_alg_not_allowed':
      return `Server presents a banned host-key algorithm. Connection refused by policy.`;
    case 'auth_failed':
      return `SSH authentication failed. The stored private key was rejected by the server.`;
    case 'connect_timeout':
    case 'exec_timeout':
    case 'transfer_timeout':
      return `SSH ${e.code.replace('_', ' ')}: ${e.message}`;
    case 'output_too_large':
      return `Remote output exceeded the configured byte cap.`;
    case 'remote_too_large':
      return `Remote file is larger than the configured max upload/download size.`;
    case 'local_target_exists':
      return `Local target already exists — refusing to overwrite. Choose a fresh path under output/.`;
    case 'forbidden_address':
    case 'invalid_host':
    case 'dns_failed':
    case 'connect_failed':
      return `SSH connection failed (${e.code}): ${e.message}`;
    default:
      return `SSH error (${e.code}): ${e.message}`;
  }
}

// ──────────────────────────────────────────────────────────────────────
// Pre-flight (steps 1-7): runs before audit row insertion. Returns either an
// error tool result (caller returns it directly) or the validated connection
// + computed extras needed for the tool body.
// ──────────────────────────────────────────────────────────────────────

interface PreflightSuccess {
  ok: true;
  connection: SshConnection;
  actingUserId: string;
  pieceName: string | undefined;
}

interface PreflightFailure {
  ok: false;
  error: ToolResult;
}

type PreflightResult = PreflightSuccess | PreflightFailure;

interface PreflightArgs {
  toolName: 'SshExec' | 'SshUpload' | 'SshDownload';
  connectionId: string;
  ctx: ToolContext;
  sub: SshSubsystem;
  /** Audit action label, e.g. 'ssh.exec' / 'ssh.upload' / 'ssh.download'. */
  auditAction: string;
}

function preflight(args: PreflightArgs): PreflightResult {
  const { toolName, connectionId, ctx, sub, auditAction } = args;

  const actingUserId = (ctx.userId ?? ctx.ownerId ?? '').toString();
  if (actingUserId.length === 0) {
    return {
      ok: false,
      error: err(`${toolName} requires an authenticated user context (none found).`),
    };
  }

  const pieceName = ctx.pieceName;
  const allowed = ctx.allowedSshConnections;

  // Step 2: piece must declare allowed_ssh_connections
  if (allowed === undefined) {
    return {
      ok: false,
      error: err(
        `${toolName}: this movement does not declare allowed_ssh_connections. Add the field to the piece YAML.`,
      ),
    };
  }

  // Step 3: piece membership check
  const wildcard = allowed.length === 1 && allowed[0] === '*';
  if (!wildcard && !allowed.includes(connectionId)) {
    // Probe the connection so we don't insert an audit row whose FK
    // (connection_id → ssh_connections.id) would fail when the LLM
    // passes a hallucinated UUID. If unknown, keep the request in
    // `detail` and clear the FK column.
    const exists = sub.connectionRepo.resolveConnection(connectionId) !== null;
    sub.auditRepo.beginAndComplete(
      {
        action: auditAction,
        connectionId: exists ? connectionId : undefined,
        actingUserId,
        pieceName,
        jobId: ctx.jobId ?? undefined,
        detail: exists
          ? { reason: 'piece_not_allowed', allowed }
          : { reason: 'piece_not_allowed', requested_connection_id: connectionId, allowed },
      },
      'denied',
    );
    return {
      ok: false,
      error: err(
        `${toolName}: connection ${connectionId} is not in this piece's allowed_ssh_connections list.`,
      ),
    };
  }

  // Step 4: resolveConnection
  const connection = sub.connectionRepo.resolveConnection(connectionId);
  if (!connection) {
    // The audit table FKs connection_id → ssh_connections(id). Passing an
    // unknown UUID/label here (the LLM occasionally hallucinates one)
    // would raise SqliteError "FOREIGN KEY constraint failed" — the
    // whole job aborts as a worker error. Record the requested id in
    // `detail` and pass undefined so the FK column is NULL.
    sub.auditRepo.beginAndComplete(
      {
        action: auditAction,
        connectionId: undefined,
        actingUserId,
        pieceName,
        jobId: ctx.jobId ?? undefined,
        detail: { reason: 'unknown_connection', requested_connection_id: connectionId },
      },
      'denied',
    );
    return {
      ok: false,
      error: err(`${toolName}: connection ${connectionId} does not exist.`),
    };
  }

  // Step 5: access decision
  const access = sub.getUserAccess(actingUserId);
  const decision = sub.accessResolver.resolveAccess({
    connection,
    userId: actingUserId,
    isAdmin: access.isAdmin,
    pieceName: pieceName ?? '',
    orgIds: access.orgIds,
  });
  if (!decision.allowed) {
    sub.auditRepo.beginAndComplete(
      {
        action: auditAction,
        connectionId,
        ownerId: connection.ownerId,
        actingUserId,
        pieceName,
        jobId: ctx.jobId ?? undefined,
        detail: { reason: decision.reason ?? 'no_grant' },
      },
      'denied',
    );
    return {
      ok: false,
      error: err(
        `${toolName}: access denied (${decision.reason ?? 'no_grant'}) for connection ${connectionId}.`,
      ),
    };
  }

  // Step 6 (shared): admin-disabled + abuse lock
  // (enabled/disabled was already handled inside accessResolver via the
  //  'disabled' deny reason, but disabled_by_admin = enabled false; double-check
  //  here in case logic shifts later.)
  if (!connection.enabled || connection.disabledByAdmin) {
    sub.auditRepo.beginAndComplete(
      {
        action: auditAction,
        connectionId,
        ownerId: connection.ownerId,
        actingUserId,
        pieceName,
        jobId: ctx.jobId ?? undefined,
        detail: {
          reason: 'disabled',
          disabled_by_admin: connection.disabledByAdmin,
          admin_reason: connection.disabledByAdminReason,
        },
      },
      'denied',
    );
    return {
      ok: false,
      error: err(
        `${toolName}: connection ${connectionId} is disabled${connection.disabledByAdmin ? ' (by admin: ' + (connection.disabledByAdminReason ?? 'no reason') + ')' : ''}.`,
      ),
    };
  }
  const lock = sub.abuseRepo.isLocked(connectionId);
  if (lock.locked) {
    sub.auditRepo.beginAndComplete(
      {
        action: auditAction,
        connectionId,
        ownerId: connection.ownerId,
        actingUserId,
        pieceName,
        jobId: ctx.jobId ?? undefined,
        detail: { reason: 'abuse_locked', until: lock.until },
      },
      'denied',
    );
    return {
      ok: false,
      error: err(
        `${toolName}: connection ${connectionId} is temporarily locked due to repeated failures (until ${lock.until}).`,
      ),
    };
  }

  return { ok: true, connection, actingUserId, pieceName };
}

// ──────────────────────────────────────────────────────────────────────
// SshExec
// ──────────────────────────────────────────────────────────────────────

async function runExec(
  input: Record<string, unknown>,
  ctx: ToolContext,
  sub: SshSubsystem,
): Promise<ToolResult> {
  const connectionId = typeof input.connection_id === 'string' ? input.connection_id : '';
  const command = typeof input.command === 'string' ? input.command : '';
  if (!connectionId || !command) {
    return err('SshExec: connection_id and command are required.');
  }
  const timeoutMs = typeof input.timeout_ms === 'number' && Number.isFinite(input.timeout_ms)
    ? Math.min(input.timeout_ms, sub.config.callTimeoutSeconds * 1000)
    : sub.config.callTimeoutSeconds * 1000;

  const pre = preflight({ toolName: 'SshExec', connectionId, ctx, sub, auditAction: 'ssh.exec' });
  if (!pre.ok) return pre.error;
  const { connection, actingUserId, pieceName } = pre;

  // Step 7: command policy (deny-list + custom rules)
  const customDeny = compilePatterns(connection.commandDenyPatterns);
  const customAllow = compilePatterns(connection.commandAllowPatterns);
  const cmdCheck = checkCommand({ command, customDenyPatterns: customDeny, customAllowPatterns: customAllow });
  if (!cmdCheck.allowed) {
    sub.auditRepo.beginAndComplete(
      {
        action: 'ssh.exec',
        connectionId,
        ownerId: connection.ownerId,
        actingUserId,
        pieceName,
        jobId: ctx.jobId ?? undefined,
        detail: {
          reason: 'deny_pattern',
          deny_reason: cmdCheck.reason,
          matched: cmdCheck.matched,
          command_hash: commandHash(command),
        },
      },
      'denied',
    );
    return err(
      `SshExec: command rejected by ${cmdCheck.reason === 'builtin_deny' ? 'built-in deny pattern' : cmdCheck.reason} (${cmdCheck.matched ?? 'n/a'}).`,
    );
  }

  // Step 8: open pending audit row.
  const auditId = sub.auditRepo.begin({
    action: 'ssh.exec',
    connectionId,
    ownerId: connection.ownerId,
    actingUserId,
    pieceName,
    jobId: ctx.jobId ?? undefined,
    detail: { command_hash: commandHash(command) },
  });

  // Step 9-11: decrypt → exec → success/failure book-keeping.
  let pemBuf: Buffer | null = null;
  let passBuf: Buffer | null = null;
  try {
    try {
      pemBuf = sub.decryptKeyMaterial(connection.ownerId, connection.privateKeyEnc);
      passBuf = sub.decryptPassphrase(connection.ownerId, connection.passphraseEnc);
    } catch (e) {
      sub.auditRepo.complete(auditId, 'failed', {
        error: 'decrypt_failed',
        msg: sanitizeError(e).message,
      });
      return err('SshExec: failed to decrypt stored key material (rotation may have invalidated it).');
    }
    const resolved = buildResolved(connection, pemBuf, passBuf, sub.config.allowPrivateAddresses);
    const hooks = buildHooks(sub, connection, actingUserId, pieceName, ctx.jobId ?? undefined);
    let result: ExecResult;
    try {
      result = await sub.sshExec(
        {
          connection: resolved,
          command,
          timeoutMs,
          maxOutputBytes: sub.config.maxOutputBytes,
        },
        hooks,
      );
    } catch (e) {
      const isSession = e instanceof SshSessionError;
      const code = isSession ? e.code : 'unknown_error';
      sub.abuseRepo.checkAndRecordFailure({
        connectionId,
        ownerId: connection.ownerId,
        userId: actingUserId,
        host: connection.host,
        username: connection.username,
      });
      sub.auditRepo.complete(auditId, 'failed', {
        error: code,
        msg: isSession ? e.message : String(e),
      });
      if (isSession) {
        return err(formatSessionError(e, connectionId));
      }
      return err(`SshExec failed: ${sanitizeError(e).message}`);
    }

    // Steps 10-11: success book-keeping.
    sub.abuseRepo.recordSuccess(connectionId);
    sub.auditRepo.complete(auditId, 'success', {
      exit_code: result.exitCode,
      duration_ms: result.durationMs,
      host_fingerprint: result.hostFingerprint,
    });
    return ok(result.outputJson);
  } finally {
    clearBuffer(pemBuf);
    clearBuffer(passBuf);
  }
}

// ──────────────────────────────────────────────────────────────────────
// SshUpload / SshDownload (shared transfer flow)
// ──────────────────────────────────────────────────────────────────────

async function runUpload(
  input: Record<string, unknown>,
  ctx: ToolContext,
  sub: SshSubsystem,
): Promise<ToolResult> {
  const connectionId = typeof input.connection_id === 'string' ? input.connection_id : '';
  const localPathInput = typeof input.local_path === 'string' ? input.local_path : '';
  const remotePathInput = typeof input.remote_path === 'string' ? input.remote_path : '';
  if (!connectionId || !localPathInput || !remotePathInput) {
    return err('SshUpload: connection_id, local_path, remote_path are required.');
  }
  const maxBytes = sub.config.maxUploadSizeMb * 1024 * 1024;
  const timeoutMs = typeof input.timeout_ms === 'number' && Number.isFinite(input.timeout_ms)
    ? Math.min(input.timeout_ms, sub.config.callTimeoutSeconds * 1000)
    : sub.config.callTimeoutSeconds * 1000;

  const pre = preflight({ toolName: 'SshUpload', connectionId, ctx, sub, auditAction: 'ssh.upload' });
  if (!pre.ok) return pre.error;
  const { connection, actingUserId, pieceName } = pre;

  // For Upload/Download we additionally require that the host key has been
  // user-verified — there is no LLM-actionable recovery from an unverified
  // host on a transfer (Exec has the verifier closure that captures the
  // observation for the user). Reject before opening an audit row.
  if (connection.hostKeyVerifiedAt === null) {
    sub.auditRepo.beginAndComplete(
      {
        action: 'ssh.upload',
        connectionId,
        ownerId: connection.ownerId,
        actingUserId,
        pieceName,
        jobId: ctx.jobId ?? undefined,
        detail: { reason: 'host_key_not_verified' },
      },
      'denied',
    );
    return err(
      `SshUpload: host key for connection ${connectionId} is not user-verified. Run SshExec first to surface the verify prompt.`,
    );
  }

  // Step 7: path policy
  const localCheck = await validateLocalPath({
    localPath: localPathInput,
    workspaceRoot: await resolveWorkspaceRoot(ctx),
    mustExist: true,
  });
  if (!localCheck.ok) {
    sub.auditRepo.beginAndComplete(
      {
        action: 'ssh.upload',
        connectionId,
        ownerId: connection.ownerId,
        actingUserId,
        pieceName,
        jobId: ctx.jobId ?? undefined,
        detail: { reason: 'local_path', detail: localCheck.reason },
      },
      'denied',
    );
    return err(`SshUpload: local path rejected (${localCheck.reason}).`);
  }
  const localPath = localCheck.resolved as string;
  let remoteResolved = remotePathInput;
  if (!connection.allowRemoteUnrestricted) {
    const remoteCheck = validateRemotePath(remotePathInput, connection.remotePathPrefix);
    if (!remoteCheck.ok) {
      sub.auditRepo.beginAndComplete(
        {
          action: 'ssh.upload',
          connectionId,
          ownerId: connection.ownerId,
          actingUserId,
          pieceName,
          jobId: ctx.jobId ?? undefined,
          detail: { reason: 'remote_path', detail: remoteCheck.reason },
        },
        'denied',
      );
      return err(`SshUpload: remote path rejected (${remoteCheck.reason}).`);
    }
    remoteResolved = remoteCheck.normalized as string;
  }

  // Pre-check the local size cap to avoid streaming a too-large file.
  const st = fs.statSync(localPath);
  if (!st.isFile()) {
    return err('SshUpload: local path is not a regular file.');
  }
  if (st.size > maxBytes) {
    return err(`SshUpload: local file (${st.size}B) exceeds max upload size (${maxBytes}B).`);
  }

  // Step 8: pending audit row.
  const auditId = sub.auditRepo.begin({
    action: 'ssh.upload',
    connectionId,
    ownerId: connection.ownerId,
    actingUserId,
    pieceName,
    jobId: ctx.jobId ?? undefined,
    detail: { local: path.basename(localPath), remote: remoteResolved, bytes: st.size },
  });

  // Step 9-11.
  let pemBuf: Buffer | null = null;
  let passBuf: Buffer | null = null;
  try {
    try {
      pemBuf = sub.decryptKeyMaterial(connection.ownerId, connection.privateKeyEnc);
      passBuf = sub.decryptPassphrase(connection.ownerId, connection.passphraseEnc);
    } catch (e) {
      sub.auditRepo.complete(auditId, 'failed', {
        error: 'decrypt_failed',
        msg: sanitizeError(e).message,
      });
      return err('SshUpload: failed to decrypt stored key material.');
    }
    const resolved = buildResolved(connection, pemBuf, passBuf, sub.config.allowPrivateAddresses);
    const hooks = buildHooks(sub, connection, actingUserId, pieceName, ctx.jobId ?? undefined);
    let result: TransferResult;
    try {
      result = await sub.sshUpload(
        { connection: resolved, localPath, remotePath: remoteResolved, timeoutMs, maxBytes },
        hooks,
      );
    } catch (e) {
      const isSession = e instanceof SshSessionError;
      const code = isSession ? e.code : 'unknown_error';
      sub.abuseRepo.checkAndRecordFailure({
        connectionId,
        ownerId: connection.ownerId,
        userId: actingUserId,
        host: connection.host,
        username: connection.username,
      });
      sub.auditRepo.complete(auditId, 'failed', {
        error: code,
        msg: isSession ? e.message : String(e),
      });
      if (isSession) {
        return err(formatSessionError(e, connectionId));
      }
      return err(`SshUpload failed: ${sanitizeError(e).message}`);
    }
    sub.abuseRepo.recordSuccess(connectionId);
    sub.auditRepo.complete(auditId, 'success', {
      bytes: result.bytes,
      duration_ms: result.durationMs,
      host_fingerprint: result.hostFingerprint,
    });
    return ok(
      JSON.stringify(
        {
          ok: true,
          bytes: result.bytes,
          duration_ms: result.durationMs,
          remote: remoteResolved,
          host_fingerprint: result.hostFingerprint,
        },
        null,
        2,
      ),
    );
  } finally {
    clearBuffer(pemBuf);
    clearBuffer(passBuf);
  }
}

async function runDownload(
  input: Record<string, unknown>,
  ctx: ToolContext,
  sub: SshSubsystem,
): Promise<ToolResult> {
  const connectionId = typeof input.connection_id === 'string' ? input.connection_id : '';
  const remotePathInput = typeof input.remote_path === 'string' ? input.remote_path : '';
  const localPathInput = typeof input.local_path === 'string' ? input.local_path : '';
  if (!connectionId || !remotePathInput || !localPathInput) {
    return err('SshDownload: connection_id, remote_path, local_path are required.');
  }
  const maxBytes = sub.config.maxDownloadSizeMb * 1024 * 1024;
  const timeoutMs = typeof input.timeout_ms === 'number' && Number.isFinite(input.timeout_ms)
    ? Math.min(input.timeout_ms, sub.config.callTimeoutSeconds * 1000)
    : sub.config.callTimeoutSeconds * 1000;

  const pre = preflight({ toolName: 'SshDownload', connectionId, ctx, sub, auditAction: 'ssh.download' });
  if (!pre.ok) return pre.error;
  const { connection, actingUserId, pieceName } = pre;

  if (connection.hostKeyVerifiedAt === null) {
    sub.auditRepo.beginAndComplete(
      {
        action: 'ssh.download',
        connectionId,
        ownerId: connection.ownerId,
        actingUserId,
        pieceName,
        jobId: ctx.jobId ?? undefined,
        detail: { reason: 'host_key_not_verified' },
      },
      'denied',
    );
    return err(
      `SshDownload: host key for connection ${connectionId} is not user-verified. Run SshExec first to surface the verify prompt.`,
    );
  }

  // Path policy.
  const localCheck = await validateLocalPath({
    localPath: localPathInput,
    workspaceRoot: await resolveWorkspaceRoot(ctx),
    mustExist: false,
  });
  if (!localCheck.ok) {
    sub.auditRepo.beginAndComplete(
      {
        action: 'ssh.download',
        connectionId,
        ownerId: connection.ownerId,
        actingUserId,
        pieceName,
        jobId: ctx.jobId ?? undefined,
        detail: { reason: 'local_path', detail: localCheck.reason },
      },
      'denied',
    );
    return err(`SshDownload: local path rejected (${localCheck.reason}).`);
  }
  const localPath = localCheck.resolved as string;
  let remoteResolved = remotePathInput;
  if (!connection.allowRemoteUnrestricted) {
    const remoteCheck = validateRemotePath(remotePathInput, connection.remotePathPrefix);
    if (!remoteCheck.ok) {
      sub.auditRepo.beginAndComplete(
        {
          action: 'ssh.download',
          connectionId,
          ownerId: connection.ownerId,
          actingUserId,
          pieceName,
          jobId: ctx.jobId ?? undefined,
          detail: { reason: 'remote_path', detail: remoteCheck.reason },
        },
        'denied',
      );
      return err(`SshDownload: remote path rejected (${remoteCheck.reason}).`);
    }
    remoteResolved = remoteCheck.normalized as string;
  }

  const auditId = sub.auditRepo.begin({
    action: 'ssh.download',
    connectionId,
    ownerId: connection.ownerId,
    actingUserId,
    pieceName,
    jobId: ctx.jobId ?? undefined,
    detail: { remote: remoteResolved, local: path.basename(localPath) },
  });

  let pemBuf: Buffer | null = null;
  let passBuf: Buffer | null = null;
  try {
    try {
      pemBuf = sub.decryptKeyMaterial(connection.ownerId, connection.privateKeyEnc);
      passBuf = sub.decryptPassphrase(connection.ownerId, connection.passphraseEnc);
    } catch (e) {
      sub.auditRepo.complete(auditId, 'failed', {
        error: 'decrypt_failed',
        msg: sanitizeError(e).message,
      });
      return err('SshDownload: failed to decrypt stored key material.');
    }
    const resolved = buildResolved(connection, pemBuf, passBuf, sub.config.allowPrivateAddresses);
    const hooks = buildHooks(sub, connection, actingUserId, pieceName, ctx.jobId ?? undefined);
    let result: TransferResult;
    try {
      result = await sub.sshDownload(
        { connection: resolved, remotePath: remoteResolved, localPath, timeoutMs, maxBytes },
        hooks,
      );
    } catch (e) {
      const isSession = e instanceof SshSessionError;
      const code = isSession ? e.code : 'unknown_error';
      sub.abuseRepo.checkAndRecordFailure({
        connectionId,
        ownerId: connection.ownerId,
        userId: actingUserId,
        host: connection.host,
        username: connection.username,
      });
      sub.auditRepo.complete(auditId, 'failed', {
        error: code,
        msg: isSession ? e.message : String(e),
      });
      if (isSession) {
        return err(formatSessionError(e, connectionId));
      }
      return err(`SshDownload failed: ${sanitizeError(e).message}`);
    }
    sub.abuseRepo.recordSuccess(connectionId);
    sub.auditRepo.complete(auditId, 'success', {
      bytes: result.bytes,
      duration_ms: result.durationMs,
      host_fingerprint: result.hostFingerprint,
    });
    return ok(
      JSON.stringify(
        {
          ok: true,
          bytes: result.bytes,
          duration_ms: result.durationMs,
          local: localPath,
          host_fingerprint: result.hostFingerprint,
        },
        null,
        2,
      ),
    );
  } finally {
    clearBuffer(pemBuf);
    clearBuffer(passBuf);
  }
}

async function resolveWorkspaceRoot(ctx: ToolContext): Promise<string> {
  // Caller's workspacePath is the absolute path; fs.realpath collapses any
  // ancestral symlinks (validateLocalPath's contract).
  return fs.promises.realpath(ctx.workspacePath);
}

// ──────────────────────────────────────────────────────────────────────
// SshListConnections
// ──────────────────────────────────────────────────────────────────────

async function runListConnections(
  _input: Record<string, unknown>,
  ctx: ToolContext,
  sub: SshSubsystem,
): Promise<ToolResult> {
  const actingUserId = (ctx.userId ?? ctx.ownerId ?? '').toString();
  if (actingUserId.length === 0) {
    return err('SshListConnections requires an authenticated user context (none found).');
  }
  const allowed = ctx.allowedSshConnections;
  if (allowed === undefined) {
    return err(
      'SshListConnections: this movement does not declare allowed_ssh_connections. Add the field to the piece YAML.',
    );
  }

  const wildcard = allowed.length === 1 && allowed[0] === '*';
  const access = sub.getUserAccess(actingUserId);

  // Filter chain: enabled + piece-membership + access-grant. We deliberately
  // skip the abuse lock here — a temporarily-locked connection is still a
  // valid choice once the lock expires, and surfacing it lets the LLM tell
  // the user "this one is locked until X" instead of going silent.
  const visible: Array<Record<string, unknown>> = [];
  for (const conn of sub.connectionRepo.listAll()) {
    if (!conn.enabled || conn.disabledByAdmin) continue;
    if (!wildcard && !allowed.includes(conn.id)) continue;
    const decision = sub.accessResolver.resolveAccess({
      connection: conn,
      userId: actingUserId,
      isAdmin: access.isAdmin,
      pieceName: ctx.pieceName ?? '',
      orgIds: access.orgIds,
    });
    if (!decision.allowed) continue;
    visible.push({
      id: conn.id,
      label: conn.label,
      host: conn.host,
      port: conn.port,
      username: conn.username,
      host_key_verified: conn.hostKeyVerifiedAt !== null,
      host_key_pending: conn.hostKeyPending,
    });
  }

  sub.auditRepo.beginAndComplete(
    {
      action: 'ssh.list_connections',
      actingUserId,
      pieceName: ctx.pieceName,
      jobId: ctx.jobId ?? undefined,
      detail: { count: visible.length, wildcard },
    },
    'success',
  );

  return ok(JSON.stringify({ connections: visible }, null, 2));
}

// ──────────────────────────────────────────────────────────────────────
// Dispatcher
// ──────────────────────────────────────────────────────────────────────

export async function executeTool(
  name: string,
  input: Record<string, unknown>,
  ctx: ToolContext,
): Promise<ToolResult | null> {
  if (!SSH_TOOL_NAMES.has(name)) return null;

  if (!_subsystem) {
    return err('SSH subsystem is not initialised (ssh.enabled=false or MCP_ENCRYPTION_KEY missing).');
  }
  if (_subsystem.maintenance.isActive()) {
    return err('SSH subsystem is in maintenance — retry in a moment.');
  }

  if (name === 'SshExec') return runExec(input, ctx, _subsystem);
  if (name === 'SshUpload') return runUpload(input, ctx, _subsystem);
  if (name === 'SshDownload') return runDownload(input, ctx, _subsystem);
  if (name === 'SshListConnections') return runListConnections(input, ctx, _subsystem);
  return null;
}

// Re-export BeginAuditInput to ease test fixture imports without pulling in
// the full audit-repo path.
export type { BeginAuditInput };

// Phase 3 (SSH Console): expose preflight + a few helpers so the new console
// tool module can reuse the exact same access / state checks as SshExec.
export { preflight, buildResolved, buildHooks };
export type { PreflightResult, PreflightArgs };
