fusion/packages/cli/src/commands/chat.ts

import { AgentStore } from "@fusion/core";
import type { Message } from "@fusion/core";
import { createMessageStore, formatParticipant, formatTime, CLI_USER_ID } from "./message.js";
import { resolveProject } from "../project-context.js";
import { createInterface } from "node:readline/promises";

const MAX_MESSAGE_LENGTH = 8192;
const DEFAULT_POLL_MS = 1000;
const HISTORY_LIMIT = 20;

export interface ChatInteractiveOptions {
  project?: string;
  pollIntervalMs?: number;
  once?: boolean;
  nonInteractive?: boolean;
  input?: NodeJS.ReadableStream;
  output?: NodeJS.WritableStream;
}

async function getProjectPath(projectName?: string): Promise<string> {
  if (projectName) {
    const context = await resolveProject(projectName);
    return context.projectPath;
  }

  try {
    const context = await resolveProject(undefined);
    return context.projectPath;
  } catch {
    return process.cwd();
  }
}

async function createAgentStore(projectName?: string): Promise<AgentStore> {
  const projectPath = await getProjectPath(projectName);
  const store = new AgentStore({ rootDir: `${projectPath}/.fusion` });
  await store.init();
  return store;
}

function parsePollMs(options: ChatInteractiveOptions): number {
  const envValue = process.env.FUSION_CHAT_POLL_MS;
  const envPollMs = envValue ? Number.parseInt(envValue, 10) : Number.NaN;
  const candidate = options.pollIntervalMs ?? (Number.isFinite(envPollMs) ? envPollMs : DEFAULT_POLL_MS);
  return Number.isFinite(candidate) && candidate > 0 ? candidate : DEFAULT_POLL_MS;
}

function printMessage(output: NodeJS.WritableStream, message: Message): void {
  const fromLabel = formatParticipant(message.fromId, message.fromType);
  const time = formatTime(message.createdAt);
  output.write(`${fromLabel} — ${time}\n`);
  output.write(`${message.content}\n\n`);
}

function printConversationTail(output: NodeJS.WritableStream, messages: Message[]): void {
  if (messages.length === 0) {
    output.write("\nNo messages yet.\n\n");
    return;
  }

  output.write("\nRecent conversation:\n\n");
  for (const message of messages) {
    printMessage(output, message);
  }
}

function sleep(ms: number, signal: AbortSignal): Promise<void> {
  return new Promise((resolve, reject) => {
    const timer = setTimeout(resolve, ms);
    const onAbort = () => {
      clearTimeout(timer);
      reject(new Error("aborted"));
    };
    if (signal.aborted) {
      onAbort();
      return;
    }
    signal.addEventListener("abort", onAbort, { once: true });
  });
}

async function waitForReply(
  messageStore: Awaited<ReturnType<typeof createMessageStore>>["store"],
  agentId: string,
  printedIds: Set<string>,
  output: NodeJS.WritableStream,
  pollIntervalMs: number,
  timeoutMs: number,
): Promise<boolean> {
  const started = Date.now();
  while (Date.now() - started < timeoutMs) {
    const inbox = messageStore.getInbox(CLI_USER_ID, "user", { limit: 50 });
    for (const message of inbox.slice().reverse()) {
      if (message.fromId !== agentId || message.fromType !== "agent") continue;
      if (printedIds.has(message.id)) continue;
      printedIds.add(message.id);
      printMessage(output, message);
      messageStore.markAsRead(message.id);
      return true;
    }
    await new Promise((resolve) => setTimeout(resolve, pollIntervalMs));
  }
  return false;
}

