add whatsapp and telegram support

apps/x/packages/core/package.json

@@ -30,6 +30,7 @@
     "@x/shared": "workspace:*",
     "ai": "^5.0.133",
     "awilix": "^12.0.5",
+    "baileys": "7.0.0-rc13",
     "chokidar": "^4.0.3",
     "cors": "^2.8.6",
     "cron-parser": "^5.5.0",
@@ -45,6 +46,7 @@
     "papaparse": "^5.5.3",
     "pdf-parse": "^2.4.5",
     "posthog-node": "^4.18.0",
+    "qrcode": "^1.5.4",
     "react": "^19.2.3",
     "xlsx": "^0.18.5",
     "yaml": "^2.8.2",
@@ -56,6 +58,7 @@
     "@types/node": "^25.0.3",
     "@types/papaparse": "^5.5.2",
     "@types/pdf-parse": "^1.1.5",
+    "@types/qrcode": "^1.5.6",
     "vitest": "catalog:"
   }
 }

apps/x/packages/core/src/channels/bridge.ts

@@ -0,0 +1,396 @@
+import type { SessionIndexEntry } from "@x/shared/dist/sessions.js";
+import { reduceTurn, type TurnStreamEvent } from "@x/shared/dist/turns.js";
+import { assistantText, lastAssistantText } from "../agents/headless.js";
+import { TurnNotSettledError, type ISessions } from "../sessions/api.js";
+import type { EmitterSessionBus } from "../sessions/bus.js";
+
+// Transport-agnostic command layer: inbound texts from a messaging channel
+// are parsed into commands (list / resume / new / stop / status) or forwarded
+// into a regular chat session; the turn's final assistant text is sent back
+// through the transport's reply callback. Turns run with autoPermission and
+// show up live in the desktop UI like any other session.
+
+const AGENT_ID = "copilot";
+const TURN_TIMEOUT_MS = 30 * 60 * 1000;
+const LIST_LIMIT = 10;
+// Telegram caps messages at 4096 chars; WhatsApp is far higher. Long replies
+// are chunked, then truncated — the desktop app has the full text.
+const REPLY_CHUNK_SIZE = 3500;
+const MAX_REPLY_CHUNKS = 3;
+
+const HELP_TEXT = [
+    "🤖 Rowboat commands:",
+    "• list — recent chats",
+    "• resume N — continue chat N from the list",
+    "• new [message] — start a fresh chat",
+    "• status — current chat and what it's doing",
+    "• stop — cancel the running task",
+    "",
+    "Anything else is sent to the current chat.",
+].join("\n");
+
+export type ReplyFn = (text: string) => Promise<void>;
+
+interface SenderState {
+    activeSessionId: string | null;
+    // sessionIds as last shown by `list` (1-based indexing for `resume N`).
+    lastList: string[];
+    pendingAsk: { turnId: string; toolCallId: string } | null;
+    busy: boolean;
+}
+
+type Settled =
+    | { kind: "completed"; text: string | null }
+    | { kind: "failed"; error: string }
+    | { kind: "cancelled" }
+    | { kind: "ask_human"; toolCallId: string; query: string; options?: string[] }
+    | { kind: "suspended" }
+    | { kind: "timeout" };
+
+function settleOf(event: TurnStreamEvent): Settled | null {
+    switch (event.type) {
+        case "turn_completed":
+            return { kind: "completed", text: assistantText(event.output) };
+        case "turn_failed":
+            return { kind: "failed", error: event.error };
+        case "turn_cancelled":
+            return { kind: "cancelled" };
+        case "turn_suspended": {
+            const ask = event.pendingAsyncTools.find(
+                (t) => t.toolId === "builtin:ask-human" || t.toolName === "ask-human",
+            );
+            if (ask) {
+                const input = ask.input as { query?: unknown; options?: unknown } | null;
+                const query =
+                    typeof input?.query === "string" && input.query
+                        ? input.query
+                        : "The agent needs your input.";
+                const options = Array.isArray(input?.options)
+                    ? input.options.filter((o): o is string => typeof o === "string")
+                    : undefined;
+                return { kind: "ask_human", toolCallId: ask.toolCallId, query, options };
+            }
+            // Other async tools settle on their own and the turn resumes;
+            // keep waiting. Pending permissions need the desktop.
+            if (event.pendingAsyncTools.length === 0 && event.pendingPermissions.length > 0) {
+                return { kind: "suspended" };
+            }
+            return null;
+        }
+        default:
+            return null;
+    }
+}
+
+function relativeTime(iso: string): string {
+    const then = Date.parse(iso);
+    if (!Number.isFinite(then)) return "";
+    const diffSec = Math.round((Date.now() - then) / 1000);
+    if (diffSec < 60) return "just now";
+    const diffMin = Math.round(diffSec / 60);
+    if (diffMin < 60) return `${diffMin}m ago`;
+    const diffHr = Math.round(diffMin / 60);
+    if (diffHr < 24) return `${diffHr}h ago`;
+    return `${Math.round(diffHr / 24)}d ago`;
+}
+
+function chunkReply(text: string): string[] {
+    if (text.length <= REPLY_CHUNK_SIZE) return [text];
+    const parts: string[] = [];
+    let rest = text;
+    while (rest.length > 0 && parts.length < MAX_REPLY_CHUNKS) {
+        parts.push(rest.slice(0, REPLY_CHUNK_SIZE));
+        rest = rest.slice(REPLY_CHUNK_SIZE);
+    }
+    if (rest.length > 0) {
+        parts[parts.length - 1] += "\n… (truncated — open Rowboat for the full reply)";
+    }
+    return parts;
+}
+
+export class ChannelBridge {
+    private senders = new Map<string, SenderState>();
+
+    constructor(
+        private readonly deps: {
+            sessions: ISessions;
+            sessionBus: EmitterSessionBus;
+        },
+    ) {}
+
+    async handleInbound(senderKey: string, text: string, reply: ReplyFn): Promise<void> {
+        const trimmed = text.trim();
+        if (!trimmed) return;
+        const state = this.senderState(senderKey);
+        const lower = trimmed.toLowerCase();
+
+        try {
+            if (lower === "help" || lower === "?") {
+                await reply(HELP_TEXT);
+                return;
+            }
+            if (lower === "list" || lower === "chats") {
+                await reply(this.renderList(state));
+                return;
+            }
+            const resume = /^(?:resume|open)\s+(\d+)$/.exec(lower);
+            if (resume) {
+                await reply(this.resumeSession(state, Number(resume[1])));
+                return;
+            }
+            if (lower === "status") {
+                await reply(this.renderStatus(state));
+                return;
+            }
+            if (lower === "stop") {
+                await reply(await this.stopActive(state));
+                return;
+            }
+            if (lower === "new") {
+                state.activeSessionId = null;
+                state.pendingAsk = null;
+                await reply("🆕 Fresh chat — send your first message.");
+                return;
+            }
+            const newWithText = /^new\s+([\s\S]+)$/i.exec(trimmed);
+            if (newWithText) {
+                state.activeSessionId = null;
+                state.pendingAsk = null;
+                await this.runMessage(state, newWithText[1].trim(), reply);
+                return;
+            }
+            await this.runMessage(state, trimmed, reply);
+        } catch (error) {
+            const message = error instanceof Error ? error.message : String(error);
+            await reply(`❌ ${message}`).catch(() => undefined);
+        }
+    }
+
+    private senderState(senderKey: string): SenderState {
+        let state = this.senders.get(senderKey);
+        if (!state) {
+            state = { activeSessionId: null, lastList: [], pendingAsk: null, busy: false };
+            this.senders.set(senderKey, state);
+        }
+        return state;
+    }
+
+    private sortedSessions(): SessionIndexEntry[] {
+        return this.deps.sessions
+            .listSessions()
+            .filter((e) => !e.error)
+            .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt));
+    }
+
+    private renderList(state: SenderState): string {
+        const entries = this.sortedSessions().slice(0, LIST_LIMIT);
+        if (entries.length === 0) {
+            return "No chats yet — just send a message to start one.";
+        }
+        state.lastList = entries.map((e) => e.sessionId);
+        const lines = entries.map((e, i) => {
+            const marker =
+                e.latestTurnStatus === "suspended" ? " ⚠️" :
+                e.latestTurnStatus === "idle" ? " ⏳" : "";
+            const active = e.sessionId === state.activeSessionId ? " ← current" : "";
+            return `${i + 1}. ${e.title ?? "Untitled"}${marker} (${relativeTime(e.updatedAt)})${active}`;
+        });
+        return [
+            "Recent chats:",
+            ...lines,
+            "",
+            `Reply "resume N" to continue one.`,
+        ].join("\n");
+    }
+
+    private resumeSession(state: SenderState, index: number): string {
+        if (state.lastList.length === 0) {
+            state.lastList = this.sortedSessions()
+                .slice(0, LIST_LIMIT)
+                .map((e) => e.sessionId);
+        }
+        const sessionId = state.lastList[index - 1];
+        if (!sessionId) {
+            return `No chat #${index} — send "list" to see recent chats.`;
+        }
+        state.activeSessionId = sessionId;
+        state.pendingAsk = null;
+        const entry = this.sortedSessions().find((e) => e.sessionId === sessionId);
+        return `▶️ Resumed "${entry?.title ?? "Untitled"}" — send a message to continue.`;
+    }
+
+    private renderStatus(state: SenderState): string {
+        if (!state.activeSessionId) {
+            return "No current chat — your next message starts a new one.";
+        }
+        const entry = this.sortedSessions().find(
+            (e) => e.sessionId === state.activeSessionId,
+        );
+        if (!entry) return "Current chat no longer exists — send a message to start fresh.";
+        const status = state.busy
+            ? "working"
+            : entry.latestTurnStatus === "suspended"
+              ? "waiting on input"
+              : entry.latestTurnStatus;
+        return `Current chat: "${entry.title ?? "Untitled"}" — ${status}.`;
+    }
+
+    private async stopActive(state: SenderState): Promise<string> {
+        state.pendingAsk = null;
+        if (!state.activeSessionId) return "Nothing to stop.";
+        const entry = this.deps.sessions
+            .listSessions()
+            .find((e) => e.sessionId === state.activeSessionId);
+        if (!entry?.latestTurnId) return "Nothing to stop.";
+        if (
+            entry.latestTurnStatus === "completed" ||
+            entry.latestTurnStatus === "failed" ||
+            entry.latestTurnStatus === "cancelled"
+        ) {
+            return "Nothing running in the current chat.";
+        }
+        await this.deps.sessions.stopTurn(entry.latestTurnId, "stopped from mobile channel");
+        return "🛑 Stop requested.";
+    }
+
+    private async runMessage(state: SenderState, text: string, reply: ReplyFn): Promise<void> {
+        if (state.busy) {
+            await reply('⏳ Still working on the previous message — send "stop" to cancel it.');
+            return;
+        }
+        state.busy = true;
+        // Subscribe before advancing so no settle event can slip past.
+        const watcher = this.watchBus();
+        try {
+            let turnId: string;
+            if (state.pendingAsk) {
+                const ask = state.pendingAsk;
+                state.pendingAsk = null;
+                turnId = ask.turnId;
+                await reply("⏳ Working on it…");
+                await this.deps.sessions.respondToAskHuman(ask.turnId, ask.toolCallId, text);
+            } else {
+                if (!state.activeSessionId) {
+                    state.activeSessionId = await this.deps.sessions.createSession();
+                }
+                await reply("⏳ Working on it…");
+                const sent = await this.deps.sessions.sendMessage(
+                    state.activeSessionId,
+                    { role: "user", content: text },
+                    { agent: { agentId: AGENT_ID }, autoPermission: true },
+                );
+                turnId = sent.turnId;
+            }
+            const settled = await watcher.waitFor(turnId, TURN_TIMEOUT_MS);
+            await this.deliverSettled(state, turnId, settled, reply);
+        } catch (error) {
+            if (error instanceof TurnNotSettledError) {
+                await reply(
+                    '⏳ That chat is still working on something — send "stop" to cancel it, or "new" to start a fresh chat.',
+                );
+                return;
+            }
+            throw error;
+        } finally {
+            watcher.dispose();
+            state.busy = false;
+        }
+    }
+
+    private async deliverSettled(
+        state: SenderState,
+        turnId: string,
+        settled: Settled,
+        reply: ReplyFn,
+    ): Promise<void> {
+        switch (settled.kind) {
+            case "completed": {
+                let text = settled.text;
+                if (!text) {
+                    // Rare: final message had no text parts; recover the last
+                    // assistant text from the persisted turn.
+                    try {
+                        const turn = await this.deps.sessions.getTurn(turnId);
+                        text = lastAssistantText(reduceTurn(turn.events));
+                    } catch {
+                        text = null;
+                    }
+                }
+                for (const chunk of chunkReply(text ?? "✅ Done (no text reply).")) {
+                    await reply(chunk);
+                }
+                return;
+            }
+            case "failed":
+                await reply(`❌ Task failed: ${settled.error}`);
+                return;
+            case "cancelled":
+                await reply("🛑 Stopped.");
+                return;
+            case "ask_human": {
+                state.pendingAsk = { turnId, toolCallId: settled.toolCallId };
+                const lines = [`❓ ${settled.query}`];
+                if (settled.options?.length) {
+                    lines.push(...settled.options.map((o, i) => `${i + 1}. ${o}`));
+                }
+                lines.push("", "Reply with your answer.");
+                await reply(lines.join("\n"));
+                return;
+            }
+            case "suspended":
+                await reply(
+                    "⚠️ The agent is waiting for a permission approval — open Rowboat on your desktop to continue.",
+                );
+                return;
+            case "timeout":
+                await reply(
+                    "⏱️ Still running after 30 minutes — check the desktop app for progress.",
+                );
+                return;
+        }
+    }
+
+    // Buffers turn events from the moment of subscription so a settle that
+    // fires between sendMessage() returning and waitFor() attaching is never
+    // lost. One watcher per in-flight message; disposed in runMessage.
+    private watchBus() {
+        const buffered: Array<{ turnId: string; event: TurnStreamEvent }> = [];
+        let waiter: { turnId: string; resolve: (settled: Settled) => void } | null = null;
+        const unsubscribe = this.deps.sessionBus.subscribe((event) => {
+            if (event.kind !== "turn-event") return;
+            if (waiter) {
+                if (event.turnId !== waiter.turnId) return;
+                const settled = settleOf(event.event);
+                if (settled) waiter.resolve(settled);
+                return;
+            }
+            buffered.push({ turnId: event.turnId, event: event.event });
+        });
+        return {
+            waitFor: (turnId: string, timeoutMs: number): Promise<Settled> =>
+                new Promise<Settled>((resolve) => {
+                    for (const b of buffered) {
+                        if (b.turnId !== turnId) continue;
+                        const settled = settleOf(b.event);
+                        if (settled) {
+                            resolve(settled);
+                            return;
+                        }
+                    }
+                    buffered.length = 0;
+                    const timer = setTimeout(
+                        () => resolve({ kind: "timeout" }),
+                        timeoutMs,
+                    );
+                    waiter = {
+                        turnId,
+                        resolve: (settled) => {
+                            clearTimeout(timer);
+                            resolve(settled);
+                        },
+                    };
+                }),
+            dispose: unsubscribe,
+        };
+    }
+}

