diff --git a/bun.lock b/bun.lock index 9bc3889ae..69b18865d 100644 --- a/bun.lock +++ b/bun.lock @@ -104,6 +104,7 @@ "name": "@opencode-ai/desktop-electron", "version": "2026.6.6", "dependencies": { + "@opencode-ai/remote-bridge": "workspace:*", "@opencode-ai/util": "workspace:*", "electron-context-menu": "4.1.2", "electron-log": "^5", diff --git a/packages/app/e2e/settings/settings-shell.spec.ts b/packages/app/e2e/settings/settings-shell.spec.ts index 0e3c4b0f3..0fa977a0b 100644 --- a/packages/app/e2e/settings/settings-shell.spec.ts +++ b/packages/app/e2e/settings/settings-shell.spec.ts @@ -1,8 +1,7 @@ import { test, expect } from "../fixtures" import { closeSettingsPanel, openSettings } from "../actions" -// Foundation lock for the settings route (nav in the sidebar slot, content in the main slot), -// migrating the existing pages in place (remote / integrations hidden until ready). +// Foundation lock for the settings route (nav in the sidebar slot, content in the main slot). test("@smoke settings shell shows the migrated nav and switches pages", async ({ page, gotoSession }) => { await gotoSession() @@ -15,12 +14,14 @@ test("@smoke settings shell shows the migrated nav and switches pages", async ({ await expect(settings.getByRole("tab", { name: "General" })).toHaveAttribute("aria-selected", "true") await expect(settings.locator('[data-action="settings-language"]')).toBeVisible() - // Currently 6 tabs: General / Shortcuts / Models / Integrations / Worktrees / Memory - for (const name of ["General", "Shortcuts", "Models", "Integrations", "Worktrees", "Memory"]) { + // Currently 7 tabs: General / Shortcuts / Models / Integrations / Remote access / Worktrees / Memory + for (const name of ["General", "Shortcuts", "Models", "Integrations", "Remote access", "Worktrees", "Memory"]) { await expect(settings.getByRole("tab", { name })).toBeVisible() } - // Remote access stays hidden until its page is ready - await expect(settings.getByRole("tab", { name: "Remote access" })).toHaveCount(0) + // Remote access is a real page now: clicking it renders the Telegram connection row. + await settings.getByRole("tab", { name: "Remote access" }).click() + await expect(settings.getByRole("button", { name: "Connect" })).toBeVisible() + await expect(settings.getByText("Telegram", { exact: true })).toBeVisible() // Models page = providers + models stacked: both blocks render await settings.getByRole("tab", { name: "Models" }).click() diff --git a/packages/app/e2e/snap/settings-remote.snap.ts b/packages/app/e2e/snap/settings-remote.snap.ts new file mode 100644 index 000000000..e70999c9c --- /dev/null +++ b/packages/app/e2e/snap/settings-remote.snap.ts @@ -0,0 +1,117 @@ +import { test } from "../fixtures" +import { openSettings } from "../actions" +import { composeGrid, snapOutputPath, type Shot } from "./_compose" + +test.use({ viewport: { width: 1440, height: 900 }, deviceScaleFactor: 2 }) + +// Remote access settings page + connect flow. snap runs in web Chromium, which has +// no Electron preload, so window.api.remote is injected here as a stub. The stub +// holds a mutable status the test drives via window.__remote.set(...), and parks +// startPairing() until window.__remote.capture(...) — so one run can snapshot every +// state the user sees: the page (disconnected / connected / degraded) and the +// connect-flow dialog (token / waiting / confirm / disconnect). The IPC behaviour +// behind that API is covered by unit tests in desktop-electron and remote-bridge. +test("settings-remote", async ({ page, project }) => { + test.setTimeout(180_000) + + await page.addInitScript(() => { + let status = { state: "disconnected", platform: null, identity: null, error: null } as Record + const listeners = new Set<(s: unknown) => void>() + let resolvePairing: ((v: unknown) => void) | undefined + ;(window as any).api = { + ...(window as any).api, + remote: { + getStatus: () => Promise.resolve(status), + // Stays pending so the dialog parks on the "waiting" step until the test + // releases it via __remote.capture(). + startPairing: () => new Promise((resolve) => (resolvePairing = resolve)), + cancelPairing: () => Promise.resolve(), + confirmPairing: () => Promise.resolve(), + disconnect: () => Promise.resolve(), + onStatus: (cb: (s: unknown) => void) => { + listeners.add(cb) + return () => listeners.delete(cb) + }, + }, + } + ;(window as any).__remote = { + set: (s: Record) => { + status = s + listeners.forEach((cb) => cb(status)) + }, + capture: (sender: unknown) => resolvePairing?.(sender), + } + }) + + await project.open() + const settings = await openSettings(page) + await settings.getByRole("tab", { name: "Remote access" }).click() + + const shots: Shot[] = [] + + // 1) Disconnected page. + const connect = settings.getByRole("button", { name: "Connect" }).first() + await connect.waitFor({ state: "visible", timeout: 30_000 }) + shots.push({ name: "disconnected", buf: await settings.screenshot() }) + + // 2) Connect dialog — paste the bot token. + await connect.click() + const dialog = page.getByRole("dialog") + await dialog.waitFor({ state: "visible", timeout: 30_000 }) + shots.push({ name: "token", buf: await dialog.screenshot() }) + + // 3) Waiting — token accepted, awaiting the first message (startPairing pending). + await dialog.getByRole("textbox").first().fill("8403172:AAExampleBotTokenForPreview") + await dialog.getByRole("button", { name: "Continue" }).click() + await dialog.getByText("Send your bot a message").waitFor({ state: "visible", timeout: 30_000 }) + shots.push({ name: "waiting", buf: await dialog.screenshot() }) + + // 4) Confirm — the captured sender awaits approval. + await page.evaluate(() => + (window as any).__remote.capture({ userId: "8403172", userName: "yuhan", botUsername: "my_pawwork_bot" }), + ) + await dialog.getByText("Allow this account?").waitFor({ state: "visible", timeout: 30_000 }) + shots.push({ name: "confirm", buf: await dialog.screenshot() }) + + // Close the connect dialog via its Close (X) control. The confirm step's Cancel + // only steps back to the token step (it doesn't close), and the pairing flow + // keeps Escape from tearing the dialog down — so drive the explicit X. + await dialog.getByRole("button", { name: "Close" }).click() + await dialog.waitFor({ state: "hidden", timeout: 30_000 }) + + // 5) Connected page — green status rule + paired identity. + await page.evaluate(() => + (window as any).__remote.set({ + state: "connected", + platform: "telegram", + identity: { userId: "8403172", userName: "yuhan" }, + error: null, + }), + ) + await settings.getByText("Connected", { exact: true }).waitFor({ state: "visible", timeout: 30_000 }) + shots.push({ name: "connected", buf: await settings.screenshot() }) + + // 6) Disconnect confirm dialog — only reachable while connected. + await settings.getByRole("button", { name: "Disconnect" }).click() + const disconnectDialog = page.getByRole("dialog") + await disconnectDialog.waitFor({ state: "visible", timeout: 30_000 }) + shots.push({ name: "disconnect", buf: await disconnectDialog.screenshot() }) + await disconnectDialog.getByRole("button", { name: "Cancel" }).click() + await disconnectDialog.waitFor({ state: "hidden", timeout: 30_000 }) + + // 7) Degraded page — red status rule + error detail. + await page.evaluate(() => + (window as any).__remote.set({ + state: "degraded", + platform: "telegram", + identity: { userId: "8403172", userName: "yuhan" }, + error: "Lost connection to Telegram", + }), + ) + await settings.getByText("Lost connection to Telegram").first().waitFor({ state: "visible", timeout: 30_000 }) + shots.push({ name: "degraded", buf: await settings.screenshot() }) + + const out = snapOutputPath("settings-remote") + await composeGrid(shots, out) + process.stdout.write(`\n[snap] settings-remote grid -> ${out}\n\n`) +}) diff --git a/packages/app/src/app.tsx b/packages/app/src/app.tsx index f71da187f..7e107fc8e 100644 --- a/packages/app/src/app.tsx +++ b/packages/app/src/app.tsx @@ -45,7 +45,7 @@ import { SettingsProvider } from "@/context/settings" import { TerminalProvider } from "@/context/terminal" import { AppStartupPending } from "@/components/app-startup-pending" import { AboutModal } from "@/components/about-modal" -import type { AboutInfo, RendererDiagnosticInput, RendererDiagnosticsExportResult, WebSearchStatus } from "@/desktop-api-contract" +import type { AboutInfo, RemoteBridge, RendererDiagnosticInput, RendererDiagnosticsExportResult, WebSearchStatus } from "@/desktop-api-contract" import DirectoryLayout from "@/pages/directory-layout" import Layout from "@/pages/layout" import AutomationsRoute from "@/pages/automations/automations-route" @@ -114,6 +114,7 @@ declare global { webSearchStatus?: () => Promise saveExaApiKey?: (key: string) => Promise removeExaApiKey?: () => Promise + remote?: RemoteBridge } } } diff --git a/packages/app/src/components/dialog-connect-remote.tsx b/packages/app/src/components/dialog-connect-remote.tsx new file mode 100644 index 000000000..7fe8871e0 --- /dev/null +++ b/packages/app/src/components/dialog-connect-remote.tsx @@ -0,0 +1,157 @@ +import { Button } from "@opencode-ai/ui/button" +import { useDialog } from "@opencode-ai/ui/context/dialog" +import { Dialog } from "@opencode-ai/ui/dialog" +import { Icon } from "@opencode-ai/ui/icon" +import { Spinner } from "@opencode-ai/ui/spinner" +import { TextField } from "@opencode-ai/ui/text-field" +import { Match, onCleanup, Switch } from "solid-js" +import { createStore } from "solid-js/store" +import type { RemotePairingResult } from "@/desktop-api-contract" +import { useLanguage } from "@/context/language" + +// Connect flow for the mobile companion. Reuses the connect-provider dialog +// SHAPE — a small state machine inside a Dialog — but its own backend (the +// main-process bridge over window.api.remote), since this connects a chat bot, +// not an LLM provider. Pairing is paste-token → message-the-bot → approve. +export function DialogConnectRemote(props: { onApproved?: () => void }) { + const language = useLanguage() + const dialog = useDialog() + + const [store, setStore] = createStore({ + step: "token" as "token" | "waiting" | "confirm", + token: "", + captured: undefined as RemotePairingResult | undefined, + error: undefined as string | undefined, + busy: false, + }) + + // Guard against late async resolutions after the dialog is gone, and stop the + // bot's capture poll when the dialog closes mid-pairing. + const alive = { value: true } + // Monotonic pairing-attempt id: cancelling (or restarting) pairing bumps it, so + // a startPairing() that resolves just after the user cancels is ignored instead + // of flipping the dialog back to the confirm step. + let attempt = 0 + onCleanup(() => { + alive.value = false + void window.api?.remote?.cancelPairing() + }) + + const remote = () => window.api?.remote + + async function startPairing(event?: Event) { + event?.preventDefault() + const api = remote() + const token = store.token.trim() + if (!api || token === "" || store.busy) return + const mine = ++attempt + setStore({ step: "waiting", error: undefined }) + try { + const captured = await api.startPairing(token) + if (!alive.value || mine !== attempt) return + // null = cancelled before a sender arrived; fall back to the token step. + if (!captured) return setStore({ step: "token" }) + setStore({ step: "confirm", captured }) + } catch (err) { + if (!alive.value || mine !== attempt) return + setStore({ step: "token", error: errorMessage(err) }) + } + } + + async function allow() { + const api = remote() + const captured = store.captured + if (!api || !captured || store.busy) return + setStore("busy", true) + try { + // The main process holds the token + captured identity from startPairing; + // confirm just approves it — we never resend the secret. + await api.confirmPairing() + // The bridge is starting but not yet serving. Hand the success signal to the + // page, which fires the toast only when status actually reaches "connected" + // (and never when a 409 ends in "degraded") — so we don't claim "connected" + // here, a step before it is true. Page-side, so it survives this close(). + props.onApproved?.() + if (!alive.value) return + dialog.close() + } catch (err) { + if (!alive.value) return + setStore({ busy: false, step: "token", error: errorMessage(err) }) + } + } + + function backToToken() { + attempt++ // invalidate any in-flight startPairing so its late resolve is dropped + void remote()?.cancelPairing() + setStore({ step: "token", captured: undefined }) + } + + return ( + +
+ + +
+ setStore("token", value)} + validationState={store.error ? "invalid" : undefined} + error={store.error} + /> +

{language.t("settings.remote.connect.token.help")}

+
+ +
+ +
+ + +
+
+ + {language.t("settings.remote.connect.waiting.title")} +
+

{language.t("settings.remote.connect.waiting.body")}

+
+ +
+
+
+ + +
+
+ + {language.t("settings.remote.connect.confirm.title")} +
+

+ {language.t("settings.remote.connect.confirm.body", { name: store.captured?.userName ?? "" })} +

