AkurAI-Mail / tools/akurai-mail-index.ts

Latest change 614540419b314114a0a52d8a7c30fd13de559e2f - Add semantic multi-account mail indexer by Ólafur Búi Ólafsson

#!/usr/bin/env bun
import { SQL } from "bun";
import { createHash } from "node:crypto";
import { readdir, readFile } from "node:fs/promises";
import { homedir } from "node:os";
import { join } from "node:path";
import { ImapFlow, type FetchMessageObject, type MessageAddressObject } from "imapflow";
import PostalMime, { type Address } from "postal-mime";

const HOME = homedir();
const ACCOUNTS_DIR = process.env.AKURAI_MAIL_ACCOUNTS_DIR ?? join(HOME, ".omp/agent/skills/akurai-mail/accounts");
const DB_PASSWORD_FILE = process.env.AKURAI_MAIL_DB_PASSWORD_FILE ?? join(HOME, ".config/akurai-mail-index/db.pass");
const EMBED_URL = process.env.AKURAI_MAIL_EMBED_URL ?? "http://127.0.0.1:8081/v1/embeddings";
const RERANK_URL = process.env.AKURAI_MAIL_RERANK_URL ?? "http://127.0.0.1:8082/rerank";
const EMBED_MODEL = "qwen3-embedding-0.6b";
const EMBED_DIMENSIONS = 1024;
const SEARCH_TEXT_LIMIT = 1_500;

interface AccountConfig {
  alias: string;
  address: string;
  displayName: string;
  host: string;
  port: number;
  secure: boolean;
  login: string;
  passwordFile: string;
}

interface IndexedMessage {
  id: string;
  account: string;
  address: string;
  folder: string;
  uid: number;
  subject: string;
  senderName: string;
  senderAddress: string;
  sentAt: string | null;
  body: string;
  contentSha: Buffer;
}

interface SearchRow {
  id: string | number | bigint;
  account: string;
  address: string;
  folder: string;
  uid: number;
  subject: string;
  sender_name: string | null;
  sender_address: string | null;
  sent_at: string | Date | null;
  body_text: string | null;
  distance: number;
}

interface CliOptions {
  account?: string;
  folder?: string;
  from?: string;
  after?: string;
  before?: string;
  limit: number;
  json: boolean;
}