apps/x/packages/core/src/channels/repo.ts

@@ -0,0 +1,40 @@
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import type { z } from 'zod';
+import { WorkDir } from '../config/config.js';
+import { ChannelsConfig, DEFAULT_CHANNELS_CONFIG } from '@x/shared/dist/channels.js';
+
+export interface IChannelsConfigRepo {
+    getConfig(): Promise<z.infer<typeof ChannelsConfig>>;
+    setConfig(config: z.infer<typeof ChannelsConfig>): Promise<void>;
+}
+
+export class FSChannelsConfigRepo implements IChannelsConfigRepo {
+    private readonly configPath = path.join(WorkDir, 'config', 'channels.json');
+
+    constructor() {
+        this.ensureConfigFile();
+    }
+
+    private async ensureConfigFile(): Promise<void> {
+        try {
+            await fs.access(this.configPath);
+        } catch {
+            await fs.writeFile(this.configPath, JSON.stringify(DEFAULT_CHANNELS_CONFIG, null, 2));
+        }
+    }
+
+    async getConfig(): Promise<z.infer<typeof ChannelsConfig>> {
+        try {
+            const content = await fs.readFile(this.configPath, 'utf8');
+            return ChannelsConfig.parse(JSON.parse(content));
+        } catch {
+            return DEFAULT_CHANNELS_CONFIG;
+        }
+    }
+
+    async setConfig(config: z.infer<typeof ChannelsConfig>): Promise<void> {
+        const validated = ChannelsConfig.parse(config);
+        await fs.writeFile(this.configPath, JSON.stringify(validated, null, 2));
+    }
+}

