atlas-cli.ts

#!/usr/bin/env node
import { Command, CommanderError } from "commander";
import fs from "node:fs";
import path from "node:path";
import { fileURLToPath } from "node:url";

import { runAtlasHarvestCli } from "./atlas/harvest-cli.js";
import { collectMissingRequiredEnv } from "./config.js";

const DEFAULT_TOOL = "atlas-search";
const DEFAULT_FEEDBACK_TOOL = "submit-feedback";
const FEEDBACK_RATINGS = ["helpful", "not_helpful"] as const;
const DEFAULT_MCP_URL = "https://mcp.pathfinder.copilotkit.dev/mcp";
const INTEGER_PATTERN = /^[1-9]\d*$/;
const NUMBER_PATTERN = /^[+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?$/;

type WriteFn = (text: string) => void;

interface AtlasCliIo {
  stdout?: WriteFn;
  stderr?: WriteFn;
}

interface SearchOptions {
  json?: boolean;
  limit?: string;
  minScore?: string;
  token?: string;
  tool?: string;
  url?: string;
}

interface FeedbackOptions {
  comment?: string;
  for?: string;
  json?: boolean;
  rating?: string;
  token?: string;
  tool?: string;
  url?: string;
}

interface JsonRpcMessage {
  jsonrpc?: string;
  // JSON-RPC permits numeric, string, or null ids; a proxy may echo "1" for 1,
  // and error frames may legitimately carry a null id. Match by coerced id and
  // never let a strict numeric type discard a real response frame.
  id?: number | string | null;
  result?: unknown;
  error?: {
    message?: string;
  };
}

interface McpPostResult {
  messages: JsonRpcMessage[];
  sessionId: string | null;
}

function parseSseMessages(text: string): JsonRpcMessage[] {
  const messages: JsonRpcMessage[] = [];
  let dataLines: string[] = [];

  const flushEvent = () => {
    if (dataLines.length === 0) {
      return;
    }

    const data = dataLines.join("\n");
    dataLines = [];

    // Skip keepalive / empty `data:` frames: an empty value yields `""`,
    // which would otherwise crash on JSON.parse("").
    if (data.trim() === "") {
      return;
    }

    try {
      messages.push(JSON.parse(data) as JsonRpcMessage);
    } catch {
      // Skip unparseable keepalive frames rather than crashing the search.
    }
  };

  for (const rawLine of text.split(/\r?\n/)) {
    if (rawLine === "") {
      flushEvent();
      continue;
    }

    if (!rawLine.startsWith("data:")) {
      continue;
    }

    const data = rawLine.slice(5);
    dataLines.push(data.startsWith(" ") ? data.slice(1) : data);
  }

  flushEvent();

  return messages;
}

async function mcpPost(
  url: string,
  body: unknown,
  options: {
    onSessionId?: (sessionId: string) => void;
    sessionId?: string;
    token?: string;
  } = {},
): Promise<McpPostResult> {
  const headers: Record<string, string> = {
    "Content-Type": "application/json",
    Accept: "application/json, text/event-stream",
  };

  if (options.sessionId) {
    headers["Mcp-Session-Id"] = options.sessionId;
  }

  if (options.token) {
    headers.Authorization = `Bearer ${options.token}`;
  }

  const response = await fetch(url, {
    method: "POST",
    headers,
    body: JSON.stringify(body),
  });

  const nextSessionId =
    response.headers.get("mcp-session-id") ?? options.sessionId ?? null;
  if (nextSessionId) {
    options.onSessionId?.(nextSessionId);
  }

  if (response.status === 202) {
    return { messages: [], sessionId: nextSessionId };
  }

  if (!response.ok) {
    throw new Error(`HTTP ${response.status}: ${await response.text()}`);
  }

  const text = await response.text();
  const messages = parseSseMessages(text);

  if (messages.length > 0) {
    return { messages, sessionId: nextSessionId };
  }

  if (!text.trim()) {
    return { messages: [], sessionId: nextSessionId };
  }

  try {
    return {
      messages: [JSON.parse(text) as JsonRpcMessage],
      sessionId: nextSessionId,
    };
  } catch {
    throw new Error(`Unparseable response: ${text.slice(0, 200)}`);
  }
}

async function closeMcpSession(
  url: string,
  options: {
    sessionId?: string;
    token?: string;
  },
): Promise<void> {
  if (!options.sessionId) {
    return;
  }

  const headers: Record<string, string> = {
    "Mcp-Session-Id": options.sessionId,
  };

  if (options.token) {
    headers.Authorization = `Bearer ${options.token}`;
  }

  try {
    await fetch(url, {
      method: "DELETE",
      headers,
    });
  } catch {
    // Best-effort cleanup must not mask the search result or original failure.
  }
}

function buildToolArguments(
  query: string,
  options: SearchOptions,
): Record<string, unknown> {
  const args: Record<string, unknown> = { query };

  if (options.limit !== undefined) {
    if (!INTEGER_PATTERN.test(options.limit)) {
      throw new Error("limit must be a positive integer");
    }

    const limit = Number(options.limit);
    if (!Number.isSafeInteger(limit)) {
      throw new Error("limit must be a positive integer");
    }
    args.limit = limit;
  }

  if (options.minScore !== undefined) {
    if (!NUMBER_PATTERN.test(options.minScore)) {
      throw new Error("min-score must be a finite number in [0, 1]");
    }

    const minScore = Number(options.minScore);
    if (!Number.isFinite(minScore) || minScore < 0 || minScore > 1) {
      throw new Error("min-score must be a finite number in [0, 1]");
    }
    args.min_score = minScore;
  }

  return args;
}

export function buildFeedbackArguments(
  query: string,
  options: FeedbackOptions,
): Record<string, unknown> {
  const rating = options.rating;
  if (
    rating === undefined ||
    !(FEEDBACK_RATINGS as readonly string[]).includes(rating)
  ) {
    throw new Error(`rating must be one of: ${FEEDBACK_RATINGS.join(", ")}`);
  }

  const comment = options.comment;
  if (comment === undefined || comment.trim() === "") {
    throw new Error("comment must not be empty");
  }

  // --for always resolves via its Commander default, so a missing value here
  // means the default was dropped — fail loud rather than sending an undefined
  // tool_name (symmetric with the --tool guard in feedback()/search()).
  if (options.for === undefined) {
    throw new Error("atlas: --for is required");
  }

  return {
    tool_name: options.for,
    query,
    rating,
    comment,
  };
}

function printToolText(message: JsonRpcMessage, write: WriteFn): void {
  const result = message.result as
    { content?: Array<{ type?: string; text?: string }> } | undefined;
  const content = Array.isArray(result?.content) ? result.content : [];
  const textItems = content
    .map((item) => item.text)
    .filter((text): text is string => typeof text === "string");

  if (textItems.length === 0) {
    write("No results.\n");
    return;
  }

  for (const text of textItems) {
    write(`${text}\n`);
  }
}

interface CallToolOptions {
  json?: boolean;
  token?: string;
  url?: string;
}

async function callTool(
  toolName: string,
  toolArguments: Record<string, unknown>,
  options: CallToolOptions,
  write: WriteFn,
): Promise<void> {
  const url = options.url ?? process.env.ATLAS_MCP_URL ?? DEFAULT_MCP_URL;
  const token = options.token ?? process.env.ATLAS_TOKEN;
  let sessionId: string | undefined;
  const recordSessionId = (nextSessionId: string) => {
    sessionId = nextSessionId;
  };

  try {
    const init = await mcpPost(
      url,
      {
        jsonrpc: "2.0",
        method: "initialize",
        id: 0,
        params: {
          protocolVersion: "2025-03-26",
          capabilities: {},
          clientInfo: { name: "atlas", version: "1.0.0" },
        },
      },
      { onSessionId: recordSessionId, token },
    );

    sessionId = init.sessionId ?? undefined;
    const initError = init.messages.find((item) => item.error)?.error;
    if (initError) {
      throw new Error(initError.message ?? "MCP initialize failed");
    }

    await mcpPost(
      url,
      { jsonrpc: "2.0", method: "notifications/initialized" },
      { onSessionId: recordSessionId, sessionId, token },
    );

    const toolsCallRequestId = 1;
    const response = await mcpPost(
      url,
      {
        jsonrpc: "2.0",
        method: "tools/call",
        id: toolsCallRequestId,
        params: {
          name: toolName,
          arguments: toolArguments,
        },
      },
      { onSessionId: recordSessionId, sessionId, token },
    );

    // Select the response frame with an exhaustive three-tier fallback:
    //   1. the frame whose coerced id matches the request (a proxy echoing "1"
    //      for 1 still matches) and which actually carries a result or error;
    //   2. else an error frame whose id is null/omitted, surfacing a real
    //      server error that legitimately dropped its id;
    //   3. else a result frame whose id is null/omitted, tolerating an omitted
    //      id on a sole result frame.
    // Tiers 2 and 3 only consider id-less frames so a frame bearing a clearly
    // different explicit id is never surfaced for this request.
    // Only when none of these exist do we declare "no response".
    const message =
      response.messages.find(
        (item) =>
          String(item.id) === String(toolsCallRequestId) &&
          ("result" in item || "error" in item),
      ) ??
      response.messages.find((item) => "error" in item && item.id == null) ??
      response.messages.find((item) => "result" in item && item.id == null);
    if (!message) {
      throw new Error("atlas: no response from server for tools/call");
    }

    if (message.error) {
      throw new Error(message.error.message ?? "MCP tool call failed");
    }

    const toolResult = message.result as
      | { isError?: boolean; content?: Array<{ type?: string; text?: string }> }
      | null
      | undefined;
    if (toolResult?.isError === true) {
      const errorContent = Array.isArray(toolResult.content)
        ? toolResult.content
        : [];
      const errorText = errorContent
        .map((item) => item.text)
        .filter((text): text is string => typeof text === "string")
        .join("\n");
      throw new Error(errorText || "MCP tool call reported an error");
    }

    if (options.json) {
      write(`${JSON.stringify(message, null, 2)}\n`);
      return;
    }

    printToolText(message, write);
  } finally {
    await closeMcpSession(url, { sessionId, token });
  }
}

async function search(
  query: string,
  options: SearchOptions,
  write: WriteFn,
): Promise<void> {
  // --tool always resolves via its Commander default, so a missing value here
  // means the default was dropped — fail loud rather than silently re-default.
  if (options.tool === undefined) {
    throw new Error("atlas: --tool is required");
  }
  const toolArguments = buildToolArguments(query, options);

  await callTool(options.tool, toolArguments, options, write);
}

async function feedback(
  query: string,
  options: FeedbackOptions,
  write: WriteFn,
): Promise<void> {
  // --tool always resolves via its Commander default, so a missing value here
  // means the default was dropped — fail loud rather than silently re-default.
  if (options.tool === undefined) {
    throw new Error("atlas: --tool is required");
  }
  const toolArguments = buildFeedbackArguments(query, options);

  await callTool(options.tool, toolArguments, options, write);
}

/**
 * Env preflight — enumerate EVERY missing required environment variable in
 * one pass and report them before any harvest/search work runs. Intentionally
 * lives here in atlas-cli.ts (NOT a harvest-cli subcommand) so it stays off
 * the harvest-cli serialization chain.
 *
 * Fails-loud with a non-zero exit when anything is missing so an operator
 * running `atlas preflight` in a deploy check sees the full list up front —
 * unlike server boot, which throws on the first failing group. Returns 0 and
 * prints an OK line when the environment is fully configured.
 */
export function runPreflight(writeOut: WriteFn, writeErr: WriteFn): number {
  const missing = collectMissingRequiredEnv();
  if (missing.length === 0) {
    writeOut("atlas preflight: OK — all required environment variables set.\n");
    return 0;
  }
  writeErr(
    `atlas preflight: missing required environment variables:\n${missing
      .map((m) => `  - ${m}`)
      .join(
        "\n",
      )}\nSet them before starting the server or running a harvest.\n`,
  );
  return 1;
}

export async function runAtlasCli(
  argv: string[] = process.argv.slice(2),
  io: AtlasCliIo = {},
): Promise<number> {
  const writeOut = io.stdout ?? ((text: string) => process.stdout.write(text));
  const writeErr = io.stderr ?? ((text: string) => process.stderr.write(text));

  // `harvest` short-circuits BEFORE commander parses: the raw tail is
  // forwarded to the harvest driver genuinely verbatim — order intact,
  // INCLUDING a leading `--`, which commander's `[args...]` variadic would
  // otherwise consume as its own operand separator and silently drop,
  // turning standalone-inert operands back into parsed options. The driver
  // (src/atlas/harvest-cli.ts) owns its own commander program, io wiring,
  // exit codes, and stderr formatting (formatCliError), so `atlas harvest
  // run --run-id ...` behaves exactly like running the driver module
  // directly.
  if (argv[0] === "harvest") {
    return runAtlasHarvestCli(argv.slice(1), {
      stdout: writeOut,
      stderr: writeErr,
    });
  }

  const program = new Command();
  program
    .name("atlas")
    .description("Agent-facing Atlas search over Pathfinder MCP")
    .exitOverride()
    // Required so the `harvest` mount below can use passThroughOptions():
    // option processing stops at the first subcommand, leaving each verb to
    // parse its own flags (search/feedback already declare all their options
    // locally, so their behavior is unchanged).
    .enablePositionalOptions()
    .configureOutput({
      writeOut,
      writeErr,
      outputError: (text, write) => write(text),
    });

  program
    .command("search")
    .description("Search Atlas knowledge through a Pathfinder MCP endpoint")
    .argument("<query>", "Search query")
    .option("--url <url>", "Pathfinder MCP URL")
    .option("--token <token>", "Bearer token for the MCP endpoint")
    .option("--tool <name>", "MCP tool name", DEFAULT_TOOL)
    .option("--limit <n>", "Maximum number of results")
    .option("--min-score <score>", "Minimum search score")
    .option("--json", "Print the raw MCP JSON-RPC response")
    .action(async (query: string, options: SearchOptions) => {
      await search(query, options, writeOut);
    });

  program
    .command("feedback")
    .description(
      "Submit Atlas retrieval feedback through a Pathfinder MCP endpoint",
    )
    .argument("<query>", "The query the feedback is about")
    .requiredOption(
      "--rating <rating>",
      "Feedback rating (helpful or not_helpful)",
    )
    .requiredOption("--comment <text>", "Free-form feedback comment")
    .option("--for <tool_name>", "Tool the feedback is about", DEFAULT_TOOL)
    .option("--url <url>", "Pathfinder MCP URL")
    .option("--token <token>", "Bearer token for the MCP endpoint")
    .option("--tool <name>", "MCP tool name", DEFAULT_FEEDBACK_TOOL)
    .option("--json", "Print the raw MCP JSON-RPC response")
    .action(async (query: string, options: FeedbackOptions) => {
      await feedback(query, options, writeOut);
    });

  let preflightExitCode: number | undefined;
  program
    .command("preflight")
    .description(
      "Report every missing required environment variable in one pass " +
        "(fatal-in-production keys + source-gated tokens) before running the server or a harvest",
    )
    .action(() => {
      preflightExitCode = runPreflight(writeOut, writeErr);
    });

  // The harvest DRIVER (src/atlas/harvest-cli.ts) as a registered verb.
  // Execution is handled by the pre-parse short-circuit at the top of
  // runAtlasCli (which forwards the raw tail verbatim, leading `--`
  // included), so this registration is UNREACHABLE for `atlas harvest ...`
  // invocations. It is kept so `atlas --help` still lists the verb, and as a
  // correct fallback for any commander-routed path.
  let harvestExitCode: number | undefined;
  program
    .command("harvest")
    .description(
      "Atlas harvest driver — run the pipeline over a fragment corpus and " +
        "drive ratification/index (subcommands: run, artifact, sync, reindex)",
    )
    .helpOption(false)
    .allowUnknownOption()
    .allowExcessArguments(true)
    .passThroughOptions()
    .argument("[args...]", "Arguments forwarded to the harvest driver")
    .action(async (args: string[]) => {
      harvestExitCode = await runAtlasHarvestCli(args, {
        stdout: writeOut,
        stderr: writeErr,
      });
    });

  try {
    await program.parseAsync(argv, { from: "user" });
    return preflightExitCode ?? harvestExitCode ?? 0;
  } catch (error) {
    if (error instanceof CommanderError) {
      return error.exitCode;
    }

    const message = error instanceof Error ? error.message : String(error);
    writeErr(`error: ${message}\n`);
    return 1;
  }
}

export function isAtlasCliEntrypoint(
  moduleUrl: string,
  argvPath: string | undefined,
): boolean {
  if (!argvPath) {
    return false;
  }

  return (
    resolveEntrypointPath(fileURLToPath(moduleUrl)) ===
    resolveEntrypointPath(argvPath)
  );
}

function resolveEntrypointPath(candidatePath: string): string {
  const normalizedPath = path.resolve(candidatePath);

  try {
    return fs.realpathSync(normalizedPath);
  } catch {
    return normalizedPath;
  }
}

if (isAtlasCliEntrypoint(import.meta.url, process.argv[1])) {
  runAtlasCli()
    .then((exitCode) => {
      process.exitCode = exitCode;
    })
    .catch((error) => {
      process.stderr.write(
        `error: ${error instanceof Error ? error.message : String(error)}\n`,
      );
      process.exitCode = 1;
    });
}