const schema = [
  `CREATE TABLE IF NOT EXISTS accounts (
    id INT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
    alias VARCHAR(191) NOT NULL UNIQUE,
    address VARCHAR(320) NOT NULL,
    display_name VARCHAR(320) NOT NULL DEFAULT '',
    last_sync_at DATETIME(6) NULL,
    last_error TEXT NULL,
    created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
    INDEX accounts_address (address)
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
  `CREATE TABLE IF NOT EXISTS folders (
    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
    account_id INT UNSIGNED NOT NULL,
    path VARCHAR(512) NOT NULL,
    uid_validity BIGINT UNSIGNED NOT NULL DEFAULT 0,
    last_uid BIGINT UNSIGNED NOT NULL DEFAULT 0,
    message_count INT UNSIGNED NOT NULL DEFAULT 0,
    last_sync_at DATETIME(6) NULL,
    created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
    UNIQUE KEY folders_account_path (account_id, path),
    CONSTRAINT folders_account_fk FOREIGN KEY (account_id) REFERENCES accounts(id) ON DELETE CASCADE
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
  `CREATE TABLE IF NOT EXISTS messages (
    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
    account_id INT UNSIGNED NOT NULL,
    folder_id BIGINT UNSIGNED NOT NULL,
    uid BIGINT UNSIGNED NOT NULL,
    message_id VARCHAR(998) NULL,
    in_reply_to VARCHAR(998) NULL,
    subject TEXT NOT NULL,
    sender_name VARCHAR(512) NULL,
    sender_address VARCHAR(320) NULL,
    recipients JSON NOT NULL,
    cc JSON NOT NULL,
    sent_at DATETIME(6) NULL,
    internal_date DATETIME(6) NULL,
    flags JSON NOT NULL,
    size_bytes BIGINT UNSIGNED NOT NULL DEFAULT 0,
    body_text MEDIUMTEXT NOT NULL,
    body_html MEDIUMTEXT NOT NULL,
    attachments JSON NOT NULL,
    raw_message LONGBLOB NOT NULL,
    content_sha BINARY(32) NOT NULL,
    deleted_at DATETIME(6) NULL,
    created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
    UNIQUE KEY messages_folder_uid (folder_id, uid),
    INDEX messages_account_date (account_id, sent_at),
    INDEX messages_folder_date (folder_id, sent_at),
    INDEX messages_sender (sender_address),
    INDEX messages_message_id (message_id(191)),
    FULLTEXT INDEX messages_text (subject, sender_name, sender_address, body_text),
    CONSTRAINT messages_account_fk FOREIGN KEY (account_id) REFERENCES accounts(id) ON DELETE CASCADE,
    CONSTRAINT messages_folder_fk FOREIGN KEY (folder_id) REFERENCES folders(id) ON DELETE CASCADE
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
  `CREATE TABLE IF NOT EXISTS embeddings (
    message_id BIGINT UNSIGNED NOT NULL PRIMARY KEY,
    content_sha BINARY(32) NOT NULL,
    model VARCHAR(191) NOT NULL,
    embedding VECTOR(1024) NOT NULL,
    embedded_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    VECTOR INDEX embedding_vector (embedding) M=8 DISTANCE=cosine,
    CONSTRAINT embeddings_message_fk FOREIGN KEY (message_id) REFERENCES messages(id) ON DELETE CASCADE
  ) ENGINE=InnoDB`,
  `CREATE TABLE IF NOT EXISTS sync_runs (
    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
    started_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    finished_at DATETIME(6) NULL,
    status ENUM('running','succeeded','failed','skipped') NOT NULL DEFAULT 'running',
    accounts_scanned INT UNSIGNED NOT NULL DEFAULT 0,
    folders_scanned INT UNSIGNED NOT NULL DEFAULT 0,
    messages_added INT UNSIGNED NOT NULL DEFAULT 0,
    messages_updated INT UNSIGNED NOT NULL DEFAULT 0,
    messages_deleted INT UNSIGNED NOT NULL DEFAULT 0,
    error_text TEXT NULL,
    INDEX sync_runs_started (started_at)
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`
];

async function database(): Promise<SQL> {
  const password = (await readFile(DB_PASSWORD_FILE, "utf8")).trim();
  if (!password) throw new Error(`empty database password file: ${DB_PASSWORD_FILE}`);
  return new SQL({ adapter: "mysql", hostname: "127.0.0.1", port: 3306, database: "akurai_mail", username: "mail_index", password, max: 4, idleTimeout: 30 });
}

async function ensureSchema(db: SQL): Promise<void> {
  for (const statement of schema) await db.unsafe(statement);
}

async function loadAccounts(selected?: string): Promise<AccountConfig[]> {
  const files = (await readdir(ACCOUNTS_DIR)).filter((name) => name.endsWith(".conf")).sort();
  const accounts: AccountConfig[] = [];
  for (const file of files) {
    const parsed = Bun.TOML.parse(await readFile(join(ACCOUNTS_DIR, file), "utf8")) as Record<string, unknown>;
    const root = parsed.accounts as Record<string, Record<string, unknown>> | undefined;
    if (!root) continue;
    for (const [alias, value] of Object.entries(root)) {
      if (selected && alias !== selected) continue;
      const backend = value.backend as Record<string, unknown>;
      const auth = backend.auth as Record<string, unknown>;
      const encryption = backend.encryption as Record<string, unknown>;
      const command = String(auth.cmd ?? "");
      const match = command.match(/^cat\s+(.+)$/);
      if (!match) throw new Error(`${alias}: backend.auth.cmd must be a simple cat command`);
      accounts.push({
        alias,
        address: String(value.email ?? backend.login ?? ""),
        displayName: String(value["display-name"] ?? ""),
        host: String(backend.host ?? ""),
        port: Number(backend.port ?? 993),
        secure: encryption.type === "tls",
        login: String(backend.login ?? value.email ?? ""),
        passwordFile: match[1]!.trim()
      });
    }
  }
  if (selected && accounts.length === 0) throw new Error(`unknown account: ${selected}`);
  if (accounts.length === 0) throw new Error(`no mail accounts found in ${ACCOUNTS_DIR}`);
  return accounts;
}

function mysqlDate(value: Date | string | undefined): string | null {
  if (!value) return null;
  const date = value instanceof Date ? value : new Date(value);
  return Number.isNaN(date.getTime()) ? null : date.toISOString().slice(0, 23).replace("T", " ");
}

function addressList(values: MessageAddressObject[] | undefined): Array<{ name: string; address: string }> {
  return (values ?? []).map((item) => ({ name: item.name ?? "", address: item.address ?? "" }));
}

function postalAddresses(values: Address[] | undefined): Array<{ name: string; address: string }> {
  const result: Array<{ name: string; address: string }> = [];
  for (const value of values ?? []) {
    const mailboxes = "group" in value && value.group ? value.group : [value];
    for (const mailbox of mailboxes) {
      if (mailbox.address) result.push({ name: mailbox.name, address: mailbox.address });
    }
  }
  return result;
}

function searchableText(message: Pick<IndexedMessage, "subject" | "senderName" | "senderAddress" | "body">): string {
  return [`Subject: ${message.subject}`, `From: ${message.senderName} <${message.senderAddress}>`, message.body].join("\n").slice(0, SEARCH_TEXT_LIMIT);
}

async function embeddingBatch(texts: string[]): Promise<number[][]> {
  if (texts.length === 0) return [];
  const response = await fetch(EMBED_URL, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ model: EMBED_MODEL, input: texts }), signal: AbortSignal.timeout(120_000) });
  const body = await response.text();
  if (!response.ok) throw new Error(`embedding service returned ${response.status}: ${body.slice(0, 300)}`);
  const payload: unknown = JSON.parse(body);
  if (!payload || typeof payload !== "object" || !("data" in payload) || !Array.isArray(payload.data)) throw new Error("embedding service returned an invalid response");
  const ordered = payload.data.map((item, index) => {
    if (!item || typeof item !== "object" || !(("embedding" in item) && Array.isArray(item.embedding))) throw new Error(`embedding ${index} is invalid`);
    return item.embedding.map(Number);
  });
  if (ordered.length !== texts.length || ordered.some((vector) => vector.length !== EMBED_DIMENSIONS)) throw new Error("embedding service returned an invalid vector batch");
  return ordered;
}

async function rerank(query: string, texts: string[]): Promise<Array<{ index: number; score: number }>> {
  const response = await fetch(RERANK_URL, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ query, texts, return_text: false }), signal: AbortSignal.timeout(120_000) });
  if (!response.ok) throw new Error(`reranker returned ${response.status}`);
  return await response.json() as Array<{ index: number; score: number }>;
}

async function upsertAccount(db: SQL, account: AccountConfig): Promise<number> {
  await db`INSERT INTO accounts (alias, address, display_name) VALUES (${account.alias}, ${account.address}, ${account.displayName}) ON DUPLICATE KEY UPDATE address=VALUES(address), display_name=VALUES(display_name), last_error=NULL`;
  const rows = await db`SELECT id FROM accounts WHERE alias=${account.alias}` as Array<{ id: number }>;
  return Number(rows[0]!.id);
}

async function upsertFolder(db: SQL, accountId: number, path: string, uidValidity: bigint | number): Promise<{ id: number; uidValidity: bigint }> {
  await db`INSERT INTO folders (account_id, path, uid_validity) VALUES (${accountId}, ${path}, ${String(uidValidity)}) ON DUPLICATE KEY UPDATE path=VALUES(path)`;
  const rows = await db`SELECT id, uid_validity FROM folders WHERE account_id=${accountId} AND path=${path}` as Array<{ id: number; uid_validity: bigint }>;
  return { id: Number(rows[0]!.id), uidValidity: BigInt(rows[0]!.uid_validity) };
}

async function storeMessage(db: SQL, accountId: number, folderId: number, account: AccountConfig, folder: string, item: FetchMessageObject): Promise<IndexedMessage> {
  if (!item.source) throw new Error(`${account.alias}/${folder}/${item.uid}: source missing`);
  const parsed = await PostalMime.parse(item.source, { attachmentEncoding: "arraybuffer" });
  const envelope = item.envelope;
  const from = postalAddresses(parsed.from ? [parsed.from] : undefined)[0] ?? addressList(envelope?.from)[0] ?? { name: "", address: "" };
  const body = (parsed.text ?? "").trim();
  const html = parsed.html ?? "";
  const subject = parsed.subject ?? envelope?.subject ?? "";
  const contentSha = createHash("sha256").update(item.source).digest();
  const recipients = postalAddresses(parsed.to).length ? postalAddresses(parsed.to) : addressList(envelope?.to);
  const cc = postalAddresses(parsed.cc).length ? postalAddresses(parsed.cc) : addressList(envelope?.cc);
  const attachments = parsed.attachments.map((attachment) => ({ filename: attachment.filename, mimeType: attachment.mimeType, disposition: attachment.disposition, contentId: attachment.contentId, size: typeof attachment.content === "string" ? Buffer.byteLength(attachment.content) : attachment.content.byteLength }));
  await db`INSERT INTO messages (account_id, folder_id, uid, message_id, in_reply_to, subject, sender_name, sender_address, recipients, cc, sent_at, internal_date, flags, size_bytes, body_text, body_html, attachments, raw_message, content_sha, deleted_at)
    VALUES (${accountId}, ${folderId}, ${item.uid}, ${parsed.messageId ?? envelope?.messageId ?? null}, ${parsed.inReplyTo ?? envelope?.inReplyTo ?? null}, ${subject}, ${from.name}, ${from.address}, ${JSON.stringify(recipients)}, ${JSON.stringify(cc)}, ${mysqlDate(parsed.date ?? envelope?.date)}, ${mysqlDate(item.internalDate)}, ${JSON.stringify([...item.flags ?? []])}, ${item.size ?? item.source.byteLength}, ${body}, ${html}, ${JSON.stringify(attachments)}, ${item.source}, ${contentSha}, NULL)
    ON DUPLICATE KEY UPDATE message_id=VALUES(message_id), in_reply_to=VALUES(in_reply_to), subject=VALUES(subject), sender_name=VALUES(sender_name), sender_address=VALUES(sender_address), recipients=VALUES(recipients), cc=VALUES(cc), sent_at=VALUES(sent_at), internal_date=VALUES(internal_date), flags=VALUES(flags), size_bytes=VALUES(size_bytes), body_text=VALUES(body_text), body_html=VALUES(body_html), attachments=VALUES(attachments), raw_message=VALUES(raw_message), content_sha=VALUES(content_sha), deleted_at=NULL`;
  const rows = await db`SELECT id FROM messages WHERE folder_id=${folderId} AND uid=${item.uid}` as Array<{ id: string | number | bigint }>;
  return { id: String(rows[0]!.id), account: account.alias, address: account.address, folder, uid: item.uid, subject, senderName: from.name, senderAddress: from.address, sentAt: mysqlDate(parsed.date ?? envelope?.date), body, contentSha };
}

async function storeEmbeddings(db: SQL, messages: IndexedMessage[]): Promise<void> {
  for (let offset = 0; offset < messages.length; offset += 4) {
    const batch = messages.slice(offset, offset + 4);
    const vectors = await embeddingBatch(batch.map(searchableText));
    for (let index = 0; index < batch.length; index++) {
      const message = batch[index]!;
      await db`INSERT INTO embeddings (message_id, content_sha, model, embedding) VALUES (${message.id}, ${message.contentSha}, ${EMBED_MODEL}, VEC_FromText(${JSON.stringify(vectors[index])})) ON DUPLICATE KEY UPDATE content_sha=VALUES(content_sha), model=VALUES(model), embedding=VALUES(embedding), embedded_at=CURRENT_TIMESTAMP(6)`;
    }
  }
}

async function embedPending(db: SQL): Promise<number> {
  let embedded = 0;
  while (true) {
    const rows = await db`SELECT m.id, a.alias AS account, a.address, f.path AS folder, m.uid, m.subject, m.sender_name, m.sender_address, m.sent_at, m.body_text, m.content_sha
      FROM messages m JOIN accounts a ON a.id=m.account_id JOIN folders f ON f.id=m.folder_id LEFT JOIN embeddings e ON e.message_id=m.id
      WHERE m.deleted_at IS NULL AND (e.message_id IS NULL OR e.content_sha<>m.content_sha OR e.model<>${EMBED_MODEL})
      ORDER BY m.id LIMIT 100` as Array<{ id: string | number | bigint; account: string; address: string; folder: string; uid: number; subject: string; sender_name: string | null; sender_address: string | null; sent_at: string | null; body_text: string | null; content_sha: Uint8Array }>;
    if (rows.length === 0) return embedded;
    const messages = rows.map((row) => ({ id: String(row.id), account: row.account, address: row.address, folder: row.folder, uid: Number(row.uid), subject: row.subject, senderName: row.sender_name ?? "", senderAddress: row.sender_address ?? "", sentAt: row.sent_at, body: row.body_text ?? "", contentSha: Buffer.from(row.content_sha) }));
    await storeEmbeddings(db, messages);
    embedded += messages.length;
  }
}

async function syncAccount(db: SQL, account: AccountConfig): Promise<{ folders: number; added: IndexedMessage[]; deleted: number }> {
  const accountId = await upsertAccount(db, account);
  const password = (await readFile(account.passwordFile, "utf8")).trim();
  const client = new ImapFlow({ host: account.host, port: account.port, secure: account.secure, auth: { user: account.login, pass: password }, logger: false });
  const added: IndexedMessage[] = [];
  let deleted = 0;
  let scannedFolders = 0;
  try {
    await client.connect();
    const mailboxes = (await client.list()).filter((mailbox) => !mailbox.flags.has("\\Noselect"));
    for (const mailboxInfo of mailboxes) {
      const mailbox = await client.mailboxOpen(mailboxInfo.path, { readOnly: true });
      scannedFolders++;
      const folder = await upsertFolder(db, accountId, mailboxInfo.path, mailbox.uidValidity);
      if (folder.uidValidity !== 0n && folder.uidValidity !== BigInt(mailbox.uidValidity)) {
        await db`UPDATE messages SET deleted_at=CURRENT_TIMESTAMP(6) WHERE folder_id=${folder.id} AND deleted_at IS NULL`;
      }
      const existingRows = await db`SELECT uid, HEX(content_sha) AS sha FROM messages WHERE folder_id=${folder.id} AND deleted_at IS NULL` as Array<{ uid: number; sha: string }>;
      const existing = new Map(existingRows.map((row) => [Number(row.uid), row.sha]));
      const current = new Set<number>();
      const newUids: number[] = [];
      if (mailbox.exists > 0) {
        for await (const message of client.fetch("1:*", { uid: true })) {
          current.add(message.uid);
          if (!existing.has(message.uid)) newUids.push(message.uid);
        }
      }
      for (let offset = 0; offset < newUids.length; offset += 25) {
        const chunk = newUids.slice(offset, offset + 25);
        const fetched: FetchMessageObject[] = [];
        for await (const message of client.fetch(chunk, { uid: true, flags: true, envelope: true, internalDate: true, size: true, source: true }, { uid: true })) fetched.push(message);
        for (const message of fetched) added.push(await storeMessage(db, accountId, folder.id, account, mailboxInfo.path, message));
      }
      const removed = [...existing.keys()].filter((uid) => !current.has(uid));
      for (const uid of removed) await db`UPDATE messages SET deleted_at=CURRENT_TIMESTAMP(6) WHERE folder_id=${folder.id} AND uid=${uid} AND deleted_at IS NULL`;
      deleted += removed.length;
      const lastUid = current.size ? Math.max(...current) : 0;
      await db`UPDATE folders SET uid_validity=${String(mailbox.uidValidity)}, last_uid=${lastUid}, message_count=${mailbox.exists}, last_sync_at=CURRENT_TIMESTAMP(6) WHERE id=${folder.id}`;
    }
    await db`UPDATE accounts SET last_sync_at=CURRENT_TIMESTAMP(6), last_error=NULL WHERE id=${accountId}`;
    return { folders: scannedFolders, added, deleted };
  } catch (error) {
    await db`UPDATE accounts SET last_error=${String(error)} WHERE id=${accountId}`;
    throw error;
  } finally {
    await client.logout().catch(() => undefined);
  }
}

async function sync(options: CliOptions): Promise<void> {
  const db = await database();
  await ensureSchema(db);
  const lockRows = await db`SELECT GET_LOCK('akurai_mail_sync', 0) AS acquired` as Array<{ acquired: number }>;
  if (Number(lockRows[0]?.acquired) !== 1) {
    console.log(options.json ? JSON.stringify({ status: "skipped", reason: "sync already running" }) : "Mail sync already running; skipped.");
    await db.close();
    return;
  }
  await db`INSERT INTO sync_runs () VALUES ()`;
  const runRows = await db`SELECT LAST_INSERT_ID() AS id` as Array<{ id: string | number | bigint }>;
  const runId = String(runRows[0]!.id);
  const allAdded: IndexedMessage[] = [];
  let accountsScanned = 0;
  let foldersScanned = 0;
  let deleted = 0;
  try {
    for (const account of await loadAccounts(options.account)) {
      const result = await syncAccount(db, account);
      accountsScanned++;
      foldersScanned += result.folders;
      deleted += result.deleted;
      allAdded.push(...result.added);
    }
    const embeddingsUpdated = await embedPending(db);
    await db`UPDATE sync_runs SET finished_at=CURRENT_TIMESTAMP(6), status='succeeded', accounts_scanned=${accountsScanned}, folders_scanned=${foldersScanned}, messages_added=${allAdded.length}, messages_updated=${embeddingsUpdated}, messages_deleted=${deleted} WHERE id=${runId}`;
    const report = { status: "succeeded", accountsScanned, foldersScanned, messagesAdded: allAdded.length, embeddingsUpdated, messagesDeleted: deleted, newMessages: allAdded.map(({ body: _body, contentSha: _sha, ...message }) => message) };
    console.log(options.json ? JSON.stringify(report) : `Synced ${accountsScanned} account(s), ${foldersScanned} folder(s); added ${allAdded.length}, embedded ${embeddingsUpdated}, removed ${deleted}.`);
  } catch (error) {
    await db`UPDATE sync_runs SET finished_at=CURRENT_TIMESTAMP(6), status='failed', accounts_scanned=${accountsScanned}, folders_scanned=${foldersScanned}, messages_added=${allAdded.length}, messages_deleted=${deleted}, error_text=${String(error)} WHERE id=${runId}`;
    throw error;
  } finally {
    await db`SELECT RELEASE_LOCK('akurai_mail_sync')`;
    await db.close();
  }
}

function senderFromNaturalLanguage(query: string): string | null {
  const match = query.match(/\b(?:from|frá)\s+["“]?(.+?)["”]?\s*$/iu);
  if (!match?.[1]) return null;
  return match[1].replace(/[?.!,;:]+$/, "").trim() || null;
}

async function searchMail(query: string, options: CliOptions): Promise<void> {
  const db = await database();
  await ensureSchema(db);
  const [vector] = await embeddingBatch([query]);
  const account = options.account ?? null;
  const folder = options.folder ?? null;
  const senderName = options.from ?? senderFromNaturalLanguage(query);
  const sender = senderName ? `%${senderName}%` : null;
  const after = options.after ? `${options.after} 00:00:00` : null;
  const before = options.before ? `${options.before} 23:59:59` : null;
  const candidateLimit = Math.max(options.limit * 5, 30);
  const rows = await db`SELECT m.id, a.alias AS account, a.address, f.path AS folder, m.uid, m.subject, m.sender_name, m.sender_address, m.sent_at, m.body_text, VEC_DISTANCE_COSINE(e.embedding, VEC_FromText(${JSON.stringify(vector)})) AS distance
    FROM embeddings e JOIN messages m ON m.id=e.message_id JOIN accounts a ON a.id=m.account_id JOIN folders f ON f.id=m.folder_id
    WHERE m.deleted_at IS NULL
      AND (${account} IS NULL OR a.alias=${account})
      AND (${folder} IS NULL OR f.path=${folder})
      AND (${sender} IS NULL OR m.sender_address LIKE ${sender} OR m.sender_name LIKE ${sender})
      AND (${after} IS NULL OR COALESCE(m.sent_at,m.internal_date)>=${after})
      AND (${before} IS NULL OR COALESCE(m.sent_at,m.internal_date)<=${before})
    ORDER BY VEC_DISTANCE_COSINE(e.embedding, VEC_FromText(${JSON.stringify(vector)})) ASC LIMIT ${candidateLimit}` as SearchRow[];
  const ranked = rows.length > 1 ? await rerank(query, rows.map((row) => searchableText({ subject: row.subject, senderName: row.sender_name ?? "", senderAddress: row.sender_address ?? "", body: (row.body_text ?? "").slice(0, 4000) }))) : rows.map((_row, index) => ({ index, score: 1 }));
  const results = ranked.slice(0, options.limit).map((rank) => {
    const row = rows[rank.index]!;
    return { id: String(row.id), account: row.account, address: row.address, folder: row.folder, uid: row.uid, subject: row.subject, from: { name: row.sender_name ?? "", address: row.sender_address ?? "" }, date: row.sent_at, score: rank.score, semanticDistance: Number(row.distance), preview: (row.body_text ?? "").replace(/\s+/g, " ").slice(0, 240) };
  });
  if (options.json) console.log(JSON.stringify({ query, results }));
  else if (results.length === 0) console.log("No matching mail.");
  else for (const [index, result] of results.entries()) console.log(`${index + 1}. [${result.id}] ${result.subject || "(no subject)"}\n   ${result.from.name || result.from.address} <${result.from.address}> · ${result.account}/${result.folder} · ${result.date ?? "unknown date"}\n   ${result.preview}`);
  await db.close();
}

async function showMessage(id: string, json: boolean): Promise<void> {
  const db = await database();
  const rows = await db`SELECT m.id, a.alias AS account, a.address, f.path AS folder, m.uid, m.subject, m.sender_name, m.sender_address, m.recipients, m.cc, m.sent_at, m.internal_date, m.flags, m.body_text, m.attachments, m.deleted_at FROM messages m JOIN accounts a ON a.id=m.account_id JOIN folders f ON f.id=m.folder_id WHERE m.id=${id}` as Array<Record<string, unknown>>;
  if (!rows[0]) throw new Error(`message ${id} not found`);
  console.log(json ? JSON.stringify(rows[0]) : [`[${rows[0].id}] ${rows[0].subject}`, `From: ${rows[0].sender_name ?? ""} <${rows[0].sender_address ?? ""}>`, `Account: ${rows[0].account} (${rows[0].address})`, `Folder: ${rows[0].folder} · UID ${rows[0].uid}`, `Date: ${rows[0].sent_at ?? rows[0].internal_date ?? "unknown"}`, "", String(rows[0].body_text ?? "")].join("\n"));
  await db.close();
}

async function status(json: boolean): Promise<void> {
  const db = await database();
  await ensureSchema(db);
  const accounts = await db`SELECT a.alias, a.address, a.last_sync_at, a.last_error, COUNT(DISTINCT f.id) AS folders, COUNT(m.id) AS messages, COALESCE(SUM(m.id IS NOT NULL AND m.deleted_at IS NULL), 0) AS active_messages FROM accounts a LEFT JOIN folders f ON f.account_id=a.id LEFT JOIN messages m ON m.folder_id=f.id GROUP BY a.id ORDER BY a.alias` as Array<Record<string, unknown>>;
  const pending = await db`SELECT COUNT(*) AS count FROM messages m LEFT JOIN embeddings e ON e.message_id=m.id WHERE m.deleted_at IS NULL AND (e.message_id IS NULL OR e.content_sha<>m.content_sha OR e.model<>${EMBED_MODEL})` as Array<{ count: number }>;
  const runs = await db`SELECT id, started_at, finished_at, status, messages_added, messages_deleted, error_text FROM sync_runs ORDER BY id DESC LIMIT 5` as Array<Record<string, unknown>>;
  const result = { accounts, pendingEmbeddings: Number(pending[0]?.count ?? 0), recentRuns: runs };
  if (json) console.log(JSON.stringify(result));
  else {
    for (const account of accounts) console.log(`${account.alias} (${account.address}): ${account.active_messages ?? 0} active mail in ${account.folders ?? 0} folders; last sync ${account.last_sync_at ?? "never"}${account.last_error ? `; ERROR ${account.last_error}` : ""}`);
    console.log(`Pending embeddings: ${result.pendingEmbeddings}`);
  }
  await db.close();
}

async function doctor(): Promise<void> {
  const db = await database();
  await ensureSchema(db);
  await db`SELECT 1`;
  const vector = await embeddingBatch(["mail index health check"]);
  await rerank("important mail", ["A security alert needs attention.", "A weekly newsletter."]);
  for (const account of await loadAccounts()) {
    const password = (await readFile(account.passwordFile, "utf8")).trim();
    const client = new ImapFlow({ host: account.host, port: account.port, secure: account.secure, auth: { user: account.login, pass: password }, logger: false });
    await client.connect();
    await client.logout();
    console.log(`${account.alias}: IMAP ready`);
  }
  console.log(`MariaDB ready; embeddings ${vector[0]!.length}d; reranker ready`);
  await db.close();
}

async function installWorkbenchSchedule(): Promise<void> {
  const env = await readFile(join(HOME, ".config/omp-workbench/env"), "utf8");
  const token = env.match(/^WORKBENCH_TOKEN=(.+)$/m)?.[1]?.trim();
  const origin = env.match(/^WORKBENCH_ORIGIN=(.+)$/m)?.[1]?.trim() ?? "http://192.168.1.10:4173";
  if (!token) throw new Error("Workbench token is not configured");
  const login = await fetch(`${origin}/api/login`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ token }) });
  if (!login.ok) throw new Error(`Workbench login returned ${login.status}`);
  const cookie = login.headers.get("set-cookie")?.match(/omp_workbench=[^;]+/)?.[0];
  if (!cookie) throw new Error("Workbench login did not return a session cookie");
  const request = async (path: string, init: RequestInit = {}) => {
    const response = await fetch(`${origin}${path}`, { ...init, headers: { ...init.headers, cookie, origin } });
    const text = await response.text();
    if (!response.ok) throw new Error(`Workbench ${path} returned ${response.status}: ${text.slice(0, 240)}`);
    const payload: unknown = text ? JSON.parse(text) : {};
    if (!payload || typeof payload !== "object") throw new Error(`Workbench ${path} returned invalid JSON`);
    return payload as Record<string, unknown>;
  };
  const projectsPayload = await request("/api/projects");
  const projects = Array.isArray(projectsPayload.projects) ? projectsPayload.projects.filter((value): value is Record<string, unknown> => Boolean(value && typeof value === "object")) : [];
  const project = projects.find((value) => value.name === "AkurAI-Mail");
  if (!project || typeof project.id !== "string") throw new Error("AkurAI-Mail is not registered in Workbench projects");
  let channelsPayload = await request("/api/channels");
  let channels = Array.isArray(channelsPayload.channels) ? channelsPayload.channels.filter((value): value is Record<string, unknown> => Boolean(value && typeof value === "object")) : [];
  let channel = channels.find((value) => value.name === "general");
  if (!channel) {
    channel = await request("/api/channels", { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ name: "general" }) });
    channelsPayload = await request("/api/channels");
    channels = Array.isArray(channelsPayload.channels) ? channelsPayload.channels.filter((value): value is Record<string, unknown> => Boolean(value && typeof value === "object")) : [];
  }
  if (typeof channel.id !== "string") throw new Error("Workbench general channel is invalid");
  const tasks = Array.isArray(channelsPayload.tasks) ? channelsPayload.tasks.filter((value): value is Record<string, unknown> => Boolean(value && typeof value === "object")) : [];
  const existing = tasks.find((value) => value.title === "Mail inbox sync" && value.projectId === project.id);
  if (existing) {
    console.log("Workbench schedule already installed.");
    return;
  }
  const prompt = `Use the akurai-mail-semantic skill. Run exactly: akurai-mail-index sync --json
Parse the single JSON object. Reply with exactly workbench:mail-report:<that JSON object> and no Markdown or commentary. If the command fails, report the error plainly instead. Do not use subagents.`;
  await request("/api/scheduled-tasks", { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ channelId: channel.id, projectId: project.id, title: "Mail inbox sync", prompt, scheduleKind: "five_minutes", readOnly: false, nextRunAt: Date.now() + 60_000 }) });
  console.log("Installed five-minute Mail inbox sync in Workbench #general.");
}

function parseOptions(args: string[]): { positional: string[]; options: CliOptions } {
  const positional: string[] = [];
  const options: CliOptions = { limit: 10, json: false };
  for (let index = 0; index < args.length; index++) {
    const arg = args[index]!;
    if (arg === "--json") options.json = true;
    else if (arg === "--account") options.account = args[++index];
    else if (arg === "--folder") options.folder = args[++index];
    else if (arg === "--from") options.from = args[++index];
    else if (arg === "--after") options.after = args[++index];
    else if (arg === "--before") options.before = args[++index];
    else if (arg === "--limit") options.limit = Math.max(1, Math.min(50, Number(args[++index] ?? 10)));
    else positional.push(arg);
  }
  return { positional, options };
}

function usage(): never {
  console.error(`Usage:
  akurai-mail-index sync [--account ALIAS] [--json]
  akurai-mail-index search <natural language query> [--account ALIAS] [--folder NAME] [--from TEXT] [--after YYYY-MM-DD] [--before YYYY-MM-DD] [--limit N] [--json]
  akurai-mail-index show <MESSAGE_ID> [--json]
  akurai-mail-index status [--json]
  akurai-mail-index accounts [--json]
  akurai-mail-index schedule-install
  akurai-mail-index doctor`);
  process.exit(2);
}

const { positional, options } = parseOptions(process.argv.slice(2));
const command = positional.shift();
try {
  if (command === "sync") await sync(options);
  else if (command === "search") {
    const query = positional.join(" ").trim();
    if (!query) usage();
    await searchMail(query, options);
  } else if (command === "show") {
    if (!positional[0]) usage();
    await showMessage(positional[0], options.json);
  } else if (command === "status") await status(options.json);
  else if (command === "accounts") {
    const accounts = await loadAccounts(options.account);
    console.log(options.json ? JSON.stringify(accounts.map(({ passwordFile: _secret, ...account }) => account)) : accounts.map((account) => `${account.alias}\t${account.address}\t${account.host}:${account.port}`).join("\n"));
  } else if (command === "schedule-install") await installWorkbenchSchedule();
  else if (command === "doctor") await doctor();
  else usage();
} catch (error) {
  console.error(`akurai-mail-index: ${error instanceof Error ? error.message : String(error)}`);
  process.exit(1);
}