apps/x/packages/core/src/channels/service.ts

@@ -0,0 +1,183 @@
+import path from "node:path";
+import fs from "node:fs/promises";
+import QRCode from "qrcode";
+import type { z } from "zod";
+import type { ChannelsConfig, ChannelsStatus } from "@x/shared/dist/channels.js";
+import container from "../di/container.js";
+import { WorkDir } from "../config/config.js";
+import type { ISessions } from "../sessions/api.js";
+import type { EmitterSessionBus } from "../sessions/bus.js";
+import { ChannelBridge } from "./bridge.js";
+import type { IChannelsConfigRepo } from "./repo.js";
+import { TelegramTransport } from "./transports/telegram.js";
+import { WhatsAppTransport } from "./transports/whatsapp.js";
+
+// Lifecycle owner for the mobile channels: reads config, runs the enabled
+// transports against one shared ChannelBridge, and fans status out to the
+// renderer (QR pairing, connection state). init() from main after
+// sessions.initialize(); applyChannelsConfig() on every settings save.
+
+type Config = z.infer<typeof ChannelsConfig>;
+type Status = z.infer<typeof ChannelsStatus>;
+
+const WHATSAPP_AUTH_DIR = path.join(WorkDir, "channels", "whatsapp-auth");
+
+let bridge: ChannelBridge | null = null;
+let whatsapp: WhatsAppTransport | null = null;
+let telegram: TelegramTransport | null = null;
+
+const status: Status = {
+    whatsapp: { state: "disabled" },
+    telegram: { state: "disabled" },
+};
+
+const statusListeners = new Set<(status: Status) => void>();
+
+// Serializes apply/logout so a fast settings double-save can't interleave
+// transport teardown and startup.
+let lifecycle: Promise<void> = Promise.resolve();
+
+function notifyStatus(): void {
+    for (const listener of statusListeners) {
+        try {
+            listener(structuredClone(status));
+        } catch {
+            // observers must never affect the channels
+        }
+    }
+}
+
+function setWhatsAppStatus(next: Status["whatsapp"]): void {
+    status.whatsapp = next;
+    notifyStatus();
+}
+
+function setTelegramStatus(next: Status["telegram"]): void {
+    status.telegram = next;
+    notifyStatus();
+}
+
+export function getChannelsStatus(): Status {
+    return structuredClone(status);
+}
+
+export function subscribeChannelsStatus(listener: (status: Status) => void): () => void {
+    statusListeners.add(listener);
+    return () => statusListeners.delete(listener);
+}
+
+function ensureBridge(): ChannelBridge {
+    if (!bridge) {
+        bridge = new ChannelBridge({
+            sessions: container.resolve<ISessions>("sessions"),
+            sessionBus: container.resolve<EmitterSessionBus>("sessionBus"),
+        });
+    }
+    return bridge;
+}
+
+async function stopWhatsApp(): Promise<void> {
+    if (!whatsapp) return;
+    await whatsapp.stop().catch(() => undefined);
+    whatsapp = null;
+}
+
+function stopTelegram(): void {
+    if (!telegram) return;
+    telegram.stop();
+    telegram = null;
+}
+
+function startWhatsApp(config: Config["whatsapp"]): void {
+    if (!config.enabled) {
+        setWhatsAppStatus({ state: "disabled" });
+        return;
+    }
+    const channelBridge = ensureBridge();
+    const transport = new WhatsAppTransport({
+        authDir: WHATSAPP_AUTH_DIR,
+        allowFrom: config.allowFrom,
+        onInbound: (senderKey, text, reply) => {
+            void channelBridge.handleInbound(senderKey, text, reply);
+        },
+        onStatus: (update) => {
+            if (update.state === "qr" && update.qr) {
+                // Render the pairing QR main-side so the renderer just shows
+                // an <img>; the raw pairing string never leaves core.
+                QRCode.toDataURL(update.qr, { margin: 1, width: 256 })
+                    .then((qrDataUrl) => setWhatsAppStatus({ state: "qr", qrDataUrl }))
+                    .catch(() => setWhatsAppStatus({ state: "error", error: "Failed to render pairing QR" }));
+                return;
+            }
+            setWhatsAppStatus({
+                state: update.state,
+                ...(update.self ? { self: update.self } : {}),
+                ...(update.error ? { error: update.error } : {}),
+            });
+        },
+    });
+    whatsapp = transport;
+    transport.start().catch((error) => {
+        setWhatsAppStatus({
+            state: "error",
+            error: error instanceof Error ? error.message : String(error),
+        });
+    });
+}
+
+function startTelegram(config: Config["telegram"]): void {
+    if (!config.enabled) {
+        setTelegramStatus({ state: "disabled" });
+        return;
+    }
+    if (!config.botToken) {
+        setTelegramStatus({ state: "error", error: "Bot token missing — create one with @BotFather" });
+        return;
+    }
+    const channelBridge = ensureBridge();
+    const transport = new TelegramTransport({
+        botToken: config.botToken,
+        allowFrom: config.allowFrom,
+        onInbound: (senderKey, text, reply) => {
+            void channelBridge.handleInbound(senderKey, text, reply);
+        },
+        onStatus: setTelegramStatus,
+    });
+    telegram = transport;
+    void transport.start();
+}
+
+export function applyChannelsConfig(config: Config): Promise<void> {
+    lifecycle = lifecycle.then(async () => {
+        await stopWhatsApp();
+        stopTelegram();
+        startWhatsApp(config.whatsapp);
+        startTelegram(config.telegram);
+    });
+    return lifecycle;
+}
+
+// Unlink the WhatsApp device and, if the channel is still enabled, restart it
+// so a fresh pairing QR appears. Telegram is left untouched.
+export function logoutWhatsApp(): Promise<void> {
+    lifecycle = lifecycle.then(async () => {
+        if (whatsapp) {
+            await whatsapp.logout().catch(() => undefined);
+            whatsapp = null;
+        } else {
+            await fs.rm(WHATSAPP_AUTH_DIR, { recursive: true, force: true });
+        }
+        const config = await container
+            .resolve<IChannelsConfigRepo>("channelsConfigRepo")
+            .getConfig();
+        startWhatsApp(config.whatsapp);
+    });
+    return lifecycle;
+}
+
+export async function init(): Promise<void> {
+    const config = await container
+        .resolve<IChannelsConfigRepo>("channelsConfigRepo")
+        .getConfig();
+    await applyChannelsConfig(config);
+}