export async function runChatInteractive(agentId: string, options: ChatInteractiveOptions = {}): Promise<number> {
  const output = options.output ?? process.stdout;
  const input = options.input ?? process.stdin;
  const pollIntervalMs = parsePollMs(options);

  const agentStore = await createAgentStore(options.project);
  const agent = await agentStore.getAgent(agentId);
  if (!agent) {
    console.error(`Agent ${agentId} not found`);
    return 1;
  }

  const { store: messageStore, db } = await createMessageStore(options.project);
  const printedIds = new Set<string>();

  const conversation = messageStore.getConversation(
    { id: CLI_USER_ID, type: "user" },
    { id: agentId, type: "agent" },
  );
  const tail = conversation.slice(-HISTORY_LIMIT);
  for (const message of tail) printedIds.add(message.id);

  output.write(`Chat with Agent ${agentId} — type /exit or Ctrl-C to quit, /help for commands\n`);
  output.write("Replies appear when this project's engine is running (fn dashboard or fn serve).\n");
  printConversationTail(output, tail);

  const runOnce = options.once === true;
  try {
    if (runOnce) {
      const content = await readSingleMessage(input, output, options.nonInteractive);
      if (!content.trim()) return 0;

      if (content.length > MAX_MESSAGE_LENGTH) {
        console.error(`Message too long; max ${MAX_MESSAGE_LENGTH} chars`);
        return 0;
      }

      messageStore.sendMessage({
        fromId: CLI_USER_ID,
        fromType: "user",
        toId: agentId,
        toType: "agent",
        content,
        type: "user-to-agent",
        metadata: { wakeRecipient: true },
      });

      output.write(`you → ${agentId}: ${content}\n`);
      const timeoutMs = Math.max(pollIntervalMs * 10, 30_000);
      const replied = await waitForReply(messageStore, agentId, printedIds, output, pollIntervalMs, timeoutMs);
      if (!replied) {
        console.error(`No reply within ${Math.ceil(timeoutMs / 1000)}s`);
      }
      return 0;
    }

    const abortController = new AbortController();
    const poller = (async () => {
      while (!abortController.signal.aborted) {
        const inbox = messageStore.getInbox(CLI_USER_ID, "user", { limit: 50 });
        for (const message of inbox.slice().reverse()) {
          if (message.fromId !== agentId || message.fromType !== "agent") continue;
          if (printedIds.has(message.id)) continue;
          printedIds.add(message.id);
          printMessage(output, message);
          messageStore.markAsRead(message.id);
        }
        await sleep(pollIntervalMs, abortController.signal);
      }
    })().catch(() => undefined);

    const rl = createInterface({ input, output });
    rl.on("close", () => abortController.abort());

    while (true) {
      let line: string;
      try {
        line = (await rl.question("> ")).trim();
      } catch {
        break;
      }
      if (!line) continue;
      if (line === "/exit" || line === "/quit") break;
      if (line === "/help") {
        output.write("Commands: /help, /history, /clear, /exit, /quit\n");
        continue;
      }
      if (line === "/history") {
        const history = messageStore.getConversation(
          { id: CLI_USER_ID, type: "user" },
          { id: agentId, type: "agent" },
        ).slice(-HISTORY_LIMIT);
        for (const message of history) printedIds.add(message.id);
        printConversationTail(output, history);
        continue;
      }
      if (line === "/clear") {
        output.write("\x1b[2J\x1b[H");
        continue;
      }
      if (line.length > MAX_MESSAGE_LENGTH) {
        console.error(`Message too long; max ${MAX_MESSAGE_LENGTH} chars`);
        continue;
      }

      messageStore.sendMessage({
        fromId: CLI_USER_ID,
        fromType: "user",
        toId: agentId,
        toType: "agent",
        content: line,
        type: "user-to-agent",
        metadata: { wakeRecipient: true },
      });
      output.write(`you → ${agentId}: ${line}\n`);
    }

    abortController.abort();
    rl.close();
    await poller;
    return 0;
  } finally {
    db.close();
  }
}

async function readSingleMessage(
  input: NodeJS.ReadableStream,
  output: NodeJS.WritableStream,
  nonInteractive?: boolean,
): Promise<string> {
  if (nonInteractive) {
    const chunks: Buffer[] = [];
    for await (const chunk of input) {
      chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk)));
    }
    return Buffer.concat(chunks).toString("utf8").trimEnd();
  }

  const rl = createInterface({ input, output });
  try {
    return await rl.question("");
  } finally {
    rl.close();
  }
}