From c80a09daba7042c424782bc34578009691f1bc98 Mon Sep 17 00:00:00 2001 From: Arjun <6592213+arkml@users.noreply.github.com> Date: Fri, 3 Jul 2026 20:17:31 +0530 Subject: [PATCH] =?UTF-8?q?feat(channels):=20mobile=20bridge=20=E2=80=94?= =?UTF-8?q?=20chat=20with=20Rowboat=20from=20WhatsApp/Telegram?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a transport-agnostic messaging bridge in @x/core that drives the session/turn runtime from your phone: WhatsApp links as a companion device via Baileys QR pairing (self-chat is the command channel), Telegram long-polls the user's own bot token. Commands: list, resume N, new, status, stop; anything else runs a turn in the current session and replies with the final assistant text. Ask-human questions are relayed and answerable from the phone. - packages/core/src/channels/: bridge (commands, settle watcher, ask-human relay), WhatsApp + Telegram transports, config repo, service (lifecycle, status fan-out, QR rendering, current-transport reply routing, dynamic baileys import) - Settings β†’ Mobile tab: enable toggles, QR pairing, bot token, sender allowlists - Strict sender authorization: WhatsApp self-chat + allowlisted numbers (LID-aware), Telegram allowlisted chat IDs only - Telegram offset persisted across restarts; generation-guarded WhatsApp socket lifecycle; vitest coverage for the bridge Co-Authored-By: Claude Fable 5 --- .../settings/mobile-channels-settings.tsx | 8 +- .../packages/core/src/channels/bridge.test.ts | 208 ++++++++++++++++++ apps/x/packages/core/src/channels/bridge.ts | 162 +++++++++----- apps/x/packages/core/src/channels/service.ts | 92 ++++++-- .../core/src/channels/transports/telegram.ts | 151 ++++++++++--- .../core/src/channels/transports/whatsapp.ts | 64 ++++-- 6 files changed, 552 insertions(+), 133 deletions(-) create mode 100644 apps/x/packages/core/src/channels/bridge.test.ts diff --git a/apps/x/apps/renderer/src/components/settings/mobile-channels-settings.tsx b/apps/x/apps/renderer/src/components/settings/mobile-channels-settings.tsx index 4f8fc2a4..9ca775e5 100644 --- a/apps/x/apps/renderer/src/components/settings/mobile-channels-settings.tsx +++ b/apps/x/apps/renderer/src/components/settings/mobile-channels-settings.tsx @@ -13,11 +13,13 @@ import type { ChannelsConfig, ChannelsStatus } from "@x/shared/src/channels.js" type Config = z.infer type Status = z.infer -// Comma/space/newline separated ids ↔ list. +// Comma/newline separated entries; each entry keeps digits only, so a +// formatted number like "+1 (415) 555-1234" survives as one entry instead of +// being shattered at the spaces. function parseIdList(draft: string): string[] { return draft - .split(/[\s,]+/) - .map((s) => s.replace(/[^\d]/g, "")) + .split(/[,;\n]+/) + .map((s) => s.replace(/\D/g, "")) .filter(Boolean) } diff --git a/apps/x/packages/core/src/channels/bridge.test.ts b/apps/x/packages/core/src/channels/bridge.test.ts new file mode 100644 index 00000000..97c71ddf --- /dev/null +++ b/apps/x/packages/core/src/channels/bridge.test.ts @@ -0,0 +1,208 @@ +import { describe, expect, it, vi } from "vitest"; +import type { SessionIndexEntry } from "@x/shared/dist/sessions.js"; +import type { TurnStreamEvent } from "@x/shared/dist/turns.js"; +import { EmitterSessionBus } from "../sessions/bus.js"; +import { TurnInputError } from "../turns/api.js"; +import type { ISessions } from "../sessions/api.js"; +import { ChannelBridge } from "./bridge.js"; + +const SENDER = "test:1"; + +function entry(overrides: Partial): SessionIndexEntry { + return { + sessionId: "s1", + createdAt: "2026-07-01T00:00:00Z", + updatedAt: "2026-07-01T00:00:00Z", + turnCount: 1, + latestTurnStatus: "completed", + ...overrides, + }; +} + +function completedEvent(turnId: string, text: string): TurnStreamEvent { + return { + type: "turn_completed", + turnId, + ts: "2026-07-01T00:00:00Z", + output: { role: "assistant", content: text }, + finishReason: "stop", + usage: {}, + } as unknown as TurnStreamEvent; +} + +function askEvent(turnId: string, question: string, options?: string[]): TurnStreamEvent { + return { + type: "turn_suspended", + turnId, + ts: "2026-07-01T00:00:00Z", + pendingPermissions: [], + pendingAsyncTools: [ + { + toolCallId: "call_1", + toolId: "builtin:ask-human", + toolName: "ask-human", + input: { question, ...(options ? { options } : {}) }, + }, + ], + usage: {}, + } as unknown as TurnStreamEvent; +} + +interface Harness { + bridge: ChannelBridge; + bus: EmitterSessionBus; + replies: string[]; + reply: (text: string) => Promise; + sessions: { + createSession: ReturnType; + sendMessage: ReturnType; + respondToAskHuman: ReturnType; + stopTurn: ReturnType; + getTurn: ReturnType; + listSessions: ReturnType; + }; + publish: (turnId: string, event: TurnStreamEvent) => void; +} + +function harness(entries: SessionIndexEntry[] = []): Harness { + const bus = new EmitterSessionBus(); + const publish = (turnId: string, event: TurnStreamEvent) => + bus.publish({ kind: "turn-event", sessionId: "s1", turnId, event }); + const sessions = { + createSession: vi.fn(async () => "s1"), + sendMessage: vi.fn(async () => ({ turnId: "t1" })), + respondToAskHuman: vi.fn(async () => undefined), + stopTurn: vi.fn(async () => undefined), + getTurn: vi.fn(async () => ({ turnId: "t1", events: [] })), + listSessions: vi.fn(() => entries), + }; + const bridge = new ChannelBridge({ + sessions: sessions as unknown as ISessions, + sessionBus: bus, + }); + const replies: string[] = []; + const reply = async (text: string) => { + replies.push(text); + }; + return { bridge, bus, replies, reply, sessions, publish }; +} + +// Settle the turn as soon as sendMessage is called: the watcher subscribes +// before sendMessage, so a synchronous publish lands in its buffer β€” the +// exact race the buffering exists for. +function settleOnSend(h: Harness, event: TurnStreamEvent, turnId = "t1"): void { + h.sessions.sendMessage.mockImplementation(async () => { + h.publish(turnId, event); + return { turnId }; + }); +} + +describe("ChannelBridge commands", () => { + it("replies with help text", async () => { + const h = harness(); + await h.bridge.handleInbound(SENDER, "help", h.reply); + expect(h.replies).toHaveLength(1); + expect(h.replies[0]).toContain("Rowboat commands"); + }); + + it("lists recent sessions newest-first and resumes by index", async () => { + const h = harness([ + entry({ sessionId: "old", title: "Old chat", updatedAt: "2026-07-01T00:00:00Z" }), + entry({ sessionId: "new", title: "New chat", updatedAt: "2026-07-02T00:00:00Z" }), + ]); + await h.bridge.handleInbound(SENDER, "list", h.reply); + expect(h.replies[0]).toContain("1. New chat"); + expect(h.replies[0]).toContain("2. Old chat"); + + await h.bridge.handleInbound(SENDER, "resume 2", h.reply); + expect(h.replies[1]).toContain('Resumed "Old chat"'); + + settleOnSend(h, completedEvent("t1", "done")); + await h.bridge.handleInbound(SENDER, "hello again", h.reply); + expect(h.sessions.sendMessage).toHaveBeenCalledWith( + "old", + expect.objectContaining({ content: "hello again" }), + expect.objectContaining({ autoPermission: true }), + ); + }); + + it("rejects resume with an out-of-range index", async () => { + const h = harness([entry({ sessionId: "s1", title: "Only chat" })]); + await h.bridge.handleInbound(SENDER, "resume 5", h.reply); + expect(h.replies[0]).toContain("No chat #5"); + }); +}); + +describe("ChannelBridge message flow", () => { + it("creates a session, acks, and delivers the completed text", async () => { + const h = harness(); + settleOnSend(h, completedEvent("t1", "The answer is 42.")); + await h.bridge.handleInbound(SENDER, "what is the answer?", h.reply); + expect(h.sessions.createSession).toHaveBeenCalledOnce(); + expect(h.replies[0]).toContain("Working on it"); + expect(h.replies[1]).toBe("The answer is 42."); + }); + + it("delivers a failed turn as an error reply", async () => { + const h = harness(); + settleOnSend(h, { + type: "turn_failed", + turnId: "t1", + ts: "2026-07-01T00:00:00Z", + error: "model exploded", + usage: {}, + } as unknown as TurnStreamEvent); + await h.bridge.handleInbound(SENDER, "do a thing", h.reply); + expect(h.replies[1]).toContain("model exploded"); + }); + + it("relays the ask-human question text and options, then routes the answer", async () => { + const h = harness(); + settleOnSend(h, askEvent("t1", "Which lane?", ["fast", "slow"])); + await h.bridge.handleInbound(SENDER, "start task", h.reply); + expect(h.replies[1]).toContain("Which lane?"); + expect(h.replies[1]).toContain("1. fast"); + + // The answer resolves the ask; the turn then completes. + h.sessions.respondToAskHuman.mockImplementation(async () => { + h.publish("t1", completedEvent("t1", "Took the fast lane.")); + }); + await h.bridge.handleInbound(SENDER, "fast", h.reply); + expect(h.sessions.respondToAskHuman).toHaveBeenCalledWith("t1", "call_1", "fast"); + expect(h.replies[3]).toBe("Took the fast lane."); + }); + + it("falls back to a normal message when the ask was already answered elsewhere", async () => { + const h = harness(); + settleOnSend(h, askEvent("t1", "Which lane?")); + await h.bridge.handleInbound(SENDER, "start task", h.reply); + + h.sessions.respondToAskHuman.mockRejectedValue( + new TurnInputError("no pending async tool call call_1"), + ); + settleOnSend(h, completedEvent("t2", "Handled as a new message."), "t2"); + await h.bridge.handleInbound(SENDER, "actually do this instead", h.reply); + expect(h.sessions.sendMessage).toHaveBeenLastCalledWith( + "s1", + expect.objectContaining({ content: "actually do this instead" }), + expect.anything(), + ); + expect(h.replies.at(-1)).toBe("Handled as a new message."); + }); + + it("reports busy while a turn is in flight", async () => { + const h = harness(); + let releaseTurn!: () => void; + h.sessions.sendMessage.mockImplementation(async () => { + releaseTurn = () => h.publish("t1", completedEvent("t1", "finally")); + return { turnId: "t1" }; + }); + const first = h.bridge.handleInbound(SENDER, "slow task", h.reply); + await new Promise((resolve) => setTimeout(resolve, 0)); + await h.bridge.handleInbound(SENDER, "impatient follow-up", h.reply); + expect(h.replies.some((r) => r.includes("Still working"))).toBe(true); + releaseTurn(); + await first; + expect(h.replies.at(-1)).toBe("finally"); + }); +}); diff --git a/apps/x/packages/core/src/channels/bridge.ts b/apps/x/packages/core/src/channels/bridge.ts index 187331cc..41643676 100644 --- a/apps/x/packages/core/src/channels/bridge.ts +++ b/apps/x/packages/core/src/channels/bridge.ts @@ -1,6 +1,8 @@ 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 { TurnInputError } from "../turns/api.js"; +import { ASK_HUMAN_TOOL } from "../turns/bridges/real-agent-resolver.js"; import { TurnNotSettledError, type ISessions } from "../sessions/api.js"; import type { EmitterSessionBus } from "../sessions/bus.js"; @@ -18,6 +20,8 @@ const LIST_LIMIT = 10; const REPLY_CHUNK_SIZE = 3500; const MAX_REPLY_CHUNKS = 3; +const ASK_HUMAN_TOOL_ID = `builtin:${ASK_HUMAN_TOOL}`; + const HELP_TEXT = [ "πŸ€– Rowboat commands:", "β€’ list β€” recent chats", @@ -57,13 +61,13 @@ function settleOf(event: TurnStreamEvent): Settled | null { return { kind: "cancelled" }; case "turn_suspended": { const ask = event.pendingAsyncTools.find( - (t) => t.toolId === "builtin:ask-human" || t.toolName === "ask-human", + (t) => t.toolId === ASK_HUMAN_TOOL_ID || t.toolName === ASK_HUMAN_TOOL, ); if (ask) { - const input = ask.input as { query?: unknown; options?: unknown } | null; + const input = ask.input as { question?: unknown; options?: unknown } | null; const query = - typeof input?.query === "string" && input.query - ? input.query + typeof input?.question === "string" && input.question + ? input.question : "The agent needs your input."; const options = Array.isArray(input?.options) ? input.options.filter((o): o is string => typeof o === "string") @@ -108,6 +112,11 @@ function chunkReply(text: string): string[] { return parts; } +interface TurnWatcher { + waitFor(turnId: string, timeoutMs: number): Promise; + dispose(): void; +} + export class ChannelBridge { private senders = new Map(); @@ -175,15 +184,20 @@ export class ChannelBridge { return state; } - private sortedSessions(): SessionIndexEntry[] { + private sessionEntry(sessionId: string): SessionIndexEntry | undefined { + return this.deps.sessions.listSessions().find((e) => e.sessionId === sessionId); + } + + private recentSessions(): SessionIndexEntry[] { return this.deps.sessions .listSessions() .filter((e) => !e.error) - .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt)); + .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt)) + .slice(0, LIST_LIMIT); } private renderList(state: SenderState): string { - const entries = this.sortedSessions().slice(0, LIST_LIMIT); + const entries = this.recentSessions(); if (entries.length === 0) { return "No chats yet β€” just send a message to start one."; } @@ -205,9 +219,7 @@ export class ChannelBridge { private resumeSession(state: SenderState, index: number): string { if (state.lastList.length === 0) { - state.lastList = this.sortedSessions() - .slice(0, LIST_LIMIT) - .map((e) => e.sessionId); + state.lastList = this.recentSessions().map((e) => e.sessionId); } const sessionId = state.lastList[index - 1]; if (!sessionId) { @@ -215,7 +227,7 @@ export class ChannelBridge { } state.activeSessionId = sessionId; state.pendingAsk = null; - const entry = this.sortedSessions().find((e) => e.sessionId === sessionId); + const entry = this.sessionEntry(sessionId); return `▢️ Resumed "${entry?.title ?? "Untitled"}" β€” send a message to continue.`; } @@ -223,9 +235,7 @@ export class ChannelBridge { if (!state.activeSessionId) { return "No current chat β€” your next message starts a new one."; } - const entry = this.sortedSessions().find( - (e) => e.sessionId === state.activeSessionId, - ); + const entry = this.sessionEntry(state.activeSessionId); if (!entry) return "Current chat no longer exists β€” send a message to start fresh."; const status = state.busy ? "working" @@ -238,9 +248,7 @@ export class ChannelBridge { private async stopActive(state: SenderState): Promise { state.pendingAsk = null; if (!state.activeSessionId) return "Nothing to stop."; - const entry = this.deps.sessions - .listSessions() - .find((e) => e.sessionId === state.activeSessionId); + const entry = this.sessionEntry(state.activeSessionId); if (!entry?.latestTurnId) return "Nothing to stop."; if ( entry.latestTurnStatus === "completed" || @@ -259,30 +267,18 @@ export class ChannelBridge { return; } state.busy = true; - // Subscribe before advancing so no settle event can slip past. - const watcher = this.watchBus(); try { - let turnId: string; + await reply("⏳ Working on it…"); 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 answered = await this.answerAsk(state, ask, text, reply); + if (answered) return; + // The ask was already resolved elsewhere (e.g. answered in the + // desktop UI) or the turn is terminal β€” treat the text as a + // normal message instead of discarding it. } - const settled = await watcher.waitFor(turnId, TURN_TIMEOUT_MS); - await this.deliverSettled(state, turnId, settled, reply); + await this.sendToSession(state, text, reply); } catch (error) { if (error instanceof TurnNotSettledError) { await reply( @@ -292,11 +288,61 @@ export class ChannelBridge { } throw error; } finally { - watcher.dispose(); state.busy = false; } } + private async sendToSession(state: SenderState, text: string, reply: ReplyFn): Promise { + // Subscribe before advancing so no settle event can slip past. + const watcher = this.watchBus(); + try { + if (!state.activeSessionId) { + state.activeSessionId = await this.deps.sessions.createSession(); + } + const sent = await this.deps.sessions.sendMessage( + state.activeSessionId, + { role: "user", content: text }, + { agent: { agentId: AGENT_ID }, autoPermission: true }, + ); + const settled = await watcher.waitFor(sent.turnId, TURN_TIMEOUT_MS); + await this.deliverSettled(state, sent.turnId, settled, reply); + } finally { + watcher.dispose(); + } + } + + // Returns false when the ask was stale (already answered on the desktop / + // turn terminal) β€” the caller then routes the text as a normal message. + private async answerAsk( + state: SenderState, + ask: { turnId: string; toolCallId: string }, + text: string, + reply: ReplyFn, + ): Promise { + const watcher = this.watchBus(); + try { + const settledPromise = watcher.waitFor(ask.turnId, TURN_TIMEOUT_MS); + // respondToAskHuman resolves only when the whole advance settles, + // so it must not be awaited ahead of the watcher (that would + // bypass TURN_TIMEOUT_MS). Race instead: its rejection (stale + // ask) must beat the 30-minute timeout; its success defers to the + // settle event. + const settled = await Promise.race([ + settledPromise, + this.deps.sessions + .respondToAskHuman(ask.turnId, ask.toolCallId, text) + .then(() => settledPromise), + ]); + await this.deliverSettled(state, ask.turnId, settled, reply); + return true; + } catch (error) { + if (error instanceof TurnInputError) return false; + throw error; + } finally { + watcher.dispose(); + } + } + private async deliverSettled( state: SenderState, turnId: string, @@ -350,38 +396,35 @@ export class ChannelBridge { } } - // 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 }> = []; + // Buffers settle-relevant events (β‰ˆ1 per turn) from the moment of + // subscription so a settle firing between advance-start and waitFor() is + // never lost β€” without retaining the per-token delta stream of every + // concurrent session. One watcher per in-flight message. + private watchBus(): TurnWatcher { + const buffered: Array<{ turnId: string; settled: Settled }> = []; let waiter: { turnId: string; resolve: (settled: Settled) => void } | null = null; + let cancelTimer: (() => void) | null = null; const unsubscribe = this.deps.sessionBus.subscribe((event) => { if (event.kind !== "turn-event") return; + const settled = settleOf(event.event); + if (!settled) return; if (waiter) { - if (event.turnId !== waiter.turnId) return; - const settled = settleOf(event.event); - if (settled) waiter.resolve(settled); + if (event.turnId === waiter.turnId) waiter.resolve(settled); return; } - buffered.push({ turnId: event.turnId, event: event.event }); + buffered.push({ turnId: event.turnId, settled }); }); return { waitFor: (turnId: string, timeoutMs: number): Promise => new Promise((resolve) => { - for (const b of buffered) { - if (b.turnId !== turnId) continue; - const settled = settleOf(b.event); - if (settled) { - resolve(settled); - return; - } + const hit = buffered.find((b) => b.turnId === turnId); + if (hit) { + resolve(hit.settled); + return; } buffered.length = 0; - const timer = setTimeout( - () => resolve({ kind: "timeout" }), - timeoutMs, - ); + const timer = setTimeout(() => resolve({ kind: "timeout" }), timeoutMs); + cancelTimer = () => clearTimeout(timer); waiter = { turnId, resolve: (settled) => { @@ -390,7 +433,10 @@ export class ChannelBridge { }, }; }), - dispose: unsubscribe, + dispose: () => { + unsubscribe(); + cancelTimer?.(); + }, }; } } diff --git a/apps/x/packages/core/src/channels/service.ts b/apps/x/packages/core/src/channels/service.ts index d7fb8754..164d8efa 100644 --- a/apps/x/packages/core/src/channels/service.ts +++ b/apps/x/packages/core/src/channels/service.ts @@ -1,6 +1,5 @@ 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"; @@ -10,7 +9,9 @@ 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"; +// Type-only: the real module (which pulls the ~9MB baileys dependency) is +// loaded dynamically in startWhatsApp, so boot pays nothing while disabled. +import type { 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 @@ -21,6 +22,7 @@ type Config = z.infer; type Status = z.infer; const WHATSAPP_AUTH_DIR = path.join(WorkDir, "channels", "whatsapp-auth"); +const TELEGRAM_STATE_FILE = path.join(WorkDir, "channels", "telegram-state.json"); let bridge: ChannelBridge | null = null; let whatsapp: WhatsAppTransport | null = null; @@ -34,13 +36,22 @@ const status: Status = { const statusListeners = new Set<(status: Status) => void>(); // Serializes apply/logout so a fast settings double-save can't interleave -// transport teardown and startup. +// transport teardown and startup. enqueue() recovers the chain before adding +// a step β€” a rejected step must fail only its own caller, never poison every +// later settings save. let lifecycle: Promise = Promise.resolve(); +function enqueue(step: () => Promise): Promise { + const run = lifecycle.catch(() => undefined).then(step); + lifecycle = run.catch(() => undefined); + return run; +} + function notifyStatus(): void { + const snapshot = structuredClone(status); for (const listener of statusListeners) { try { - listener(structuredClone(status)); + listener(snapshot); } catch { // observers must never affect the channels } @@ -78,37 +89,67 @@ function ensureBridge(): ChannelBridge { async function stopWhatsApp(): Promise { if (!whatsapp) return; - await whatsapp.stop().catch(() => undefined); + const stopping = whatsapp; whatsapp = null; + await stopping.stop().catch(() => undefined); + setWhatsAppStatus({ state: "disabled" }); } function stopTelegram(): void { if (!telegram) return; - telegram.stop(); + const stopping = telegram; telegram = null; + stopping.stop(); + setTelegramStatus({ state: "disabled" }); } -function startWhatsApp(config: Config["whatsapp"]): void { +// Invalidates pending async QR renders whenever a newer status lands. +let qrSeq = 0; + +async function startWhatsApp(config: Config["whatsapp"]): Promise { if (!config.enabled) { setWhatsAppStatus({ state: "disabled" }); return; } const channelBridge = ensureBridge(); - const transport = new WhatsAppTransport({ + const [{ WhatsAppTransport: Transport }, QRCode] = await Promise.all([ + import("./transports/whatsapp.js"), + import("qrcode").then((m) => m.default), + ]); + const transport = new Transport({ authDir: WHATSAPP_AUTH_DIR, allowFrom: config.allowFrom, - onInbound: (senderKey, text, reply) => { + onInbound: (senderKey, chatJid, text) => { + // Route replies through whichever transport is current at send + // time β€” the originating instance may have been replaced by a + // settings save while the turn was running. + const reply = async (replyText: string) => { + const current = whatsapp; + if (!current) throw new Error("WhatsApp channel is disabled"); + await current.send(chatJid, replyText); + }; void channelBridge.handleInbound(senderKey, text, reply); }, onStatus: (update) => { + if (whatsapp !== transport) return; // superseded instance if (update.state === "qr" && update.qr) { + const seq = ++qrSeq; // Render the pairing QR main-side so the renderer just shows // an ; 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" })); + .then((qrDataUrl) => { + if (whatsapp === transport && seq === qrSeq) { + setWhatsAppStatus({ state: "qr", qrDataUrl }); + } + }) + .catch(() => { + if (whatsapp === transport && seq === qrSeq) { + setWhatsAppStatus({ state: "error", error: "Failed to render pairing QR" }); + } + }); return; } + qrSeq++; setWhatsAppStatus({ state: update.state, ...(update.self ? { self: update.self } : {}), @@ -118,6 +159,7 @@ function startWhatsApp(config: Config["whatsapp"]): void { }); whatsapp = transport; transport.start().catch((error) => { + if (whatsapp !== transport) return; setWhatsAppStatus({ state: "error", error: error instanceof Error ? error.message : String(error), @@ -138,41 +180,49 @@ function startTelegram(config: Config["telegram"]): void { const transport = new TelegramTransport({ botToken: config.botToken, allowFrom: config.allowFrom, - onInbound: (senderKey, text, reply) => { + stateFile: TELEGRAM_STATE_FILE, + onInbound: (senderKey, chatId, text) => { + const reply = async (replyText: string) => { + const current = telegram; + if (!current) throw new Error("Telegram channel is disabled"); + await current.send(chatId, replyText); + }; void channelBridge.handleInbound(senderKey, text, reply); }, - onStatus: setTelegramStatus, + onStatus: (update) => { + if (telegram !== transport) return; // superseded instance + setTelegramStatus(update); + }, }); telegram = transport; void transport.start(); } export function applyChannelsConfig(config: Config): Promise { - lifecycle = lifecycle.then(async () => { + return enqueue(async () => { await stopWhatsApp(); stopTelegram(); - startWhatsApp(config.whatsapp); + await 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 { - lifecycle = lifecycle.then(async () => { + return enqueue(async () => { if (whatsapp) { - await whatsapp.logout().catch(() => undefined); + const out = whatsapp; whatsapp = null; + await out.logout().catch(() => undefined); } else { - await fs.rm(WHATSAPP_AUTH_DIR, { recursive: true, force: true }); + await fs.rm(WHATSAPP_AUTH_DIR, { recursive: true, force: true }).catch(() => undefined); } const config = await container .resolve("channelsConfigRepo") .getConfig(); - startWhatsApp(config.whatsapp); + await startWhatsApp(config.whatsapp); }); - return lifecycle; } export async function init(): Promise { diff --git a/apps/x/packages/core/src/channels/transports/telegram.ts b/apps/x/packages/core/src/channels/transports/telegram.ts index ccb0069f..b4616fd7 100644 --- a/apps/x/packages/core/src/channels/transports/telegram.ts +++ b/apps/x/packages/core/src/channels/transports/telegram.ts @@ -1,16 +1,39 @@ +import fs from "node:fs/promises"; +import path from "node:path"; 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). +// +// The getUpdates offset is persisted to disk after each processed batch: +// Telegram only marks updates confirmed when a LATER getUpdates call passes a +// higher offset, so without persistence every transport restart (app relaunch +// or settings save) would redeliver β€” and re-execute β€” the last batch. const POLL_TIMEOUT_S = 50; const RETRY_DELAY_MS = 5000; +const MAX_RETRY_DELAY_MS = 60_000; type Status = z.infer; +class TelegramApiError extends Error { + constructor( + message: string, + readonly code?: number, + ) { + super(message); + this.name = "TelegramApiError"; + } +} + +// 401 = token revoked/invalid, 404 = bot deleted / malformed token. Retrying +// these forever would hammer the API and show a misleading "polling" status. +function isTerminal(error: unknown): boolean { + return error instanceof TelegramApiError && (error.code === 401 || error.code === 404); +} + interface TelegramUpdate { update_id: number; message?: { @@ -24,7 +47,10 @@ interface TelegramUpdate { export interface TelegramTransportOptions { botToken: string; allowFrom: string[]; - onInbound: (senderKey: string, text: string, reply: ReplyFn) => void; + // JSON file holding { offset } across restarts. + stateFile: string; + // chatId is the address to reply to; the caller owns reply routing. + onInbound: (senderKey: string, chatId: string, text: string) => void; onStatus: (status: Status) => void; } @@ -32,6 +58,7 @@ export class TelegramTransport { private abort: AbortController | null = null; private stopped = false; private offset = 0; + private botUsername: string | undefined; constructor(private readonly opts: TelegramTransportOptions) {} @@ -47,40 +74,85 @@ export class TelegramTransport { 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 { - const res = await fetch(this.api(method), { + const res = await fetch(`https://api.telegram.org/bot${this.opts.botToken}/${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 }; + const payload = (await res.json()) as { + ok: boolean; + result?: unknown; + description?: string; + error_code?: number; + }; if (!payload.ok) { - throw new Error(payload.description ?? `Telegram API error (${method})`); + throw new TelegramApiError( + payload.description ?? `Telegram API error (${method})`, + payload.error_code, + ); } return payload.result; } - private async run(): Promise { - // Validate the token up front so a typo surfaces in settings - // immediately instead of as a silent poll loop failure. + private async loadOffset(): Promise { 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; + const raw = await fs.readFile(this.opts.stateFile, "utf8"); + const parsed = JSON.parse(raw) as { offset?: unknown }; + if (typeof parsed.offset === "number" && Number.isFinite(parsed.offset)) { + this.offset = parsed.offset; + } + } catch { + // first run or unreadable state β€” start from 0 + } + } + + private async saveOffset(): Promise { + try { + await fs.mkdir(path.dirname(this.opts.stateFile), { recursive: true }); + await fs.writeFile(this.opts.stateFile, JSON.stringify({ offset: this.offset })); + } catch { + // best effort β€” worst case is one redelivered batch after restart + } + } + + private async sleep(ms: number): Promise { + await new Promise((resolve) => setTimeout(resolve, ms)); + } + + private async run(): Promise { + await this.loadOffset(); + + // Validate the token, retrying transient failures (the app often + // starts at login before the network is up). Only a definitive + // API rejection is terminal. + let delay = RETRY_DELAY_MS; + while (!this.stopped) { + try { + const me = (await this.call("getMe")) as { username?: string }; + if (this.stopped) return; + this.botUsername = me.username; + this.opts.onStatus({ state: "polling", botUsername: me.username }); + break; + } catch (error) { + if (this.stopped) return; + const message = error instanceof Error ? error.message : String(error); + if (isTerminal(error)) { + this.opts.onStatus({ + state: "error", + error: `Bot token rejected (${message}) β€” create a new token with @BotFather.`, + }); + return; + } + this.opts.onStatus({ state: "error", error: message }); + await this.sleep(delay); + delay = Math.min(delay * 2, MAX_RETRY_DELAY_MS); + } } + delay = RETRY_DELAY_MS; + let healthy = true; while (!this.stopped) { this.abort = new AbortController(); try { @@ -94,18 +166,32 @@ export class TelegramTransport { this.abort.signal, )) as TelegramUpdate[]; for (const update of updates) { - this.offset = Math.max(this.offset, update.update_id + 1); + this.offset = update.update_id + 1; this.handleUpdate(update); } + if (updates.length > 0) { + await this.saveOffset(); + } + if (!healthy) { + // Restore the healthy status only after a successful poll. + healthy = true; + this.opts.onStatus({ state: "polling", botUsername: this.botUsername }); + } + delay = RETRY_DELAY_MS; } 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" }); + const message = error instanceof Error ? error.message : String(error); + if (isTerminal(error)) { + this.opts.onStatus({ + state: "error", + error: `Bot token rejected (${message}) β€” create a new token with @BotFather.`, + }); + return; + } + healthy = false; + this.opts.onStatus({ state: "error", error: message }); + await this.sleep(delay); + delay = Math.min(delay * 2, MAX_RETRY_DELAY_MS); } } } @@ -116,7 +202,6 @@ export class TelegramTransport { // 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, @@ -124,10 +209,10 @@ export class TelegramTransport { ).catch(() => undefined); return; } - this.opts.onInbound(`telegram:${chatId}`, message.text, reply); + this.opts.onInbound(`telegram:${chatId}`, chatId, message.text); } - private async send(chatId: string, text: string): Promise { + async send(chatId: string, text: string): Promise { await this.call("sendMessage", { chat_id: chatId, text }); } } diff --git a/apps/x/packages/core/src/channels/transports/whatsapp.ts b/apps/x/packages/core/src/channels/transports/whatsapp.ts index 4e70f412..52852a52 100644 --- a/apps/x/packages/core/src/channels/transports/whatsapp.ts +++ b/apps/x/packages/core/src/channels/transports/whatsapp.ts @@ -6,7 +6,6 @@ import makeWASocket, { 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 @@ -34,7 +33,9 @@ export interface WhatsAppTransportStatus { export interface WhatsAppTransportOptions { authDir: string; allowFrom: string[]; - onInbound: (senderKey: string, text: string, reply: ReplyFn) => void; + // chatJid is the address to reply to; the caller owns reply routing so a + // reply can go through whichever transport instance is current by then. + onInbound: (senderKey: string, chatJid: string, text: string) => void; onStatus: (status: WhatsAppTransportStatus) => void; } @@ -45,7 +46,13 @@ interface TextishMessage { } interface InboundWAMessage { - key?: { remoteJid?: string | null; fromMe?: boolean | null; id?: string | null }; + key?: { + remoteJid?: string | null; + // Phone-number JID when remoteJid is a LID (anonymized) JID. + remoteJidAlt?: string | null; + fromMe?: boolean | null; + id?: string | null; + }; message?: unknown; } @@ -60,6 +67,11 @@ function messageText(message: unknown): string | null { export class WhatsAppTransport { private sock: WASocket | null = null; private stopped = false; + // Bumped on every connect/stop/logout; handlers close over their own + // generation and go inert the moment they are superseded, so a stop() + // racing an await inside connect() cannot leave a zombie socket + // processing messages alongside its replacement. + private generation = 0; private sentIds = new Set(); constructor(private readonly opts: WhatsAppTransportOptions) {} @@ -72,6 +84,7 @@ export class WhatsAppTransport { async stop(): Promise { this.stopped = true; + this.generation++; try { this.sock?.end(undefined); } catch { @@ -85,6 +98,7 @@ export class WhatsAppTransport { // local credentials so the next start shows a fresh QR. async logout(): Promise { this.stopped = true; + this.generation++; try { await this.sock?.logout(); } catch { @@ -97,18 +111,22 @@ export class WhatsAppTransport { private async connect(): Promise { if (this.stopped) return; + const generation = ++this.generation; const { state, saveCreds } = await useMultiFileAuthState(this.opts.authDir); + if (this.stopped || generation !== this.generation) return; const sock = makeWASocket({ auth: state, syncFullHistory: false, markOnlineOnConnect: false, }); this.sock = sock; + const isCurrent = () => + !this.stopped && generation === this.generation && this.sock === sock; sock.ev.on("creds.update", saveCreds); sock.ev.on("connection.update", (update) => { - if (this.stopped) return; + if (!isCurrent()) return; if (update.qr) { this.opts.onStatus({ state: "qr", qr: update.qr }); } @@ -129,6 +147,7 @@ export class WhatsAppTransport { return; } setTimeout(() => { + if (!isCurrent()) return; this.connect().catch((error) => { this.opts.onStatus({ state: "error", @@ -140,7 +159,7 @@ export class WhatsAppTransport { }); sock.ev.on("messages.upsert", ({ messages, type }) => { - if (type !== "notify") return; + if (!isCurrent() || type !== "notify") return; for (const msg of messages) { this.handleMessage(sock, msg); } @@ -155,24 +174,37 @@ export class WhatsAppTransport { 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), + // LID-addressed chats put the anonymized id in remoteJid and (when + // the server supplies it) the real phone-number JID in remoteJidAlt. + // Identity checks must consider both. + const altJid: string | undefined = msg.key?.remoteJidAlt ?? undefined; + const chatJids = altJid ? [jid, altJid] : [jid]; + const user = sock.user as { id?: string; lid?: string } | undefined; + const selfIds = [user?.id, user?.lid].filter((v): v is string => Boolean(v)); + const isSelfChat = chatJids.some((j) => + selfIds.some((selfId) => areJidsSameUser(j, selfId)), ); - const isSelfChat = selfIds.some((selfId) => areJidsSameUser(jid, selfId)); - const senderNumber = jidDecode(jid)?.user ?? ""; + const senderNumbers = chatJids.flatMap((j) => { + const decoded = jidDecode(j)?.user; + return decoded ? [decoded] : []; + }); // 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; + if (!senderNumbers.some((n) => this.opts.allowFrom.includes(n))) return; } - const reply: ReplyFn = (replyText) => this.send(jid, replyText); - this.opts.onInbound(`whatsapp:${senderNumber || jid}`, text, reply); + // Prefer the phone number (altJid decodes to it when present) as the + // stable sender identity. + const senderId = altJid + ? (jidDecode(altJid)?.user ?? senderNumbers[0] ?? jid) + : (senderNumbers[0] ?? jid); + this.opts.onInbound(`whatsapp:${senderId}`, jid, text); } - private async send(jid: string, text: string): Promise { + async send(jid: string, text: string): Promise { const sock = this.sock; if (!sock) throw new Error("WhatsApp is not connected"); const sent = await sock.sendMessage(jid, { text: `${REPLY_MARKER}${text}` }); @@ -180,11 +212,7 @@ export class WhatsAppTransport { 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; - } + this.sentIds = new Set(Array.from(this.sentIds).slice(-250)); } } }