apps/x/packages/core/src/channels/transports/telegram.ts

@@ -0,0 +1,133 @@
+import type { z } from "zod";
+import type { TelegramChannelStatus } from "@x/shared/dist/channels.js";
+import type { ReplyFn } from "../bridge.js";
+
+// Telegram Bot API transport. Deliberately dependency-free: the Bot API is
+// plain HTTPS — getUpdates long polling (outbound connection, works behind
+// NAT) plus sendMessage. The user supplies their own bot token (@BotFather).
+
+const POLL_TIMEOUT_S = 50;
+const RETRY_DELAY_MS = 5000;
+
+type Status = z.infer<typeof TelegramChannelStatus>;
+
+interface TelegramUpdate {
+    update_id: number;
+    message?: {
+        message_id: number;
+        text?: string;
+        chat: { id: number; type: string };
+        from?: { id: number; is_bot?: boolean };
+    };
+}
+
+export interface TelegramTransportOptions {
+    botToken: string;
+    allowFrom: string[];
+    onInbound: (senderKey: string, text: string, reply: ReplyFn) => void;
+    onStatus: (status: Status) => void;
+}
+
+export class TelegramTransport {
+    private abort: AbortController | null = null;
+    private stopped = false;
+    private offset = 0;
+
+    constructor(private readonly opts: TelegramTransportOptions) {}
+
+    async start(): Promise<void> {
+        this.stopped = false;
+        this.opts.onStatus({ state: "starting" });
+        void this.run();
+    }
+
+    stop(): void {
+        this.stopped = true;
+        this.abort?.abort();
+        this.opts.onStatus({ state: "disabled" });
+    }
+
+    private api(method: string): string {
+        return `https://api.telegram.org/bot${this.opts.botToken}/${method}`;
+    }
+
+    private async call(method: string, body?: unknown, signal?: AbortSignal): Promise<unknown> {
+        const res = await fetch(this.api(method), {
+            method: "POST",
+            headers: { "content-type": "application/json" },
+            body: body === undefined ? undefined : JSON.stringify(body),
+            ...(signal ? { signal } : {}),
+        });
+        const payload = (await res.json()) as { ok: boolean; result?: unknown; description?: string };
+        if (!payload.ok) {
+            throw new Error(payload.description ?? `Telegram API error (${method})`);
+        }
+        return payload.result;
+    }
+
+    private async run(): Promise<void> {
+        // Validate the token up front so a typo surfaces in settings
+        // immediately instead of as a silent poll loop failure.
+        try {
+            const me = (await this.call("getMe")) as { username?: string };
+            if (this.stopped) return;
+            this.opts.onStatus({ state: "polling", botUsername: me.username });
+        } catch (error) {
+            if (this.stopped) return;
+            this.opts.onStatus({
+                state: "error",
+                error: error instanceof Error ? error.message : String(error),
+            });
+            return;
+        }
+
+        while (!this.stopped) {
+            this.abort = new AbortController();
+            try {
+                const updates = (await this.call(
+                    "getUpdates",
+                    {
+                        timeout: POLL_TIMEOUT_S,
+                        offset: this.offset,
+                        allowed_updates: ["message"],
+                    },
+                    this.abort.signal,
+                )) as TelegramUpdate[];
+                for (const update of updates) {
+                    this.offset = Math.max(this.offset, update.update_id + 1);
+                    this.handleUpdate(update);
+                }
+            } catch (error) {
+                if (this.stopped) return;
+                this.opts.onStatus({
+                    state: "error",
+                    error: error instanceof Error ? error.message : String(error),
+                });
+                await new Promise((resolve) => setTimeout(resolve, RETRY_DELAY_MS));
+                if (this.stopped) return;
+                this.opts.onStatus({ state: "polling" });
+            }
+        }
+    }
+
+    private handleUpdate(update: TelegramUpdate): void {
+        const message = update.message;
+        if (!message?.text || message.from?.is_bot) return;
+        // DMs only: group chats would let any member drive the bridge.
+        if (message.chat.type !== "private") return;
+        const chatId = String(message.chat.id);
+        const reply: ReplyFn = (text) => this.send(chatId, text);
+        if (!this.opts.allowFrom.includes(chatId)) {
+            void this.send(
+                chatId,
+                `⛔ Not authorized. Your chat ID is ${chatId} — add it under Rowboat → Settings → Mobile to pair this chat.`,
+            ).catch(() => undefined);
+            return;
+        }
+        this.opts.onInbound(`telegram:${chatId}`, message.text, reply);
+    }
+
+    private async send(chatId: string, text: string): Promise<void> {
+        await this.call("sendMessage", { chat_id: chatId, text });
+    }
+}

