mirror of
https://github.com/rowboatlabs/rowboat.git
synced 2026-07-12 21:02:17 +02:00
Merge pull request #662 from rowboatlabs/mobile
add whatsapp and telegram support
This commit is contained in:
arkml
GitHub
commit
7b25cfe33f
16 changed files with 2849 additions and 9 deletions
|
|
@ -51,6 +51,8 @@ import type { CodeSession } from '@x/shared/dist/code-sessions.js';
|
|||
import { invalidateCopilotInstructionsCache } from '@x/core/dist/application/assistant/instructions.js';
|
||||
import { triggerSync as triggerGranolaSync } from '@x/core/dist/knowledge/granola/sync.js';
|
||||
import { ISlackConfigRepo } from '@x/core/dist/slack/repo.js';
|
||||
import { IChannelsConfigRepo } from '@x/core/dist/channels/repo.js';
|
||||
import { applyChannelsConfig, getChannelsStatus, logoutWhatsApp, subscribeChannelsStatus } from '@x/core/dist/channels/service.js';
|
||||
import { runAgentSlack, getAgentSlackCliStatus, AgentSlackRunError } from '@x/core/dist/slack/agent-slack-exec.js';
|
||||
import { knowledgeSourcesRepo } from '@x/core/dist/knowledge/sources/repo.js';
|
||||
import { rankSlackHomeMessages } from '@x/core/dist/knowledge/sources/rank_slack_home.js';
|
||||
|
|
@ -646,6 +648,20 @@ function emitSessionEvent(event: SessionBusEvent): void {
|
|||
}
|
||||
}
|
||||
|
||||
// Mobile channels: status changes (QR pairing, connect/disconnect) → renderer.
|
||||
let channelsWatcher: (() => void) | null = null;
|
||||
export function startChannelsWatcher(): void {
|
||||
if (channelsWatcher) return;
|
||||
channelsWatcher = subscribeChannelsStatus((status) => {
|
||||
const windows = BrowserWindow.getAllWindows();
|
||||
for (const win of windows) {
|
||||
if (!win.isDestroyed() && win.webContents) {
|
||||
win.webContents.send('channels:status', status);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
let sessionsWatcher: (() => void) | null = null;
|
||||
export function startSessionsWatcher(): void {
|
||||
if (sessionsWatcher) {
|
||||
|
|
@ -1208,6 +1224,22 @@ export function setupIpcHandlers() {
|
|||
|
||||
return { success: true };
|
||||
},
|
||||
// ── Mobile channels (WhatsApp / Telegram bridge) ─────────────
|
||||
'channels:getConfig': async () => {
|
||||
return container.resolve<IChannelsConfigRepo>('channelsConfigRepo').getConfig();
|
||||
},
|
||||
'channels:setConfig': async (_event, args) => {
|
||||
await container.resolve<IChannelsConfigRepo>('channelsConfigRepo').setConfig(args);
|
||||
await applyChannelsConfig(args);
|
||||
return { success: true };
|
||||
},
|
||||
'channels:getStatus': async () => {
|
||||
return getChannelsStatus();
|
||||
},
|
||||
'channels:whatsappLogout': async () => {
|
||||
await logoutWhatsApp();
|
||||
return { success: true };
|
||||
},
|
||||
'slack:getConfig': async () => {
|
||||
const repo = container.resolve<ISlackConfigRepo>('slackConfigRepo');
|
||||
const config = await repo.getConfig();
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ import path from "node:path";
|
|||
import {
|
||||
setupIpcHandlers,
|
||||
startRunsWatcher, startSessionsWatcher,
|
||||
startChannelsWatcher,
|
||||
startCodeSessionStatusWatcher,
|
||||
startServicesWatcher,
|
||||
startLiveNoteAgentWatcher,
|
||||
|
|
@ -25,6 +26,7 @@ import { init as initEmailLabeling } from "@x/core/dist/knowledge/label_emails.j
|
|||
import { init as initNoteTagging } from "@x/core/dist/knowledge/tag_notes.js";
|
||||
import { init as initInlineTasks } from "@x/core/dist/knowledge/inline_tasks.js";
|
||||
import { init as initAgentRunner } from "@x/core/dist/agent-schedule/runner.js";
|
||||
import { init as initChannels } from "@x/core/dist/channels/service.js";
|
||||
import { init as initAgentNotes } from "@x/core/dist/knowledge/agent_notes.js";
|
||||
import { init as initCalendarNotifications } from "@x/core/dist/knowledge/notify_calendar_meetings.js";
|
||||
import { init as initMeetingPrep } from "@x/core/dist/knowledge/meeting_prep_scheduler.js";
|
||||
|
|
@ -414,6 +416,13 @@ app.whenReady().then(async () => {
|
|||
await container.resolve<ISessions>('sessions').initialize();
|
||||
startSessionsWatcher();
|
||||
|
||||
// Mobile channels (WhatsApp/Telegram bridge): needs the session index, so
|
||||
// start after initialize(). Failures must never block boot.
|
||||
startChannelsWatcher();
|
||||
initChannels().catch((error) => {
|
||||
console.error('[Channels] Failed to start mobile channels:', error);
|
||||
});
|
||||
|
||||
// start code-session status tracker (derives working/needs-you/idle + notifications)
|
||||
startCodeSessionStatusWatcher();
|
||||
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@
|
|||
|
||||
import * as React from "react"
|
||||
import { useState, useEffect, useCallback, useMemo } from "react"
|
||||
import { Server, Key, Shield, Palette, Monitor, Sun, Moon, Loader2, CheckCircle2, Plus, X, Wrench, Search, ChevronRight, Link2, Tags, Mail, BookOpen, User, Plug, HelpCircle, MessageCircle, Bug, Terminal, AlertTriangle, RefreshCw, PanelRight, Bell } from "lucide-react"
|
||||
import { Server, Key, Shield, Palette, Monitor, Sun, Moon, Loader2, CheckCircle2, Plus, X, Wrench, Search, ChevronRight, Link2, Tags, Mail, BookOpen, User, Plug, HelpCircle, MessageCircle, Bug, Terminal, AlertTriangle, RefreshCw, PanelRight, Bell, Smartphone } from "lucide-react"
|
||||
|
||||
import {
|
||||
Dialog,
|
||||
|
|
@ -25,10 +25,11 @@ import { useTheme } from "@/contexts/theme-context"
|
|||
import { toast } from "sonner"
|
||||
import { AccountSettings } from "@/components/settings/account-settings"
|
||||
import { ConnectedAccountsSettings } from "@/components/settings/connected-accounts-settings"
|
||||
import { MobileChannelsSettings } from "@/components/settings/mobile-channels-settings"
|
||||
import type { ApprovalPolicy } from "@x/shared/src/code-mode.js"
|
||||
import { startProvisioning, useProvisioning, enabledOptimistic, type AgentStatus, type CodeModeAgentStatus } from "@/lib/code-mode-provisioning"
|
||||
|
||||
type ConfigTab = "account" | "connections" | "models" | "mcp" | "security" | "code-mode" | "appearance" | "notifications" | "note-tagging" | "help"
|
||||
type ConfigTab = "account" | "connections" | "mobile" | "models" | "mcp" | "security" | "code-mode" | "appearance" | "notifications" | "note-tagging" | "help"
|
||||
|
||||
interface TabConfig {
|
||||
id: ConfigTab
|
||||
|
|
@ -51,6 +52,12 @@ const tabs: TabConfig[] = [
|
|||
icon: Plug,
|
||||
description: "Manage accounts and tools",
|
||||
},
|
||||
{
|
||||
id: "mobile",
|
||||
label: "Mobile",
|
||||
icon: Smartphone,
|
||||
description: "Chat with Rowboat from WhatsApp or Telegram",
|
||||
},
|
||||
{
|
||||
id: "models",
|
||||
label: "Models",
|
||||
|
|
@ -2216,7 +2223,7 @@ export function SettingsDialog({ children, defaultTab = "account", open: control
|
|||
</div>
|
||||
|
||||
{/* Content */}
|
||||
<div className={cn("flex-1 p-4 min-h-0", (activeTab === "models" || activeTab === "connections" || activeTab === "account" || activeTab === "code-mode" || activeTab === "notifications") ? "overflow-y-auto" : activeTab === "note-tagging" ? "overflow-hidden flex flex-col" : "overflow-hidden")}>
|
||||
<div className={cn("flex-1 p-4 min-h-0", (activeTab === "models" || activeTab === "connections" || activeTab === "mobile" || activeTab === "account" || activeTab === "code-mode" || activeTab === "notifications") ? "overflow-y-auto" : activeTab === "note-tagging" ? "overflow-hidden flex flex-col" : "overflow-hidden")}>
|
||||
{activeTab === "account" ? (
|
||||
<AccountSettings dialogOpen={open} />
|
||||
) : activeTab === "connections" ? (
|
||||
|
|
@ -2231,6 +2238,8 @@ export function SettingsDialog({ children, defaultTab = "account", open: control
|
|||
<ToolsLibrarySettings dialogOpen={open} rowboatConnected={rowboatConnected} />
|
||||
</div>
|
||||
</div>
|
||||
) : activeTab === "mobile" ? (
|
||||
<MobileChannelsSettings dialogOpen={open} />
|
||||
) : activeTab === "models" ? (
|
||||
rowboatConnected
|
||||
? <RowboatModelSettings dialogOpen={open} />
|
||||
|
|
|
|||
|
|
@ -0,0 +1,287 @@
|
|||
"use client"
|
||||
|
||||
import { useCallback, useEffect, useState } from "react"
|
||||
import type { z } from "zod"
|
||||
import { Loader2, MessageCircle, Send, Smartphone } from "lucide-react"
|
||||
import { Button } from "@/components/ui/button"
|
||||
import { Input } from "@/components/ui/input"
|
||||
import { Separator } from "@/components/ui/separator"
|
||||
import { Switch } from "@/components/ui/switch"
|
||||
import { toast } from "sonner"
|
||||
import type { ChannelsConfig, ChannelsStatus } from "@x/shared/src/channels.js"
|
||||
|
||||
type Config = z.infer<typeof ChannelsConfig>
|
||||
type Status = z.infer<typeof ChannelsStatus>
|
||||
|
||||
// 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(/[,;\n]+/)
|
||||
.map((s) => s.replace(/\D/g, ""))
|
||||
.filter(Boolean)
|
||||
}
|
||||
|
||||
export function MobileChannelsSettings({ dialogOpen }: { dialogOpen: boolean }) {
|
||||
const [config, setConfig] = useState<Config | null>(null)
|
||||
const [status, setStatus] = useState<Status | null>(null)
|
||||
const [tokenDraft, setTokenDraft] = useState("")
|
||||
const [waAllowDraft, setWaAllowDraft] = useState("")
|
||||
const [tgAllowDraft, setTgAllowDraft] = useState("")
|
||||
const [saving, setSaving] = useState(false)
|
||||
|
||||
useEffect(() => {
|
||||
if (!dialogOpen) return
|
||||
let cancelled = false
|
||||
void (async () => {
|
||||
try {
|
||||
const [cfg, st] = await Promise.all([
|
||||
window.ipc.invoke("channels:getConfig", null),
|
||||
window.ipc.invoke("channels:getStatus", null),
|
||||
])
|
||||
if (cancelled) return
|
||||
setConfig(cfg)
|
||||
setStatus(st)
|
||||
setTokenDraft(cfg.telegram.botToken)
|
||||
setWaAllowDraft(cfg.whatsapp.allowFrom.join(", "))
|
||||
setTgAllowDraft(cfg.telegram.allowFrom.join(", "))
|
||||
} catch {
|
||||
if (!cancelled) toast.error("Failed to load mobile channel settings")
|
||||
}
|
||||
})()
|
||||
const unsubscribe = window.ipc.on("channels:status", (st) => {
|
||||
setStatus(st)
|
||||
})
|
||||
return () => {
|
||||
cancelled = true
|
||||
unsubscribe()
|
||||
}
|
||||
}, [dialogOpen])
|
||||
|
||||
const save = useCallback(async (next: Config) => {
|
||||
setConfig(next)
|
||||
setSaving(true)
|
||||
try {
|
||||
await window.ipc.invoke("channels:setConfig", next)
|
||||
} catch (error) {
|
||||
toast.error(error instanceof Error ? error.message : "Failed to save channel settings")
|
||||
} finally {
|
||||
setSaving(false)
|
||||
}
|
||||
}, [])
|
||||
|
||||
if (!config) {
|
||||
return (
|
||||
<div className="flex items-center justify-center py-8 text-muted-foreground">
|
||||
<Loader2 className="size-4 animate-spin" />
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
const wa = status?.whatsapp
|
||||
const tg = status?.telegram
|
||||
|
||||
return (
|
||||
<div className="space-y-6">
|
||||
<div className="flex items-start gap-2.5 rounded-md bg-muted/50 px-3 py-2.5">
|
||||
<Smartphone className="size-4 mt-0.5 shrink-0 text-muted-foreground" />
|
||||
<p className="text-xs text-muted-foreground">
|
||||
Chat with Rowboat from your phone. Send <span className="font-mono">help</span> for
|
||||
commands (<span className="font-mono">list</span>, <span className="font-mono">resume 2</span>,{" "}
|
||||
<span className="font-mono">new</span>, <span className="font-mono">stop</span>) — anything
|
||||
else is a message to the current chat. Your computer must be on with Rowboat running.
|
||||
</p>
|
||||
</div>
|
||||
|
||||
{/* WhatsApp */}
|
||||
<div className="space-y-3">
|
||||
<div className="flex items-center justify-between">
|
||||
<div className="flex items-center gap-2.5">
|
||||
<div className="flex size-8 items-center justify-center rounded-md bg-muted">
|
||||
<MessageCircle className="size-4" />
|
||||
</div>
|
||||
<div className="flex flex-col">
|
||||
<span className="text-sm font-medium">WhatsApp</span>
|
||||
<span className="text-xs text-muted-foreground">
|
||||
{wa?.state === "connected"
|
||||
? `Linked as +${wa.self ?? "?"} — message yourself to use it`
|
||||
: wa?.state === "qr"
|
||||
? "Scan the QR below with your phone"
|
||||
: wa?.state === "starting"
|
||||
? "Connecting…"
|
||||
: wa?.state === "error"
|
||||
? wa.error ?? "Error"
|
||||
: "Links your own WhatsApp as a companion device"}
|
||||
</span>
|
||||
</div>
|
||||
</div>
|
||||
<div className="flex items-center gap-2">
|
||||
{wa?.state === "connected" && (
|
||||
<Button
|
||||
variant="outline"
|
||||
size="sm"
|
||||
className="h-7 px-3 text-xs"
|
||||
onClick={() => {
|
||||
void window.ipc.invoke("channels:whatsappLogout", null).catch(() => {
|
||||
toast.error("Failed to unlink WhatsApp")
|
||||
})
|
||||
}}
|
||||
>
|
||||
Unlink
|
||||
</Button>
|
||||
)}
|
||||
<Switch
|
||||
checked={config.whatsapp.enabled}
|
||||
disabled={saving}
|
||||
onCheckedChange={(enabled) =>
|
||||
void save({ ...config, whatsapp: { ...config.whatsapp, enabled } })
|
||||
}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{config.whatsapp.enabled && wa?.state === "qr" && wa.qrDataUrl && (
|
||||
<div className="flex items-center gap-4 rounded-md border p-3">
|
||||
<img src={wa.qrDataUrl} alt="WhatsApp pairing QR" className="size-40 rounded" />
|
||||
<div className="text-xs text-muted-foreground space-y-1">
|
||||
<p className="font-medium text-foreground">Link Rowboat to WhatsApp</p>
|
||||
<p>1. Open WhatsApp on your phone</p>
|
||||
<p>2. Settings → Linked Devices → Link a Device</p>
|
||||
<p>3. Scan this code</p>
|
||||
<p className="pt-1">
|
||||
Then message <span className="font-medium">yourself</span> (your own contact) to talk
|
||||
to Rowboat.
|
||||
</p>
|
||||
</div>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{config.whatsapp.enabled && (
|
||||
<div className="space-y-1.5">
|
||||
<label className="text-xs font-medium">Additional allowed numbers</label>
|
||||
<div className="flex gap-2">
|
||||
<Input
|
||||
value={waAllowDraft}
|
||||
onChange={(e) => setWaAllowDraft(e.target.value)}
|
||||
placeholder="e.g. 14155551234, 919876543210"
|
||||
className="h-8 text-xs"
|
||||
/>
|
||||
<Button
|
||||
variant="outline"
|
||||
size="sm"
|
||||
className="h-8 px-3 text-xs"
|
||||
disabled={saving}
|
||||
onClick={() =>
|
||||
void save({
|
||||
...config,
|
||||
whatsapp: { ...config.whatsapp, allowFrom: parseIdList(waAllowDraft) },
|
||||
})
|
||||
}
|
||||
>
|
||||
Save
|
||||
</Button>
|
||||
</div>
|
||||
<p className="text-xs text-muted-foreground">
|
||||
Your own number (self-chat) is always allowed. Digits only, with country code.
|
||||
</p>
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
|
||||
<Separator />
|
||||
|
||||
{/* Telegram */}
|
||||
<div className="space-y-3">
|
||||
<div className="flex items-center justify-between">
|
||||
<div className="flex items-center gap-2.5">
|
||||
<div className="flex size-8 items-center justify-center rounded-md bg-muted">
|
||||
<Send className="size-4" />
|
||||
</div>
|
||||
<div className="flex flex-col">
|
||||
<span className="text-sm font-medium">Telegram</span>
|
||||
<span className="text-xs text-muted-foreground">
|
||||
{tg?.state === "polling"
|
||||
? `Listening${tg.botUsername ? ` as @${tg.botUsername}` : ""}`
|
||||
: tg?.state === "starting"
|
||||
? "Connecting…"
|
||||
: tg?.state === "error"
|
||||
? tg.error ?? "Error"
|
||||
: "Uses your own bot — create one with @BotFather"}
|
||||
</span>
|
||||
</div>
|
||||
</div>
|
||||
<Switch
|
||||
checked={config.telegram.enabled}
|
||||
disabled={saving}
|
||||
onCheckedChange={(enabled) =>
|
||||
void save({ ...config, telegram: { ...config.telegram, enabled } })
|
||||
}
|
||||
/>
|
||||
</div>
|
||||
|
||||
{config.telegram.enabled && (
|
||||
<>
|
||||
<div className="space-y-1.5">
|
||||
<label className="text-xs font-medium">Bot token</label>
|
||||
<div className="flex gap-2">
|
||||
<Input
|
||||
type="password"
|
||||
value={tokenDraft}
|
||||
onChange={(e) => setTokenDraft(e.target.value)}
|
||||
placeholder="123456789:AAF…"
|
||||
className="h-8 text-xs font-mono"
|
||||
/>
|
||||
<Button
|
||||
variant="outline"
|
||||
size="sm"
|
||||
className="h-8 px-3 text-xs"
|
||||
disabled={saving}
|
||||
onClick={() =>
|
||||
void save({
|
||||
...config,
|
||||
telegram: { ...config.telegram, botToken: tokenDraft.trim() },
|
||||
})
|
||||
}
|
||||
>
|
||||
Save
|
||||
</Button>
|
||||
</div>
|
||||
<p className="text-xs text-muted-foreground">
|
||||
Message @BotFather on Telegram → /newbot → paste the token here.
|
||||
</p>
|
||||
</div>
|
||||
<div className="space-y-1.5">
|
||||
<label className="text-xs font-medium">Allowed chat IDs</label>
|
||||
<div className="flex gap-2">
|
||||
<Input
|
||||
value={tgAllowDraft}
|
||||
onChange={(e) => setTgAllowDraft(e.target.value)}
|
||||
placeholder="e.g. 123456789"
|
||||
className="h-8 text-xs"
|
||||
/>
|
||||
<Button
|
||||
variant="outline"
|
||||
size="sm"
|
||||
className="h-8 px-3 text-xs"
|
||||
disabled={saving}
|
||||
onClick={() =>
|
||||
void save({
|
||||
...config,
|
||||
telegram: { ...config.telegram, allowFrom: parseIdList(tgAllowDraft) },
|
||||
})
|
||||
}
|
||||
>
|
||||
Save
|
||||
</Button>
|
||||
</div>
|
||||
<p className="text-xs text-muted-foreground">
|
||||
Message your bot once — it replies with your chat ID to add here.
|
||||
</p>
|
||||
</div>
|
||||
</>
|
||||
)}
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
|
@ -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:"
|
||||
}
|
||||
}
|
||||
|
|
|
|||
278
apps/x/packages/core/src/channels/bridge.test.ts
Normal file
278
apps/x/packages/core/src/channels/bridge.test.ts
Normal file
|
|
@ -0,0 +1,278 @@
|
|||
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, type ModelChoice } from "./bridge.js";
|
||||
|
||||
const SENDER = "test:1";
|
||||
|
||||
function entry(overrides: Partial<SessionIndexEntry>): 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<void>;
|
||||
sessions: {
|
||||
createSession: ReturnType<typeof vi.fn>;
|
||||
sendMessage: ReturnType<typeof vi.fn>;
|
||||
respondToAskHuman: ReturnType<typeof vi.fn>;
|
||||
stopTurn: ReturnType<typeof vi.fn>;
|
||||
getTurn: ReturnType<typeof vi.fn>;
|
||||
listSessions: ReturnType<typeof vi.fn>;
|
||||
};
|
||||
listModels: ReturnType<typeof vi.fn>;
|
||||
publish: (turnId: string, event: TurnStreamEvent) => void;
|
||||
}
|
||||
|
||||
const MODELS: ModelChoice[] = [
|
||||
{ provider: "anthropic", model: "claude-fable-5", label: "Fable 5 — Anthropic" },
|
||||
{ provider: "anthropic", model: "claude-sonnet-5", label: "Sonnet 5 — Anthropic" },
|
||||
{ provider: "openai", model: "gpt-5", label: "GPT-5 — OpenAI" },
|
||||
];
|
||||
|
||||
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 listModels = vi.fn(async () => MODELS);
|
||||
const bridge = new ChannelBridge({
|
||||
sessions: sessions as unknown as ISessions,
|
||||
sessionBus: bus,
|
||||
listModels,
|
||||
});
|
||||
const replies: string[] = [];
|
||||
const reply = async (text: string) => {
|
||||
replies.push(text);
|
||||
};
|
||||
return { bridge, bus, replies, reply, sessions, listModels, 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 model selection", () => {
|
||||
it("lists models and applies a by-index selection to later turns", async () => {
|
||||
const h = harness();
|
||||
await h.bridge.handleInbound(SENDER, "model", h.reply);
|
||||
expect(h.replies[0]).toContain("1. Fable 5 — Anthropic");
|
||||
expect(h.replies[0]).toContain("app default");
|
||||
|
||||
await h.bridge.handleInbound(SENDER, "model 3", h.reply);
|
||||
expect(h.replies[1]).toContain("GPT-5");
|
||||
|
||||
settleOnSend(h, completedEvent("t1", "done"));
|
||||
await h.bridge.handleInbound(SENDER, "hello", h.reply);
|
||||
expect(h.sessions.sendMessage).toHaveBeenCalledWith(
|
||||
"s1",
|
||||
expect.anything(),
|
||||
expect.objectContaining({
|
||||
agent: {
|
||||
agentId: "copilot",
|
||||
overrides: { model: { provider: "openai", model: "gpt-5" } },
|
||||
},
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("selects by unique name match and resets on 'model default'", async () => {
|
||||
const h = harness();
|
||||
await h.bridge.handleInbound(SENDER, "model fable", h.reply);
|
||||
expect(h.replies[0]).toContain("Fable 5");
|
||||
|
||||
settleOnSend(h, completedEvent("t1", "done"));
|
||||
await h.bridge.handleInbound(SENDER, "hi", h.reply);
|
||||
expect(h.sessions.sendMessage).toHaveBeenCalledWith(
|
||||
"s1",
|
||||
expect.anything(),
|
||||
expect.objectContaining({
|
||||
agent: expect.objectContaining({
|
||||
overrides: { model: { provider: "anthropic", model: "claude-fable-5" } },
|
||||
}),
|
||||
}),
|
||||
);
|
||||
|
||||
await h.bridge.handleInbound(SENDER, "model default", h.reply);
|
||||
settleOnSend(h, completedEvent("t2", "done"), "t2");
|
||||
await h.bridge.handleInbound(SENDER, "hi again", h.reply);
|
||||
expect(h.sessions.sendMessage).toHaveBeenLastCalledWith(
|
||||
"s1",
|
||||
expect.anything(),
|
||||
expect.objectContaining({ agent: { agentId: "copilot" } }),
|
||||
);
|
||||
});
|
||||
|
||||
it("asks for disambiguation on an ambiguous name and rejects unknown ones", async () => {
|
||||
const h = harness();
|
||||
await h.bridge.handleInbound(SENDER, "model claude", h.reply);
|
||||
expect(h.replies[0]).toContain("matches 2 models");
|
||||
|
||||
await h.bridge.handleInbound(SENDER, "model llama", h.reply);
|
||||
expect(h.replies[1]).toContain('No model matching "llama"');
|
||||
});
|
||||
});
|
||||
|
||||
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");
|
||||
});
|
||||
});
|
||||
540
apps/x/packages/core/src/channels/bridge.ts
Normal file
540
apps/x/packages/core/src/channels/bridge.ts
Normal file
|
|
@ -0,0 +1,540 @@
|
|||
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";
|
||||
|
||||
// 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 ASK_HUMAN_TOOL_ID = `builtin:${ASK_HUMAN_TOOL}`;
|
||||
|
||||
const HELP_TEXT = [
|
||||
"🤖 Rowboat commands:",
|
||||
"• list — recent chats",
|
||||
"• resume N — continue chat N from the list",
|
||||
"• new [message] — start a fresh chat",
|
||||
"• model [N or name] — pick the model (\"model default\" resets)",
|
||||
"• status — current chat and what it's doing",
|
||||
"• stop — cancel the running task",
|
||||
"",
|
||||
"Anything else is sent to the current chat.",
|
||||
].join("\n");
|
||||
|
||||
const MODEL_LIST_LIMIT = 20;
|
||||
|
||||
export type ReplyFn = (text: string) => Promise<void>;
|
||||
|
||||
export interface ModelChoice {
|
||||
provider: string;
|
||||
model: string;
|
||||
label: string;
|
||||
}
|
||||
|
||||
interface SenderState {
|
||||
activeSessionId: string | null;
|
||||
// sessionIds as last shown by `list` (1-based indexing for `resume N`).
|
||||
lastList: string[];
|
||||
// Choices as last shown by `model` (1-based indexing for `model N`).
|
||||
lastModels: ModelChoice[];
|
||||
// Per-sender override passed on every turn; null = app default model.
|
||||
model: { provider: string; model: string } | null;
|
||||
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 === ASK_HUMAN_TOOL_ID || t.toolName === ASK_HUMAN_TOOL,
|
||||
);
|
||||
if (ask) {
|
||||
const input = ask.input as { question?: unknown; options?: unknown } | null;
|
||||
const 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")
|
||||
: 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;
|
||||
}
|
||||
|
||||
interface TurnWatcher {
|
||||
waitFor(turnId: string, timeoutMs: number): Promise<Settled>;
|
||||
dispose(): void;
|
||||
}
|
||||
|
||||
export class ChannelBridge {
|
||||
private senders = new Map<string, SenderState>();
|
||||
|
||||
constructor(
|
||||
private readonly deps: {
|
||||
sessions: ISessions;
|
||||
sessionBus: EmitterSessionBus;
|
||||
listModels: () => Promise<ModelChoice[]>;
|
||||
},
|
||||
) {}
|
||||
|
||||
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 === "model" || lower === "models") {
|
||||
await reply(await this.renderModelList(state));
|
||||
return;
|
||||
}
|
||||
const model = /^model\s+(.+)$/i.exec(trimmed);
|
||||
if (model) {
|
||||
await reply(await this.selectModel(state, model[1].trim()));
|
||||
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: [],
|
||||
lastModels: [],
|
||||
model: null,
|
||||
pendingAsk: null,
|
||||
busy: false,
|
||||
};
|
||||
this.senders.set(senderKey, state);
|
||||
}
|
||||
return state;
|
||||
}
|
||||
|
||||
private isCurrentModel(state: SenderState, choice: ModelChoice): boolean {
|
||||
return (
|
||||
state.model?.provider === choice.provider && state.model?.model === choice.model
|
||||
);
|
||||
}
|
||||
|
||||
private async renderModelList(state: SenderState): Promise<string> {
|
||||
const choices = await this.deps.listModels();
|
||||
if (choices.length === 0) {
|
||||
return "No models available — configure one in Rowboat → Settings → Models.";
|
||||
}
|
||||
state.lastModels = choices;
|
||||
const shown = choices.slice(0, MODEL_LIST_LIMIT);
|
||||
const lines = shown.map((c, i) => {
|
||||
const current = this.isCurrentModel(state, c) ? " ← current" : "";
|
||||
return `${i + 1}. ${c.label}${current}`;
|
||||
});
|
||||
if (choices.length > shown.length) {
|
||||
lines.push(`… and ${choices.length - shown.length} more — pick by name.`);
|
||||
}
|
||||
return [
|
||||
`Models${state.model ? "" : " (using app default)"}:`,
|
||||
...lines,
|
||||
"",
|
||||
`Reply "model N" or "model <name>" to switch, "model default" to reset.`,
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
private async selectModel(state: SenderState, arg: string): Promise<string> {
|
||||
const lower = arg.toLowerCase();
|
||||
if (lower === "default" || lower === "reset") {
|
||||
state.model = null;
|
||||
return "✅ Using the app default model.";
|
||||
}
|
||||
if (state.lastModels.length === 0) {
|
||||
state.lastModels = await this.deps.listModels();
|
||||
}
|
||||
let choice: ModelChoice | undefined;
|
||||
if (/^\d+$/.test(lower)) {
|
||||
choice = state.lastModels[Number(lower) - 1];
|
||||
if (!choice) {
|
||||
return `No model #${lower} — send "model" to see the list.`;
|
||||
}
|
||||
} else {
|
||||
const matches = state.lastModels.filter(
|
||||
(c) =>
|
||||
c.label.toLowerCase().includes(lower) ||
|
||||
c.model.toLowerCase().includes(lower),
|
||||
);
|
||||
if (matches.length === 0) {
|
||||
return `No model matching "${arg}" — send "model" to see the list.`;
|
||||
}
|
||||
if (matches.length > 1) {
|
||||
const preview = matches.slice(0, 5).map((c) => `• ${c.label}`);
|
||||
return [`"${arg}" matches ${matches.length} models:`, ...preview, "", "Be more specific."].join("\n");
|
||||
}
|
||||
choice = matches[0];
|
||||
}
|
||||
state.model = { provider: choice.provider, model: choice.model };
|
||||
return `✅ Model set to ${choice.label} for your chats from here.`;
|
||||
}
|
||||
|
||||
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))
|
||||
.slice(0, LIST_LIMIT);
|
||||
}
|
||||
|
||||
private renderList(state: SenderState): string {
|
||||
const entries = this.recentSessions();
|
||||
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.recentSessions().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.sessionEntry(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.sessionEntry(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.sessionEntry(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;
|
||||
try {
|
||||
await reply("⏳ Working on it…");
|
||||
if (state.pendingAsk) {
|
||||
const ask = state.pendingAsk;
|
||||
state.pendingAsk = null;
|
||||
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.
|
||||
}
|
||||
await this.sendToSession(state, text, 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 {
|
||||
state.busy = false;
|
||||
}
|
||||
}
|
||||
|
||||
private async sendToSession(state: SenderState, text: string, reply: ReplyFn): Promise<void> {
|
||||
// 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,
|
||||
...(state.model ? { overrides: { model: state.model } } : {}),
|
||||
},
|
||||
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<boolean> {
|
||||
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,
|
||||
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 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) waiter.resolve(settled);
|
||||
return;
|
||||
}
|
||||
buffered.push({ turnId: event.turnId, settled });
|
||||
});
|
||||
return {
|
||||
waitFor: (turnId: string, timeoutMs: number): Promise<Settled> =>
|
||||
new Promise<Settled>((resolve) => {
|
||||
const hit = buffered.find((b) => b.turnId === turnId);
|
||||
if (hit) {
|
||||
resolve(hit.settled);
|
||||
return;
|
||||
}
|
||||
buffered.length = 0;
|
||||
const timer = setTimeout(() => resolve({ kind: "timeout" }), timeoutMs);
|
||||
cancelTimer = () => clearTimeout(timer);
|
||||
waiter = {
|
||||
turnId,
|
||||
resolve: (settled) => {
|
||||
clearTimeout(timer);
|
||||
resolve(settled);
|
||||
},
|
||||
};
|
||||
}),
|
||||
dispose: () => {
|
||||
unsubscribe();
|
||||
cancelTimer?.();
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
40
apps/x/packages/core/src/channels/repo.ts
Normal file
40
apps/x/packages/core/src/channels/repo.ts
Normal file
|
|
@ -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));
|
||||
}
|
||||
}
|
||||
251
apps/x/packages/core/src/channels/service.ts
Normal file
251
apps/x/packages/core/src/channels/service.ts
Normal file
|
|
@ -0,0 +1,251 @@
|
|||
import path from "node:path";
|
||||
import fs from "node:fs/promises";
|
||||
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 { isSignedIn } from "../account/account.js";
|
||||
import { listGatewayModels } from "../models/gateway.js";
|
||||
import { listOnboardingModels } from "../models/models-dev.js";
|
||||
import { ChannelBridge, type ModelChoice } from "./bridge.js";
|
||||
import type { IChannelsConfigRepo } from "./repo.js";
|
||||
import { TelegramTransport } from "./transports/telegram.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
|
||||
// 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");
|
||||
const TELEGRAM_STATE_FILE = path.join(WorkDir, "channels", "telegram-state.json");
|
||||
|
||||
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. 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<void> = Promise.resolve();
|
||||
|
||||
function enqueue(step: () => Promise<void>): Promise<void> {
|
||||
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(snapshot);
|
||||
} 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);
|
||||
}
|
||||
|
||||
// Same catalog the desktop model picker uses (models:list IPC).
|
||||
async function listBridgeModels(): Promise<ModelChoice[]> {
|
||||
const catalog = (await isSignedIn())
|
||||
? await listGatewayModels()
|
||||
: await listOnboardingModels();
|
||||
return catalog.providers.flatMap((provider) =>
|
||||
provider.models.map((m) => ({
|
||||
provider: provider.id,
|
||||
model: m.id,
|
||||
label: `${m.name ?? m.id} — ${provider.name}`,
|
||||
})),
|
||||
);
|
||||
}
|
||||
|
||||
function ensureBridge(): ChannelBridge {
|
||||
if (!bridge) {
|
||||
bridge = new ChannelBridge({
|
||||
sessions: container.resolve<ISessions>("sessions"),
|
||||
sessionBus: container.resolve<EmitterSessionBus>("sessionBus"),
|
||||
listModels: listBridgeModels,
|
||||
});
|
||||
}
|
||||
return bridge;
|
||||
}
|
||||
|
||||
async function stopWhatsApp(): Promise<void> {
|
||||
if (!whatsapp) return;
|
||||
const stopping = whatsapp;
|
||||
whatsapp = null;
|
||||
await stopping.stop().catch(() => undefined);
|
||||
setWhatsAppStatus({ state: "disabled" });
|
||||
}
|
||||
|
||||
function stopTelegram(): void {
|
||||
if (!telegram) return;
|
||||
const stopping = telegram;
|
||||
telegram = null;
|
||||
stopping.stop();
|
||||
setTelegramStatus({ state: "disabled" });
|
||||
}
|
||||
|
||||
// Invalidates pending async QR renders whenever a newer status lands.
|
||||
let qrSeq = 0;
|
||||
|
||||
async function startWhatsApp(config: Config["whatsapp"]): Promise<void> {
|
||||
if (!config.enabled) {
|
||||
setWhatsAppStatus({ state: "disabled" });
|
||||
return;
|
||||
}
|
||||
const channelBridge = ensureBridge();
|
||||
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, 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 <img>; the raw pairing string never leaves core.
|
||||
QRCode.toDataURL(update.qr, { margin: 1, width: 256 })
|
||||
.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 } : {}),
|
||||
...(update.error ? { error: update.error } : {}),
|
||||
});
|
||||
},
|
||||
});
|
||||
whatsapp = transport;
|
||||
transport.start().catch((error) => {
|
||||
if (whatsapp !== transport) return;
|
||||
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,
|
||||
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: (update) => {
|
||||
if (telegram !== transport) return; // superseded instance
|
||||
setTelegramStatus(update);
|
||||
},
|
||||
});
|
||||
telegram = transport;
|
||||
void transport.start();
|
||||
}
|
||||
|
||||
export function applyChannelsConfig(config: Config): Promise<void> {
|
||||
return enqueue(async () => {
|
||||
await stopWhatsApp();
|
||||
stopTelegram();
|
||||
await startWhatsApp(config.whatsapp);
|
||||
startTelegram(config.telegram);
|
||||
});
|
||||
}
|
||||
|
||||
// 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> {
|
||||
return enqueue(async () => {
|
||||
if (whatsapp) {
|
||||
const out = whatsapp;
|
||||
whatsapp = null;
|
||||
await out.logout().catch(() => undefined);
|
||||
} else {
|
||||
await fs.rm(WHATSAPP_AUTH_DIR, { recursive: true, force: true }).catch(() => undefined);
|
||||
}
|
||||
const config = await container
|
||||
.resolve<IChannelsConfigRepo>("channelsConfigRepo")
|
||||
.getConfig();
|
||||
await startWhatsApp(config.whatsapp);
|
||||
});
|
||||
}
|
||||
|
||||
export async function init(): Promise<void> {
|
||||
const config = await container
|
||||
.resolve<IChannelsConfigRepo>("channelsConfigRepo")
|
||||
.getConfig();
|
||||
await applyChannelsConfig(config);
|
||||
}
|
||||
218
apps/x/packages/core/src/channels/transports/telegram.ts
Normal file
218
apps/x/packages/core/src/channels/transports/telegram.ts
Normal file
|
|
@ -0,0 +1,218 @@
|
|||
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";
|
||||
|
||||
// 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<typeof TelegramChannelStatus>;
|
||||
|
||||
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?: {
|
||||
message_id: number;
|
||||
text?: string;
|
||||
chat: { id: number; type: string };
|
||||
from?: { id: number; is_bot?: boolean };
|
||||
};
|
||||
}
|
||||
|
||||
export interface TelegramTransportOptions {
|
||||
botToken: string;
|
||||
allowFrom: string[];
|
||||
// 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;
|
||||
}
|
||||
|
||||
export class TelegramTransport {
|
||||
private abort: AbortController | null = null;
|
||||
private stopped = false;
|
||||
private offset = 0;
|
||||
private botUsername: string | undefined;
|
||||
|
||||
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 async call(method: string, body?: unknown, signal?: AbortSignal): Promise<unknown> {
|
||||
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;
|
||||
error_code?: number;
|
||||
};
|
||||
if (!payload.ok) {
|
||||
throw new TelegramApiError(
|
||||
payload.description ?? `Telegram API error (${method})`,
|
||||
payload.error_code,
|
||||
);
|
||||
}
|
||||
return payload.result;
|
||||
}
|
||||
|
||||
private async loadOffset(): Promise<void> {
|
||||
try {
|
||||
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<void> {
|
||||
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<void> {
|
||||
await new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
private async run(): Promise<void> {
|
||||
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 {
|
||||
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 = 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;
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
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}`, chatId, message.text);
|
||||
}
|
||||
|
||||
async send(chatId: string, text: string): Promise<void> {
|
||||
await this.call("sendMessage", { chat_id: chatId, text });
|
||||
}
|
||||
}
|
||||
219
apps/x/packages/core/src/channels/transports/whatsapp.ts
Normal file
219
apps/x/packages/core/src/channels/transports/whatsapp.ts
Normal file
|
|
@ -0,0 +1,219 @@
|
|||
import fs from "node:fs/promises";
|
||||
import makeWASocket, {
|
||||
DisconnectReason,
|
||||
areJidsSameUser,
|
||||
isJidGroup,
|
||||
jidDecode,
|
||||
useMultiFileAuthState,
|
||||
} from "baileys";
|
||||
|
||||
// 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[];
|
||||
// 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;
|
||||
}
|
||||
|
||||
interface TextishMessage {
|
||||
conversation?: unknown;
|
||||
extendedTextMessage?: { text?: unknown };
|
||||
ephemeralMessage?: { message?: TextishMessage };
|
||||
}
|
||||
|
||||
interface InboundWAMessage {
|
||||
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;
|
||||
}
|
||||
|
||||
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;
|
||||
// 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<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;
|
||||
this.generation++;
|
||||
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;
|
||||
this.generation++;
|
||||
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 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 (!isCurrent()) 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(() => {
|
||||
if (!isCurrent()) return;
|
||||
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 (!isCurrent() || 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;
|
||||
|
||||
// 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 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 (!senderNumbers.some((n) => this.opts.allowFrom.includes(n))) return;
|
||||
}
|
||||
|
||||
// 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);
|
||||
}
|
||||
|
||||
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) {
|
||||
this.sentIds = new Set(Array.from(this.sentIds).slice(-250));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
||||
|
|
|
|||
54
apps/x/packages/shared/src/channels.ts
Normal file
54
apps/x/packages/shared/src/channels.ts
Normal file
|
|
@ -0,0 +1,54 @@
|
|||
import { z } from "zod";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Mobile messaging channels (WhatsApp / Telegram bridges).
|
||||
//
|
||||
// Each channel links the user's OWN account/bot to the local app: WhatsApp
|
||||
// connects as a linked device (QR pairing), Telegram long-polls the user's
|
||||
// own bot token. There is no central relay — the desktop app must be running.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export const WhatsAppChannelConfig = z.object({
|
||||
enabled: z.boolean(),
|
||||
// Extra sender phone numbers (digits only, no '+') allowed to drive the
|
||||
// bridge. The linked account itself (self-chat) is always allowed.
|
||||
allowFrom: z.array(z.string()),
|
||||
});
|
||||
|
||||
export const TelegramChannelConfig = z.object({
|
||||
enabled: z.boolean(),
|
||||
botToken: z.string(),
|
||||
// Telegram chat ids (numeric strings) allowed to drive the bridge.
|
||||
// Unknown senders are told their chat id so it can be added here.
|
||||
allowFrom: z.array(z.string()),
|
||||
});
|
||||
|
||||
export const ChannelsConfig = z.object({
|
||||
whatsapp: WhatsAppChannelConfig,
|
||||
telegram: TelegramChannelConfig,
|
||||
});
|
||||
|
||||
export const DEFAULT_CHANNELS_CONFIG: z.infer<typeof ChannelsConfig> = {
|
||||
whatsapp: { enabled: false, allowFrom: [] },
|
||||
telegram: { enabled: false, botToken: "", allowFrom: [] },
|
||||
};
|
||||
|
||||
export const WhatsAppChannelStatus = z.object({
|
||||
state: z.enum(["disabled", "starting", "qr", "connected", "error"]),
|
||||
// PNG data URL of the pairing QR (present while state === "qr").
|
||||
qrDataUrl: z.string().optional(),
|
||||
// Linked account phone number (digits) once connected.
|
||||
self: z.string().optional(),
|
||||
error: z.string().optional(),
|
||||
});
|
||||
|
||||
export const TelegramChannelStatus = z.object({
|
||||
state: z.enum(["disabled", "starting", "polling", "error"]),
|
||||
botUsername: z.string().optional(),
|
||||
error: z.string().optional(),
|
||||
});
|
||||
|
||||
export const ChannelsStatus = z.object({
|
||||
whatsapp: WhatsAppChannelStatus,
|
||||
telegram: TelegramChannelStatus,
|
||||
});
|
||||
|
|
@ -19,4 +19,5 @@ export * as browserControl from './browser-control.js';
|
|||
export * as billing from './billing.js';
|
||||
export * as notificationSettings from './notification-settings.js';
|
||||
export * as codeSessions from './code-sessions.js';
|
||||
export * as channels from './channels.js';
|
||||
export { PrefixLogger };
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ import { EmailBlockSchema, GmailThreadSchema } from './blocks.js';
|
|||
import { PermissionDecision, ApprovalPolicy, CodingAgent } from './code-mode.js';
|
||||
import { NotificationSettingsSchema } from './notification-settings.js';
|
||||
import { CodeProject, CodeSession, CodeSessionMode, CodeSessionStatus, GitRepoInfo, GitStatusFile, CodeAgentModelOptions } from './code-sessions.js';
|
||||
import { ChannelsConfig, ChannelsStatus } from './channels.js';
|
||||
|
||||
// ============================================================================
|
||||
// Runtime Validation Schemas (Single Source of Truth)
|
||||
|
|
@ -950,6 +951,28 @@ const ipcSchemas = {
|
|||
success: z.literal(true),
|
||||
}),
|
||||
},
|
||||
// ── Mobile channels (WhatsApp / Telegram bridge) ─────────────
|
||||
'channels:getConfig': {
|
||||
req: z.null(),
|
||||
res: ChannelsConfig,
|
||||
},
|
||||
'channels:setConfig': {
|
||||
req: ChannelsConfig,
|
||||
res: z.object({ success: z.literal(true) }),
|
||||
},
|
||||
'channels:getStatus': {
|
||||
req: z.null(),
|
||||
res: ChannelsStatus,
|
||||
},
|
||||
'channels:whatsappLogout': {
|
||||
req: z.null(),
|
||||
res: z.object({ success: z.literal(true) }),
|
||||
},
|
||||
// Push: main → renderer status updates (QR rotation, connect/disconnect).
|
||||
'channels:status': {
|
||||
req: ChannelsStatus,
|
||||
res: z.null(),
|
||||
},
|
||||
'slack:getConfig': {
|
||||
req: z.null(),
|
||||
res: z.object({
|
||||
|
|
|
|||
886
apps/x/pnpm-lock.yaml
generated
886
apps/x/pnpm-lock.yaml
generated
File diff suppressed because it is too large
Load diff
Loading…
Add table
Add a link
Reference in a new issue