diff --git a/.env.example b/.env.example index 832d07b..2ea283d 100644 --- a/.env.example +++ b/.env.example @@ -28,12 +28,19 @@ CASE_LOCAL=1 # DESK_RESOLUTION=1280x800x24 # Browser Drive still takes the key per request (x-openai-key / x-anthropic-key). -# Box key is only for phone ntfy chat (CASE_NTFY_CHAT=1). Topic is a password: -# openssl rand -hex 32 -# CASE_NTFY_CHAT=1 +# The box key below is only for phone chat (Telegram or ntfy). # CASE_DRIVE_PROVIDER=openai # CASE_DRIVE_API_KEY= # CASE_DRIVE_MODEL= + +# Phone chat over Telegram: make a bot with @BotFather, paste its token, start +# the ui, send /start to the bot and it replies with the chat id to put here. +# CASE_TELEGRAM_TOKEN= +# CASE_TELEGRAM_CHAT_ID= + +# Phone chat over ntfy. Topic is a password: +# openssl rand -hex 32 +# CASE_NTFY_CHAT=1 # CASE_NTFY_URL=https://ntfy.sh # CASE_NTFY_TOPIC= # CASE_NTFY_TOKEN= diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6d87181..c86d3ba 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -44,6 +44,8 @@ jobs: npm --prefix web ci --omit=dev # serve.mjs imports openai at module load node web/web-ui/test_serve.mjs node web/web-ui/test_ntfy.mjs + node web/web-ui/test_phone.mjs + node web/web-ui/test_telegram.mjs node web/web-ui/test_nav.mjs node web/web-ui/test_deploy.mjs diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 9e9051e..2e29b1b 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -32,6 +32,9 @@ python3 -m venv .venv && .venv/bin/pip install -r requirements-dev.txt .venv/bin/python tests/test_deskd.py .venv/bin/python tests/test_browse.py node web/web-ui/test_serve.mjs +node web/web-ui/test_phone.mjs +node web/web-ui/test_ntfy.mjs +node web/web-ui/test_telegram.mjs node web/web-ui/test_nav.mjs node web/web-ui/test_deploy.mjs ``` diff --git a/README.md b/README.md index 31c6cdc..63e871b 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,8 @@ What the agent gets, over MCP: link); the machine types it into the site's own login page. The agent and the API never see the password. - **Human handoff**: 2FA codes, captchas and approvals pause the run and reach a - human — in Drive, or on their phone via a one-shot Assist link (ntfy). + human — in Drive, on their phone over Telegram (Approve / Deny buttons, reply + with the code), or via a one-shot Assist link (ntfy). - **Skills**: the agent saves a completed task as a SKILL.md on the computer and follows it next run. Procedural memory that survives reboots. - **Schedules**: recurring headless runs on the computer's own identity. @@ -106,19 +107,72 @@ host and are never copied onto the computer. ### More knobs -Phone notifications for 2FA/approvals (ntfy), CAPTCHA auto-solve, scheduled -runs: all optional, all documented in [.env.example](.env.example). +Phone chat (Telegram or ntfy), CAPTCHA auto-solve, scheduled runs: all +optional, all documented in [.env.example](.env.example). ### Phone chat (optional) -Drive can take tasks from your phone through [ntfy](https://ntfy.sh). Off by -default. Nothing gets exposed: Drive dials out to the ntfy server and posts -replies back. Phone messages run through the same brain and `threads.json` as -the laptop UI, in a thread named `Phone`. +Drive can take tasks from your phone. Off by default. Nothing gets exposed: +Drive dials out and posts replies back. Phone messages run through the same +brain and `threads.json` as the laptop UI, in a thread named `Phone`. Both +channels need a box-side key, since there is no browser to hold one: -ntfy is a pub-sub service. The public server has no accounts: a topic is just -a name, and anyone who knows the name can post and read. The topic name is -your only credential, so mint a long random one and treat it like a password: +``` +CASE_DRIVE_PROVIDER=openai # or anthropic +CASE_DRIVE_API_KEY= +``` + +A pending handoff (2FA code, approval) consumes the next phone message. With +several open, prefix the answer with the handoff id: `h_ab12 483920`. +`approve`, `deny`, `done`, or a bare code with nothing waiting gets back +"Nothing waiting." Text sent while a Phone turn is running steers that turn; +otherwise it starts a task on the box's first computer. + +This is a live channel, not a queue. Telegram holds messages for a Drive that +is down and reports the ones older than ten minutes back as skipped; ntfy +drops them, so send again. + +#### Telegram + +1. In Telegram, open [@BotFather](https://t.me/BotFather), send `/newbot`, + pick any name, and copy the token it gives you. Keep the bot private: + `/setjoingroups` → Disable. +2. Put the token in `.env` and start the UI: + +``` +CASE_TELEGRAM_TOKEN=123456:ABC… +``` + +```bash +docker compose up -d ui +``` + +3. Send `/start` to your bot. It answers with your chat id and the line to + add. Add it to `.env` and restart the UI: + +``` +CASE_TELEGRAM_CHAT_ID=123456789 +``` + +```bash +docker compose up -d ui +``` + +4. Send a task: `what is on the screen?`. The bot shows "typing" while it + works and posts the result (or the error), split at Telegram's message + limit. + +Only your chat can drive the box; every other chat is ignored. Approval +handoffs arrive with Approve / Deny buttons; code handoffs arrive as a prompt +you reply to. Restarting the UI never loses a pending handoff: it is sent +again on reconnect. + +#### ntfy + +[ntfy](https://ntfy.sh) is a pub-sub service. The public server has no +accounts: a topic is just a name, and anyone who knows the name can post and +read. The topic name is your only credential, so mint a long random one and +treat it like a password: ```bash openssl rand -hex 32 @@ -135,8 +189,6 @@ CASE_NTFY_CHAT=1 CASE_NTFY_URL=https://ntfy.sh # or your ntfy server CASE_NTFY_TOPIC= CASE_NTFY_TOKEN= # self-hosted ntfy auth only -CASE_DRIVE_PROVIDER=openai # or anthropic -CASE_DRIVE_API_KEY= ``` ```bash @@ -156,16 +208,6 @@ docker compose up -d ui Drive posts `Working`, then the final text or the error, back to the same topic. Its own posts are tagged so it never reads them back as instructions. -A pending handoff (2FA code, approval) consumes the next phone message. With -several open, prefix the answer with the handoff id: `h_ab12 483920`. -`approve`, `deny`, `done`, or a bare code with nothing waiting gets back -"Nothing waiting." Text sent while a Phone turn is running steers that turn; -otherwise it starts a task on the box's first computer. - -This is a live channel, not a queue. If Drive was down when you sent -something, send it again. The API key sits in the box env for this feature; -the laptop Drive page still uses the key you paste in the page. - ### Token hardening (optional) Copy `.env.example` to `.env`, generate a token, and set `CASE_TOKEN` before @@ -242,7 +284,7 @@ No Docker: ```bash python3 -m venv .venv && .venv/bin/pip install -r requirements-dev.txt for t in tests/test_*.py; do [ "$t" = tests/test_acceptance.py ] || .venv/bin/python "$t"; done -(cd web && npm ci && node web-ui/test_serve.mjs && node web-ui/test_nav.mjs && node web-ui/test_deploy.mjs) +(cd web && npm ci && npm test && node web-ui/test_nav.mjs && node web-ui/test_deploy.mjs) ``` Acceptance tests need a running stack (`tests/test_acceptance.py`). diff --git a/SECURITY.md b/SECURITY.md index 1021094..f471310 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -45,9 +45,12 @@ sandbox; `no-new-privileges` is deliberately not set because passwordless sudo requires setuid. Do not run untrusted code inside a desktop you also use for personal browsing. -(d) **ntfy topics are bearer secrets.** Anyone who knows a topic name can post or -subscribe. Treat topic names like passwords; use random names and rotate if leaked. +(d) **ntfy topics and Telegram bot tokens are bearer secrets.** Anyone who knows +a topic name can post or subscribe; anyone who holds the bot token can read and +send as the bot. Treat both like passwords; use random topic names, and revoke +the token in @BotFather if it leaks. Report vulnerabilities privately via GitHub Security Advisories: https://github.com/case-computers/case/security/advisories/new -Do not file public issues that include tokens, ntfy topics, or vault contents. +Do not file public issues that include tokens, ntfy topics, bot tokens, or vault +contents. diff --git a/compose.yaml b/compose.yaml index d4e440c..084986e 100644 --- a/compose.yaml +++ b/compose.yaml @@ -118,6 +118,8 @@ services: CASE_DRIVE_PROVIDER: ${CASE_DRIVE_PROVIDER:-} CASE_DRIVE_API_KEY: ${CASE_DRIVE_API_KEY:-} CASE_DRIVE_MODEL: ${CASE_DRIVE_MODEL:-} + CASE_TELEGRAM_TOKEN: ${CASE_TELEGRAM_TOKEN:-} + CASE_TELEGRAM_CHAT_ID: ${CASE_TELEGRAM_CHAT_ID:-} volumes: - ui-data:/data ports: diff --git a/web/package.json b/web/package.json index 8be1f52..967637b 100644 --- a/web/package.json +++ b/web/package.json @@ -4,7 +4,7 @@ "type": "module", "scripts": { "start": "node web-ui/serve.mjs", - "test": "node web-ui/test_serve.mjs && node web-ui/test_ntfy.mjs" + "test": "node web-ui/test_serve.mjs && node web-ui/test_phone.mjs && node web-ui/test_ntfy.mjs && node web-ui/test_telegram.mjs" }, "dependencies": { "@anthropic-ai/sdk": "^0.117.1", diff --git a/web/web-ui/ntfy.mjs b/web/web-ui/ntfy.mjs index 9224c88..e5b5768 100644 --- a/web/web-ui/ntfy.mjs +++ b/web/web-ui/ntfy.mjs @@ -1,8 +1,5 @@ // SPDX-License-Identifier: MIT export const OUTBOUND_TAG = 'case-outbound'; -export const PHONE_THREAD_ID = 't_phone'; -const HANDOFF_RE = /^(h_\w+)\s+(.+)$/s; -const RESERVED_RE = /^(approve|deny|done|i'm done|im done|i am done|\d+)$/i; export function ntfyConfig(env = process.env) { const chat = ['1', 'true'].includes(String(env.CASE_NTFY_CHAT || '').trim().toLowerCase()); @@ -50,33 +47,6 @@ export function parseSseData(chunk, carry = '') { return { events, rest }; } -export function parseHandoffReply(text) { - const m = HANDOFF_RE.exec(String(text || '').trim()); - if (m) return { hid: m[1], value: m[2].trim() }; - return { hid: null, value: String(text || '').trim() }; -} - -export function routePhone({ text, pendingIds = [], busy = false }) { - const raw = String(text || '').trim(); - if (!raw) return { type: 'ignore' }; - const parsed = parseHandoffReply(raw); - if (parsed.hid) { - if (!pendingIds.includes(parsed.hid)) { - return { type: 'error', error: `no pending handoff ${parsed.hid}` }; - } - return { type: 'handoff', hid: parsed.hid, value: parsed.value }; - } - if (pendingIds.length === 1) { - return { type: 'handoff', hid: pendingIds[0], value: parsed.value }; - } - if (pendingIds.length > 1) { - return { type: 'error', error: `${pendingIds.length} pending handoffs; prefix with handoff id` }; - } - if (RESERVED_RE.test(parsed.value)) return { type: 'error', error: 'Nothing waiting.' }; - if (busy) return { type: 'steer', text: parsed.value }; - return { type: 'task', text: parsed.value }; -} - export function clipNtfy(s, n = 3500) { const t = String(s || ''); return t.length <= n ? t : `${t.slice(0, n)}\n…open Drive for the rest`; diff --git a/web/web-ui/phone.mjs b/web/web-ui/phone.mjs new file mode 100644 index 0000000..6336e80 --- /dev/null +++ b/web/web-ui/phone.mjs @@ -0,0 +1,35 @@ +// SPDX-License-Identifier: MIT +/** + * Phone chat routing shared by every transport (ntfy, Telegram). A transport + * turns its wire format into text; this decides what the text means. + */ +export const PHONE_THREAD_ID = 't_phone'; +const HANDOFF_RE = /^(h_\w+)\s+(.+)$/s; +const RESERVED_RE = /^(approve|deny|done|i'm done|im done|i am done|\d+)$/i; + +export function parseHandoffReply(text) { + const m = HANDOFF_RE.exec(String(text || '').trim()); + if (m) return { hid: m[1], value: m[2].trim() }; + return { hid: null, value: String(text || '').trim() }; +} + +export function routePhone({ text, pendingIds = [], busy = false }) { + const raw = String(text || '').trim(); + if (!raw) return { type: 'ignore' }; + const parsed = parseHandoffReply(raw); + if (parsed.hid) { + if (!pendingIds.includes(parsed.hid)) { + return { type: 'error', error: `no pending handoff ${parsed.hid}` }; + } + return { type: 'handoff', hid: parsed.hid, value: parsed.value }; + } + if (pendingIds.length === 1) { + return { type: 'handoff', hid: pendingIds[0], value: parsed.value }; + } + if (pendingIds.length > 1) { + return { type: 'error', error: `${pendingIds.length} pending handoffs; prefix with handoff id` }; + } + if (RESERVED_RE.test(parsed.value)) return { type: 'error', error: 'Nothing waiting.' }; + if (busy) return { type: 'steer', text: parsed.value }; + return { type: 'task', text: parsed.value }; +} diff --git a/web/web-ui/serve.mjs b/web/web-ui/serve.mjs index 311cfca..d048738 100644 --- a/web/web-ui/serve.mjs +++ b/web/web-ui/serve.mjs @@ -22,6 +22,8 @@ import { fileURLToPath } from 'node:url'; import OpenAI from 'openai'; import { CASE_TOOLS, caseCall, caseToolPlan, runCaseTool, streamEventToNdjson, tracesFromOutput, chatAuth, envDriveAuth, resolveChatModel, histToAnthropicMessages, anthropicToolLoop, withRateRetry } from './case-tools.mjs'; import * as ntfy from './ntfy.mjs'; +import { PHONE_THREAD_ID, routePhone } from './phone.mjs'; +import * as telegram from './telegram.mjs'; const DIR = path.dirname(fileURLToPath(import.meta.url)); const PORT = Number(process.env.PORT || 4174); @@ -881,10 +883,10 @@ async function steer(req, res) { return json(res, 200, { queued: true }); } export function phoneThread() { - let t = THREADS.get(ntfy.PHONE_THREAD_ID); + let t = THREADS.get(PHONE_THREAD_ID); if (t) return t; t = { - id: ntfy.PHONE_THREAD_ID, title: 'Phone', agent: '', items: [], + id: PHONE_THREAD_ID, title: 'Phone', agent: '', items: [], created: Date.now(), updated: Date.now(), }; THREADS.set(t.id, t); @@ -1215,9 +1217,13 @@ async function chat(req, res) { } } -async function pendingHandoffIds() { +async function pendingHandoffs() { const r = await api('GET', '/handoffs?status=pending', { timeoutMs: 8000 }); - return (r.json?.handoffs || []).map((h) => h.id).filter(Boolean); + return r.json?.handoffs || []; +} + +async function pendingHandoffIds() { + return (await pendingHandoffs()).map((r) => r.id).filter(Boolean); } async function answerHandoff(hid, value) { @@ -1226,43 +1232,46 @@ async function answerHandoff(hid, value) { }); } -async function onPhoneMessage(cfg, auth, model, text) { +const ANSWERED = { approve: 'Approved.', deny: 'Denied.' }; + +async function phoneContext() { const thread = phoneThread(); - let pending = []; - try { pending = await pendingHandoffIds(); } + let pendingIds = []; + try { pendingIds = await pendingHandoffIds(); } catch (err) { console.warn('phone handoffs:', err.message || err); } - const decision = ntfy.routePhone({ - text, pendingIds: pending, busy: CHAT_BUSY.has(thread.id), - }); - const say = (title, message) => ntfy.publish(cfg, { title, message }).catch((err) => { - console.warn('ntfy publish:', err.message || err); - }); + return { thread, pendingIds, busy: CHAT_BUSY.has(thread.id) }; +} + +/** Carry out a routed phone decision. say(kind, text) is the transport's + * reply; kind ∈ error | answered | queued | working | done. */ +async function phoneAct(decision, { auth, model, say }) { + const thread = phoneThread(); if (decision.type === 'ignore') return; - if (decision.type === 'error') return say('[Case] error', decision.error); + if (decision.type === 'error') return say('error', decision.error); if (decision.type === 'handoff') { try { const r = await answerHandoff(decision.hid, decision.value); if (r.status >= 400) { - return say('[Case] error', r.json?.error?.message || `handoff ${r.status}`); + return say('error', r.json?.error?.message || `handoff ${r.status}`); } - return say('[Case] answered', `Answered ${decision.hid}`); + return say('answered', ANSWERED[decision.value] || 'Submitted.'); } catch (err) { - return say('[Case] error', err.message || 'handoff failed'); + return say('error', err.message || 'handoff failed'); } } if (decision.type === 'steer') { const q = STEER.get(thread.id) || []; q.push(decision.text); STEER.set(thread.id, q); - return say('[Case] queued', 'Queued on the running turn'); + return say('queued', 'Queued on the running turn'); } let computerId; try { computerId = await cid(); } - catch (err) { return say('[Case] error', err.message || 'cased unreachable'); } - if (!computerId) return say('[Case] error', 'no computer — create one first'); - if (CHAT_BUSY.has(thread.id)) return say('[Case] error', 'this thread is still running a turn'); + catch (err) { return say('error', err.message || 'cased unreachable'); } + if (!computerId) return say('error', 'no computer — create one first'); + if (CHAT_BUSY.has(thread.id)) return say('error', 'this thread is still running a turn'); CHAT_BUSY.add(thread.id); - await say('[Case] working', 'Working'); + await say('working', 'Working'); let finalText = ''; let errText = ''; const emit = (obj) => { @@ -1282,8 +1291,21 @@ async function onPhoneMessage(cfg, auth, model, text) { } finally { CHAT_BUSY.delete(thread.id); } - if (errText) return say('[Case] error', errText); - return say('[Case] done', finalText || 'done'); + if (errText) return say('error', errText); + return say('done', finalText || 'done'); +} + +const NTFY_TITLES = { + error: '[Case] error', answered: '[Case] answered', queued: '[Case] queued', + working: '[Case] working', done: '[Case] done', +}; + +async function onPhoneMessage(cfg, auth, model, text) { + const { pendingIds, busy } = await phoneContext(); + const say = (kind, message) => ntfy.publish(cfg, { title: NTFY_TITLES[kind], message }).catch((err) => { + console.warn('ntfy publish:', err.message || err); + }); + return phoneAct(routePhone({ text, pendingIds, busy }), { auth, model, say }); } export function startPhoneNtfy(env = process.env) { @@ -1304,6 +1326,149 @@ export function startPhoneNtfy(env = process.env) { return true; } +// ---------- phone: Telegram ---------- +/** cased /v1/events framing: "event: \ndata: \n\n"; comments are + * heartbeats. Returns parsed blocks plus the unterminated tail. */ +export function sseEvents(chunk, carry = '') { + const blocks = (carry + chunk).split('\n\n'); + const rest = blocks.pop() ?? ''; + const events = []; + for (const b of blocks) { + const event = /^event: (.+)$/m.exec(b)?.[1]; + const data = /^data: (.+)$/m.exec(b)?.[1]; + if (!event || !data) continue; + try { events.push({ event, data: JSON.parse(data) }); } catch { /* malformed */ } + } + return { events, rest }; +} + +/** Follow cased's event stream and call onHandoff for each new handoff. + * Reconnects forever; cased restarts must not silence the phone. */ +function watchHandoffs(onHandoff, onConnect) { + let timer = null; + const retry = () => { clearTimeout(timer); timer = setTimeout(connect, 5000); }; + const connect = () => { + const rq = http.get({ + hostname: CASE.hostname, port: CASE.port, path: '/v1/events', + headers: { accept: 'text/event-stream', ...(TOKEN ? { authorization: 'Bearer ' + TOKEN } : {}) }, + }, (rs) => { + if (rs.statusCode !== 200) console.warn('cased events:', rs.statusCode); + else onConnect(); + let carry = ''; + rs.setEncoding('utf8'); + rs.on('data', (chunk) => { + const parsed = sseEvents(chunk, carry); + carry = parsed.rest; + for (const { event, data } of parsed.events) { + if (event === 'handoff_created') onHandoff(data); + } + }); + rs.on('error', () => { /* close follows */ }); + rs.on('close', retry); + }); + rq.on('error', retry); + }; + connect(); +} + +// Force-reply prompt message id → handoff id, so a reply names its handoff. +const CODE_PROMPTS = new Map(); +// Handoff ids already sent to the phone, so a reconnect replay does not repeat +// them. Ids answered from the Drive UI linger; handoffs are rare, it is bytes. +const PUSHED = new Set(); + +async function pushHandoff(cfg, h) { + if (PUSHED.has(h.id)) return; + PUSHED.add(h.id); + const m = telegram.handoffMessage(h); + try { + const sent = await telegram.tgApi(cfg.token, 'sendMessage', { chat_id: cfg.chatId, ...m }); + if (m.reply_markup.force_reply) CODE_PROMPTS.set(Number(sent.message_id), h.id); + } catch (err) { + PUSHED.delete(h.id); + console.warn('telegram handoff:', err.message || err); + } +} + +/** Reply function for phoneAct: "working" shows the typing indicator until + * the next reply; everything else is text, split at Telegram's limit. */ +function telegramSay(cfg) { + let typing = null; + const action = () => telegram.tgApi(cfg.token, 'sendChatAction', { chat_id: cfg.chatId, action: 'typing' }) + .catch(() => { /* cosmetic */ }); + return async (kind, text) => { + clearInterval(typing); + typing = null; + if (kind === 'working') { + action(); + typing = setInterval(action, 4500); + return; + } + for (const part of telegram.chunk(kind === 'error' ? `Error: ${text}` : text)) { + await telegram.tgApi(cfg.token, 'sendMessage', { chat_id: cfg.chatId, text: part }) + .catch((err) => console.warn('telegram send:', err.message || err)); + } + }; +} + +async function onTelegramUpdate(cfg, auth, model, update) { + const parsed = telegram.parseUpdate(update); + if (!parsed) return; + const { chatId, msg } = parsed; + if (chatId !== cfg.chatId) { + if (!cfg.chatId && msg.kind === 'text' && msg.text === '/start') { + await telegram.tgApi(cfg.token, 'sendMessage', { + chat_id: chatId, + text: `This chat's id is ${chatId}. Put CASE_TELEGRAM_CHAT_ID=${chatId} in .env and restart the ui container.`, + }); + } + return; + } + if (msg.kind === 'callback') { + await telegram.tgApi(cfg.token, 'answerCallbackQuery', { callback_query_id: msg.callbackId }) + .catch(() => { /* spinner only */ }); + } + const { pendingIds, busy } = await phoneContext(); + const decision = telegram.routeTelegram({ msg, codePrompts: CODE_PROMPTS, pendingIds, busy }); + const say = telegramSay(cfg); + if (decision.type === 'start') return say('done', 'Paired. Send a task, or answer a prompt here.'); + if (decision.type === 'stale') { + return say('error', `skipped a message sent ${Math.round(decision.age / 60)} min ago while Drive was down:\n${decision.text.slice(0, 300)}\nSend it again if you still want it.`); + } + if (decision.type === 'handoff') { + for (const [mid, hid] of CODE_PROMPTS) if (hid === decision.hid) CODE_PROMPTS.delete(mid); + PUSHED.delete(decision.hid); + } + return phoneAct(decision, { auth, model, say }); +} + +export function startPhoneTelegram(env = process.env) { + const cfg = telegram.telegramConfig(env); + if (!cfg.token) { + if (env.CASE_TELEGRAM_TOKEN) console.warn('CASE_TELEGRAM_TOKEN is not a BotFather token'); + return false; + } + const auth = envDriveAuth(env); + if (!auth.key) { + console.warn('CASE_TELEGRAM_TOKEN set but CASE_DRIVE_API_KEY unset'); + return false; + } + const model = resolveChatModel(env.CASE_DRIVE_MODEL || '', auth.provider); + // A webhook left on the bot makes getUpdates 409 forever; clearing it is idempotent. + telegram.tgApi(cfg.token, 'deleteWebhook', {}).catch(() => { /* poll reports 409 if this failed */ }) + .then(() => telegram.poll(cfg, (u) => onTelegramUpdate(cfg, auth, model, u))); + if (cfg.chatId) { + watchHandoffs((h) => pushHandoff(cfg, h), () => { + pendingHandoffs().then((rows) => rows.forEach((h) => pushHandoff(cfg, h))) + .catch((err) => console.warn('phone handoffs:', err.message || err)); + }); + } + console.log(cfg.chatId + ? 'drive telegram chat on' + : 'drive telegram: send /start to the bot, then set CASE_TELEGRAM_CHAT_ID'); + return true; +} + // ---------- noVNC proxy (compose network or host-mapped vnc_port) ---------- function vncUpstream(req) { if (livePathHasDotDot(req.url)) return null; @@ -1423,5 +1588,6 @@ if (isMain) { server.listen(PORT, BIND, () => { process.stdout.write(`drive http://${BIND}:${PORT}/ deploy http://${BIND}:${PORT}/deploy (cased ${CASE.hostname}:${CASE.port}${LOCAL ? ', local' : ''})\n`); startPhoneNtfy(); + startPhoneTelegram(); }); } diff --git a/web/web-ui/telegram.mjs b/web/web-ui/telegram.mjs new file mode 100644 index 0000000..0ea1893 --- /dev/null +++ b/web/web-ui/telegram.mjs @@ -0,0 +1,135 @@ +// SPDX-License-Identifier: MIT +/** + * Telegram transport for phone chat. The box owns the bot (CASE_TELEGRAM_TOKEN) + * and long-polls getUpdates, so nothing listens on a public port. Exactly one + * chat (CASE_TELEGRAM_CHAT_ID) may drive it; every other chat is ignored. + */ +import { routePhone } from './phone.mjs'; + +export const STALE_S = 600; +export const MAX_TEXT = 4000; +const CALLBACK_RE = /^h:(h_\w+):(approve|deny)$/; +const TOKEN_RE = /^\d+:[\w-]+$/; // BotFather shape; anything else never reaches a URL + +export function telegramConfig(env = process.env) { + const raw = String(env.CASE_TELEGRAM_TOKEN || '').trim(); + return { + token: TOKEN_RE.test(raw) ? raw : '', + chatId: Number(env.CASE_TELEGRAM_CHAT_ID) || 0, + }; +} + +export async function tgApi(token, method, body, fetchImpl = fetch, timeoutMs = 10000) { + const r = await fetchImpl(`https://api.telegram.org/bot${token}/${method}`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(body), + signal: AbortSignal.timeout(timeoutMs), + }); + const j = await r.json().catch(() => ({})); + if (!j.ok) { + const err = new Error(`telegram ${method} ${r.status}: ${j.description || ''}`.trim()); + err.status = r.status; + throw err; + } + return j.result; +} + +/** One Bot API update → { chatId, msg }, or null for anything that is not + * text or a button tap in a private chat. */ +export function parseUpdate(update) { + const cq = update?.callback_query; + if (cq?.message?.chat?.type === 'private' && typeof cq.data === 'string') { + return { + chatId: Number(cq.message.chat.id), + msg: { kind: 'callback', data: cq.data.slice(0, 64), callbackId: String(cq.id) }, + }; + } + const m = update?.message; + if (!m || m.chat?.type !== 'private' || typeof m.text !== 'string') return null; + return { + chatId: Number(m.chat.id), + msg: { + kind: 'text', + text: m.text.slice(0, MAX_TEXT).trim(), + date: Number(m.date) || 0, + replyTo: Number(m.reply_to_message?.message_id) || 0, + }, + }; +} + +/** Telegram-only shapes first (buttons, reply-to-a-prompt, /start, stale), + * then the shared phone router. Replies to a code prompt carry their handoff + * id, so they skip the stale check and the one-pending guess. */ +export function routeTelegram({ + msg, codePrompts = new Map(), pendingIds = [], busy = false, now = Math.floor(Date.now() / 1000), +}) { + if (msg.kind === 'callback') { + const m = CALLBACK_RE.exec(msg.data); + return m ? { type: 'handoff', hid: m[1], value: m[2] } : { type: 'ignore' }; + } + if (!msg.text) return { type: 'ignore' }; + if (msg.text === '/start') return { type: 'start' }; + if (msg.replyTo && codePrompts.has(msg.replyTo)) { + return { type: 'handoff', hid: codePrompts.get(msg.replyTo), value: msg.text }; + } + const age = now - msg.date; + if (age > STALE_S) return { type: 'stale', text: msg.text, age }; + return routePhone({ text: msg.text, pendingIds, busy }); +} + +export function handoffMessage(h) { + const who = h.domain || 'Your computer'; + const prompt = String(h.prompt || '').slice(0, 1500); + if (h.kind === 'approval') { + return { + text: `${who} needs approval:\n${prompt}`, + reply_markup: { inline_keyboard: [[ + { text: 'Approve', callback_data: `h:${h.id}:approve` }, + { text: 'Deny', callback_data: `h:${h.id}:deny` }, + ]] }, + }; + } + return { + text: `${who} needs you:\n${prompt}\n\nReply to this message with the code, or "done" once you have handled it.`, + reply_markup: { force_reply: true, input_field_placeholder: 'Code' }, + }; +} + +export function chunk(text, n = 4000) { + const s = String(text || ''); + const out = []; + for (let i = 0; i < s.length; i += n) out.push(s.slice(i, i + n)); + return out; +} + +/** getUpdates loop. Updates are handed to onUpdate without awaiting it, so a + * running turn never blocks the next message (that is how steering works). + * Confirmed by the next offset; unconfirmed ones return after a restart and + * routeTelegram drops the stale ones. A 401 means the token is wrong: stop. */ +export async function poll(cfg, onUpdate, { + fetchImpl = fetch, + sleep = (ms) => new Promise((r) => setTimeout(r, ms)), + signal, +} = {}) { + let offset = 0; + while (!signal?.aborted) { + try { + const updates = await tgApi(cfg.token, 'getUpdates', { + offset, timeout: 30, allowed_updates: ['message', 'callback_query'], + }, fetchImpl, 40000); + for (const u of updates) { + offset = u.update_id + 1; + Promise.resolve().then(() => onUpdate(u)).catch((err) => { + console.warn('telegram update:', err?.message || err); + }); + } + continue; + } catch (err) { + if (signal?.aborted) return; + console.warn('telegram poll:', err?.message || err); + if (err?.status === 401) return; + } + await sleep(5000); + } +} diff --git a/web/web-ui/test_ntfy.mjs b/web/web-ui/test_ntfy.mjs index 0688d50..926e8e3 100644 --- a/web/web-ui/test_ntfy.mjs +++ b/web/web-ui/test_ntfy.mjs @@ -4,7 +4,7 @@ import assert from 'node:assert/strict'; import { envDriveAuth } from './case-tools.mjs'; import { OUTBOUND_TAG, authHeaders, clipNtfy, inboundText, isOutbound, listen, - ntfyConfig, parseHandoffReply, parseSseData, publish, routePhone, tagsOf, + ntfyConfig, parseSseData, publish, tagsOf, } from './ntfy.mjs'; import { startPhoneNtfy } from './serve.mjs'; @@ -45,25 +45,6 @@ assert.equal(inboundText({ event: 'message', message: ' hi ' }), 'hi'); assert.match(rest, /"id":"b"/); } -assert.deepEqual(parseHandoffReply('h_abc 482910'), { hid: 'h_abc', value: '482910' }); -assert.deepEqual(parseHandoffReply('approve'), { hid: null, value: 'approve' }); - -assert.deepEqual(routePhone({ text: 'h_1 approve', pendingIds: ['h_1'] }), - { type: 'handoff', hid: 'h_1', value: 'approve' }); -assert.deepEqual(routePhone({ text: '482910', pendingIds: ['h_9'] }), - { type: 'handoff', hid: 'h_9', value: '482910' }); -assert.equal(routePhone({ text: 'ok', pendingIds: ['h_1', 'h_2'] }).type, 'error'); -assert.equal(routePhone({ text: 'h_nope x', pendingIds: ['h_1'] }).type, 'error'); -assert.equal(routePhone({ text: 'approve', pendingIds: [] }).error, 'Nothing waiting.'); -assert.equal(routePhone({ text: 'done', pendingIds: [] }).error, 'Nothing waiting.'); -assert.equal(routePhone({ text: '123456', pendingIds: [] }).error, 'Nothing waiting.'); -assert.deepEqual(routePhone({ text: 'check gmail', pendingIds: [], busy: true }), - { type: 'steer', text: 'check gmail' }); -assert.deepEqual(routePhone({ text: 'check gmail', pendingIds: [] }), - { type: 'task', text: 'check gmail' }); -assert.equal(routePhone({ text: 'check gmail', pendingIds: ['h_1'] }).type, 'handoff'); -assert.equal(routePhone({ text: '' }).type, 'ignore'); - assert.ok(clipNtfy('x'.repeat(4000)).includes('open Drive for the rest')); assert.equal(clipNtfy('short'), 'short'); diff --git a/web/web-ui/test_phone.mjs b/web/web-ui/test_phone.mjs new file mode 100644 index 0000000..c5fa9c9 --- /dev/null +++ b/web/web-ui/test_phone.mjs @@ -0,0 +1,27 @@ +#!/usr/bin/env node +// SPDX-License-Identifier: MIT +import assert from 'node:assert/strict'; +import { PHONE_THREAD_ID, parseHandoffReply, routePhone } from './phone.mjs'; + +assert.equal(PHONE_THREAD_ID, 't_phone'); + +assert.deepEqual(parseHandoffReply('h_abc 482910'), { hid: 'h_abc', value: '482910' }); +assert.deepEqual(parseHandoffReply('approve'), { hid: null, value: 'approve' }); + +assert.deepEqual(routePhone({ text: 'h_1 approve', pendingIds: ['h_1'] }), + { type: 'handoff', hid: 'h_1', value: 'approve' }); +assert.deepEqual(routePhone({ text: '482910', pendingIds: ['h_9'] }), + { type: 'handoff', hid: 'h_9', value: '482910' }); +assert.equal(routePhone({ text: 'ok', pendingIds: ['h_1', 'h_2'] }).type, 'error'); +assert.equal(routePhone({ text: 'h_nope x', pendingIds: ['h_1'] }).type, 'error'); +assert.equal(routePhone({ text: 'approve', pendingIds: [] }).error, 'Nothing waiting.'); +assert.equal(routePhone({ text: 'done', pendingIds: [] }).error, 'Nothing waiting.'); +assert.equal(routePhone({ text: '123456', pendingIds: [] }).error, 'Nothing waiting.'); +assert.deepEqual(routePhone({ text: 'check gmail', pendingIds: [], busy: true }), + { type: 'steer', text: 'check gmail' }); +assert.deepEqual(routePhone({ text: 'check gmail', pendingIds: [] }), + { type: 'task', text: 'check gmail' }); +assert.equal(routePhone({ text: 'check gmail', pendingIds: ['h_1'] }).type, 'handoff'); +assert.equal(routePhone({ text: '' }).type, 'ignore'); + +console.log('ok test_phone.mjs'); diff --git a/web/web-ui/test_serve.mjs b/web/web-ui/test_serve.mjs index 0f36269..549dcb1 100644 --- a/web/web-ui/test_serve.mjs +++ b/web/web-ui/test_serve.mjs @@ -7,7 +7,7 @@ import fs from 'node:fs'; import os from 'node:os'; import path from 'node:path'; import { fileURLToPath } from 'node:url'; -import { shq, pathOk, parseFind, mimeFor, histTrim, histCloseOpenCalls, normHost, threadTurns, parseCaseUrl, liveCid, liveDestPath, livePathHasDotDot, tokenMatches, liveTarget, extraPlan, isLocalMode, pageFile, clip, snapshotElide, stashShot, hydrateShots, migrateShots, stashAttach, resolveAttach, hydrateAttaches, attachKind, ATTACH_MAX } from './serve.mjs'; +import { shq, pathOk, parseFind, mimeFor, histTrim, histCloseOpenCalls, normHost, threadTurns, parseCaseUrl, liveCid, liveDestPath, livePathHasDotDot, tokenMatches, liveTarget, extraPlan, isLocalMode, pageFile, clip, snapshotElide, stashShot, hydrateShots, migrateShots, stashAttach, resolveAttach, hydrateAttaches, attachKind, ATTACH_MAX, sseEvents } from './serve.mjs'; import { CASE_TOOLS, chatAuth, resolveChatModel, openaiToolsToAnthropic, newAnthropicStreamCtx, anthropicEventToNdjson, tracesFromAnthropicMessage, @@ -507,4 +507,18 @@ assert.equal(pageFile('/deploy.html'), '/deploy.html'); for (const { log } of [a, b, c]) fs.rmSync(log, { force: true }); } +{ + const { events, rest } = sseEvents( + ': connected\n\n' + + 'event: handoff_created\ndata: {"id":"h_1","kind":"approval"}\n\n' + + ': hb\n\n' + + 'event: credential_added\ndata: {"name":"x"}\n\nevent: handoff_cre', + ); + assert.deepEqual(events, [ + { event: 'handoff_created', data: { id: 'h_1', kind: 'approval' } }, + { event: 'credential_added', data: { name: 'x' } }, + ]); + assert.equal(rest, 'event: handoff_cre'); +} + console.log('web-ui serve: all checks pass'); diff --git a/web/web-ui/test_telegram.mjs b/web/web-ui/test_telegram.mjs new file mode 100644 index 0000000..9ebe175 --- /dev/null +++ b/web/web-ui/test_telegram.mjs @@ -0,0 +1,136 @@ +#!/usr/bin/env node +// SPDX-License-Identifier: MIT +import assert from 'node:assert/strict'; +import { + MAX_TEXT, STALE_S, chunk, handoffMessage, parseUpdate, poll, routeTelegram, telegramConfig, tgApi, +} from './telegram.mjs'; +import { startPhoneTelegram } from './serve.mjs'; + +assert.deepEqual(telegramConfig({}), { token: '', chatId: 0 }); +assert.deepEqual(telegramConfig({ CASE_TELEGRAM_TOKEN: ' 1:tok ', CASE_TELEGRAM_CHAT_ID: '42' }), + { token: '1:tok', chatId: 42 }); +assert.equal(telegramConfig({ CASE_TELEGRAM_CHAT_ID: 'abc' }).chatId, 0); +assert.equal(telegramConfig({ CASE_TELEGRAM_TOKEN: '12 34:abc' }).token, ''); + +{ + const calls = []; + const fetchImpl = async (url, opts) => { + calls.push({ url, opts }); + return { status: 200, json: async () => ({ ok: true, result: { message_id: 7 } }) }; + }; + const r = await tgApi('tok', 'sendMessage', { chat_id: 1, text: 'hi' }, fetchImpl); + assert.deepEqual(r, { message_id: 7 }); + assert.equal(calls[0].url, 'https://api.telegram.org/bottok/sendMessage'); + assert.equal(calls[0].opts.method, 'POST'); + assert.equal(JSON.parse(calls[0].opts.body).text, 'hi'); +} +{ + const fetchImpl = async () => ({ status: 409, json: async () => ({ ok: false, description: 'Conflict: webhook active' }) }); + await assert.rejects(() => tgApi('tok', 'getUpdates', {}, fetchImpl), (err) => { + assert.equal(err.status, 409); + assert.match(err.message, /getUpdates 409: Conflict: webhook active/); + assert.doesNotMatch(err.message, /tok/); + return true; + }); +} + +const text = (t, extra = {}) => ({ + message: { message_id: 5, date: 1000, text: t, chat: { id: 42, type: 'private' }, ...extra }, +}); +assert.deepEqual(parseUpdate(text('check mail')), { + chatId: 42, msg: { kind: 'text', text: 'check mail', date: 1000, replyTo: 0 }, +}); +assert.deepEqual(parseUpdate(text(' 483920 ', { reply_to_message: { message_id: 3 } })).msg, + { kind: 'text', text: '483920', date: 1000, replyTo: 3 }); +assert.equal(parseUpdate(text('x'.repeat(MAX_TEXT + 5))).msg.text.length, MAX_TEXT); +assert.equal(parseUpdate({ message: { message_id: 1, date: 1, text: 'x', chat: { id: 1, type: 'group' } } }), null); +assert.equal(parseUpdate({ message: { message_id: 1, date: 1, chat: { id: 1, type: 'private' }, photo: [] } }), null); +assert.equal(parseUpdate({ edited_message: {} }), null); +assert.deepEqual(parseUpdate({ + callback_query: { id: 'cb1', data: 'h:h_ab12:approve', message: { message_id: 9, chat: { id: 42, type: 'private' } } }, +}), { chatId: 42, msg: { kind: 'callback', data: 'h:h_ab12:approve', callbackId: 'cb1' } }); + +const now = 5000; +const msg = (t, extra = {}) => ({ kind: 'text', text: t, date: now - 5, replyTo: 0, ...extra }); +assert.deepEqual(routeTelegram({ msg: msg(''), now }), { type: 'ignore' }); +assert.deepEqual(routeTelegram({ msg: msg('/start'), now }), { type: 'start' }); +assert.deepEqual(routeTelegram({ msg: msg('check mail'), now }), { type: 'task', text: 'check mail' }); +assert.deepEqual(routeTelegram({ msg: msg('faster'), busy: true, now }), { type: 'steer', text: 'faster' }); +assert.deepEqual(routeTelegram({ msg: msg('old', { date: now - STALE_S - 1 }), now }), + { type: 'stale', text: 'old', age: STALE_S + 1 }); +assert.deepEqual(routeTelegram({ msg: msg('483920'), pendingIds: ['h_9'], now }), + { type: 'handoff', hid: 'h_9', value: '483920' }); +// A reply to a code prompt names its handoff, even when stale or ambiguous. +assert.deepEqual(routeTelegram({ + msg: msg('483920', { replyTo: 11, date: now - 900 }), codePrompts: new Map([[11, 'h_ab12']]), + pendingIds: ['h_ab12', 'h_zz'], now, +}), { type: 'handoff', hid: 'h_ab12', value: '483920' }); +assert.deepEqual(routeTelegram({ msg: { kind: 'callback', data: 'h:h_ab12:deny', callbackId: 'c1' }, now }), + { type: 'handoff', hid: 'h_ab12', value: 'deny' }); +assert.deepEqual(routeTelegram({ msg: { kind: 'callback', data: 'junk', callbackId: 'c2' }, now }), { type: 'ignore' }); + +assert.deepEqual(handoffMessage({ id: 'h_1', kind: 'approval', prompt: 'Buy 3 licences?', domain: 'coupa.com' }), { + text: 'coupa.com needs approval:\nBuy 3 licences?', + reply_markup: { inline_keyboard: [[ + { text: 'Approve', callback_data: 'h:h_1:approve' }, + { text: 'Deny', callback_data: 'h:h_1:deny' }, + ]] }, +}); +assert.deepEqual(handoffMessage({ id: 'h_2', kind: 'question', prompt: 'Enter the 6-digit code', domain: null }), { + text: 'Your computer needs you:\nEnter the 6-digit code\n\nReply to this message with the code, or "done" once you have handled it.', + reply_markup: { force_reply: true, input_field_placeholder: 'Code' }, +}); +assert.equal(handoffMessage({ id: 'h_3', kind: 'approval', prompt: 'p'.repeat(2000) }).text.length, 'Your computer needs approval:\n'.length + 1500); + +assert.deepEqual(chunk('a'.repeat(8100)).map((s) => s.length), [4000, 4000, 100]); +assert.deepEqual(chunk(''), []); +assert.deepEqual(chunk('x', 1), ['x']); + +{ + // One batch, then a 409 that stops the loop through sleep(). + const bodies = []; + const got = []; + const ac = new AbortController(); + let n = 0; + const fetchImpl = async (url, opts) => { + bodies.push(JSON.parse(opts.body)); + n += 1; + if (n === 1) return { status: 200, json: async () => ({ ok: true, result: [{ update_id: 10, a: 1 }, { update_id: 11, a: 2 }] }) }; + return { status: 409, json: async () => ({ ok: false, description: 'Conflict' }) }; + }; + await poll({ token: 'tok' }, (u) => { got.push(u.update_id); }, { + fetchImpl, signal: ac.signal, sleep: async () => { ac.abort(); }, + }); + assert.deepEqual(got, [10, 11]); + assert.equal(bodies[0].offset, 0); + assert.equal(bodies[0].timeout, 30); + assert.deepEqual(bodies[0].allowed_updates, ['message', 'callback_query']); + assert.equal(bodies[1].offset, 12); +} +{ + // A bad token ends the loop instead of retrying forever. + let n = 0; + const fetchImpl = async () => { n += 1; return { status: 401, json: async () => ({ ok: false, description: 'Unauthorized' }) }; }; + await poll({ token: 'tok' }, () => {}, { fetchImpl, sleep: async () => { throw new Error('should not sleep'); } }); + assert.equal(n, 1); +} +{ + // Handler failures are logged, not fatal, and do not block the next update. + const got = []; + const ac = new AbortController(); + let n = 0; + const fetchImpl = async () => { + n += 1; + if (n === 1) return { status: 200, json: async () => ({ ok: true, result: [{ update_id: 1 }, { update_id: 2 }] }) }; + ac.abort(); + return { status: 200, json: async () => ({ ok: true, result: [] }) }; + }; + await poll({ token: 'tok' }, async (u) => { got.push(u.update_id); if (u.update_id === 1) throw new Error('boom'); }, + { fetchImpl, signal: ac.signal, sleep: async () => {} }); + assert.deepEqual(got, [1, 2]); +} + +assert.equal(startPhoneTelegram({}), false); +assert.equal(startPhoneTelegram({ CASE_TELEGRAM_TOKEN: '1:tok' }), false); + +console.log('ok test_telegram.mjs');