apps/x/packages/core/src/channels/transports/whatsapp.ts

@@ -0,0 +1,191 @@
+import fs from "node:fs/promises";
+import makeWASocket, {
+    DisconnectReason,
+    areJidsSameUser,
+    isJidGroup,
+    jidDecode,
+    useMultiFileAuthState,
+} from "baileys";
+import type { ReplyFn } from "../bridge.js";
+
+// WhatsApp transport via Baileys: the app links to the user's own WhatsApp
+// account as a linked device (QR pairing, same as WhatsApp Web) over an
+// outbound WebSocket — no server, no port forwarding.
+//
+// Access model: the linked account's own self-chat ("message yourself") is
+// always allowed; other senders must be explicitly allowlisted by phone
+// number. Group chats are ignored entirely.
+
+type WASocket = ReturnType<typeof makeWASocket>;
+
+const RECONNECT_DELAY_MS = 3000;
+// Marks bridge-sent messages. In the self-chat our own replies come back on
+// messages.upsert like any other message; the marker (plus sent-id tracking)
+// keeps the bridge from answering itself in a loop.
+const REPLY_MARKER = "🤖 ";
+
+export interface WhatsAppTransportStatus {
+    state: "starting" | "qr" | "connected" | "error" | "disabled";
+    qr?: string;
+    self?: string;
+    error?: string;
+}
+
+export interface WhatsAppTransportOptions {
+    authDir: string;
+    allowFrom: string[];
+    onInbound: (senderKey: string, text: string, reply: ReplyFn) => void;
+    onStatus: (status: WhatsAppTransportStatus) => void;
+}
+
+interface TextishMessage {
+    conversation?: unknown;
+    extendedTextMessage?: { text?: unknown };
+    ephemeralMessage?: { message?: TextishMessage };
+}
+
+interface InboundWAMessage {
+    key?: { remoteJid?: string | null; fromMe?: boolean | null; id?: string | null };
+    message?: unknown;
+}
+
+function messageText(message: unknown): string | null {
+    if (!message || typeof message !== "object") return null;
+    const m = message as TextishMessage;
+    const unwrapped = m.ephemeralMessage?.message ?? m;
+    const text: unknown = unwrapped.conversation ?? unwrapped.extendedTextMessage?.text;
+    return typeof text === "string" && text ? text : null;
+}
+
+export class WhatsAppTransport {
+    private sock: WASocket | null = null;
+    private stopped = false;
+    private sentIds = new Set<string>();
+
+    constructor(private readonly opts: WhatsAppTransportOptions) {}
+
+    async start(): Promise<void> {
+        this.stopped = false;
+        this.opts.onStatus({ state: "starting" });
+        await this.connect();
+    }
+
+    async stop(): Promise<void> {
+        this.stopped = true;
+        try {
+            this.sock?.end(undefined);
+        } catch {
+            // already closed
+        }
+        this.sock = null;
+        this.opts.onStatus({ state: "disabled" });
+    }
+
+    // Unlink this device: invalidates the pairing on the phone and clears
+    // local credentials so the next start shows a fresh QR.
+    async logout(): Promise<void> {
+        this.stopped = true;
+        try {
+            await this.sock?.logout();
+        } catch {
+            // best effort — clearing creds below is what actually unpairs us
+        }
+        this.sock = null;
+        await fs.rm(this.opts.authDir, { recursive: true, force: true });
+        this.opts.onStatus({ state: "disabled" });
+    }
+
+    private async connect(): Promise<void> {
+        if (this.stopped) return;
+        const { state, saveCreds } = await useMultiFileAuthState(this.opts.authDir);
+        const sock = makeWASocket({
+            auth: state,
+            syncFullHistory: false,
+            markOnlineOnConnect: false,
+        });
+        this.sock = sock;
+
+        sock.ev.on("creds.update", saveCreds);
+
+        sock.ev.on("connection.update", (update) => {
+            if (this.stopped) return;
+            if (update.qr) {
+                this.opts.onStatus({ state: "qr", qr: update.qr });
+            }
+            if (update.connection === "open") {
+                const self = jidDecode(sock.user?.id ?? "")?.user;
+                this.opts.onStatus({ state: "connected", ...(self ? { self } : {}) });
+            }
+            if (update.connection === "close") {
+                const statusCode = (update.lastDisconnect?.error as { output?: { statusCode?: number } } | undefined)
+                    ?.output?.statusCode;
+                if (statusCode === DisconnectReason.loggedOut) {
+                    // Unlinked from the phone; stale creds would loop forever.
+                    void fs.rm(this.opts.authDir, { recursive: true, force: true });
+                    this.opts.onStatus({
+                        state: "error",
+                        error: "Logged out from the phone — toggle WhatsApp off and on to pair again.",
+                    });
+                    return;
+                }
+                setTimeout(() => {
+                    this.connect().catch((error) => {
+                        this.opts.onStatus({
+                            state: "error",
+                            error: error instanceof Error ? error.message : String(error),
+                        });
+                    });
+                }, RECONNECT_DELAY_MS);
+            }
+        });
+
+        sock.ev.on("messages.upsert", ({ messages, type }) => {
+            if (type !== "notify") return;
+            for (const msg of messages) {
+                this.handleMessage(sock, msg);
+            }
+        });
+    }
+
+    private handleMessage(sock: WASocket, msg: InboundWAMessage): void {
+        const jid: string | undefined = msg.key?.remoteJid ?? undefined;
+        if (!jid || isJidGroup(jid) || jid === "status@broadcast") return;
+        const messageId: string | undefined = msg.key?.id ?? undefined;
+        if (messageId && this.sentIds.has(messageId)) return;
+        const text = messageText(msg.message);
+        if (!text || text.startsWith(REPLY_MARKER)) return;
+
+        const selfIds = [sock.user?.id, (sock.user as { lid?: string } | undefined)?.lid].filter(
+            (v): v is string => Boolean(v),
+        );
+        const isSelfChat = selfIds.some((selfId) => areJidsSameUser(jid, selfId));
+        const senderNumber = jidDecode(jid)?.user ?? "";
+
+        // Self-chat is the owner by definition. Anyone else must be
+        // allowlisted — this bridge is remote control over the desktop agent.
+        if (!isSelfChat) {
+            if (msg.key?.fromMe) return;
+            if (!this.opts.allowFrom.includes(senderNumber)) return;
+        }
+
+        const reply: ReplyFn = (replyText) => this.send(jid, replyText);
+        this.opts.onInbound(`whatsapp:${senderNumber || jid}`, text, reply);
+    }
+
+    private async send(jid: string, text: string): Promise<void> {
+        const sock = this.sock;
+        if (!sock) throw new Error("WhatsApp is not connected");
+        const sent = await sock.sendMessage(jid, { text: `${REPLY_MARKER}${text}` });
+        const id = sent?.key?.id;
+        if (id) {
+            this.sentIds.add(id);
+            if (this.sentIds.size > 500) {
+                // Trim oldest; Set iteration order is insertion order.
+                for (const old of this.sentIds) {
+                    this.sentIds.delete(old);
+                    if (this.sentIds.size <= 250) break;
+                }
+            }
+        }
+    }
+}