+
+ + +
+
+
+
+
+
+ ) +} + +function errorMessage(err: unknown): string { + return err instanceof Error ? err.message : String(err) +} diff --git a/packages/app/src/desktop-api-contract.ts b/packages/app/src/desktop-api-contract.ts index ca06bfeb3..4d8a1ad54 100644 --- a/packages/app/src/desktop-api-contract.ts +++ b/packages/app/src/desktop-api-contract.ts @@ -77,3 +77,39 @@ export type WebSearchStatus = { needsAttention: boolean quotaExceeded: boolean } + +/** Mobile-companion connection state, as the renderer sees it. */ +export type RemoteState = "disconnected" | "connecting" | "connected" | "degraded" + +/** Masked status of the mobile-companion bridge — never includes the bot token. */ +export type RemoteStatus = { + state: RemoteState + platform: "telegram" | null + identity: { userId: string; userName: string } | null + error: string | null +} + +/** The sender captured during pairing, for the user to approve before connecting. */ +export type RemotePairingResult = { + userId: string + userName: string + botUsername?: string +} + +/** + * Control surface for the mobile-companion bridge (connect a phone chat app to + * this desktop's agent). Desktop/Electron only. Pairing is two steps: start + * (paste token, then message the bot from your phone — resolves with the + * captured sender, or null if cancelled) then confirm (approve that identity). + * The token crosses this boundary only on `startPairing`; the main process holds + * it from there, so `confirmPairing` approves the captured identity with no args + * and never resends the secret. + */ +export type RemoteBridge = { + getStatus(): Promise + startPairing(token: string): Promise + cancelPairing(): Promise + confirmPairing(): Promise + disconnect(): Promise + onStatus(handler: (status: RemoteStatus) => void): () => void +} diff --git a/packages/app/src/desktop-api.ts b/packages/app/src/desktop-api.ts index 3e8f21bb9..b60b8e9f2 100644 --- a/packages/app/src/desktop-api.ts +++ b/packages/app/src/desktop-api.ts @@ -1,6 +1,10 @@ export { buildDesktopContext, desktopWindowTitle, type DesktopContext } from "./utils/desktop-context" export type { AboutInfo, + RemoteBridge, + RemotePairingResult, + RemoteState, + RemoteStatus, RendererDiagnosticInput, RendererDiagnosticsExportResult, RendererErrorDetails, diff --git a/packages/app/src/i18n/en.ts b/packages/app/src/i18n/en.ts index 3682dd0ac..2798b16fa 100644 --- a/packages/app/src/i18n/en.ts +++ b/packages/app/src/i18n/en.ts @@ -958,7 +958,38 @@ export const dict = { "settings.tab.integrations": "Integrations", "settings.tab.memory": "Memory", "settings.backToApp": "Back to app", - "settings.remote.placeholder": "Remote access settings will live here.", + "settings.remote.description": + "Message your agent from a chat app on your phone. Your computer connects out — no public address or server setup.", + "settings.remote.section.telegram": "Telegram", + "settings.remote.status.disconnected": "Not connected", + "settings.remote.status.connecting": "Connecting…", + "settings.remote.status.connected": "Connected", + "settings.remote.status.degraded": "Connection problem", + "settings.remote.unsupported": "Remote access requires the desktop app.", + "settings.remote.pairedWith": "Paired with {{name}}", + "settings.remote.action.connect": "Connect", + "settings.remote.action.disconnect": "Disconnect", + "settings.remote.capabilities.title": "What you can do", + "settings.remote.capabilities.prompts": "Send prompts and read your agent's replies", + "settings.remote.capabilities.permissions": "Approve or deny permission requests", + "settings.remote.capabilities.sessions": "Switch sessions or stop a running turn", + "settings.remote.connect.title": "Connect Telegram", + "settings.remote.connect.token.label": "Bot token", + "settings.remote.connect.token.placeholder": "123456789:ABCdef…", + "settings.remote.connect.token.help": + "Create a bot with @BotFather in Telegram, then paste the token it gives you. It's stored encrypted on this computer.", + "settings.remote.connect.waiting.title": "Send your bot a message", + "settings.remote.connect.waiting.body": + "Open Telegram on your phone and send any message to your bot. The first sender becomes the only person who can drive your agent.", + "settings.remote.connect.confirm.title": "Allow this account?", + "settings.remote.connect.confirm.body": + "{{name}} will be able to send prompts and answer permission requests in your agent sessions. Only allow yourself.", + "settings.remote.connect.action.allow": "Allow", + "settings.remote.connect.toast.title": "Telegram connected", + "settings.remote.connect.toast.body": "You can now message your agent from Telegram.", + "settings.remote.disconnect.title": "Disconnect Telegram?", + "settings.remote.disconnect.body": + "This removes the saved bot token from this computer. You can connect again any time.", "settings.integrations.description": "Servers PawWork connects to, plus MCP, language servers, and plugins available to your sessions.", "settings.integrations.empty": "None configured.", diff --git a/packages/app/src/i18n/remote-placeholders.test.ts b/packages/app/src/i18n/remote-placeholders.test.ts new file mode 100644 index 000000000..8ea9767f3 --- /dev/null +++ b/packages/app/src/i18n/remote-placeholders.test.ts @@ -0,0 +1,25 @@ +import * as i18n from "@solid-primitives/i18n" +import { expect, test } from "bun:test" +import { dict as en } from "./en" +import { dict as zh } from "./zh" + +// The app resolves params with @solid-primitives/i18n's {{name}} syntax (see +// context/language.tsx, which wires t() with i18n.resolveTemplate). A single +// {name} would render literally to the user. Guard the real resolver against the +// real remote strings, in both shipped locales. +const resolve = (template: string, params: Record) => i18n.resolveTemplate(template, params) + +for (const [locale, dict] of [ + ["en", en], + ["zh", zh], +] as const) { + test(`remote i18n placeholders interpolate in ${locale}`, () => { + const paired = resolve(dict["settings.remote.pairedWith"], { name: "Ada" }) + expect(paired).toContain("Ada") + expect(paired).not.toContain("{") + + const confirm = resolve(dict["settings.remote.connect.confirm.body"], { name: "Ada" }) + expect(confirm).toContain("Ada") + expect(confirm).not.toContain("{") + }) +} diff --git a/packages/app/src/i18n/zh.ts b/packages/app/src/i18n/zh.ts index ebf948155..c6cfa772c 100644 --- a/packages/app/src/i18n/zh.ts +++ b/packages/app/src/i18n/zh.ts @@ -843,7 +843,33 @@ export const dict = { "settings.tab.integrations": "集成", "settings.tab.memory": "记忆", "settings.backToApp": "返回应用", - "settings.remote.placeholder": "远程访问设置即将在此处提供。", + "settings.remote.description": "在手机的聊天软件里和你的 agent 对话。电脑主动向外连接,不需要公网地址或服务器配置。", + "settings.remote.section.telegram": "Telegram", + "settings.remote.status.disconnected": "未连接", + "settings.remote.status.connecting": "连接中…", + "settings.remote.status.connected": "已连接", + "settings.remote.status.degraded": "连接异常", + "settings.remote.unsupported": "远程访问需要桌面应用。", + "settings.remote.pairedWith": "已配对:{{name}}", + "settings.remote.capabilities.title": "可以做什么", + "settings.remote.capabilities.prompts": "发送指令,接收 agent 的回复", + "settings.remote.capabilities.permissions": "批准或拒绝授权请求", + "settings.remote.capabilities.sessions": "切换会话,或停止正在进行的运行", + "settings.remote.action.connect": "连接", + "settings.remote.action.disconnect": "断开", + "settings.remote.connect.title": "连接 Telegram", + "settings.remote.connect.token.label": "机器人 Token", + "settings.remote.connect.token.placeholder": "123456789:ABCdef…", + "settings.remote.connect.token.help": "在 Telegram 里用 @BotFather 创建一个机器人,把它给你的 Token 粘贴进来。Token 会加密保存在本机。", + "settings.remote.connect.waiting.title": "给机器人发条消息", + "settings.remote.connect.waiting.body": "在手机上打开 Telegram,给你的机器人发任意一条消息。第一个发消息的人会成为唯一能操控你 agent 的人。", + "settings.remote.connect.confirm.title": "允许这个账号?", + "settings.remote.connect.confirm.body": "{{name}} 将能在你的 agent 会话里发送指令、回应授权请求。只允许你自己。", + "settings.remote.connect.action.allow": "允许", + "settings.remote.connect.toast.title": "Telegram 已连接", + "settings.remote.connect.toast.body": "现在可以在 Telegram 里和你的 agent 对话了。", + "settings.remote.disconnect.title": "断开 Telegram?", + "settings.remote.disconnect.body": "这会从本机删除已保存的机器人 Token。你随时可以重新连接。", "settings.integrations.description": "爪印连接的服务器,以及当前项目可用的 MCP、语言服务器、插件。", "settings.integrations.empty": "尚未配置。", "connectionHealth.toast.server.title": "服务器连接异常", diff --git a/packages/app/src/pages/settings/remote-connect-toast.test.ts b/packages/app/src/pages/settings/remote-connect-toast.test.ts new file mode 100644 index 000000000..5bdf1c951 --- /dev/null +++ b/packages/app/src/pages/settings/remote-connect-toast.test.ts @@ -0,0 +1,24 @@ +import { expect, test } from "bun:test" +import { connectToastAction } from "./remote-connect-toast" + +test("fires the success toast only when an armed connect actually reaches connected", () => { + expect(connectToastAction(true, "connected")).toBe("fire") +}) + +test("a 409 that ends in degraded disarms without a success toast", () => { + // The P2 case: Allow was clicked (armed), but the bridge never serves (another + // client owns the token). No false "connected" toast — the status row shows it. + expect(connectToastAction(true, "degraded")).toBe("disarm") + expect(connectToastAction(true, "disconnected")).toBe("disarm") +}) + +test("stays armed while still connecting", () => { + expect(connectToastAction(true, "connecting")).toBe("none") +}) + +test("launch-time auto-reconnect (not armed) never toasts", () => { + // startIfConfigured connects without a user Allow; awaiting is false, so reaching + // connected must stay silent. + expect(connectToastAction(false, "connected")).toBe("none") + expect(connectToastAction(false, "degraded")).toBe("none") +}) diff --git a/packages/app/src/pages/settings/remote-connect-toast.ts b/packages/app/src/pages/settings/remote-connect-toast.ts new file mode 100644 index 000000000..0058c721d --- /dev/null +++ b/packages/app/src/pages/settings/remote-connect-toast.ts @@ -0,0 +1,20 @@ +import type { RemoteStatus } from "@/desktop-api-contract" + +export type ConnectToastAction = "fire" | "disarm" | "none" + +/** + * Decide what the deferred connect success toast should do on a status change. + * + * The toast is armed only after the user clicks Allow (`awaiting`), so launch-time + * auto-reconnect stays silent. It fires exactly when status reaches "connected" — + * not the moment Allow returns, which is a step before the bridge is actually + * serving. A terminal non-connected outcome ("degraded" from a 409 where another + * client owns the token, or "disconnected") disarms with no toast (the status row + * already shows the cause); "connecting" keeps waiting. + */ +export function connectToastAction(awaiting: boolean, next: RemoteStatus["state"]): ConnectToastAction { + if (!awaiting) return "none" + if (next === "connected") return "fire" + if (next === "degraded" || next === "disconnected") return "disarm" + return "none" // "connecting": still pending, stay armed +} diff --git a/packages/app/src/pages/settings/remote.tsx b/packages/app/src/pages/settings/remote.tsx index f28aee6b2..f2d56db8c 100644 --- a/packages/app/src/pages/settings/remote.tsx +++ b/packages/app/src/pages/settings/remote.tsx @@ -1,15 +1,217 @@ -import { type Component } from "solid-js" +import { Button } from "@opencode-ai/ui/button" +import { useDialog } from "@opencode-ai/ui/context/dialog" +import { Dialog } from "@opencode-ai/ui/dialog" +import { Icon } from "@opencode-ai/ui/icon" +import { showToast } from "@opencode-ai/ui/toast" +import { type Component, type ComponentProps, Match, Show, Switch, createSignal, onCleanup, onMount } from "solid-js" +import { SettingsList } from "@/components/settings-list" import { useLanguage } from "@/context/language" +import type { RemoteStatus } from "@/desktop-api-contract" +import { connectToastAction } from "./remote-connect-toast" -// Remote access page. PR1 body is a placeholder so the nav entry is not empty and the whole thing ships. -// Later it gains the real remote-server connect / manage feature (the server part of the old Connections -// + Manage Servers). +const DISCONNECTED: RemoteStatus = { state: "disconnected", platform: null, identity: null, error: null } + +type IconName = ComponentProps["name"] + +// Remote access (mobile companion): connect a phone chat app to this desktop's +// agent. The bridge lives in the main process; this page shows masked status, +// what the connection enables, and opens the connect / disconnect dialogs. The +// Telegram row carries a 2px status left-rule (no box) per the cards rule in +// docs/DESIGN.md — green when connected, red when degraded, neutral otherwise. export const RemotePage: Component = () => { const language = useLanguage() + const dialog = useDialog() + const [status, setStatus] = createSignal(DISCONNECTED) + // The bridge lives in the main process; without its preload API (web preview, + // or a preload regression) there is nothing to drive, so the page shows a + // disabled "needs the desktop app" state instead of a Connect button that + // would silently no-op. + const supported = !!window.api?.remote + // Armed by the connect dialog's Allow; the success toast fires only when status + // then actually reaches "connected" (see handleStatus), never the moment Allow + // returns. So a bridge that ends up "degraded" (e.g. a 409: another client owns + // the token) shows no false "connected" toast. + const [awaitingConnect, setAwaitingConnect] = createSignal(false) + + // Defer the connect success toast to the real "connected" transition (see + // connectToastAction): a terminal non-connected outcome just disarms — the + // status row (red "Connection problem" + the cause) already carries the failure. + const handleStatus = (next: RemoteStatus) => { + const action = connectToastAction(awaitingConnect(), next.state) + if (action !== "none") setAwaitingConnect(false) + if (action === "fire") { + showToast({ + variant: "success", + icon: "circle-check", + title: language.t("settings.remote.connect.toast.title"), + description: language.t("settings.remote.connect.toast.body"), + }) + } + setStatus(next) + } + + onMount(() => { + const api = window.api?.remote + if (!api) return + void api.getStatus().then(setStatus) + onCleanup(api.onStatus(handleStatus)) + }) + + const openConnect = () => { + setAwaitingConnect(false) // fresh attempt; arm only when Allow is clicked + void import("@/components/dialog-connect-remote").then((m) => + dialog.show(() => setAwaitingConnect(true)} />), + ) + } + const openDisconnect = () => { + // Inline, not a lazy module: it is a trivial confirm whose deps are already in + // this page's bundle, so a separate dynamic chunk bought nothing. The heavy + // connect/pairing flow stays lazily imported. + dialog.show(() => ) + } + + const statusLabel = () => { + if (!supported) return language.t("settings.remote.unsupported") + switch (status().state) { + case "connected": + return language.t("settings.remote.status.connected") + case "connecting": + return language.t("settings.remote.status.connecting") + case "degraded": + return language.t("settings.remote.status.degraded") + case "disconnected": + return language.t("settings.remote.status.disconnected") + } + } + + // The 2px left-rule and the status word share one color: success when + // connected, error when degraded, neutral otherwise (idle / connecting / + // unsupported). Mirrors the Integrations status mapping. + const live = () => supported && status().state === "connected" + const bad = () => supported && status().state === "degraded" + const ruleClass = () => (live() ? "bg-icon-success-base" : bad() ? "bg-error" : "bg-border-weak") + const labelClass = () => (live() ? "text-icon-success-base" : bad() ? "text-error" : "text-fg-weak") + + const detail = () => { + const current = status() + if (current.state === "degraded") return current.error ?? undefined + if (current.identity) return language.t("settings.remote.pairedWith", { name: current.identity.userName }) + return undefined + } + + const connectable = () => status().state === "disconnected" + + return ( + +
+

{language.t("settings.tab.remoteAccess")}

+

{language.t("settings.remote.description")}

+
+ +
+ + +
+

{language.t("settings.remote.capabilities.title")}

+
+ + + + +
+ + ) +} + +// The Telegram vendor mark keeps its brand color — a sanctioned exception to the +// one-icon-DNA chrome rule (docs/DESIGN.md: app/vendor logos keep brand colors). +function TelegramMark() { + return ( + + ) +} + +function Capability(props: { icon: IconName; text: string }) { return ( -
-

{language.t("settings.tab.remoteAccess")}

-

{language.t("settings.remote.placeholder")}