apps/x/packages/core/src/di/container.ts

@@ -18,6 +18,7 @@ import { IAbortRegistry, InMemoryAbortRegistry } from "../runs/abort-registry.js
 import { FSAgentScheduleRepo, IAgentScheduleRepo } from "../agent-schedule/repo.js";
 import { FSAgentScheduleStateRepo, IAgentScheduleStateRepo } from "../agent-schedule/state-repo.js";
 import { FSSlackConfigRepo, ISlackConfigRepo } from "../slack/repo.js";
+import { FSChannelsConfigRepo, IChannelsConfigRepo } from "../channels/repo.js";
 import { CodeModeManager } from "../code-mode/acp/manager.js";
 import { CodePermissionRegistry } from "../code-mode/acp/permission-registry.js";
 import { FSCodeProjectsRepo, ICodeProjectsRepo } from "../code-mode/projects/repo.js";
@@ -90,6 +91,7 @@ container.register({
     agentScheduleRepo: asClass<IAgentScheduleRepo>(FSAgentScheduleRepo).singleton(),
     agentScheduleStateRepo: asClass<IAgentScheduleStateRepo>(FSAgentScheduleStateRepo).singleton(),
     slackConfigRepo: asClass<ISlackConfigRepo>(FSSlackConfigRepo).singleton(),
+    channelsConfigRepo: asClass<IChannelsConfigRepo>(FSChannelsConfigRepo).singleton(),
 
     // ACP code-mode engine: the manager holds a live agent connection per chat only
     // around an active turn (torn down after a short idle grace; resumed via