+
+ + {props.text}
) } + +// Irreversible confirm: disconnecting wipes the saved token, so it gets a +// two-button Dialog with a danger action, per the design system. Inlined here +// (not a separate lazy module): it is a trivial confirm whose deps already ship +// with this page, so it never pulls in the connect flow's pairing state machine. +function DialogDisconnectRemote() { + const language = useLanguage() + const dialog = useDialog() + const [busy, setBusy] = createSignal(false) + + const handleDisconnect = async () => { + if (busy()) return + setBusy(true) + try { + await window.api?.remote?.disconnect() + dialog.close() + } finally { + setBusy(false) + } + } + + return ( + +
+ {language.t("settings.remote.disconnect.body")} +
+
+ + +
+
+ ) +} diff --git a/packages/app/src/pages/settings/settings-shell.tsx b/packages/app/src/pages/settings/settings-shell.tsx index 7e2d371fe..b7c878a16 100644 --- a/packages/app/src/pages/settings/settings-shell.tsx +++ b/packages/app/src/pages/settings/settings-shell.tsx @@ -9,18 +9,16 @@ import { SettingsMemory } from "@/components/settings-memory" import { SettingsWorktrees } from "@/components/settings-worktrees" import { IntegrationsPage } from "./integrations" import { ModelsPage } from "./models" +import { RemotePage } from "./remote" // Settings is a real route whose nav renders into the shell's sidebar slot // (SettingsNav) while the page fills the main slot (SettingsContent). Geometry // (width / background / border) is inherited from the shell slots instead of being // re-declared, which removes the alignment drift the old standalone overlay had // (its fixed 200px nav + surface-raised content diverged from the real sidebar). -// Remote access is not part of this surface yet: its page has no content branch, so it is -// intentionally absent from both the type and TAB_VALUES. It comes back — type, TAB_VALUES, -// NAV_ITEMS and a content Match together — when its page lands. -export type SettingsTab = "general" | "shortcuts" | "models" | "integrations" | "worktrees" | "memory" +export type SettingsTab = "general" | "shortcuts" | "models" | "integrations" | "remoteAccess" | "worktrees" | "memory" -const TAB_VALUES: SettingsTab[] = ["general", "shortcuts", "models", "integrations", "worktrees", "memory"] +const TAB_VALUES: SettingsTab[] = ["general", "shortcuts", "models", "integrations", "remoteAccess", "worktrees", "memory"] export function isSettingsTab(value: string): value is SettingsTab { return (TAB_VALUES as string[]).includes(value) @@ -31,6 +29,7 @@ const NAV_ITEMS = [ { value: "shortcuts", icon: "keyboard", labelKey: "settings.tab.shortcuts" }, { value: "models", icon: "models", labelKey: "settings.tab.models" }, { value: "integrations", icon: "link", labelKey: "settings.tab.integrations" }, + { value: "remoteAccess", icon: "remote-control", labelKey: "settings.tab.remoteAccess" }, { value: "worktrees", icon: "worktree", labelKey: "settings.tab.worktrees" }, { value: "memory", icon: "brain", labelKey: "settings.tab.memory" }, ] as const satisfies ReadonlyArray<{ value: SettingsTab; icon: string; labelKey: string }> @@ -202,6 +201,9 @@ export const SettingsContent: Component<{ + + + diff --git a/packages/desktop-electron/electron-vite.config.test.ts b/packages/desktop-electron/electron-vite.config.test.ts index 7ebe0310c..2d61f6160 100644 --- a/packages/desktop-electron/electron-vite.config.test.ts +++ b/packages/desktop-electron/electron-vite.config.test.ts @@ -19,6 +19,11 @@ test("renderer dedupes the ui workspace package", () => { test("main build does not externalize OpenCLI from the desktop bundle", () => { const source = readFileSync(path.join(import.meta.dir, "electron.vite.config.ts"), "utf8") - expect(source).toContain("externalizeDeps: { include: [nodePtyPkg] }") + // node-pty is the only dependency force-externalized (a native module); OpenCLI + // stays bundled. remote-bridge ships .ts source, so it is force-BUNDLED via + // exclude — leaving it external would leak a bare .ts import the runtime guard + // rejects (see desktop-smoke). + expect(source).toContain("include: [nodePtyPkg]") + expect(source).toContain('exclude: ["@opencode-ai/remote-bridge"]') expect(source).not.toContain("OPENCLI_EXTERNALS") }) diff --git a/packages/desktop-electron/electron.vite.config.ts b/packages/desktop-electron/electron.vite.config.ts index 6c5408ace..8b0c3c28a 100644 --- a/packages/desktop-electron/electron.vite.config.ts +++ b/packages/desktop-electron/electron.vite.config.ts @@ -63,7 +63,12 @@ export default defineConfig({ rollupOptions: { input: { index: "src/main/index.ts" }, }, - externalizeDeps: { include: [nodePtyPkg] }, + // remote-bridge ships TypeScript source only (exports map → ./src/*.ts) and + // has no runtime deps, so it must be BUNDLED into out/main rather than + // externalized — a bare `import "@opencode-ai/remote-bridge/..."` left in the + // output would resolve to .ts at runtime, which Electron cannot execute (and + // the runtime-import-guard fails the build on exactly that). + externalizeDeps: { include: [nodePtyPkg], exclude: ["@opencode-ai/remote-bridge"] }, }, plugins: [ { diff --git a/packages/desktop-electron/package.json b/packages/desktop-electron/package.json index 4efd2f9d4..55a36a1cb 100644 --- a/packages/desktop-electron/package.json +++ b/packages/desktop-electron/package.json @@ -30,6 +30,7 @@ }, "main": "./out/main/index.js", "dependencies": { + "@opencode-ai/remote-bridge": "workspace:*", "@opencode-ai/util": "workspace:*", "electron-context-menu": "4.1.2", "electron-log": "^5", diff --git a/packages/desktop-electron/src/main/index.ts b/packages/desktop-electron/src/main/index.ts index 96fac1e35..a483bb7dc 100644 --- a/packages/desktop-electron/src/main/index.ts +++ b/packages/desktop-electron/src/main/index.ts @@ -6,7 +6,7 @@ import { createServer } from "node:net" import os, { homedir } from "node:os" import { dirname, join } from "node:path" import type { Event } from "electron" -import { app, BrowserWindow, clipboard, dialog, shell } from "electron" +import { app, BrowserWindow, clipboard, dialog, safeStorage, shell } from "electron" import pkg from "electron-updater" import { buildDesktopContext } from "@opencode-ai/app/desktop-api" @@ -77,6 +77,7 @@ import { createFeedbackHandler, feedbackDialogLabels } from "./feedback" import { registerIpcHandlers, sendDeepLinks, sendMenuCommand, sendSqliteMigrationProgress } from "./ipc" import { registerAboutIpc, triggerAbout } from "./ipc/about" import { registerBrowserIpc } from "./ipc/browser" +import { registerRemoteIpc } from "./ipc/remote" import { createDesktopBrowserBridgeHost } from "./browser/automation-host" import { browserControllers } from "./browser/controller-automation" import { diagnosticsLogTail, filePath, initLogging } from "./logging" @@ -92,6 +93,8 @@ import { SESSION_EXPORT_RENDERER_DIAGNOSTICS_MAX_BYTES, } from "./renderer-diagnostics" import { backendLogFilePath, getDefaultServerUrl, getWslConfig, setDefaultServerUrl, setWslConfig, spawnLocalServer } from "./server" +import { createRemoteBridgeRuntime } from "./remote-bridge" +import { safeStorageCredentialStore, type CredentialStoreEnv } from "./remote-credentials" import { PAWWORK_RUNTIME } from "./runtime-namespace" import { createUpdaterController, createUpdateFeed, githubFeed, r2Feed, type FeedTarget } from "./updater" import { pendingUpdateCacheDir } from "./updater-cache" @@ -154,6 +157,26 @@ const contextWindowCleanup = new Set() const pendingDeepLinks: string[] = [] const serverReady = defer() +// The mobile-companion bridge: connects a phone chat app to this desktop's agent +// sessions. Runs in the main process against the local server; credentials are +// encrypted main-only and never cross to the renderer. +// safeStorage + the on-disk paths are injected so the credential store itself +// stays Electron-free and unit-testable with a fake env. +const remoteUserData = () => app.getPath("userData") +const remoteStateFile = () => join(remoteUserData(), "remote-bridge-state.json") +const remoteCredentialEnv: CredentialStoreEnv = { + credentialsFile: () => join(remoteUserData(), "remote-bridge-credentials.json"), + stateFile: remoteStateFile, + isEncryptionAvailable: () => safeStorage.isEncryptionAvailable(), + encryptString: (plain) => safeStorage.encryptString(plain), + decryptString: (cipher) => safeStorage.decryptString(cipher), +} +const remoteBridge = createRemoteBridgeRuntime({ + credentials: safeStorageCredentialStore(remoteCredentialEnv), + statePath: remoteStateFile(), + serverInfo: () => serverReady.promise, + locale: () => currentDesktopContext().locale, +}) const logger = initLogging() const problemReportRoot = problemReportsRoot(app.getPath("userData")) const rendererDiagnostics = createRendererDiagnosticsRecorder({ @@ -409,6 +432,9 @@ function setupApp() { }) app.on("before-quit", () => { + // Best-effort: signal the bridge to stop polling before the server it talks + // to is torn down. Its poll fetches are aborted, so this returns promptly. + void remoteBridge.stop() killSidecar() }) @@ -551,6 +577,11 @@ async function initialize() { await loadingTask setInitStep({ phase: "done" }) + // Start the mobile-companion bridge only after the server is healthy, in the + // background so it never blocks opening the main window. Failures land in the + // bridge's own status (degraded), not here. + void remoteBridge.startIfConfigured() + if (overlay) { await loadingComplete.promise } @@ -671,6 +702,7 @@ registerIpcHandlers({ registerAboutIpc() registerBrowserIpc({ sessionIDForWindow: (windowID) => desktopContexts.current(windowID).sessionID }) +registerRemoteIpc(remoteBridge) function killSidecar() { if (!server) return diff --git a/packages/desktop-electron/src/main/ipc/remote.ts b/packages/desktop-electron/src/main/ipc/remote.ts new file mode 100644 index 000000000..dd6b6cf27 --- /dev/null +++ b/packages/desktop-electron/src/main/ipc/remote.ts @@ -0,0 +1,34 @@ +import { BrowserWindow, ipcMain } from "electron" +import { PairingCancelledError, type RemoteBridgeRuntime } from "../remote-bridge" + +/** + * Wires the mobile-companion bridge IPC. The renderer pastes the bot token once, + * inbound, over remote:start-pairing; from there it stays main-only — confirm and + * disconnect carry no token, and the status read back is always masked. Status + * changes are broadcast to every window so the settings page reflects + * connect/degraded without polling. + */ +export function registerRemoteIpc(runtime: RemoteBridgeRuntime) { + runtime.onStatusChange((status) => { + for (const win of BrowserWindow.getAllWindows()) { + if (!win.isDestroyed()) win.webContents.send("remote:status", status) + } + }) + + ipcMain.handle("remote:get-status", () => runtime.getStatus()) + // Resolves with the captured sender, or null if the user cancelled (closed the + // connect dialog). A real failure (bad token) rejects so the UI can show it. + ipcMain.handle("remote:start-pairing", async (_event, token: string) => { + try { + return await runtime.startPairing(token) + } catch (err) { + if (err instanceof PairingCancelledError) return null + throw err + } + }) + ipcMain.handle("remote:cancel-pairing", () => runtime.cancelPairing()) + // No args: the token + captured identity are held main-side from start-pairing, + // so confirm only approves them — the renderer can't supply or swap the token. + ipcMain.handle("remote:confirm-pairing", () => runtime.confirmPairing()) + ipcMain.handle("remote:disconnect", () => runtime.disconnect()) +} diff --git a/packages/desktop-electron/src/main/remote-bridge.test.ts b/packages/desktop-electron/src/main/remote-bridge.test.ts new file mode 100644 index 000000000..8415eb8f0 --- /dev/null +++ b/packages/desktop-electron/src/main/remote-bridge.test.ts @@ -0,0 +1,294 @@ +import { expect, test } from "bun:test" +import { + PairingCancelledError, + RemoteBridgeRuntime, + type CapturedSender, + type CredentialStore, + type RemoteBridgeDeps, + type RemoteCredentials, +} from "./remote-bridge" + +function memoryStore( + initial: RemoteCredentials | null = null, + available = true, +): CredentialStore & { value: RemoteCredentials | null } { + return { + value: initial, + isAvailable() { + return available + }, + load() { + return this.value + }, + save(creds) { + this.value = creds + }, + clear() { + this.value = null + }, + } +} + +/** A fake bridge App whose run() stays pending until aborted, like the real one. + * It signals readiness immediately, the way a platform that has finished its + * backlog drain would, so the runtime reaches "connected". */ +function fakeApp() { + let started = false + return { + started: () => started, + app: { + run(signal?: AbortSignal, onReady?: () => void) { + started = true + onReady?.() + return new Promise((resolve) => { + if (signal?.aborted) return resolve() + signal?.addEventListener("abort", () => resolve(), { once: true }) + }) + }, + }, + } +} + +function deps(overrides: Partial = {}): RemoteBridgeDeps { + return { + credentials: memoryStore(), + statePath: "/tmp/state.json", + serverInfo: async () => ({ url: "http://localhost:1", username: "u", password: "p" }), + locale: () => "en", + buildApp: async () => fakeApp().app, + makePoller: () => ({ getMe: async () => ({ id: "1", username: "bot" }) }) as any, + capture: async () => ({ userId: "42", userName: "yu", botUsername: "bot" }) as CapturedSender, + makePlatform: () => ({ name: "telegram", start: async () => {}, reply: async () => {}, send: async () => {}, stop: async () => {} }), + ...overrides, + } +} + +test("startPairing returns the captured sender plus bot identity", async () => { + const runtime = new RemoteBridgeRuntime(deps()) + const result = await runtime.startPairing("123:abc") + expect(result).toEqual({ userId: "42", userName: "yu", botUsername: "bot" }) +}) + +test("startPairing fails fast when secure storage is unavailable, without contacting Telegram", async () => { + let pollerMade = false + let captureCalled = false + const runtime = new RemoteBridgeRuntime( + deps({ + credentials: memoryStore(null, false), + makePoller: () => { + pollerMade = true + return { getMe: async () => ({ id: "1", username: "bot" }) } as any + }, + capture: async () => { + captureCalled = true + return { userId: "42", userName: "yu" } as CapturedSender + }, + }), + ) + await expect(runtime.startPairing("123:abc")).rejects.toThrow(/secure storage is unavailable/) + // The whole point of the preflight: no poller, no getMe/getUpdates, no capture. + expect(pollerMade).toBe(false) + expect(captureCalled).toBe(false) +}) + +test("startPairing surfaces an invalid token before asking the user to message", async () => { + // The token is now proven inside capture (its drain hits a fatal 401 before any + // sender is awaited); a thrown capture maps to the reach-Telegram error. + const runtime = new RemoteBridgeRuntime( + deps({ + capture: async () => { + throw new Error("401 Unauthorized") + }, + }), + ) + await expect(runtime.startPairing("bad")).rejects.toThrow(/could not reach Telegram/) +}) + +test("cancelPairing aborts capture and surfaces a cancellation", async () => { + const runtime = new RemoteBridgeRuntime( + deps({ + capture: (_poller, signal) => + new Promise((resolve) => signal.addEventListener("abort", () => resolve(null), { once: true })), + }), + ) + const pending = runtime.startPairing("123:abc") + await Promise.resolve() + runtime.cancelPairing() + await expect(pending).rejects.toBeInstanceOf(PairingCancelledError) +}) + +test("a capture that resolves after cancelPairing does not resurrect a pending pairing", async () => { + let captureEntered = () => {} + const entered = new Promise((r) => (captureEntered = r)) + let resolveCapture: (s: CapturedSender | null) => void = () => {} + const runtime = new RemoteBridgeRuntime( + deps({ + capture: () => { + captureEntered() + return new Promise((resolve) => (resolveCapture = resolve)) + }, + }), + ) + const pending = runtime.startPairing("123:abc") + await entered + runtime.cancelPairing() + // The capture resolves with a real sender AFTER the user cancelled — it must be + // dropped, not stored, or a later confirm would pair an abandoned identity. + resolveCapture({ userId: "42", userName: "yu" }) + await expect(pending).rejects.toBeInstanceOf(PairingCancelledError) + await expect(runtime.confirmPairing()).rejects.toThrow(/no pairing/) +}) + +test("confirmPairing persists the captured pairing and reports connected", async () => { + const store = memoryStore() + const runtime = new RemoteBridgeRuntime(deps({ credentials: store })) + await runtime.startPairing("123:abc") // fake capture returns user 42 / yu + await runtime.confirmPairing() // no token passed — main-side pending is used + expect(store.value).toEqual({ token: "123:abc", allowFrom: "42", userName: "yu" }) + const status = runtime.getStatus() + expect(status.state).toBe("connected") + expect(status.platform).toBe("telegram") + expect(status.identity).toEqual({ userId: "42", userName: "yu" }) +}) + +test("status stays 'connecting' until the bridge signals it is serving (onReady)", async () => { + // The startup-race fix: app.run() merely returning a promise is not "connected". + // The real run still has to ready the stream, hydrate, and drain the Telegram + // backlog before its poll loop exists; a message sent in that window is dropped. + // Hold onReady to observe that pre-serving window, then release it. + let release: (() => void) | undefined + const runtime = new RemoteBridgeRuntime( + deps({ + buildApp: async () => ({ + run: (signal?: AbortSignal, onReady?: () => void) => { + release = onReady + return new Promise((resolve) => { + if (signal?.aborted) return resolve() + signal?.addEventListener("abort", () => resolve(), { once: true }) + }) + }, + }), + }), + ) + await runtime.startPairing("t") + await runtime.confirmPairing() + // run() has started but the platform is not serving yet — must not be connected. + expect(runtime.getStatus().state).toBe("connecting") + release?.() + expect(runtime.getStatus().state).toBe("connected") + await runtime.stop() +}) + +test("confirmPairing without a pending pairing is rejected (renderer cannot inject a token)", async () => { + const runtime = new RemoteBridgeRuntime(deps()) + await expect(runtime.confirmPairing()).rejects.toThrow(/no pairing/) +}) + +test("disconnect stops the bridge and wipes credentials", async () => { + const store = memoryStore({ token: "t", allowFrom: "42" }) + const runtime = new RemoteBridgeRuntime(deps({ credentials: store })) + await runtime.startIfConfigured() + expect(runtime.getStatus().state).toBe("connected") + await runtime.disconnect() + expect(store.value).toBeNull() + expect(runtime.getStatus().state).toBe("disconnected") +}) + +test("startIfConfigured does nothing without saved credentials", async () => { + const runtime = new RemoteBridgeRuntime(deps({ credentials: memoryStore(null) })) + await runtime.startIfConfigured() + expect(runtime.getStatus().state).toBe("disconnected") +}) + +test("a double confirm builds only one bridge (the second finds no pending)", async () => { + let built = 0 + let running = 0 + let maxRunning = 0 + const runtime = new RemoteBridgeRuntime( + deps({ + buildApp: async () => { + built++ + return { + run(signal?: AbortSignal) { + running++ + maxRunning = Math.max(maxRunning, running) + return new Promise((resolve) => { + const done = () => { + running-- + resolve() + } + if (signal?.aborted) return done() + signal?.addEventListener("abort", done, { once: true }) + }) + }, + } + }, + }), + ) + await runtime.startPairing("t") + // A double-clicked Allow: the first consumes the pending pairing and builds; + // the second finds none, so only one live bridge is ever started. + const results = await Promise.allSettled([runtime.confirmPairing(), runtime.confirmPairing()]) + expect(results.map((r) => r.status)).toEqual(["fulfilled", "rejected"]) + expect(built).toBe(1) + expect(maxRunning).toBe(1) + await runtime.stop() + expect(running).toBe(0) +}) + +test("a fatal run() failure becomes degraded status, not an unhandled rejection", async () => { + const runtime = new RemoteBridgeRuntime( + deps({ + buildApp: async () => ({ run: async () => { throw new Error("revoked token") } }), + }), + ) + await runtime.startPairing("t") + await runtime.confirmPairing() + // The run rejection is observed on a microtask; let it settle. + await new Promise((r) => setTimeout(r, 5)) + const status = runtime.getStatus() + expect(status.state).toBe("degraded") + expect(status.error).toMatch(/revoked token/) +}) + +test("a confirmPairing whose bridge fails to build lands on degraded, not stuck connecting", async () => { + // serverInfo/buildApp can throw before run() ever starts (e.g. the server is + // unreachable). confirmPairing awaits startBridge directly — unlike + // startIfConfigured — so without the degraded fallback the status would hang on + // "connecting" forever. + const runtime = new RemoteBridgeRuntime( + deps({ buildApp: async () => { throw new Error("server unreachable") } }), + ) + await runtime.startPairing("t") + await expect(runtime.confirmPairing()).rejects.toThrow(/server unreachable/) + const status = runtime.getStatus() + expect(status.state).toBe("degraded") + expect(status.error).toMatch(/server unreachable/) +}) + +test("stop() aborts the live bridge synchronously, before its returned promise settles", async () => { + // before-quit runs `void stop(); killSidecar()` without awaiting. The abort must + // fire on the synchronous prefix of stop(), so the poller is already tearing down + // before the sidecar it talks to is killed. + let abortedSync = false + const runtime = new RemoteBridgeRuntime( + deps({ + buildApp: async () => ({ + run: (signal?: AbortSignal, onReady?: () => void) => { + onReady?.() + return new Promise((resolve) => { + if (signal?.aborted) return resolve() + signal?.addEventListener("abort", () => { abortedSync = true; resolve() }, { once: true }) + }) + }, + }), + }), + ) + await runtime.startPairing("t") + await runtime.confirmPairing() + expect(runtime.getStatus().state).toBe("connected") + + const stopping = runtime.stop() // do not await yet + expect(abortedSync).toBe(true) // abort landed on the sync prefix, before any await + await stopping +}) diff --git a/packages/desktop-electron/src/main/remote-bridge.ts b/packages/desktop-electron/src/main/remote-bridge.ts new file mode 100644 index 000000000..250fe677e --- /dev/null +++ b/packages/desktop-electron/src/main/remote-bridge.ts @@ -0,0 +1,322 @@ +import { createApp, normalizeLocale, type Config, type PlatformFactory } from "@opencode-ai/remote-bridge/gateway" +import { + captureFirstSender, + type CapturedSender, + TelegramPlatform, + TelegramPoller, +} from "@opencode-ai/remote-bridge/platforms/telegram" +import type { Platform } from "@opencode-ai/remote-bridge/types" +import type { RemotePairingResult, RemoteStatus } from "@opencode-ai/app/desktop-api" + +export type { RemotePairingResult, RemoteStatus } from "@opencode-ai/app/desktop-api" + +// Bounds how long a disconnect / quit waits for the bridge to stop, so a wedged +// long-poll can never block app shutdown. The poll fetch is aborted on stop, so +// this is just a backstop. +const STOP_TIMEOUT_MS = 3_000 + +export interface RemoteCredentials { + token: string + allowFrom: string + /** Display name of the paired user, for the settings page; non-secret. */ + userName?: string +} + +/** Persistence seam for the secret credentials (safeStorage-backed in prod). */ +export interface CredentialStore { + /** Whether secrets can actually be persisted (OS encryption is available). + * Checked up front by startPairing so we never walk the user through pairing + * only to fail at the final save. */ + isAvailable(): boolean + load(): RemoteCredentials | null + save(creds: RemoteCredentials): void + clear(): void +} + +/** Raised when pairing is aborted (the connect dialog was closed) rather than + * failing — lets the IPC layer distinguish a cancel from a real error. */ +export class PairingCancelledError extends Error { + constructor() { + super("pairing cancelled") + this.name = "PairingCancelledError" + } +} + +interface ServerInfo { + url: string + // The desktop's ServerReadyData uses null for "no auth"; normalized to + // undefined when building the bridge config. + username?: string | null + password?: string | null +} + +/** A runnable bridge, as the runtime needs it (`createApp`'s `App` satisfies it). */ +interface BridgeApp { + run(signal?: AbortSignal, onReady?: () => void): Promise +} + +export interface RemoteBridgeDeps { + credentials: CredentialStore + statePath: string + serverInfo: () => Promise + /** The desktop UI language, for localizing the chat-facing copy. */ + locale: () => string + // Injected so the lifecycle can be tested without Electron or a live network. + buildApp: (config: Config, factory: PlatformFactory) => Promise + makePoller: (token: string) => TelegramPoller + capture: (poller: TelegramPoller, signal: AbortSignal) => Promise + makePlatform: (token: string, allowFrom: string) => Platform +} + +/** + * Owns the single in-process bridge for the desktop app. Every lifecycle change + * (connect / confirm-pairing / disconnect / shutdown) runs through one serial + * queue so a double-click can never start two pollers on the same token. The + * bridge runs in the background — its failures land in `status`, never as an + * unhandled rejection. Pairing (capturing the first sender before an allow_from + * exists) runs on its own poller and is always torn down before the real bridge + * starts, since two pollers on one token race a 409. + */ +export class RemoteBridgeRuntime { + private status: RemoteStatus = { state: "disconnected", platform: null, identity: null, error: null } + private readonly listeners = new Set<(status: RemoteStatus) => void>() + private queue: Promise = Promise.resolve() + private ac: AbortController | null = null + private runPromise: Promise | null = null + private pairingAc: AbortController | null = null + // The captured-but-not-yet-approved pairing. Holds the token main-side between + // startPairing and confirmPairing so the renderer never has to resend the secret. + private pending: { token: string; allowFrom: string; userName?: string } | null = null + + constructor(private readonly deps: RemoteBridgeDeps) {} + + getStatus(): RemoteStatus { + return this.status + } + + onStatusChange(listener: (status: RemoteStatus) => void): () => void { + this.listeners.add(listener) + return () => this.listeners.delete(listener) + } + + /** Start the bridge on launch if a connection was previously saved. Failures + * become degraded status, never a thrown error that breaks app startup. */ + async startIfConfigured(): Promise { + const creds = this.deps.credentials.load() + if (!creds) return + // startBridge sets degraded on any failure; this catch only keeps the + // background start from surfacing as an unhandled rejection. + await this.enqueue(() => this.startBridge(creds)).catch(() => {}) + } + + /** + * Validate the token, then wait for the first private message so we learn who + * to pair with. Blocks until a sender is captured; throws PairingCancelledError + * if `cancelPairing` aborts it first. Does not persist anything — that is + * `confirmPairing`'s job once the user approves the captured identity. + */ + async startPairing(token: string): Promise { + token = token.trim() + if (token === "") throw new Error("a bot token is required") + // Preflight secure storage before contacting Telegram: without OS encryption + // the token can never be saved, so fail here rather than after the user has + // walked the whole token → message bot → confirm flow only for confirmPairing's + // final save() to throw. + if (!this.deps.credentials.isAvailable()) { + throw new Error("secure storage is unavailable on this system, so the bot token cannot be saved") + } + if (this.status.state === "connected" || this.status.state === "connecting") { + throw new Error("disconnect the current account before pairing a new one") + } + this.pairingAc?.abort() + this.pending = null + const ac = new AbortController() + this.pairingAc = ac + const poller = this.deps.makePoller(token) + // One pairing primitive: capture drains the backlog (establishing the offset + // baseline and proving the token) BEFORE fetching the bot identity, then waits + // for the first sender. Draining first means a message the user sends during a + // slow getMe lands past the baseline and is captured, not mistaken for backlog. + let captured: CapturedSender | null + try { + captured = await this.deps.capture(poller, ac.signal) + } catch (err) { + // Cancelled (cancelPairing/disconnect) or superseded by a newer startPairing + // while we awaited: surface a cancel and leave the current handle alone — it + // belongs to whoever replaced us. Otherwise the capture's drain/getMe hit a + // bad token or an unreachable Telegram. + if (ac.signal.aborted || this.pairingAc !== ac) throw new PairingCancelledError() + this.pairingAc = null + throw new Error(`could not reach Telegram with that token: ${message(err)}`) + } + // Same guard after capture: if this attempt was cancelled or superseded while + // we waited, drop the result — never resurrect `pending` for an abandoned + // attempt, even if a sender did arrive. + if (ac.signal.aborted || this.pairingAc !== ac) throw new PairingCancelledError() + this.pairingAc = null + if (!captured) throw new PairingCancelledError() + // Hold the token main-side; confirmPairing approves this identity without the + // renderer resending the secret. + this.pending = { token, allowFrom: captured.userId, userName: captured.userName } + return { userId: captured.userId, userName: captured.userName, botUsername: captured.botUsername } + } + + cancelPairing(): void { + this.pairingAc?.abort() + this.pairingAc = null + this.pending = null + } + + /** + * Approve the pending pairing and start the bridge. The token was captured + * main-side by `startPairing`, so the renderer approves the captured identity + * without ever resending the secret. + */ + async confirmPairing(): Promise { + const pending = this.pending + if (!pending) throw new Error("no pairing is awaiting confirmation") + this.pending = null + await this.enqueue(async () => { + await this.stopBridge() + const creds: RemoteCredentials = { ...pending } + this.deps.credentials.save(creds) + await this.startBridge(creds) + }) + } + + /** Stop the bridge and wipe the saved credentials. */ + async disconnect(): Promise { + this.cancelPairing() + await this.enqueue(async () => { + await this.stopBridge() + this.deps.credentials.clear() + this.setStatus({ state: "disconnected", platform: null, identity: null, error: null }) + }) + } + + /** Idempotent stop for app shutdown — does NOT clear credentials. */ + async stop(): Promise { + this.cancelPairing() + // Abort the live bridge synchronously, before the first await. before-quit runs + // `void stop(); killSidecar()` without awaiting, so the abort must land on this + // sync prefix — otherwise the poll loop is still running against the sidecar + // (the server it talks to) at the moment it is torn down. stopBridge re-aborts + // (idempotent) and awaits the teardown. + this.ac?.abort() + await this.enqueue(() => this.stopBridge()) + } + + private async startBridge(creds: RemoteCredentials): Promise { + await this.stopBridge() + this.setStatus({ state: "connecting", platform: "telegram", error: null }) + let app: BridgeApp + try { + const server = await this.deps.serverInfo() + const config: Config = { + pawWorkBaseURL: server.url, + pawWorkUsername: server.username ?? undefined, + pawWorkPassword: server.password ?? undefined, + statePath: this.deps.statePath, + locale: normalizeLocale(this.deps.locale()), + platforms: [{ name: "telegram", enabled: true, options: { allow_from: creds.allowFrom } }], + } + // The token is captured in the factory closure, never placed in `config` + // (which may be logged): only the non-secret allow_from travels through it. + const factory: PlatformFactory = (name, options) => { + if (name !== "telegram") throw new Error(`unsupported platform ${name}`) + return this.deps.makePlatform(creds.token, String(options.allow_from ?? "")) + } + app = await this.deps.buildApp(config, factory) + } catch (err) { + // serverInfo / buildApp failed before the bridge ever ran. Land on degraded + // so confirmPairing — which awaits this directly, unlike startIfConfigured — + // can't leave the status stuck on "connecting" forever when the build throws. + this.setStatus({ state: "degraded", error: message(err) }) + throw err + } + const ac = new AbortController() + this.ac = ac + // Stay "connecting" until the bridge is actually serving: app.run() only + // returns a promise here; it still has to ready the event stream, hydrate, + // and drain the Telegram backlog before its live poll loop is installed. A + // message sent during that window would be dropped as backlog, so onReady — + // which fires once the platform is polling — is what flips us to "connected". + const run = app.run(ac.signal, () => { + if (this.ac === ac) + this.setStatus({ + state: "connected", + identity: { userId: creds.allowFrom, userName: creds.userName ?? creds.allowFrom }, + }) + }) + this.runPromise = run + // run() resolves on a clean stop and rejects on a fatal failure (revoked + // token, stream protocol error). Observe it so a failure becomes degraded + // status rather than an unhandled rejection. The `this.ac === ac` guard + // keeps a stale run from clobbering a newer connection's status. + run + .then(() => { + if (this.ac === ac) this.setStatus({ state: "disconnected", platform: null, identity: null }) + }) + .catch((err) => { + if (this.ac === ac) this.setStatus({ state: "degraded", error: message(err) }) + }) + } + + private async stopBridge(): Promise { + const ac = this.ac + const runPromise = this.runPromise + this.ac = null + this.runPromise = null + if (!ac) return + ac.abort() + if (runPromise) { + await Promise.race([runPromise.catch(() => {}), delay(STOP_TIMEOUT_MS)]) + } + } + + private setStatus(patch: Partial): void { + this.status = { ...this.status, ...patch } + for (const listener of this.listeners) { + try { + listener(this.status) + } catch { + // a listener failure must not break status propagation + } + } + } + + // Serialize lifecycle ops so concurrent calls (e.g. a double-clicked button) + // never run two startBridge/stopBridge sequences at once. + private enqueue(op: () => Promise): Promise { + const run = this.queue.then(op, op) + this.queue = run.then( + () => {}, + () => {}, + ) + return run + } +} + +/** Wire a runtime with the real remote-bridge implementation. */ +export function createRemoteBridgeRuntime(deps: { + credentials: CredentialStore + statePath: string + serverInfo: () => Promise + locale: () => string +}): RemoteBridgeRuntime { + return new RemoteBridgeRuntime({ + ...deps, + buildApp: createApp, + makePoller: (token) => new TelegramPoller(token), + capture: captureFirstSender, + makePlatform: (token, allowFrom) => new TelegramPlatform({ token, allowFrom }), + }) +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)) +} + +function message(err: unknown): string { + return err instanceof Error ? err.message : String(err) +} diff --git a/packages/desktop-electron/src/main/remote-credentials.test.ts b/packages/desktop-electron/src/main/remote-credentials.test.ts new file mode 100644 index 000000000..06dd3535f --- /dev/null +++ b/packages/desktop-electron/src/main/remote-credentials.test.ts @@ -0,0 +1,51 @@ +import { expect, test } from "bun:test" +import { chmodSync, mkdtempSync, statSync, writeFileSync } from "node:fs" +import { tmpdir } from "node:os" +import path from "node:path" +import { safeStorageCredentialStore, type CredentialStoreEnv } from "./remote-credentials" + +// A deterministic fake env: a real tmp file for IO, identity "encryption" (utf8 +// bytes <-> string) so the base64 envelope round-trips exactly like safeStorage +// would. No electron module mock — this test does not depend on which test file +// mocked "electron" first. +function fakeEnv(): CredentialStoreEnv { + const dir = mkdtempSync(path.join(tmpdir(), "remote-creds-")) + return { + credentialsFile: () => path.join(dir, "credentials.json"), + stateFile: () => path.join(dir, "state.json"), + isEncryptionAvailable: () => true, + encryptString: (plain) => Buffer.from(plain, "utf8"), + decryptString: (cipher) => cipher.toString("utf8"), + } +} + +test("credential store round-trips userName through save/load", () => { + const store = safeStorageCredentialStore(fakeEnv()) + + store.save({ token: "123:ABC", allowFrom: "42", userName: "yuhan" }) + expect(store.load()).toEqual({ token: "123:ABC", allowFrom: "42", userName: "yuhan" }) + + // A credential saved without a name still loads (userName simply absent). + store.save({ token: "123:ABC", allowFrom: "42" }) + expect(store.load()).toEqual({ token: "123:ABC", allowFrom: "42", userName: undefined }) +}) + +test("isAvailable reflects the env; save refuses when encryption is unavailable", () => { + const base = fakeEnv() + expect(safeStorageCredentialStore(base).isAvailable()).toBe(true) + + const store = safeStorageCredentialStore({ ...base, isEncryptionAvailable: () => false }) + expect(store.isAvailable()).toBe(false) + expect(() => store.save({ token: "123:ABC", allowFrom: "42" })).toThrow(/secure storage is unavailable/) +}) + +test("save re-enforces 0o600 even when the file already exists with loose perms", () => { + if (process.platform === "win32") return // POSIX mode bits only + const env = fakeEnv() + const file = env.credentialsFile() + // An older build or a copied file could have left the token world-readable. + writeFileSync(file, "stale", { mode: 0o644 }) + chmodSync(file, 0o644) // defeat umask so the precondition is genuinely loose + safeStorageCredentialStore(env).save({ token: "123:ABC", allowFrom: "42" }) + expect(statSync(file).mode & 0o777).toBe(0o600) +}) diff --git a/packages/desktop-electron/src/main/remote-credentials.ts b/packages/desktop-electron/src/main/remote-credentials.ts new file mode 100644 index 000000000..25701b4ee --- /dev/null +++ b/packages/desktop-electron/src/main/remote-credentials.ts @@ -0,0 +1,88 @@ +import { chmodSync, existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs" +import path from "node:path" +import type { CredentialStore, RemoteCredentials } from "./remote-bridge" + +// Chat credentials (bot token + paired user id) are secrets. The token crosses +// IPC exactly once, inbound, when the user pastes it into the connect dialog; +// from there it lives only in the main process, encrypted at rest with Electron +// safeStorage. It is never sent back out: the renderer only ever reads masked +// status, confirmPairing carries no token, and the stored secret never returns +// over IPC. Stored separately from electron-store (which the renderer can read) +// for that reason. + +const FILE_VERSION = 1 + +interface Envelope { + version: number + cipher: string // base64 of safeStorage.encryptString(JSON) +} + +/** + * The OS-backed bits the credential store depends on: where the files live and + * the safeStorage crypto. Injected rather than imported so this module stays free + * of Electron — the store is unit-tested with a fake env. (Importing "electron" + * in a unit test is order-dependent — other test files mock it — and throws + * outright without an installed binary.) The Electron-backed env is wired in + * index.ts. + */ +export interface CredentialStoreEnv { + credentialsFile(): string + stateFile(): string + isEncryptionAvailable(): boolean + encryptString(plain: string): Buffer + decryptString(cipher: Buffer): string +} + +/** + * safeStorage-backed credential store. Encryption is required: if the OS keyring + * is unavailable (e.g. a headless Linux box with no secret service) we refuse to + * persist rather than silently write a plaintext token. macOS and Windows always + * have it. + */ +export function safeStorageCredentialStore(env: CredentialStoreEnv): CredentialStore { + return { + isAvailable(): boolean { + return env.isEncryptionAvailable() + }, + + load(): RemoteCredentials | null { + const file = env.credentialsFile() + if (!existsSync(file)) return null + try { + const envelope = JSON.parse(readFileSync(file, "utf8")) as Envelope + if (!envelope?.cipher) return null + const plain = env.decryptString(Buffer.from(envelope.cipher, "base64")) + const parsed = JSON.parse(plain) as RemoteCredentials + if (!parsed?.token || !parsed?.allowFrom) return null + // userName is the non-secret display name approved at pairing; without it + // the settings page falls back to the raw user id after a restart. + return { token: parsed.token, allowFrom: parsed.allowFrom, userName: parsed.userName } + } catch { + // A corrupt or undecryptable file (e.g. moved between machines) is + // treated as "not connected" rather than crashing startup. + return null + } + }, + + save(creds: RemoteCredentials): void { + if (!env.isEncryptionAvailable()) { + throw new Error("secure storage is unavailable on this system, cannot save the bot token") + } + const file = env.credentialsFile() + mkdirSync(path.dirname(file), { recursive: true }) + const cipher = env.encryptString(JSON.stringify(creds)).toString("base64") + const envelope: Envelope = { version: FILE_VERSION, cipher } + writeFileSync(file, JSON.stringify(envelope), { mode: 0o600 }) + // writeFileSync's mode only applies when the file is created; rewriting an + // existing file keeps its old permissions. Re-assert 0o600 so a token file + // ever left world-readable is tightened on the next save. No-op on Windows, + // which has no POSIX mode bits. + chmodSync(file, 0o600) + }, + + clear(): void { + rmSync(env.credentialsFile(), { force: true }) + rmSync(env.stateFile(), { force: true }) + }, + } +} diff --git a/packages/desktop-electron/src/preload/index.ts b/packages/desktop-electron/src/preload/index.ts index 01185a75a..3b667ce5f 100644 --- a/packages/desktop-electron/src/preload/index.ts +++ b/packages/desktop-electron/src/preload/index.ts @@ -7,6 +7,19 @@ import { getRuntimeFlags } from "./runtime-flags" const runtimeFlags = getRuntimeFlags(process.env) const invokeSetDesktopContext = (context: DesktopContext) => ipcRenderer.invoke("set-desktop-context", context) +const remote: ElectronAPI["remote"] = { + getStatus: () => ipcRenderer.invoke("remote:get-status"), + startPairing: (token) => ipcRenderer.invoke("remote:start-pairing", token), + cancelPairing: () => ipcRenderer.invoke("remote:cancel-pairing"), + confirmPairing: () => ipcRenderer.invoke("remote:confirm-pairing"), + disconnect: () => ipcRenderer.invoke("remote:disconnect"), + onStatus: (cb) => { + const handler = (_: unknown, status: Parameters[0]) => cb(status) + ipcRenderer.on("remote:status", handler) + return () => ipcRenderer.removeListener("remote:status", handler) + }, +} + const browser: ElectronAPI["browser"] = { navigate: (target, url) => ipcRenderer.invoke("browser:navigate", target, url), goBack: (target) => ipcRenderer.invoke("browser:back", target), @@ -144,6 +157,7 @@ const api: ElectronAPI = { flashFrame: () => ipcRenderer.invoke("flash-frame"), setBadgeCount: (count: number) => ipcRenderer.invoke("set-badge-count", count), browser, + remote, } contextBridge.exposeInMainWorld("api", api) diff --git a/packages/desktop-electron/src/preload/types.ts b/packages/desktop-electron/src/preload/types.ts index 6c88258d9..28cdc3653 100644 --- a/packages/desktop-electron/src/preload/types.ts +++ b/packages/desktop-electron/src/preload/types.ts @@ -2,6 +2,7 @@ import type { AboutInfo, BrowserBridge, DesktopContext, + RemoteBridge, RendererDiagnosticInput, RendererDiagnosticsExportResult, ReportProblemInput, @@ -117,4 +118,5 @@ export type ElectronAPI = { flashFrame: () => Promise setBadgeCount: (count: number) => Promise browser: BrowserBridge + remote: RemoteBridge } diff --git a/packages/desktop-electron/tsconfig.json b/packages/desktop-electron/tsconfig.json index 9637fe03d..7802767cf 100644 --- a/packages/desktop-electron/tsconfig.json +++ b/packages/desktop-electron/tsconfig.json @@ -4,6 +4,7 @@ "module": "ESNext", "skipLibCheck": true, "moduleResolution": "bundler", + "allowImportingTsExtensions": true, "allowSyntheticDefaultImports": true, "esModuleInterop": true, "jsx": "preserve", diff --git a/packages/remote-bridge/src/engine.test.ts b/packages/remote-bridge/src/engine.test.ts index ec35d05ad..b4b74eed7 100644 --- a/packages/remote-bridge/src/engine.test.ts +++ b/packages/remote-bridge/src/engine.test.ts @@ -4,6 +4,7 @@ import { tmpdir } from "node:os" import { join } from "node:path" import { answersForQuestionText, deliveryConfig, Engine, questionPrompt } from "./engine.ts" import { SessionPointers } from "./session-pointers.ts" +import { PartialDeliveryError } from "./types.ts" import type { Message, PendingPermission, @@ -61,11 +62,15 @@ class FakePlatform implements Platform { reconstructKey = "" replyFailures = 0 replyCalls = 0 + // When set, reply throws a PartialDeliveryError instead of a plain transient + // error — used to assert the engine does NOT retry (resend) a partial delivery. + replyPartial = false constructor(readonly name = "chat") {} async start(): Promise {} async reply(_replyCtx: unknown, content: string): Promise { this.replyCalls++ + if (this.replyPartial) throw new PartialDeliveryError(new Error("a later chunk failed")) if (this.replyFailures > 0) { this.replyFailures-- throw new Error("transient delivery failure") @@ -341,7 +346,7 @@ test("surfaces a pending permission and answers it", async () => { await engine.handleMessage(platform, msg) await engine.handlePermission(perm({ id: "perm_1", sessionID: "ses_new", permission: "edit", patterns: ["/repo/app.ts"] })) - expect(lastReply(platform)).toBe("PawWork asks permission: edit\n/repo/app.ts\n\nReply yes, always, or no.") + expect(lastReply(platform)).toBe("PawWork needs your permission:\nedit\n/repo/app.ts\n\nReply yes (allow once), always (always allow), or no (deny).") msg.content = "always" await engine.handleMessage(platform, msg) @@ -357,7 +362,7 @@ test("routes a child session permission through the root conversation", async () await engine.registerSession(sessionInfo({ id: "child_1", parentID: "ses_new" })) await engine.handlePermission(perm({ id: "perm_child", sessionID: "child_1", permission: "edit", patterns: ["/repo/child.ts"] })) - expect(lastReply(platform)).toBe("PawWork asks permission: edit\n/repo/child.ts\n\nReply yes, always, or no.") + expect(lastReply(platform)).toBe("PawWork needs your permission:\nedit\n/repo/child.ts\n\nReply yes (allow once), always (always allow), or no (deny).") msg.content = "yes" await engine.handleMessage(platform, msg) @@ -433,7 +438,7 @@ test("surfaces a pending question with formatted options", async () => { ) expect(lastReply(platform)).toBe( - "Approach\nWhich path should I take?\n1. A - Small change\n2. B - Larger cleanup\n\nReply with a number or answer text.", + "Approach\nWhich path should I take?\n1. A - Small change\n2. B - Larger cleanup\n\nReply with the number of your choice.", ) }) @@ -478,6 +483,18 @@ test("assistant text retries a transient delivery failure, then gives up bounded expect(keepsFailing.replyCalls).toBe(deliveryConfig.attempts) }) +test("a partial delivery is surfaced without resending the whole message", async () => { + // When the platform reports it already delivered part of a multi-part message + // (PartialDeliveryError), retrying would duplicate the parts that arrived. The + // engine must surface it on the first failure, never resend. + const platform = new FakePlatform() + platform.replyPartial = true + const engine = new Engine(new FakeSidecar()) + await engine.handleMessage(platform, { sessionKey: "slack:dm:a", content: "hi" }) + await expect(engine.handleAssistantText("ses_new", "a long, split answer")).rejects.toBeInstanceOf(PartialDeliveryError) + expect(platform.replyCalls).toBe(1) +}) + test("permission and question prompts retry a transient delivery failure", async () => { const permPlatform = new FakePlatform() permPlatform.replyFailures = deliveryConfig.attempts - 1 @@ -591,7 +608,7 @@ test("a failed next blocker is kept and re-surfaced on recovery (question)", asy test("question prompt hints match the question type", () => { const single = questionPrompt(pendingQuestion({ sessionID: "s", questions: [question("Pick one", { options: [["A"], ["B"]] })] })) - expect(single.endsWith("Reply with a number or answer text.")).toBe(true) + expect(single.endsWith("Reply with the number of your choice.")).toBe(true) const multiSelect = questionPrompt( pendingQuestion({ sessionID: "s", questions: [question("Pick several", { multiple: true, options: [["A"], ["B"]] })] }), @@ -599,7 +616,10 @@ test("question prompt hints match the question type", () => { expect(multiSelect).toContain("separated by commas") const multiQuestion = questionPrompt(pendingQuestion({ sessionID: "s", questions: [question("First?"), question("Second?")] })) - expect(multiQuestion).toContain("one line per question") + expect(multiQuestion).toContain("on its own line") + // Multiple questions are numbered so "one answer per line" maps onto them. + expect(multiQuestion).toContain("Question 1:") + expect(multiQuestion).toContain("Question 2:") }) test("multi-select accepts full-width and ideographic commas", () => { @@ -609,6 +629,38 @@ test("multi-select accepts full-width and ideographic commas", () => { } }) +test("renders prompts and accepts replies in Chinese", async () => { + const sidecar = new FakeSidecar() + const platform = new FakePlatform() + const engine = new Engine(sidecar, undefined, "zh") + const msg: Message = { sessionKey: "telegram:chat:alice", content: "开始" } + await engine.handleMessage(platform, msg) + + await engine.handlePermission(perm({ id: "perm_1", sessionID: "ses_new", permission: "edit", patterns: ["/repo/app.ts"] })) + const reply = lastReply(platform) + expect(reply).toContain("爪印需要你的许可:") + expect(reply).toContain("/repo/app.ts") + expect(reply).toContain("允许一次") + + // A Chinese reply keyword resolves the permission, even though the rendered locale is zh. + msg.content = "是" + await engine.handleMessage(platform, msg) + expect(sidecar.permissionReplies[0].reply.reply).toBe("once") +}) + +test("localizes question numbering, hints, and the line-count error in Chinese", () => { + const zh = questionPrompt( + pendingQuestion({ sessionID: "s", questions: [question("第一题"), question("第二题")] }), + "zh", + ) + expect(zh).toContain("问题 1:") + expect(zh).toContain("问题 2:") + expect(zh).toContain("每个问题回复一行") + + const pending = pendingQuestion({ sessionID: "s", questions: [question("A"), question("B")] }) + expect(() => answersForQuestionText(pending, "只有一行", "zh")).toThrow("共 2 个问题") +}) + test("answers pending questions in arrival order, one at a time", async () => { const sidecar = new FakeSidecar() const platform = new FakePlatform() @@ -652,7 +704,7 @@ test("surfaces interleaved blockers one at a time", async () => { pendingQuestion({ sessionID: "ses_new", messageID: "msg_1", callID: "call_1", questions: [question("Pick one", { options: [["A"], ["B"]] })] }), ) expect(platform.replies).toHaveLength(1) - expect(platform.replies[0]).toContain("asks permission") + expect(platform.replies[0]).toContain("needs your permission") msg.content = "yes" await engine.handleMessage(platform, msg) diff --git a/packages/remote-bridge/src/engine.ts b/packages/remote-bridge/src/engine.ts index 25174b1ac..f9675a148 100644 --- a/packages/remote-bridge/src/engine.ts +++ b/packages/remote-bridge/src/engine.ts @@ -1,5 +1,7 @@ import type { EventHandler } from "./pawwork-events.ts" +import { t, type Locale } from "./i18n.ts" import { SessionPointers as SessionPointersStore } from "./session-pointers.ts" +import { PartialDeliveryError } from "./types.ts" import type { Message, PendingPermission, @@ -41,6 +43,7 @@ export class Engine implements EventHandler { constructor( private readonly sidecar: Sidecar, private readonly pointers: SessionPointers = SessionPointersStore.memory(), + private readonly locale: Locale = "en", ) {} currentSession(remoteKey: string): string { @@ -67,7 +70,7 @@ export class Engine implements EventHandler { try { sessionID = await this.ensureSession(key) } catch (err) { - await replyQuietly(platform, msg.replyCtx, "PawWork could not start a session: " + message(err)) + await replyQuietly(platform, msg.replyCtx, t(this.locale, "err.startSession") + message(err)) throw err } @@ -81,7 +84,7 @@ export class Engine implements EventHandler { try { await this.sidecar.sendPrompt(sessionID, text) } catch (err) { - await replyQuietly(platform, msg.replyCtx, "PawWork could not send the message: " + message(err)) + await replyQuietly(platform, msg.replyCtx, t(this.locale, "err.sendMessage") + message(err)) throw err } } @@ -177,13 +180,13 @@ export class Engine implements EventHandler { if (blocker.kind === "permission") { const reply = permissionReplyForText(text) if (reply === "") { - await platform.reply(msg.replyCtx, "Reply yes, always, or no.") + await platform.reply(msg.replyCtx, t(this.locale, "permission.notUnderstoodPrefix") + t(this.locale, "permission.replyHint")) return true } try { await this.sidecar.replyPermission(blocker.permission, { reply, message: "" }) } catch (err) { - await replyQuietly(platform, msg.replyCtx, "PawWork could not answer the permission request: " + message(err)) + await replyQuietly(platform, msg.replyCtx, t(this.locale, "err.answerPermission") + message(err)) throw err } this.clearPendingPermission(blocker.permission) @@ -193,7 +196,7 @@ export class Engine implements EventHandler { // question blocker let answers: string[][] try { - answers = answersForQuestionText(blocker.question, text) + answers = answersForQuestionText(blocker.question, text, this.locale) } catch (err) { await platform.reply(msg.replyCtx, message(err)) return true @@ -201,7 +204,7 @@ export class Engine implements EventHandler { try { await this.sidecar.submitQuestion(blocker.question, answers) } catch (err) { - await replyQuietly(platform, msg.replyCtx, "PawWork could not submit the answer: " + message(err)) + await replyQuietly(platform, msg.replyCtx, t(this.locale, "err.submitAnswer") + message(err)) throw err } this.clearPendingQuestion(blocker.question) @@ -250,10 +253,10 @@ export class Engine implements EventHandler { if (candidate.delivered) return null if (candidate.kind === "permission") { const permission = this.permissions.get(candidate.key)! - return { ref: candidate, sessionID: permission.sessionID, content: permissionPrompt(permission) } + return { ref: candidate, sessionID: permission.sessionID, content: permissionPrompt(permission, this.locale) } } const question = this.questions.get(candidate.key)! - return { ref: candidate, sessionID: question.sessionID, content: questionPrompt(question) } + return { ref: candidate, sessionID: question.sessionID, content: questionPrompt(question, this.locale) } } return null } @@ -367,17 +370,17 @@ export class Engine implements EventHandler { try { sessionID = await this.sidecar.createSession() } catch (err) { - await replyQuietly(platform, msg.replyCtx, "PawWork could not start a session: " + message(err)) + await replyQuietly(platform, msg.replyCtx, t(this.locale, "err.startSession") + message(err)) throw err } try { await this.setCurrent(key, sessionID) } catch (err) { - await replyQuietly(platform, msg.replyCtx, "PawWork could not remember the session: " + message(err)) + await replyQuietly(platform, msg.replyCtx, t(this.locale, "err.rememberSession") + message(err)) throw err } this.setActive(sessionID, platform, msg.replyCtx) - await platform.reply(msg.replyCtx, "Started a new PawWork session.") + await platform.reply(msg.replyCtx, t(this.locale, "cmd.newSession")) return true } case "/sessions": { @@ -389,21 +392,21 @@ export class Engine implements EventHandler { case "/stop": { const sessionID = this.currentSession(key) if (sessionID === "") { - await platform.reply(msg.replyCtx, "No active PawWork session.") + await platform.reply(msg.replyCtx, t(this.locale, "cmd.noActiveSession")) return true } let aborted: boolean try { aborted = await this.sidecar.abortSession(sessionID) } catch (err) { - await replyQuietly(platform, msg.replyCtx, "PawWork could not stop the run: " + message(err)) + await replyQuietly(platform, msg.replyCtx, t(this.locale, "err.stopRun") + message(err)) throw err } - await platform.reply(msg.replyCtx, aborted ? "Stopped the current PawWork run." : "No running PawWork run.") + await platform.reply(msg.replyCtx, aborted ? t(this.locale, "cmd.stopped") : t(this.locale, "cmd.noRunning")) return true } case "/help": - await platform.reply(msg.replyCtx, "Commands: /new, /sessions, /sessions N, /stop.") + await platform.reply(msg.replyCtx, t(this.locale, "cmd.help")) return true default: return false @@ -440,16 +443,16 @@ export class Engine implements EventHandler { try { sessions = await this.sidecar.listSessions(5) } catch (err) { - await replyQuietly(platform, msg.replyCtx, "PawWork could not list sessions: " + message(err)) + await replyQuietly(platform, msg.replyCtx, t(this.locale, "err.listSessions") + message(err)) throw err } if (sessions.length === 0) { - await platform.reply(msg.replyCtx, "No recent PawWork sessions.") + await platform.reply(msg.replyCtx, t(this.locale, "cmd.noRecent")) return } - const lines = ["Recent PawWork sessions:"] + const lines = [t(this.locale, "cmd.recentSessions")] sessions.forEach((session, index) => lines.push(`${index + 1}. ${sessionLabel(session)}`)) - await platform.reply(msg.replyCtx, lines.join("\n") + "\n\nSwitch with /sessions 2.") + await platform.reply(msg.replyCtx, lines.join("\n") + "\n\n" + t(this.locale, "cmd.switchHint")) } private async switchSession(platform: Platform, msg: Message, key: string, rawIndex: string): Promise { @@ -457,7 +460,7 @@ export class Engine implements EventHandler { // and "2.5" are rejected, not silently truncated to 2. const index = /^[+-]?\d+$/.test(rawIndex) ? Number.parseInt(rawIndex, 10) : Number.NaN if (Number.isNaN(index) || index < 1) { - await platform.reply(msg.replyCtx, "Choose a session with /sessions 1.") + await platform.reply(msg.replyCtx, t(this.locale, "cmd.chooseHint")) return } // Fetch the current list rather than trusting a cached picker: between @@ -467,11 +470,11 @@ export class Engine implements EventHandler { try { sessions = await this.sidecar.listSessions(5) } catch (err) { - await replyQuietly(platform, msg.replyCtx, "PawWork could not list sessions: " + message(err)) + await replyQuietly(platform, msg.replyCtx, t(this.locale, "err.listSessions") + message(err)) throw err } if (index > sessions.length) { - await platform.reply(msg.replyCtx, `Only ${sessions.length} recent PawWork sessions are available.`) + await platform.reply(msg.replyCtx, t(this.locale, "cmd.onlyN", { n: sessions.length })) return } const session = sessions[index - 1] @@ -479,11 +482,11 @@ export class Engine implements EventHandler { await this.registerSession(session) await this.setCurrent(key, session.id) } catch (err) { - await replyQuietly(platform, msg.replyCtx, "PawWork could not remember the session: " + message(err)) + await replyQuietly(platform, msg.replyCtx, t(this.locale, "err.rememberSession") + message(err)) throw err } this.setActive(session.id, platform, msg.replyCtx) - await platform.reply(msg.replyCtx, "Switched to " + sessionLabel(session) + ".") + await platform.reply(msg.replyCtx, t(this.locale, "cmd.switchedTo", { x: sessionLabel(session) })) } private setCurrent(remoteKey: string, sessionID: string): Promise { @@ -533,6 +536,10 @@ async function sendDeliveryWithRetry(target: Delivery, content: string): Promise return } catch (err) { lastError = err + // Part of a multi-part message already reached the user; resending the whole + // payload would duplicate what arrived, so surface it without retrying. The + // platform retried the failed chunk in place before giving up. + if (err instanceof PartialDeliveryError) throw err if (attempt < deliveryConfig.attempts && deliveryConfig.backoffMs > 0) { await delay(attempt * deliveryConfig.backoffMs) } @@ -547,21 +554,29 @@ function sessionLabel(session: Session): string { return session.title.trim() !== "" ? session.title : session.id } -function permissionPrompt(permission: PendingPermission): string { - const lines = ["PawWork asks permission: " + permission.permission] +function permissionPrompt(permission: PendingPermission, locale: Locale): string { + const lines = [t(locale, "permission.title")] + if (permission.permission.trim() !== "") lines.push(permission.permission) for (const pattern of permission.patterns) { if (pattern.trim() === "") continue lines.push(pattern) } - return lines.join("\n") + "\n\nReply yes, always, or no." + return lines.join("\n") + "\n\n" + t(locale, "permission.replyHint") } -export function questionPrompt(pending: PendingQuestion): string { - if (pending.questions.length === 0) return "PawWork asks a question.\n\nReply with your answer." +export function questionPrompt(pending: PendingQuestion, locale: Locale = "en"): string { + if (pending.questions.length === 0) return t(locale, "question.fallback") + const multiQuestion = pending.questions.length > 1 const blocks: string[] = [] - for (const question of pending.questions) { + pending.questions.forEach((question, qIndex) => { const lines: string[] = [] - if (question.header.trim() !== "") lines.push(question.header) + // Number each question when there are several, so "one answer per line" maps + // visibly onto them. The label is composed here, not parsed from the backend + // header — engine owns the localized "Question N" prefix, header stays as sent. + const label = multiQuestion ? t(locale, "question.label", { n: qIndex + 1 }) : "" + const header = question.header.trim() + if (label && header) lines.push(`${label} ${header}`) + else if (label || header) lines.push(label || header) lines.push(question.question) question.options.forEach((option, index) => { let line = `${index + 1}. ${option.label}` @@ -569,52 +584,64 @@ export function questionPrompt(pending: PendingQuestion): string { lines.push(line) }) blocks.push(lines.join("\n")) - } - return blocks.join("\n\n") + "\n\n" + questionReplyHint(pending.questions) + }) + return blocks.join("\n\n") + "\n\n" + questionReplyHint(pending.questions, locale) } -function questionReplyHint(questions: Question[]): string { +function questionReplyHint(questions: Question[], locale: Locale): string { const multiQuestion = questions.length > 1 const multiSelect = questions.some((question) => question.multiple) - if (multiQuestion && multiSelect) { - return "Reply with one line per question, in order. For a question that allows several choices, separate the numbers with commas (for example: 1,3)." - } - if (multiQuestion) { - return "Reply with one line per question, in order: a number or the answer text on each line." - } - if (multiSelect) { - return "Reply with the numbers separated by commas (for example: 1,3)." - } - return "Reply with a number or answer text." + if (multiQuestion && multiSelect) return t(locale, "hint.multiQuestionMulti") + if (multiQuestion) return t(locale, "hint.multiQuestion") + if (multiSelect) return t(locale, "hint.singleMulti") + return t(locale, "hint.single") } function permissionReplyForText(text: string): string { + // Accepts both English and Chinese keywords regardless of the rendered locale, + // so a reply works even if the user types in the other language. toLowerCase is + // a no-op on Chinese, harmless here. switch (text.trim().toLowerCase()) { case "yes": case "y": case "allow": case "ok": + case "是": + case "好": + case "好的": + case "允许": + case "同意": + case "可以": return "once" case "always": case "always allow": + case "总是": + case "一直": + case "始终": + case "总是允许": return "always" case "no": case "n": case "deny": case "reject": + case "否": + case "不": + case "不行": + case "拒绝": + case "不要": return "reject" default: return "" } } -export function answersForQuestionText(pending: PendingQuestion, text: string): string[][] { +export function answersForQuestionText(pending: PendingQuestion, text: string, locale: Locale = "en"): string[][] { const questions = pending.questions if (questions.length === 0) return [[text]] if (questions.length === 1) return [answerRowForQuestion(text, questions[0])] const lines = text.trim().split("\n") if (lines.length !== questions.length) { - throw new Error(`Reply with ${questions.length} lines, one answer per question.`) + throw new Error(t(locale, "answers.lineMismatch", { n: questions.length })) } return lines.map((line, index) => answerRowForQuestion(line, questions[index])) } diff --git a/packages/remote-bridge/src/gateway.test.ts b/packages/remote-bridge/src/gateway.test.ts index 611386855..6dd48770b 100644 --- a/packages/remote-bridge/src/gateway.test.ts +++ b/packages/remote-bridge/src/gateway.test.ts @@ -42,17 +42,37 @@ class FakePlatform implements Platform { sends: string[] = [] reconstructKey = "" startedAfterStream = false + readyCalls = 0 + fireReady: () => void = () => {} readonly started = deferred() private readonly stopped = deferred() constructor( readonly name: string, - private readonly opts: { sendErr?: Error; streamConnected?: () => boolean; startResolvesImmediately?: boolean } = {}, + private readonly opts: { + sendErr?: Error + streamConnected?: () => boolean + startResolvesImmediately?: boolean + readyMode?: "auto" | "double" | "manual" + } = {}, ) {} - async start(_handler: MessageHandler): Promise { + async start(_handler: MessageHandler, onReady?: () => void): Promise { this.startedAfterStream = this.opts.streamConnected ? this.opts.streamConnected() : false this.started.resolve() + // A real platform signals readiness once it is past startup and serving; model + // that here so the gateway's run-level onReady fires in tests. "double" models a + // misbehaving adapter; "manual" lets a test drive the moment via fireReady. + const fire = () => { + this.readyCalls++ + onReady?.() + } + const mode = this.opts.readyMode ?? "auto" + if (mode === "manual") this.fireReady = fire + else { + fire() + if (mode === "double") fire() + } // An event-driven adapter may register its callback and return right away; // model that with startResolvesImmediately instead of blocking until stop(). if (this.opts.startResolvesImmediately) return @@ -254,7 +274,7 @@ test("hydrate resurfaces a pending interaction through the restored target", asy // Both pending items share root ses_root; single-active surfacing shows only // the permission, leaving the question queued until it is answered. expect(platform.sends).toHaveLength(1) - expect(platform.sends[0]).toContain("PawWork asks permission: edit") + expect(platform.sends[0]).toContain("PawWork needs your permission:") } finally { server.stop() } @@ -404,6 +424,101 @@ test("run stays up when a platform start resolves on its own", async () => { } }) +test("run fires onReady only after the stream, hydrate, and platform are serving", async () => { + // The startup-race guard: a caller's "connected" must trail real readiness, so + // run's onReady cannot precede the stream, the hydrate, or the platform start. + const order: string[] = [] + const platform = new FakePlatform("runtime-test-onready-after-serving") + const server = mockServer((_req, url) => { + switch (url.pathname) { + case "/experimental/session": + order.push("hydrate") + return jsonBody([]) + case "/permission": + case "/external-result": + return jsonBody([]) + case "/global/event": + order.push("stream") + return openEventStream() + } + return undefined + }) + const controller = new AbortController() + try { + const app = await createApp( + { + pawWorkBaseURL: server.url, + statePath: await tempStatePath(), + platforms: [{ name: "runtime-test-onready-after-serving", enabled: true, options: { allow_from: "U123" } }], + }, + () => platform, + ) + const ready = deferred() + const runPromise = app.run(controller.signal, () => { + order.push("ready") + ready.resolve() + }) + await ready.promise + expect(order.indexOf("ready")).toBeGreaterThan(order.indexOf("stream")) + expect(order.indexOf("ready")).toBeGreaterThan(order.indexOf("hydrate")) + controller.abort() + await runPromise + } finally { + server.stop() + } +}) + +test("run's onReady counts each platform once, even when an adapter double-fires", async () => { + // Supervisor hardening: the contract says a platform signals readiness a single + // time, but one adapter firing onReady twice must not satisfy the whole set — the + // bridge stays "connecting" until every platform is actually serving. + let readyFired = 0 + const doubleFirer = new FakePlatform("runtime-test-double-ready", { readyMode: "double" }) + const lateComer = new FakePlatform("runtime-test-late-ready", { readyMode: "manual" }) + const server = mockServer((_req, url) => { + switch (url.pathname) { + case "/experimental/session": + case "/permission": + case "/external-result": + return jsonBody([]) + case "/global/event": + return openEventStream() + } + return undefined + }) + const controller = new AbortController() + try { + const app = await createApp( + { + pawWorkBaseURL: server.url, + statePath: await tempStatePath(), + platforms: [ + { name: "runtime-test-double-ready", enabled: true, options: { allow_from: "U123" } }, + { name: "runtime-test-late-ready", enabled: true, options: { allow_from: "U456" } }, + ], + }, + (name) => (name === "runtime-test-double-ready" ? doubleFirer : lateComer), + ) + const runPromise = app.run(controller.signal, () => { + readyFired++ + }) + // Both platforms have started: the first fired onReady twice, the second has not + // fired yet. The two duplicate fires must not stand in for the missing platform. + await doubleFirer.started.promise + await lateComer.started.promise + await waitUntil(() => doubleFirer.readyCalls === 2) + expect(readyFired).toBe(0) + // The second platform serving completes the set — run's onReady fires exactly once. + lateComer.fireReady() + await waitUntil(() => readyFired === 1) + expect(readyFired).toBe(1) + controller.abort() + await runPromise + } finally { + server.stop() + } +}) + test("run connects the event stream before the initial hydrate", async () => { const order: string[] = [] const platform = new FakePlatform("runtime-test-stream-before-hydrate") diff --git a/packages/remote-bridge/src/gateway.ts b/packages/remote-bridge/src/gateway.ts index 02615d566..ddd930d96 100644 --- a/packages/remote-bridge/src/gateway.ts +++ b/packages/remote-bridge/src/gateway.ts @@ -1,5 +1,9 @@ import { readFile } from "node:fs/promises" import { Engine } from "./engine.ts" +import { normalizeLocale, type Locale } from "./i18n.ts" + +// Re-exported so the desktop runtime can normalize its UI locale into Config.locale. +export { normalizeLocale } import { isFatalStreamError, PawWorkClient } from "./pawwork-client.ts" import type { EventHandler } from "./pawwork-events.ts" import { SessionPointers } from "./session-pointers.ts" @@ -16,6 +20,8 @@ export interface Config { pawWorkUsername?: string pawWorkPassword?: string statePath: string + /** Language for the chat-facing copy; defaults to English when unset. */ + locale?: Locale platforms: PlatformConfig[] } @@ -65,7 +71,7 @@ export async function createApp(config: Config, createPlatform: PlatformFactory) password: config.pawWorkPassword, }) client.setEventCursorStore(pointers) - const engine = new Engine(client, pointers) + const engine = new Engine(client, pointers, normalizeLocale(config.locale)) const platforms: Platform[] = [] for (const item of config.platforms ?? []) { @@ -113,7 +119,7 @@ export class App { * before pending state is loaded. Returns when `signal` aborts; rejects on a * fatal stream error or a platform failure. Ported from Go `App.Run`. */ - async run(signal?: AbortSignal): Promise { + async run(signal?: AbortSignal, onReady?: () => void): Promise { const ac = new AbortController() if (signal?.aborted) ac.abort() const onParentAbort = () => ac.abort() @@ -148,7 +154,21 @@ export class App { // Like Go's Run, stay up until abort or a fatal error even if every // platform's start() resolves on its own — a clean self-stop is not a reason // to tear the bridge down. - this.startPlatforms(childSignal, failure) + // + // Fire onReady only once every platform has drained its backlog and is + // serving, so a caller's "connected" can't precede live message delivery. + const total = this.platforms.length + let readyCount = 0 + const allReady = createDeferred() + if (total === 0) allReady.resolve() + const onPlatformReady = () => { + if (++readyCount >= total) allReady.resolve() + } + void Promise.race([allReady.promise, onAbort(childSignal)]).then(() => { + if (!childSignal.aborted) onReady?.() + }) + + this.startPlatforms(childSignal, failure, onPlatformReady) await Promise.race([onAbort(childSignal), failure.promise]) } catch (err) { // An abort is a requested stop, not a failure: any error it triggered @@ -243,18 +263,32 @@ export class App { } } - private startPlatforms(signal: AbortSignal, failure: Deferred): Promise[] { + private startPlatforms( + signal: AbortSignal, + failure: Deferred, + onPlatformReady?: () => void, + ): Promise[] { const handler = this.messageHandler() - return this.platforms.map((platform) => - Promise.resolve() - .then(() => platform.start(handler)) + return this.platforms.map((platform) => { + // Count each platform toward "all ready" at most once. The contract says a + // platform signals onReady a single time, but a misbehaving adapter that + // double-fires must not let the bridge report "connected" before every + // platform is actually serving. + let counted = false + const ready = () => { + if (counted) return + counted = true + onPlatformReady?.() + } + return Promise.resolve() + .then(() => platform.start(handler, ready)) .then( () => {}, (err) => { if (!signal.aborted) failure.reject(new Error(`${platform.name} platform failed: ${message(err)}`)) }, - ), - ) + ) + }) } private async stopPlatforms(): Promise { diff --git a/packages/remote-bridge/src/i18n.test.ts b/packages/remote-bridge/src/i18n.test.ts new file mode 100644 index 000000000..5b5cb29da --- /dev/null +++ b/packages/remote-bridge/src/i18n.test.ts @@ -0,0 +1,16 @@ +import { expect, test } from "bun:test" +import { normalizeLocale, t } from "./i18n.ts" + +test("t renders per locale and substitutes {params}", () => { + expect(t("en", "cmd.switchedTo", { x: "Foo" })).toBe("Switched to Foo.") + expect(t("zh", "cmd.switchedTo", { x: "Foo" })).toBe("已切换到 Foo。") + expect(t("zh", "cmd.onlyN", { n: 3 })).toBe("仅有 3 个近期会话。") +}) + +test("normalizeLocale coerces to a supported locale, defaulting to en", () => { + expect(normalizeLocale("zh")).toBe("zh") + expect(normalizeLocale("en")).toBe("en") + expect(normalizeLocale(undefined)).toBe("en") + expect(normalizeLocale(null)).toBe("en") + expect(normalizeLocale("fr")).toBe("en") +}) diff --git a/packages/remote-bridge/src/i18n.ts b/packages/remote-bridge/src/i18n.ts new file mode 100644 index 000000000..21cd56a0b --- /dev/null +++ b/packages/remote-bridge/src/i18n.ts @@ -0,0 +1,94 @@ +// User-facing copy for the chat protocol, in the paired user's language. This is +// a personal companion — one paired user — so the bridge renders everything in a +// single locale (the desktop UI language), not per-message. Plain tables + a tiny +// t(): no i18n framework for ~25 strings. English keeps the product name +// "PawWork"; Chinese uses "爪印" (the app's localized name) and drops it where +// chat context already makes the sender obvious (command replies), keeping it only +// where a sentence needs a subject (permission / question / errors). + +export type Locale = "en" | "zh" + +const en = { + "permission.title": "PawWork needs your permission:", + "permission.replyHint": "Reply yes (allow once), always (always allow), or no (deny).", + "permission.notUnderstoodPrefix": "Sorry, I didn't catch that. ", + "question.fallback": "PawWork has a question.\n\nReply with your answer.", + "question.label": "Question {n}:", + "hint.single": "Reply with the number of your choice.", + "hint.singleMulti": "Reply with the numbers you want, separated by commas (e.g. 1,3).", + "hint.multiQuestion": "Answer each question on its own line, in order.", + "hint.multiQuestionMulti": + "Answer each question on its own line, in order. For a multiple-choice question, separate your picks with commas (e.g. 1,3).", + "answers.lineMismatch": "There are {n} questions — reply with {n} lines, one answer per question.", + "cmd.newSession": "Started a new PawWork session.", + "cmd.noActiveSession": "No active PawWork session.", + "cmd.stopped": "Stopped the current PawWork run.", + "cmd.noRunning": "No running PawWork run.", + "cmd.help": "Commands: /new, /sessions, /sessions N, /stop.", + "cmd.recentSessions": "Recent PawWork sessions:", + "cmd.switchHint": "Switch with /sessions 2.", + "cmd.chooseHint": "Choose a session with /sessions 1.", + "cmd.onlyN": "Only {n} recent PawWork sessions are available.", + "cmd.switchedTo": "Switched to {x}.", + "cmd.noRecent": "No recent PawWork sessions.", + "err.startSession": "PawWork could not start a session: ", + "err.sendMessage": "PawWork could not send the message: ", + "err.answerPermission": "PawWork could not answer the permission request: ", + "err.submitAnswer": "PawWork could not submit the answer: ", + "err.stopRun": "PawWork could not stop the run: ", + "err.listSessions": "PawWork could not list sessions: ", + "err.rememberSession": "PawWork could not remember the session: ", +} as const + +export type MessageKey = keyof typeof en + +const zh: Record = { + "permission.title": "爪印需要你的许可:", + "permission.replyHint": "回复“是”允许一次、“总是”始终允许、“否”拒绝。", + "permission.notUnderstoodPrefix": "没看懂你的回复。", + "question.fallback": "爪印向你提问。\n\n直接回复你的答案。", + "question.label": "问题 {n}:", + "hint.single": "回复选项编号即可。", + "hint.singleMulti": "回复你选的编号,用逗号隔开(如 1,3)。", + "hint.multiQuestion": "每个问题回复一行,按顺序作答。", + "hint.multiQuestionMulti": "每个问题回复一行,按顺序作答。多选问题用逗号隔开(如 1,3)。", + "answers.lineMismatch": "共 {n} 个问题,请回复 {n} 行,每行一个答案。", + "cmd.newSession": "已新建会话。", + "cmd.noActiveSession": "没有进行中的会话。", + "cmd.stopped": "已停止当前运行。", + "cmd.noRunning": "没有正在运行的任务。", + "cmd.help": "命令:/new 新建会话,/sessions 查看会话,/sessions N 切换会话,/stop 停止运行。", + "cmd.recentSessions": "近期会话:", + "cmd.switchHint": "用 /sessions 2 切换会话。", + "cmd.chooseHint": "用 /sessions 1 选择会话。", + "cmd.onlyN": "仅有 {n} 个近期会话。", + "cmd.switchedTo": "已切换到 {x}。", + "cmd.noRecent": "没有近期会话。", + "err.startSession": "爪印无法启动会话:", + "err.sendMessage": "爪印无法发送消息:", + "err.answerPermission": "爪印无法回应权限请求:", + "err.submitAnswer": "爪印无法提交答案:", + "err.stopRun": "爪印无法停止运行:", + "err.listSessions": "爪印无法列出会话:", + "err.rememberSession": "爪印无法记住会话:", +} + +const tables: Record> = { en, zh } + +/** Render a message in `locale`, substituting `{name}` placeholders. Falls back to + * English for an unknown locale or a missing key, so a copy gap degrades to English + * rather than showing a raw key. */ +export function t(locale: Locale, key: MessageKey, params?: Record): string { + let text = (tables[locale] ?? en)[key] ?? en[key] + if (params) { + for (const [name, value] of Object.entries(params)) { + text = text.replaceAll(`{${name}}`, String(value)) + } + } + return text +} + +/** Coerce any config value to a supported locale, defaulting to English. */ +export function normalizeLocale(value: unknown): Locale { + return value === "zh" ? "zh" : "en" +} diff --git a/packages/remote-bridge/src/platforms/README.md b/packages/remote-bridge/src/platforms/README.md new file mode 100644 index 000000000..11e65d753 --- /dev/null +++ b/packages/remote-bridge/src/platforms/README.md @@ -0,0 +1,84 @@ +# Chat platform adapters + +Each file here is one **chat platform adapter** that lets a chat app on your phone +drive a PawWork agent session (send prompts, read replies, answer permission / +question prompts). An adapter implements the `Platform` seam in +[`../types.ts`](../types.ts) so the engine stays decoupled from any one chat SDK. + +**[`telegram.ts`](./telegram.ts) is the reference implementation — read it first.** + +## The contract + +A platform is one object implementing `Platform` (`../types.ts`): + +- **`name`** — stable prefix used in the remote key `::`. + Mandatory: the engine routes restored deliveries by `key.split(":")[0]`, so the + prefix must match the adapter's `name`. +- **`start(handler)`** — open the inbound transport, normalize each inbound message + to `Message`, and call `handler(this, msg)`. Resolves only when stopped; **reject + on a fatal error** (bad/revoked credential) so the runtime can surface `degraded`. +- **`reply(replyCtx, content)` / `send(replyCtx, content)`** — answer in-thread / + push proactively to a restored target. +- **`stop()`** — idempotent teardown. +- **`reconstructReplyCtx?(remoteKey)`** — rebuild a reply target after a restart + with no live inbound message, so deliveries survive relaunch. Throw if the key is + unparseable. + +Inbound messages are normalized to `Message { content, replyCtx?, channelID?, +userID?, sessionKey? }`. The engine derives the remote key from `name + channelID + +userID` — **do not hand-build session keys.** + +Adapters are constructed by the `PlatformFactory` and wired in `createApp` +(`../gateway.ts`). Before a platform starts, **`hasRemoteAudience` refuses a +wildcard/empty audience — closed by default.** Telegram requires `allow_from`; +Feishu/Lark additionally require a named group (`allow_chat` + `group_only`). + +## Interactive prompts: plain text + +Question and permission blockers are rendered to **plain text** and answered by the +user **typing** — a number / `yes` / `no`, and for a multi-question prompt one answer +per line (see `questionPrompt` / `answersForQuestionText` in [`../engine.ts`](../engine.ts)). +`reply` / `send` are deliberately text-only, so a new adapter works the moment it can +send a string. + +The cost is a clumsy answer UX: the multi-question hint asks for newline-separated +lines, but on a phone Enter _sends_ the message, so newlines are awkward to type. +Native tap-to-answer controls (Telegram inline keyboards, Feishu cards, Discord +components) are a deferred optimization — design and staging live in +[#1188](https://github.com/Astro-Han/pawwork/issues/1188). + +## Conventions every adapter must follow + +These are the lessons distilled from the most-supported agents (Hermes Agent, +OpenClaw). Follow them or you will ship the bugs they already fixed: + +1. **Outbound-only, no public IP.** This is a desktop app behind NAT. Use the + platform's long-poll / outbound-WebSocket / vendor stream mode. **Never require + an inbound webhook.** Webhook-only platforms need a hosted relay (tracked in #1188). +2. **Dedup by message id.** Long-connections redeliver on reconnect. Keep a + TTL-bounded seen-set and drop duplicates before dispatching. +3. **Reconnect with exponential backoff + a stall watchdog.** Assume the socket + drops; force-restart after prolonged silence. +4. **Classify errors.** Fatal (401/403, revoked token) → reject `start()`. + Transient (429, 5xx, network) → back off and retry. +5. **One credential, one inbound loop.** Never run two loops on the same credential + concurrently (e.g. Telegram `getUpdates` returns 409). Pairing capture and the + live bridge hand off via the ack offset; they never overlap. +6. **Credentials stay in the main process** via Electron `safeStorage`. The renderer + only ever sees masked status — never the token. + +## SDK policy + +Raw API where the protocol is simple; an **official per-platform SDK only where it +owns the reconnect / long-connection**. Never adopt an all-in-one meta-SDK (Vercel +Chat SDK, `@chat-adapter/*`) — immature and lock-in. Per-adapter SDK deps are fine +(lazy-load heavy ones); `remote-bridge`'s core stays dependency-free. + +## Planned platforms + +Telegram ships here. The rollout order for the platforms after it (Feishu, WeChat, +Discord, Slack, DingTalk, WeCom, WhatsApp), the source-verified outbound-only +transport for each (every priority platform has a no-public-IP path), and the external +references (Hermes Agent, OpenClaw, Tencent iLink) live in +[#1188](https://github.com/Astro-Han/pawwork/issues/1188). Adding a platform is a pure +adapter against the contract above — no engine change. diff --git a/packages/remote-bridge/src/platforms/telegram.test.ts b/packages/remote-bridge/src/platforms/telegram.test.ts new file mode 100644 index 000000000..6b5250e31 --- /dev/null +++ b/packages/remote-bridge/src/platforms/telegram.test.ts @@ -0,0 +1,560 @@ +import { expect, test } from "bun:test" +import { PartialDeliveryError, type MessageHandler, type Platform } from "../types.ts" +import { + captureFirstSender, + inboundMessage, + isFatalTelegramError, + MAX_CONFLICT_RETRIES, + normalizeUpdate, + parseTelegramRemoteKey, + splitForTelegram, + TelegramApiError, + TelegramPlatform, + TelegramPoller, +} from "./telegram.ts" + +// --- pure helpers ---------------------------------------------------------- + +test("splitForTelegram returns short text untouched, with no header", () => { + expect(splitForTelegram("hello")).toEqual(["hello"]) +}) + +test("splitForTelegram splits over the 4096-unit cap and headers each piece", () => { + const long = "a".repeat(9000) + const chunks = splitForTelegram(long) + expect(chunks.length).toBeGreaterThan(1) + expect(chunks[0].startsWith(`[1/${chunks.length}]\n`)).toBe(true) + // Each chunk's UTF-16 length stays within Telegram's cap. + for (const chunk of chunks) expect(chunk.length).toBeLessThanOrEqual(4096) + // Reassembling the bodies (minus headers) restores the original text. + const body = chunks.map((c) => c.replace(/^\[\d+\/\d+\]\n/, "")).join("") + expect(body).toBe(long) +}) + +test("splitForTelegram preserves a newline that falls on a split boundary", () => { + // Long enough to split, with a newline in the last 10% of the first chunk so + // the line-boundary break fires. The delimiter must survive: stripping headers + // and concatenating the bodies must reproduce the original exactly. + const text = "a".repeat(3980) + "\n" + "b".repeat(4000) + const chunks = splitForTelegram(text) + expect(chunks.length).toBeGreaterThan(1) + for (const chunk of chunks) expect(chunk.length).toBeLessThanOrEqual(4096) + const body = chunks.map((c) => c.replace(/^\[\d+\/\d+\]\n/, "")).join("") + expect(body).toBe(text) +}) + +test("splitForTelegram never splits a surrogate pair", () => { + const emoji = "😀".repeat(3000) // each emoji is 2 UTF-16 units + const chunks = splitForTelegram(emoji) + for (const chunk of chunks) { + const body = chunk.replace(/^\[\d+\/\d+\]\n/, "") + // A split surrogate would leave a lone code unit (\uD83D or \uDE00). + expect(/[\uD800-\uDBFF](?![\uDC00-\uDFFF])/.test(body)).toBe(false) + expect(/(? { + expect(parseTelegramRemoteKey("telegram:123:456")).toEqual({ chatId: "123" }) + expect(parseTelegramRemoteKey("slack:123:456")).toBeNull() + expect(parseTelegramRemoteKey("telegram:")).toBeNull() + expect(parseTelegramRemoteKey("telegram::456")).toBeNull() +}) + +test("normalizeUpdate maps a private text message and skips non-text/empty", () => { + const ok = normalizeUpdate({ update_id: 7, message: { text: "hi", from: { id: 9, username: "yu" }, chat: { id: 5, type: "private" } } }) + expect(ok).toEqual({ updateId: 7, chatId: "5", userId: "9", userName: "yu", text: "hi", isPrivate: true }) + expect(normalizeUpdate({ update_id: 8, message: { from: { id: 9 }, chat: { id: 5, type: "private" } } })).toBeNull() + expect(normalizeUpdate({ update_id: 8, message: { text: " ", from: { id: 9 }, chat: { id: 5, type: "private" } } })).toBeNull() + expect(normalizeUpdate({})).toBeNull() +}) + +test("inboundMessage enforces private-chat + allowlist as silent drops", () => { + const allowed = { update_id: 1, message: { text: "go", from: { id: 42 }, chat: { id: 100, type: "private" } } } + const stranger = { update_id: 2, message: { text: "go", from: { id: 99 }, chat: { id: 100, type: "private" } } } + const group = { update_id: 3, message: { text: "go", from: { id: 42 }, chat: { id: -100, type: "group" } } } + + expect(inboundMessage(allowed, "42")).toEqual({ + content: "go", + replyCtx: { chatId: "100" }, + channelID: "100", + userID: "42", + }) + expect(inboundMessage(stranger, "42")).toBeNull() + expect(inboundMessage(group, "42")).toBeNull() + // An empty allowFrom (pairing capture only) accepts any private sender. + expect(inboundMessage(stranger, "")).not.toBeNull() +}) + +test("isFatalTelegramError treats auth/not-found as fatal, others transient", () => { + expect(isFatalTelegramError(new TelegramApiError("getUpdates", 401, 401, "Unauthorized", undefined))).toBe(true) + expect(isFatalTelegramError(new TelegramApiError("getUpdates", 403, 403, "Forbidden", undefined))).toBe(true) + expect(isFatalTelegramError(new TelegramApiError("getUpdates", 409, 409, "Conflict", undefined))).toBe(false) + expect(isFatalTelegramError(new TelegramApiError("getUpdates", 429, 429, "Too Many", 2000))).toBe(false) + expect(isFatalTelegramError(new Error("network"))).toBe(false) +}) + +// --- against a local fake Bot API ------------------------------------------ + +interface FakeCall { + method: string + body: any +} + +/** + * A local stand-in for api.telegram.org. `getUpdates` drains a queued list of + * batches (one per call), then long-returns empty so the loop idles. Records + * every call so tests can assert the offset handoff and send payloads. + */ +function fakeBotApi(batches: any[][]) { + const calls: FakeCall[] = [] + let batchIndex = 0 + const server = Bun.serve({ + port: 0, + async fetch(req) { + const url = new URL(req.url) + const method = url.pathname.split("/").pop() ?? "" + const body = await req.json().catch(() => ({})) + calls.push({ method, body }) + if (method === "getMe") return json({ ok: true, result: { id: 1, username: "bot", first_name: "Bot" } }) + if (method === "getUpdates") { + const batch = batchIndex < batches.length ? batches[batchIndex++] : [] + return json({ ok: true, result: batch }) + } + if (method === "sendMessage") return json({ ok: true, result: { message_id: 1 } }) + return json({ ok: true, result: {} }) + }, + }) + return { url: `http://localhost:${server.port}`, calls, stop: () => server.stop(true) } +} + +const json = (value: unknown) => new Response(JSON.stringify(value), { headers: { "content-type": "application/json" } }) + +function update(id: number, userId: number, text = "hi") { + return { update_id: id, message: { text, from: { id: userId, username: `u${userId}` }, chat: { id: userId, type: "private" } } } +} + +function groupUpdate(id: number, userId: number, text = "hi") { + return { update_id: id, message: { text, from: { id: userId, username: `u${userId}` }, chat: { id: -id, type: "group" } } } +} + +async function waitFor(predicate: () => boolean, timeoutMs = 1000): Promise { + const start = Date.now() + while (!predicate()) { + if (Date.now() - start > timeoutMs) throw new Error("timed out waiting for condition") + await new Promise((r) => setTimeout(r, 5)) + } +} + +test("runLoop advances the offset past the highest update_id it received", async () => { + const api = fakeBotApi([[update(10, 42), update(11, 42)]]) + try { + const poller = new TelegramPoller("t", api.url) + poller.pollRetryMs = 0 + const ac = new AbortController() + const loop = poller.runLoop(0, () => {}, ac.signal) + // First getUpdates uses offset 0; the next must ack with maxUpdateId + 1. + await waitFor(() => api.calls.filter((c) => c.method === "getUpdates").length >= 2) + ac.abort() + await loop + const offsets = api.calls.filter((c) => c.method === "getUpdates").map((c) => c.body.offset) + expect(offsets[0]).toBe(0) + expect(offsets[1]).toBe(12) + } finally { + api.stop() + } +}) + +test("runLoop ignores a malformed update_id instead of poisoning the offset with NaN", async () => { + // A good update (id 10) then one with a non-numeric update_id. The offset must + // advance to 11 and stay finite — a NaN offset would make every later poll + // request offset=NaN (serialized to null) and replay the whole backlog. + const malformed = { update_id: "oops", message: { text: "x", from: { id: 42 }, chat: { id: 42, type: "private" } } } + const api = fakeBotApi([[update(10, 42)], [malformed]]) + try { + const poller = new TelegramPoller("t", api.url) + poller.pollRetryMs = 0 + const ac = new AbortController() + const loop = poller.runLoop(0, () => {}, ac.signal) + await waitFor(() => api.calls.filter((c) => c.method === "getUpdates").length >= 3) + ac.abort() + await loop + const offsets = api.calls.filter((c) => c.method === "getUpdates").map((c) => c.body.offset) + expect(offsets[0]).toBe(0) + expect(offsets[1]).toBe(11) // acked the good update 10 + expect(offsets[2]).toBe(11) // malformed id skipped, offset held — not NaN + expect(offsets.every((o) => Number.isFinite(o))).toBe(true) + } finally { + api.stop() + } +}) + +test("TelegramPlatform delivers only the paired user's private messages", async () => { + // Batch 0 (empty) ends the start-up drain; the live poll then returns the owner + // and a stranger, and only the owner's private message is delivered. + const api = fakeBotApi([[], [update(1, 42, "from owner"), update(2, 99, "from stranger")]]) + try { + const platform = new TelegramPlatform({ token: "t", allowFrom: "42", baseUrl: api.url }) + const received: { platform: Platform; content: string; channelID?: string }[] = [] + const handler: MessageHandler = (p, m) => received.push({ platform: p, content: m.content, channelID: m.channelID }) + const run = platform.start(handler) + await waitFor(() => received.length >= 1) + await platform.stop() + await run + expect(received).toHaveLength(1) + expect(received[0].content).toBe("from owner") + expect(received[0].channelID).toBe("42") + } finally { + api.stop() + } +}) + +test("TelegramPlatform drops the backlog on start so a queued prompt is not replayed", async () => { + // update 5 is already queued when the platform starts (a prompt sent while the + // app was down, or one left unacked when it crashed). The start-up drain must + // ack past it WITHOUT dispatching it; only the genuinely new message (update 6, + // after the empty drain poll) is delivered. + const api = fakeBotApi([[update(5, 42, "stale offline prompt")], [], [update(6, 42, "new prompt")]]) + try { + const platform = new TelegramPlatform({ token: "t", allowFrom: "42", baseUrl: api.url }) + const received: string[] = [] + const handler: MessageHandler = (_p, m) => received.push(m.content) + const run = platform.start(handler) + await waitFor(() => received.length >= 1) + await platform.stop() + await run + expect(received).toEqual(["new prompt"]) + // The drain started at offset 0 and acked the stale backlog (max id 5) at + // offset 6 before the live poll began. + const offsets = api.calls.filter((c) => c.method === "getUpdates").map((c) => c.body.offset) + expect(offsets[0]).toBe(0) + expect(offsets[1]).toBe(6) + } finally { + api.stop() + } +}) + +test("start() signals onReady only after the first live poll returns, past the backlog drain", async () => { + // The startup-race guard: onReady must not fire until the drain is done AND the + // first live getUpdates has actually returned, so a caller can't report + // "connected" while a freshly sent message would still be swept away as backlog + // (or while a 409 keeps the live loop from ever serving). update 5 is the stale + // backlog; the empty batch ends the drain; update 6 is the first live message. + const api = fakeBotApi([[update(5, 42, "stale")], [], [update(6, 42, "new")]]) + try { + const platform = new TelegramPlatform({ token: "t", allowFrom: "42", baseUrl: api.url }) + let readyCalls = 0 + let getUpdatesAtReady = -1 + const run = platform.start( + () => {}, + () => { + readyCalls++ + getUpdatesAtReady = api.calls.filter((c) => c.method === "getUpdates").length + }, + ) + await waitFor(() => readyCalls > 0) + await platform.stop() + await run + expect(readyCalls).toBe(1) + // Two drain polls (the stale batch, then the empty terminator) and then the + // first live poll that returned update 6: onReady fires only on that third + // poll's return, never during the drain and never on mere loop install. + expect(getUpdatesAtReady).toBeGreaterThanOrEqual(3) + } finally { + api.stop() + } +}) + +test("runLoop gives up after bounded 409 conflicts and never signals ready", async () => { + // Every getUpdates returns 409 — another client is long-polling this token, so + // the loop will NEVER receive anything. It must not spin forever (which would + // leave the desktop showing a healthy-looking "connecting"): after a bounded + // number of retries it rejects, and onReady never fires, so the runtime lands + // on "degraded" with a real cause. The test resolving at all proves the retry + // is bounded — an unbounded loop against this server would hang to timeout. + let polls = 0 + const server = Bun.serve({ + port: 0, + async fetch(req) { + const method = new URL(req.url).pathname.split("/").pop() ?? "" + if (method !== "getUpdates") return json({ ok: true, result: {} }) + polls++ + return json({ ok: false, error_code: 409, description: "Conflict: terminated by other getUpdates request" }) + }, + }) + try { + const poller = new TelegramPoller("t", `http://localhost:${server.port}`) + poller.pollRetryMs = 0 + let ready = 0 + await expect(poller.runLoop(0, () => {}, new AbortController().signal, () => ready++)).rejects.toThrow( + /another client is polling this bot token/i, + ) + expect(ready).toBe(0) + // MAX_CONFLICT_RETRIES retries are tolerated; the next conflict throws. + expect(polls).toBe(MAX_CONFLICT_RETRIES + 1) + } finally { + server.stop(true) + } +}) + +test("runLoop signals ready only after the first poll that returns, surviving a transient 409", async () => { + // The first two live polls 409 (a brief handoff — our own previous poller still + // releasing on reconnect), then it clears. onReady must NOT fire during the + // conflict: only after the first getUpdates that actually returns, and exactly + // once. The conflict counter resets on that success, so a later blip is tolerated. + let polls = 0 + const server = Bun.serve({ + port: 0, + async fetch(req) { + const method = new URL(req.url).pathname.split("/").pop() ?? "" + if (method !== "getUpdates") return json({ ok: true, result: {} }) + polls++ + if (polls <= 2) return json({ ok: false, error_code: 409, description: "Conflict" }) + return json({ ok: true, result: [] }) // conflict cleared: an empty live poll + }, + }) + try { + const poller = new TelegramPoller("t", `http://localhost:${server.port}`) + poller.pollRetryMs = 0 + const ac = new AbortController() + let ready = 0 + let pollsAtReady = -1 + const loop = poller.runLoop(0, () => {}, ac.signal, () => { + ready++ + pollsAtReady = polls + }) + await waitFor(() => ready > 0) + ac.abort() + await loop + expect(ready).toBe(1) + // The two 409s plus the first poll that returned: ready fires on the 3rd poll, + // never during the retries. + expect(pollsAtReady).toBe(3) + } finally { + server.stop(true) + } +}) + +test("TelegramPlatform.send splits a long reply into multiple sendMessage calls", async () => { + const api = fakeBotApi([]) + try { + const platform = new TelegramPlatform({ token: "t", allowFrom: "42", baseUrl: api.url }) + await platform.reply({ chatId: "100" }, "a".repeat(9000)) + const sends = api.calls.filter((c) => c.method === "sendMessage") + expect(sends.length).toBeGreaterThan(1) + expect(sends.every((c) => c.body.chat_id === "100")).toBe(true) + } finally { + api.stop() + } +}) + +test("sendMessage retries a transient per-chunk failure in place, never resending earlier chunks", async () => { + // A long reply → 3 chunks. The 2nd chunk's first send fails transiently (500), + // then succeeds. The retry is in place: chunk 1 is delivered exactly once and + // the user ends up with one complete, ordered set — no duplicated head. + const delivered: string[] = [] + let chunk2Attempts = 0 + const server = Bun.serve({ + port: 0, + async fetch(req) { + const method = new URL(req.url).pathname.split("/").pop() ?? "" + if (method !== "sendMessage") return json({ ok: true, result: {} }) + const body: any = await req.json().catch(() => ({})) + const header = String(body.text ?? "").match(/^\[(\d+)\/(\d+)\]/) + const idx = header ? Number(header[1]) : 1 + if (idx === 2) { + chunk2Attempts++ + if (chunk2Attempts === 1) return json({ ok: false, error_code: 500, description: "transient" }) + } + delivered.push(header ? header[0] : "") + return json({ ok: true, result: { message_id: 1 } }) + }, + }) + try { + const poller = new TelegramPoller("t", `http://localhost:${server.port}`) + poller.pollRetryMs = 0 + await poller.sendMessage("100", "a".repeat(9000)) + expect(delivered).toEqual(["[1/3]", "[2/3]", "[3/3]"]) + expect(chunk2Attempts).toBe(2) // failed once, retried once in place + } finally { + server.stop(true) + } +}) + +test("sendMessage throws PartialDeliveryError when a later chunk fails for good, without resending the head", async () => { + // The 2nd chunk fails permanently. Chunk 1 was already delivered, so a wholesale + // resend would duplicate it — sendMessage surfaces PartialDeliveryError instead, + // which the engine's delivery retry treats as terminal. + const delivered: string[] = [] + const server = Bun.serve({ + port: 0, + async fetch(req) { + const method = new URL(req.url).pathname.split("/").pop() ?? "" + if (method !== "sendMessage") return json({ ok: true, result: {} }) + const body: any = await req.json().catch(() => ({})) + const header = String(body.text ?? "").match(/^\[(\d+)\/(\d+)\]/) + const idx = header ? Number(header[1]) : 1 + if (idx >= 2) return json({ ok: false, error_code: 500, description: "down" }) + delivered.push(header ? header[0] : "") + return json({ ok: true, result: { message_id: 1 } }) + }, + }) + try { + const poller = new TelegramPoller("t", `http://localhost:${server.port}`) + poller.pollRetryMs = 0 + await expect(poller.sendMessage("100", "a".repeat(9000))).rejects.toBeInstanceOf(PartialDeliveryError) + expect(delivered).toEqual(["[1/3]"]) // head delivered once despite the in-place retries on chunk 2 + } finally { + server.stop(true) + } +}) + +test("reconstructReplyCtx rebuilds chatId from a stored remote key; throws otherwise", () => { + const platform = new TelegramPlatform({ token: "t", allowFrom: "42" }) + expect(platform.reconstructReplyCtx("telegram:100:42")).toEqual({ chatId: "100" }) + expect(() => platform.reconstructReplyCtx("nope")).toThrow() +}) + +// --- pairing capture ------------------------------------------------------- + +test("captureFirstSender drains backlog and returns the first NEW private sender", async () => { + // Batch 0 is pre-pairing backlog from user 7; the empty batch 1 marks the + // backlog as fully drained, then the long-poll delivers a genuinely new + // message from user 42. + const api = fakeBotApi([[update(5, 7, "stale")], [], [update(10, 42, "pair me")]]) + try { + const poller = new TelegramPoller("t", api.url) + poller.pollRetryMs = 0 + const captured = await captureFirstSender(poller, new AbortController().signal) + expect(captured).toEqual({ userId: "42", userName: "u42", botUsername: "bot" }) + // Drain starts at offset 0; backlog (max id 5) is acked at offset 6, where an + // empty immediate poll ends the drain. + const offsets = api.calls.filter((c) => c.method === "getUpdates").map((c) => c.body.offset) + expect(offsets[0]).toBe(0) + expect(offsets[1]).toBe(6) + // The captured message (id 10) is acked server-side with a final offset 11. + expect(offsets[offsets.length - 1]).toBe(11) + } finally { + api.stop() + } +}) + +test("captureFirstSender drains the backlog BEFORE getMe so a slow identity fetch can't drop the first message", async () => { + // The regression this guards: getMe used to run first (in startPairing), so a + // message the user sent during a slow getMe sat in the queue and was then swept + // up by capture's drain — pairing waited forever. The drain must pin the offset + // baseline first; getMe (identity only) comes after. Batch 0 is the empty drain; + // the message (id 10) is delivered by the post-getMe poll and must be captured. + const api = fakeBotApi([[], [update(10, 42, "pair me")]]) + try { + const poller = new TelegramPoller("t", api.url) + poller.pollRetryMs = 0 + const captured = await captureFirstSender(poller, new AbortController().signal) + expect(captured).toEqual({ userId: "42", userName: "u42", botUsername: "bot" }) + const firstGetUpdates = api.calls.findIndex((c) => c.method === "getUpdates") + const getMeAt = api.calls.findIndex((c) => c.method === "getMe") + expect(firstGetUpdates).toBeGreaterThanOrEqual(0) + expect(getMeAt).toBeGreaterThan(firstGetUpdates) + } finally { + api.stop() + } +}) + +test("captureFirstSender drains a MULTI-batch backlog before accepting a sender", async () => { + // The pre-pairing backlog spans two batches (stale users 7 then 8). A single + // immediate drain would stop after batch 0, and the long-poll would then hand + // back user 8's stale message as the "new" sender. The drain must consume both + // stale batches (until an empty poll) before the real pairing message (user 42). + const api = fakeBotApi([[update(5, 7, "stale")], [update(6, 8, "also stale")], [], [update(10, 42, "pair me")]]) + try { + const poller = new TelegramPoller("t", api.url) + poller.pollRetryMs = 0 + const captured = await captureFirstSender(poller, new AbortController().signal) + expect(captured).toEqual({ userId: "42", userName: "u42", botUsername: "bot" }) + } finally { + api.stop() + } +}) + +test("captureFirstSender ignores group and non-text updates, keeps waiting", async () => { + const api = fakeBotApi([[], [groupUpdate(10, 7)], [update(11, 42, "pair me")]]) + try { + const poller = new TelegramPoller("t", api.url) + poller.pollRetryMs = 0 + const captured = await captureFirstSender(poller, new AbortController().signal) + expect(captured).toEqual({ userId: "42", userName: "u42", botUsername: "bot" }) + } finally { + api.stop() + } +}) + +test("captureFirstSender returns null when the signal aborts first", async () => { + const api = fakeBotApi([]) // only ever returns empty batches + try { + const poller = new TelegramPoller("t", api.url) + poller.pollRetryMs = 0 + const ac = new AbortController() + setTimeout(() => ac.abort(), 30) + expect(await captureFirstSender(poller, ac.signal)).toBeNull() + } finally { + api.stop() + } +}) + +test("captureFirstSender retries a transient final ack so the pairing message is not replayed", async () => { + // A hand-rolled server so we can fail one specific call: the final ack of the + // captured message. offset 0 is hit twice (empty drain, then the wait loop that + // returns the pairing message); offset 11 acks update 10 — fail it once + // (transient 500), then succeed. The retry is what actually acks the message; + // if it were swallowed, the steady-state bridge (which polls from offset 0) + // would hand the pairing text back as the user's first prompt. + let zeroPolls = 0 + let ackAttempts = 0 + const offsets: number[] = [] + const server = Bun.serve({ + port: 0, + async fetch(req) { + const method = new URL(req.url).pathname.split("/").pop() ?? "" + const body: any = await req.json().catch(() => ({})) + if (method === "getMe") return json({ ok: true, result: { id: 1, username: "bot" } }) + if (method !== "getUpdates") return json({ ok: true, result: {} }) + const offset = Number(body.offset ?? 0) + offsets.push(offset) + if (offset === 0) { + zeroPolls++ + return json({ ok: true, result: zeroPolls === 1 ? [] : [update(10, 42, "pair me")] }) + } + if (offset === 11) { + ackAttempts++ + if (ackAttempts === 1) return json({ ok: false, error_code: 500, description: "transient" }) + return json({ ok: true, result: [] }) + } + return json({ ok: true, result: [] }) + }, + }) + try { + const poller = new TelegramPoller("t", `http://localhost:${server.port}`) + poller.pollRetryMs = 0 + const captured = await captureFirstSender(poller, new AbortController().signal) + expect(captured).toEqual({ userId: "42", userName: "u42", botUsername: "bot" }) + // The ack was retried after the transient failure (not swallowed): two calls + // at offset 11, so update 10 is acked and cannot be replayed to the bridge. + expect(ackAttempts).toBe(2) + expect(offsets.filter((o) => o === 11)).toEqual([11, 11]) + } finally { + server.stop(true) + } +}) + +test("start() rejects on an invalid token so the gateway can surface it", async () => { + const server = Bun.serve({ + port: 0, + fetch: () => json({ ok: false, error_code: 401, description: "Unauthorized" }), + }) + try { + const platform = new TelegramPlatform({ token: "bad", allowFrom: "42", baseUrl: `http://localhost:${server.port}` }) + await expect(platform.start(() => {})).rejects.toThrow(/401/) + } finally { + server.stop(true) + } +}) diff --git a/packages/remote-bridge/src/platforms/telegram.ts b/packages/remote-bridge/src/platforms/telegram.ts new file mode 100644 index 000000000..4ca2c2f56 --- /dev/null +++ b/packages/remote-bridge/src/platforms/telegram.ts @@ -0,0 +1,563 @@ +// Telegram as a bridge Platform. A thin wrapper over the raw Bot API +// (getUpdates long-poll + sendMessage), no SDK: the API is near-additive and a +// single desktop long-poll consumer needs offset/abort/backoff control more +// than a middleware framework. The engine drives this exactly like any other +// `Platform`; pairing (capturing the first sender before an allow_from exists) +// reuses the same `TelegramPoller` primitive — see `captureFirstSender`. + +import { PartialDeliveryError } from "../types.ts" +import type { Message, MessageHandler, Platform } from "../types.ts" + +const API_BASE = "https://api.telegram.org" +// Telegram holds a getUpdates request open for up to this many seconds before +// returning empty; the HTTP timeout below must outlast it. +const POLL_TIMEOUT_S = 25 +const REQUEST_TIMEOUT_MS = 10_000 +// Base backoff between transient getUpdates failures; a 429 overrides it with +// the server's retry_after. Lowered in tests. +const POLL_RETRY_MS = 3_000 + +// Telegram allows ONE getUpdates consumer per token; a 409 means another client +// is already long-polling it. A few retries absorb a handoff (our own previous +// poller releasing on reconnect), but a persistent 409 never clears by retrying — +// so we give up after this many and surface it, instead of spinning while the UI +// shows a "connected" bridge that silently receives nothing. +export const MAX_CONFLICT_RETRIES = 3 + +// A chunk send is retried in place up to this many times on transient failures (a +// 429's retry_after, else base backoff) so a blip on one chunk never resends the +// chunks before it. Past that, the chunk has failed for real. +const MAX_SEND_RETRIES = 3 + +// Telegram caps a message at 4096 UTF-16 code units (NOT codepoints): an +// astral-plane char (most emoji, CJK ext-B) costs 2. Splitting long assistant +// replies is correctness — an over-cap send returns 400 and the user gets +// nothing. Pulled to 4000 so an "[i/N]" continuation header fits under the cap. +const MAX_UTF16_PER_MESSAGE = 4000 + +/** A Bot API call failed. `errorCode` is Telegram's `error_code` when present, + * else the HTTP status; `retryAfterMs` is set only for 429 (clamped). */ +export class TelegramApiError extends Error { + constructor( + readonly method: string, + readonly httpStatus: number, + readonly errorCode: number | undefined, + readonly description: string, + readonly retryAfterMs: number | undefined, + ) { + super(`telegram ${method} failed: ${httpStatus}${errorCode ? ` (${errorCode})` : ""} ${description}`.trim()) + this.name = "TelegramApiError" + } +} + +/** + * Whether a Bot API error is permanent. A bad/blocked token (401/403) or a + * missing method (404) cannot be fixed by retrying, so the poll loop surfaces + * it (rejecting `start()`) instead of spinning. A 409 conflict is retried only a + * bounded number of times (see `MAX_CONFLICT_RETRIES`) — it never self-heals, so + * after that it is surfaced too. 429/5xx/network are transient and retried with backoff. + */ +export function isFatalTelegramError(err: unknown): boolean { + if (err instanceof TelegramApiError) { + const code = err.errorCode ?? err.httpStatus + return code === 401 || code === 403 || code === 404 + } + return false +} + +/** A getUpdates 409: another client is already long-polling this bot token. */ +function isConflictError(err: unknown): boolean { + return err instanceof TelegramApiError && (err.errorCode ?? err.httpStatus) === 409 +} + +/** Raised when getUpdates keeps returning 409 past `MAX_CONFLICT_RETRIES` — a + * real, non-self-healing conflict (another bridge or client owns the token). + * Surfaced so the desktop lands on "degraded" with a clear cause, never a false + * "connected" over a token that delivers nothing. */ +export class TelegramConflictError extends Error { + constructor() { + super("another client is polling this bot token (Telegram 409 conflict)") + this.name = "TelegramConflictError" + } +} + +/** Backoff for a transient poll failure: a 429's retry_after if present, else base. */ +function backoffMs(err: unknown, base: number): number { + return err instanceof TelegramApiError && err.retryAfterMs ? err.retryAfterMs : base +} + +export interface TelegramIdentity { + id: string + username?: string + displayName?: string +} + +export interface NormalizedUpdate { + updateId: number + chatId: string + userId: string + userName: string + text: string + isPrivate: boolean +} + +/** + * Normalize a raw getUpdates entry to the fields the bridge needs, or null if + * it carries no usable text. v1 is text-only: a photo/sticker/voice update + * normalizes to null and is skipped (but its update_id still advances the + * offset so it is acked, not refetched). + */ +export function normalizeUpdate(update: unknown): NormalizedUpdate | null { + if (!update || typeof update !== "object") return null + const message = (update as { message?: any }).message + if (!message || typeof message !== "object" || !message.from || !message.chat) return null + const text = typeof message.text === "string" ? message.text : "" + if (text.trim() === "") return null + return { + updateId: Number((update as { update_id?: unknown }).update_id), + chatId: String(message.chat.id), + userId: String(message.from.id), + userName: message.from.username ?? message.from.first_name ?? String(message.from.id), + text, + isPrivate: message.chat.type === "private", + } +} + +/** + * Turn a raw update into the inbound `Message` the engine expects, or null if + * it should be dropped. v1 accepts only private chats from the paired user + * (`allowFrom`); a "" allowFrom accepts any private sender (used only by the + * pairing capture, never steady state). Drops are silent by design — a bounce + * would let a stranger probe the policy. No sessionKey is set, so the engine + * derives `telegram::` and keeps the platform prefix it needs + * to restore delivery after a restart. + */ +export function inboundMessage(update: unknown, allowFrom: string): Message | null { + const norm = normalizeUpdate(update) + if (!norm) return null + if (!norm.isPrivate) return null + if (allowFrom !== "" && norm.userId !== allowFrom) return null + return { + content: norm.text, + replyCtx: { chatId: norm.chatId }, + channelID: norm.chatId, + userID: norm.userId, + } +} + +/** + * Rebuild a reply target from a remote key. The engine stores keys as + * `telegram::` (built from platform name + channelID + userID), + * so the chat id is the second segment. chat/user ids are integers, never + * embedded colons, so a plain split is safe. Returns null on any other shape. + */ +export function parseTelegramRemoteKey(key: string): { chatId: string } | null { + const parts = key.split(":") + if (parts.length < 3 || parts[0] !== "telegram") return null + const chatId = parts[1] + if (chatId.trim() === "") return null + return { chatId } +} + +/** Longest prefix of `s` whose UTF-16 length is <= `cap`, never splitting a + * surrogate pair. (`String.length` already counts UTF-16 code units.) */ +function prefixWithinUtf16(s: string, cap: number): string { + if (s.length <= cap) return s + let used = 0 + let end = 0 + for (let i = 0; i < s.length; ) { + const code = s.codePointAt(i)! + const units = code > 0xffff ? 2 : 1 + if (used + units > cap) break + used += units + i += units + end = i + } + return s.slice(0, end) +} + +/** + * Split `text` into UTF-16-bounded chunks for delivery, preferring a newline + * break near the end of each chunk. A single-chunk message is returned as-is + * with no header; a multi-chunk message gets an "[i/N]" header so the receiver + * knows it was split rather than seeing N unexplained messages. + */ +export function splitForTelegram(text: string): string[] { + if (text.length <= MAX_UTF16_PER_MESSAGE) return [text] + const HEADER_RESERVE = 12 // "[99/99]\n" + const cap = MAX_UTF16_PER_MESSAGE - HEADER_RESERVE + const pieces: string[] = [] + let remaining = text + while (remaining.length > cap) { + let chunk = prefixWithinUtf16(remaining, cap) + const minBoundary = Math.floor(chunk.length * 0.9) + const nl = chunk.lastIndexOf("\n") + // Prefer a line boundary near the end, keeping the newline as the chunk's + // last char. Every character lands in exactly one chunk — never drop the + // delimiter, or reassembling a split reply loses a newline at the seam. + if (nl >= minBoundary) chunk = chunk.slice(0, nl + 1) + pieces.push(chunk) + remaining = remaining.slice(chunk.length) + } + if (remaining.length > 0) pieces.push(remaining) + const total = pieces.length + return pieces.map((piece, idx) => `[${idx + 1}/${total}]\n${piece}`) +} + +/** + * Low-level Bot API primitive: getMe / getUpdates / sendMessage plus a long-poll + * loop. Shared by `TelegramPlatform` (steady state) and `captureFirstSender` + * (pairing) so the offset/abort/backoff handling lives in exactly one place and + * the two never poll the same token concurrently. + */ +export class TelegramPoller { + /** Base poll backoff, mutable so tests can drop it to zero. */ + pollRetryMs = POLL_RETRY_MS + private readonly baseUrl: string + + // `baseUrl` defaults to the real Bot API; tests point it at a local server. + constructor(private readonly token: string, baseUrl: string = API_BASE) { + this.baseUrl = baseUrl.replace(/\/+$/, "") + } + + async getMe(signal?: AbortSignal): Promise { + const me = await this.call("getMe", {}, signal, REQUEST_TIMEOUT_MS) + return { + id: String(me?.id ?? ""), + username: me?.username, + displayName: me?.first_name, + } + } + + /** One getUpdates call. `offset` acks every update below it; 0/omitted returns + * all currently-unconfirmed updates. `timeoutS` is Telegram's long-poll hold; + * 0 returns immediately (used to drain backlog / ack during pairing). */ + async getUpdates(offset: number, signal?: AbortSignal, timeoutS: number = POLL_TIMEOUT_S): Promise { + const result = await this.call( + "getUpdates", + { offset, timeout: timeoutS, allowed_updates: ["message"] }, + signal, + (timeoutS + 5) * 1_000, + ) + return Array.isArray(result) ? result : [] + } + + /** + * Long-poll from `startOffset`, calling `onUpdate` for each raw update. The + * offset advances past every update returned (even ones `onUpdate` ignores) + * so nothing is refetched. Resolves when `signal` aborts; rejects on a fatal + * error or a persistent 409 conflict — other transient failures back off and + * retry. `onUpdate` errors are swallowed so one bad message cannot kill the loop. + * + * `onReady` fires once, after the FIRST getUpdates that actually returns (even + * empty): that is the only proof the token is ours and live messages will be + * delivered. Firing it any earlier (e.g. when the loop is merely installed) + * would let a caller report "connected" over a token a 409 conflict keeps + * silently empty. + */ + async runLoop( + startOffset: number, + onUpdate: (update: any) => void, + signal: AbortSignal, + onReady?: () => void, + ): Promise { + let offset = startOffset + let conflicts = 0 + let ready = false + while (!signal.aborted) { + let updates: any[] + try { + updates = await this.getUpdates(offset, signal) + } catch (err) { + if (signal.aborted) return + if (isFatalTelegramError(err)) throw err + if (isConflictError(err) && ++conflicts > MAX_CONFLICT_RETRIES) throw new TelegramConflictError() + await sleep(backoffMs(err, this.pollRetryMs), signal) + continue + } + conflicts = 0 + if (!ready) { + ready = true + onReady?.() + } + for (const update of updates) { + offset = nextOffset(offset, update) + try { + onUpdate(update) + } catch { + // a handler failure must not stall the poll loop + } + } + } + } + + /** + * Send `text` to `chatId`, split over the 4096-unit cap. Each chunk is retried + * in place on transient failures (a 429's retry_after, else base backoff) so a + * blip on one chunk never resends the chunks before it. If a chunk fails for + * good after earlier chunks were already delivered, throws PartialDeliveryError + * so the engine's delivery retry won't resend the whole message and duplicate + * what arrived; a failure on the very first chunk (nothing sent yet) throws the + * raw error, so a wholesale retry is still safe. + */ + async sendMessage(chatId: string, text: string, signal?: AbortSignal): Promise { + const chunks = splitForTelegram(text) + for (let i = 0; i < chunks.length; i++) { + try { + await this.sendChunk(chatId, chunks[i], signal) + } catch (err) { + throw i === 0 ? err : new PartialDeliveryError(err) + } + } + } + + /** One sendMessage chunk, retried in place on transient failures (bounded). A + * fatal Bot API error (bad/blocked token) or an abort throws immediately. */ + private async sendChunk(chatId: string, text: string, signal?: AbortSignal): Promise { + for (let attempt = 1; ; attempt++) { + try { + await this.call("sendMessage", { chat_id: chatId, text }, signal, REQUEST_TIMEOUT_MS) + return + } catch (err) { + if (signal?.aborted || isFatalTelegramError(err) || attempt >= MAX_SEND_RETRIES) throw err + await sleep(backoffMs(err, this.pollRetryMs), signal) + } + } + } + + private async call(method: string, body: Record, signal: AbortSignal | undefined, timeoutMs: number): Promise { + const res = await fetch(`${this.baseUrl}/bot${this.token}/${method}`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + signal: withTimeout(signal, timeoutMs), + }) + const json: any = await res.json().catch(() => ({})) + if (!res.ok || json?.ok === false) { + throw new TelegramApiError( + method, + res.status, + typeof json?.error_code === "number" ? json.error_code : undefined, + typeof json?.description === "string" ? json.description : "", + retryAfterMs(json), + ) + } + return json?.result + } +} + +/** + * The steady-state Telegram platform. Accepts only private-chat messages from + * the paired `allowFrom` user (others are dropped silently — a bounce would let + * a stranger probe the policy). The inbound message carries channelID/userID + * but no sessionKey, so the engine derives `telegram::` itself, + * keeping the platform prefix it relies on to restore delivery after a restart. + */ +export class TelegramPlatform implements Platform { + readonly name = "telegram" + private readonly poller: TelegramPoller + private readonly allowFrom: string + private ac: AbortController | null = null + private loop: Promise | null = null + + constructor(opts: { token: string; allowFrom: string; baseUrl?: string }) { + this.poller = new TelegramPoller(opts.token, opts.baseUrl) + this.allowFrom = opts.allowFrom.trim() + } + + async start(handler: MessageHandler, onReady?: () => void): Promise { + if (this.ac) return + const ac = new AbortController() + this.ac = ac + try { + // Validate the token up front so a bad credential rejects start() (the + // gateway turns that into a surfaced failure) instead of looping silently. + await this.poller.getMe(ac.signal) + // Drop the backlog before serving: a prompt queued while the app was down, + // or one in flight when it crashed, must not auto-run on reconnect. Poll + // forward from the live tip so only genuinely new messages are delivered. + const startOffset = await drainBacklog(this.poller, ac.signal) + if (startOffset === null) return // aborted during startup + // onReady is handed to runLoop, which fires it only after the first live + // getUpdates actually returns — not when the loop is merely installed. A + // 409 conflict (another client owns the token) then keeps us out of + // "ready" and, past a few retries, rejects start() so the caller surfaces + // it instead of showing a "connected" bridge that receives nothing. + this.loop = this.poller.runLoop(startOffset, (update) => this.dispatch(handler, update), ac.signal, onReady) + await this.loop + } finally { + this.ac = null + this.loop = null + } + } + + private dispatch(handler: MessageHandler, update: unknown): void { + const msg = inboundMessage(update, this.allowFrom) + if (msg) handler(this, msg) + } + + reply(replyCtx: unknown, content: string): Promise { + return this.poller.sendMessage((replyCtx as { chatId: string }).chatId, content) + } + + send(replyCtx: unknown, content: string): Promise { + return this.poller.sendMessage((replyCtx as { chatId: string }).chatId, content) + } + + async stop(): Promise { + this.ac?.abort() + if (this.loop) await this.loop.catch(() => {}) + } + + reconstructReplyCtx(remoteKey: string): { chatId: string } { + const ctx = parseTelegramRemoteKey(remoteKey) + if (!ctx) throw new Error(`telegram: cannot reconstruct reply context from "${remoteKey}"`) + return ctx + } +} + +export interface CapturedSender { + userId: string + userName: string + /** The bot's @username, for the connect dialog. Fetched as part of pairing so + * the caller does not run its own getMe before the capture (which would widen + * the window where the user's first message is mistaken for backlog). */ + botUsername?: string +} + +/** + * One getUpdates call with transient-failure backoff, for pairing. Returns the + * updates, or null if the signal aborts (the caller treats that as cancelled). A + * fatal Bot API error (bad/blocked token) is rethrown — retrying cannot fix it. + * Shared by the backlog drain, the capture wait, and the final ack so a transient + * blip never silently skips the ack and lets the bridge replay the pairing message. + */ +async function getUpdatesWithRetry( + poller: TelegramPoller, + offset: number, + signal: AbortSignal, + timeoutS: number, +): Promise { + let conflicts = 0 + while (!signal.aborted) { + try { + return await poller.getUpdates(offset, signal, timeoutS) + } catch (err) { + if (signal.aborted) return null + if (isFatalTelegramError(err)) throw err + if (isConflictError(err) && ++conflicts > MAX_CONFLICT_RETRIES) throw new TelegramConflictError() + await sleep(backoffMs(err, poller.pollRetryMs), signal) + } + } + return null +} + +/** + * Advance the offset past everything already queued WITHOUT dispatching it, + * returning the next offset to poll from (or null if the signal aborts first). + * getUpdates returns at most ~100 updates per call, so the backlog can span + * several batches; loop immediate polls (timeout 0) until one comes back empty, + * each call acking what the previous one drained. + * + * Both entry points drain on (re)start: pairing, so a stale queued message can't + * mispair as the "first new sender"; and the steady-state platform, so a prompt + * queued while the app was down — or one already dispatched but not yet acked + * when the process died — is dropped instead of replayed. Telegram redelivers + * unacked updates, and replaying a prompt re-drives the agent. + */ +async function drainBacklog(poller: TelegramPoller, signal: AbortSignal): Promise { + let offset = 0 + while (!signal.aborted) { + const backlog = await getUpdatesWithRetry(poller, offset, signal, 0) + if (backlog === null) return null + if (backlog.length === 0) return offset + for (const update of backlog) offset = nextOffset(offset, update) + } + return null +} + +/** + * Pairing primitive: identify who is allowed to drive the bridge by capturing + * the first private message sent AFTER pairing begins. Used main-only by the + * connect flow before any `allow_from` exists, so it must run on its OWN poller + * and the caller must fully await it (including its final ack) before starting + * the real `TelegramPlatform` — two pollers on one token race a 409. + * + * The whole pairing handshake in one primitive: it first drains whatever is + * already queued (pre-pairing backlog) so a stale message can't mispair, then + * fetches the bot identity, then long-polls for the first new private text + * message. Draining BEFORE getMe is deliberate: it pins the offset baseline up + * front, so a message the user sends while a slow getMe is in flight lands past + * the baseline and is captured here instead of being swept up as backlog. On + * capture it acks that message server-side so the real bridge does not later + * replay it as a prompt. Returns null if the signal aborts first (the connect + * dialog was closed). A bad token surfaces as a thrown fatal error; transient + * failures back off and retry. + */ +export async function captureFirstSender(poller: TelegramPoller, signal: AbortSignal): Promise { + const drained = await drainBacklog(poller, signal) + if (drained === null) return null // aborted while draining + let offset = drained + + // botUsername is for the connect dialog. Fetched after the drain so getMe's + // latency can't widen the backlog window; the drain has already proven the token. + let botUsername: string | undefined + try { + botUsername = (await poller.getMe(signal)).username + } catch (err) { + if (signal.aborted) return null // aborted while fetching the bot identity + throw err + } + + while (!signal.aborted) { + const updates = await getUpdatesWithRetry(poller, offset, signal, POLL_TIMEOUT_S) + if (updates === null) return null // aborted while waiting + for (const update of updates) { + offset = nextOffset(offset, update) + const norm = normalizeUpdate(update) + if (norm?.isPrivate) { + // Ack the captured message so the real bridge's fresh poll won't replay it + // as a prompt. Retry transient failures: a swallowed ack here is the + // difference between a clean handoff and the pairing message resurfacing. + if ((await getUpdatesWithRetry(poller, offset, signal, 0)) === null) return null // aborted mid-ack + return { userId: norm.userId, userName: norm.userName, botUsername } + } + } + } + return null +} + +/** Advance the offset past `update`'s id, ignoring a malformed update_id rather + * than letting Number(...) poison it with NaN — a NaN offset makes every later + * getUpdates replay the backlog or stall. A non-finite id leaves the offset + * unchanged (matching the old `?? offset - 1` no-op for a missing id). */ +function nextOffset(offset: number, update: unknown): number { + const id = Number((update as { update_id?: unknown })?.update_id) + return Number.isFinite(id) ? Math.max(offset, id + 1) : offset +} + +function retryAfterMs(json: any): number | undefined { + const seconds = Number(json?.parameters?.retry_after) + if (!Number.isFinite(seconds) || seconds <= 0) return undefined + return Math.min(Math.max(seconds * 1_000, 1_000), 30_000) +} + +function withTimeout(signal: AbortSignal | undefined, ms: number): AbortSignal { + const timeout = AbortSignal.timeout(ms) + return signal ? AbortSignal.any([signal, timeout]) : timeout +} + +function sleep(ms: number, signal?: AbortSignal): Promise { + return new Promise((resolve) => { + if (signal?.aborted) return resolve() + const onAbort = () => { + clearTimeout(timer) + resolve() + } + const timer = setTimeout(() => { + signal?.removeEventListener("abort", onAbort) + resolve() + }, ms) + signal?.addEventListener("abort", onAbort, { once: true }) + }) +} diff --git a/packages/remote-bridge/src/types.ts b/packages/remote-bridge/src/types.ts index 88ef45987..1bfa98e0f 100644 --- a/packages/remote-bridge/src/types.ts +++ b/packages/remote-bridge/src/types.ts @@ -92,6 +92,19 @@ export interface Message { export type MessageHandler = (platform: Platform, msg: Message) => void +/** + * Thrown by a platform's reply/send when it has already delivered part of a + * multi-part message and then failed. The engine's delivery retry treats it as + * terminal: resending the whole payload would duplicate the parts that already + * arrived. Wraps the underlying cause for logging. + */ +export class PartialDeliveryError extends Error { + constructor(readonly reason: unknown) { + super(`partial delivery: ${reason instanceof Error ? reason.message : String(reason)}`) + this.name = "PartialDeliveryError" + } +} + /** * A chat platform the engine drives. The Vercel Chat SDK / Lark adapter is * wrapped to implement this so the engine stays decoupled from any one SDK, @@ -99,8 +112,16 @@ export type MessageHandler = (platform: Platform, msg: Message) => void */ export interface Platform { readonly name: string - /** Run the platform, delivering inbound messages to `handler`; resolves when stopped. */ - start(handler: MessageHandler): Promise + /** + * Run the platform, delivering inbound messages to `handler`; resolves when + * stopped. `onReady`, if given, fires once the platform is actually serving — + * past any backlog drain AND after the first live receive actually returns, the + * only proof inbound delivery works — so the gateway can defer "connected" until + * a message would really arrive. A platform whose receive loop cannot get going + * (e.g. a Telegram 409: another client owns the token) must reject `start()` + * without ever firing `onReady`, so the caller surfaces it instead of "connected". + */ + start(handler: MessageHandler, onReady?: () => void): Promise /** Reply in-thread to the message `replyCtx` identifies. */ reply(replyCtx: unknown, content: string): Promise /** Proactively push to a conversation (used for restored delivery targets). */