diff --git a/.env.example b/.env.example index bdacdfb..832d07b 100644 --- a/.env.example +++ b/.env.example @@ -27,14 +27,17 @@ CASE_LOCAL=1 # Desktop resolution for new/woken computers (WxH or WxHxDEPTH). # DESK_RESOLUTION=1280x800x24 -# OpenAI key stays in the Drive page (x-openai-key). Do not put it here. - -# Optional phone notifications for handoffs (2FA codes, approvals) via ntfy -# (https://ntfy.sh or self-hosted). Topic names are bearer secrets — pick long -# random ones. ANSWER_TOPIC lets you reply from the phone. Unset = disabled -# (handoffs still show up in Drive and the API). +# 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 +# CASE_DRIVE_PROVIDER=openai +# CASE_DRIVE_API_KEY= +# CASE_DRIVE_MODEL= # CASE_NTFY_URL=https://ntfy.sh # CASE_NTFY_TOPIC= +# CASE_NTFY_TOKEN= +# Legacy: separate reply topic for handoff answers only. Must not equal TOPIC. # CASE_NTFY_ANSWER_TOPIC= # Public hostname when a reverse proxy fronts the API (adds /assist links to diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ff02db2..6d87181 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -43,6 +43,7 @@ jobs: run: | 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_nav.mjs node web/web-ui/test_deploy.mjs diff --git a/LICENSE.md b/LICENSE.md index 97891e4..0b69b7e 100644 --- a/LICENSE.md +++ b/LICENSE.md @@ -57,7 +57,7 @@ license picker does not split by directory. The table above is authoritative. ## Trademark -**Case** and the Case logo are trademarks of Daemon Labs. Neither the AGPL nor +**Case** and the Case logo are trademarks of Case. Neither the AGPL nor the MIT license grants any right to use them, and this file grants none either. You may state accurately that your software is built on, derived from, or diff --git a/LICENSES/MIT.txt b/LICENSES/MIT.txt index 9353314..0a5235c 100644 --- a/LICENSES/MIT.txt +++ b/LICENSES/MIT.txt @@ -1,4 +1,4 @@ -Copyright (c) 2026 Case contributors +Copyright (c) 2026 Case Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal diff --git a/NOTICE b/NOTICE index b1ebc78..9209da3 100644 --- a/NOTICE +++ b/NOTICE @@ -1,5 +1,5 @@ Case -Copyright (c) 2026 Case contributors +Copyright (c) 2026 Case This product includes software developed by third parties. Case does not relicense any of it; each component remains under its own terms. This file diff --git a/README.md b/README.md index 21fa8b6..31c6cdc 100644 --- a/README.md +++ b/README.md @@ -109,6 +109,63 @@ host and are never copied onto the computer. Phone notifications for 2FA/approvals (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`. + +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: + +```bash +openssl rand -hex 32 +``` + +1. Install the ntfy app (Play Store / App Store) and subscribe to that topic. + Self-hosting ntfy instead? Point the app and `CASE_NTFY_URL` at your + server; `CASE_NTFY_TOKEN` carries the bearer token if your server uses + ntfy access control. +2. Configure the box (`.env`) and restart the UI container: + +``` +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 +docker compose up -d ui +``` + +3. Send a message. + - Android: the ntfy app has a message bar at the bottom of the topic view + (Settings > Show message bar if it's hidden). + - iOS: the app only receives. Make a Shortcut: Ask for Input, then Get + Contents of URL with method POST, the input as the request body, and + `https://ntfy.sh/` as the URL. Add it to the home screen or run + it with Siri. + - Any machine: `curl -d "check my mail" ntfy.sh/`. Useful to test + the bridge before involving the phone. + +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 @@ -130,8 +187,8 @@ docker compose down ### Separate database warning `bin/case up` runs the control plane on the host with its database in `~/.case`. -Compose uses a Docker volume instead. Same engine, same desktops, different -bookkeeping: computers you create one way are not listed by the other, and both +Compose uses a Docker volume instead. Same engine, same desktops, two separate +databases: computers you create one way are not listed by the other, and both want port 8787, so run one at a time. ### RAM budget diff --git a/compose.yaml b/compose.yaml index 16046d4..d4e440c 100644 --- a/compose.yaml +++ b/compose.yaml @@ -50,6 +50,7 @@ services: CASE_NTFY_URL: ${CASE_NTFY_URL:-https://ntfy.sh} CASE_NTFY_TOPIC: ${CASE_NTFY_TOPIC:-} CASE_NTFY_ANSWER_TOPIC: ${CASE_NTFY_ANSWER_TOPIC:-} + CASE_NTFY_TOKEN: ${CASE_NTFY_TOKEN:-} CASE_PUBLIC_HOST: ${CASE_PUBLIC_HOST:-} # Optional CAPTCHA auto-solve (see .env.example) CASE_DBC_AUTHTOKEN: ${CASE_DBC_AUTHTOKEN:-} @@ -110,6 +111,13 @@ services: CASE_HOME: /data CASE_LOCAL: "1" PORT: "4174" + CASE_NTFY_URL: ${CASE_NTFY_URL:-https://ntfy.sh} + CASE_NTFY_TOPIC: ${CASE_NTFY_TOPIC:-} + CASE_NTFY_TOKEN: ${CASE_NTFY_TOKEN:-} + CASE_NTFY_CHAT: ${CASE_NTFY_CHAT:-} + CASE_DRIVE_PROVIDER: ${CASE_DRIVE_PROVIDER:-} + CASE_DRIVE_API_KEY: ${CASE_DRIVE_API_KEY:-} + CASE_DRIVE_MODEL: ${CASE_DRIVE_MODEL:-} volumes: - ui-data:/data ports: diff --git a/control-plane/notify.py b/control-plane/notify.py index a50ffa6..38bc8be 100644 --- a/control-plane/notify.py +++ b/control-plane/notify.py @@ -19,6 +19,23 @@ from config import API_BASE log = logging.getLogger("cased.notify") +OUTBOUND_TAG = "case-outbound" + + +def _ntfy_token(): + return (os.environ.get("CASE_NTFY_TOKEN") or "").strip() + + +def _auth_headers(): + token = _ntfy_token() + return {"Authorization": f"Bearer {token}"} if token else {} + + +def _tags(ev): + raw = ev.get("tags") or [] + if isinstance(raw, str): + return [x.strip() for x in raw.split(",") if x.strip()] + return [str(x) for x in raw] class Ntfy: @@ -29,6 +46,11 @@ def __init__(self, url, topic, answer_topic, api_base): self.api_base = api_base if not topic: log.warning("CASE_NTFY_TOPIC unset — handoff notifications disabled") + if topic and answer_topic and topic == answer_topic: + log.warning("CASE_NTFY_ANSWER_TOPIC equals CASE_NTFY_TOPIC — answer listen disabled") + + def _same_topic(self): + return bool(self.topic and self.answer_topic and self.topic == self.answer_topic) def notify(self, handoff, computer_name): if not self.topic: @@ -38,9 +60,13 @@ def notify(self, handoff, computer_name): def _send(self, h, computer_name): try: ascii_ = lambda s: (s or "").encode("ascii", "replace").decode() + tags = [OUTBOUND_TAG] + if h.get("id"): + tags.append(h["id"]) headers = { + **_auth_headers(), "X-Title": ascii_(f"[Case] {h['kind']} — {computer_name}"), - "X-Tags": h["id"], + "X-Tags": ",".join(tags), "X-Message": ascii_(h["prompt"])[:800], } if h["kind"] == "approval": @@ -64,21 +90,24 @@ def _p(): try: requests.post(f"{self.url}/{self.topic}", data=(text or "").encode("ascii", "replace")[:1000], - headers={"X-Title": "Case run"}, timeout=15) + headers={**_auth_headers(), "X-Title": "Case run", + "X-Tags": OUTBOUND_TAG}, timeout=15) except Exception as e: log.warning("ntfy push failed: %s", e) threading.Thread(target=_p, daemon=True).start() def listen(self, on_answer): """Subscribe to the answer topic (SSE); messages are '{handoff_id} {value}' or a bare value.""" - if not self.answer_topic: + if not self.answer_topic or self._same_topic(): return threading.Thread(target=self._listen, args=(on_answer,), daemon=True).start() def _listen(self, on_answer): + headers = _auth_headers() while True: try: - r = requests.get(f"{self.url}/{self.answer_topic}/sse", stream=True, timeout=(10, None)) + r = requests.get(f"{self.url}/{self.answer_topic}/sse", stream=True, + headers=headers, timeout=(10, None)) for line in r.iter_lines(): if not line or not line.startswith(b"data: "): continue @@ -86,7 +115,7 @@ def _listen(self, on_answer): ev = json.loads(line[6:]) except ValueError: continue - if ev.get("event") != "message": + if ev.get("event") != "message" or OUTBOUND_TAG in _tags(ev): continue msg = (ev.get("message") or "").strip() m = re.match(r"^(h_\w+)\s+(.+)$", msg, re.S) diff --git a/tests/test_notify.py b/tests/test_notify.py index 1ff8499..415b62a 100644 --- a/tests/test_notify.py +++ b/tests/test_notify.py @@ -52,6 +52,52 @@ def fake_post(url, **kw): assert done.wait(2), "ntfy thread did not run" assert posted["url"] == "https://ntfy.sh/topic-x" assert "h_1" in posted["headers"].get("X-Tags", "") + assert "case-outbound" in posted["headers"].get("X-Tags", "") + + +def test_ntfy_notify_sends_bearer_token(): + os.environ["CASE_NTFY_TOKEN"] = "secret-tok" + ntfy = notify.Ntfy("https://ntfy.sh", "topic-x", None, "http://127.0.0.1:8787/v1") + done = threading.Event() + posted = {} + + def fake_post(url, **kw): + posted["headers"] = kw.get("headers") + done.set() + return mock.Mock(status_code=200) + + try: + with mock.patch.object(notify.requests, "post", side_effect=fake_post): + ntfy.notify({"id": "h_1", "kind": "question", "prompt": "hi", + "screenshot": None}, "box") + assert done.wait(2), "ntfy thread did not run" + assert posted["headers"]["Authorization"] == "Bearer secret-tok" + assert "case-outbound" in posted["headers"]["X-Tags"] + finally: + os.environ.pop("CASE_NTFY_TOKEN", None) + + +def test_same_topic_does_not_start_answer_listen(): + ntfy = notify.Ntfy("https://ntfy.sh", "same", "same", "http://127.0.0.1:8787/v1") + with mock.patch.object(notify.threading, "Thread") as th: + ntfy.listen(lambda *a: None) + th.assert_not_called() + + +def test_push_marks_outbound(): + ntfy = notify.Ntfy("https://ntfy.sh", "topic-x", None, "http://127.0.0.1:8787/v1") + done = threading.Event() + posted = {} + + def fake_post(url, **kw): + posted["headers"] = kw.get("headers") + done.set() + return mock.Mock(status_code=200) + + with mock.patch.object(notify.requests, "post", side_effect=fake_post): + ntfy.push("run finished ok") + assert done.wait(2), "ntfy thread did not run" + assert posted["headers"].get("X-Tags") == "case-outbound" def test_create_handoff_mints_assist_and_passes_url_to_notifier(): diff --git a/web/package.json b/web/package.json index 0bee8a8..8be1f52 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" + "test": "node web-ui/test_serve.mjs && node web-ui/test_ntfy.mjs" }, "dependencies": { "@anthropic-ai/sdk": "^0.117.1", diff --git a/web/web-ui/case-tools.mjs b/web/web-ui/case-tools.mjs index 51a747c..4656093 100644 --- a/web/web-ui/case-tools.mjs +++ b/web/web-ui/case-tools.mjs @@ -193,6 +193,13 @@ export function chatAuth(headers = {}) { return { provider: '', key: '' }; } +export function envDriveAuth(env = process.env) { + const key = String(env.CASE_DRIVE_API_KEY || '').trim(); + const provider = String(env.CASE_DRIVE_PROVIDER || '').trim().toLowerCase(); + if (!key || (provider !== 'openai' && provider !== 'anthropic')) return { provider: '', key: '' }; + return { provider, key }; +} + export function resolveChatModel(requested, provider) { if (provider === 'anthropic') return ANTHROPIC_MODELS[requested] || 'claude-sonnet-4-6'; return MODELS[requested] || 'gpt-5.6-terra'; diff --git a/web/web-ui/ntfy.mjs b/web/web-ui/ntfy.mjs new file mode 100644 index 0000000..9224c88 --- /dev/null +++ b/web/web-ui/ntfy.mjs @@ -0,0 +1,144 @@ +// 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()); + return { + url: String(env.CASE_NTFY_URL || 'https://ntfy.sh').replace(/\/+$/, ''), + topic: String(env.CASE_NTFY_TOPIC || '').trim(), + token: String(env.CASE_NTFY_TOKEN || '').trim(), + chat, + }; +} + +export function authHeaders(token) { + return token ? { Authorization: `Bearer ${token}` } : {}; +} + +export function tagsOf(ev) { + const t = ev?.tags; + if (Array.isArray(t)) return t.map(String); + if (typeof t === 'string') return t.split(',').map((s) => s.trim()).filter(Boolean); + return []; +} + +export function isOutbound(ev) { + return tagsOf(ev).includes(OUTBOUND_TAG); +} + +export function inboundText(ev) { + if (!ev || ev.event !== 'message' || isOutbound(ev)) return ''; + return String(ev.message || '').trim(); +} + +export function parseSseData(chunk, carry = '') { + const text = carry + chunk; + const parts = text.split('\n\n'); + const rest = parts.pop() ?? ''; + const events = []; + for (const block of parts) { + const data = block.split('\n') + .filter((l) => l.startsWith('data:')) + .map((l) => l.slice(5).trimStart()) + .join('\n'); + if (!data) continue; + try { events.push(JSON.parse(data)); } catch { /* keep-alive / malformed */ } + } + 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`; +} + +function asciiHeader(s) { + return String(s || '').replace(/[^\x20-\x7e]/g, '?'); +} + +export async function publish(cfg, { title, message, tags = [] }, fetchImpl = fetch) { + if (!cfg?.topic) return; + const headers = { + ...authHeaders(cfg.token), + 'X-Title': asciiHeader(title).slice(0, 200), + 'X-Tags': [OUTBOUND_TAG, ...tags].join(','), + 'Content-Type': 'text/plain; charset=utf-8', + }; + const r = await fetchImpl(`${cfg.url}/${encodeURIComponent(cfg.topic)}`, { + method: 'POST', + headers, + body: clipNtfy(message), + }); + if (!r.ok) throw new Error(`ntfy publish ${r.status}`); +} + +export async function listen(cfg, onMessage, { + fetchImpl = fetch, + sleep = (ms) => new Promise((r) => setTimeout(r, ms)), + signal, + now = () => Math.floor(Date.now() / 1000), +} = {}) { + if (!cfg?.topic) return; + let since = String(now()); + const seen = new Set(); + const dec = new TextDecoder(); + while (!signal?.aborted) { + try { + const r = await fetchImpl(`${cfg.url}/${encodeURIComponent(cfg.topic)}/sse?since=${encodeURIComponent(since)}`, { + headers: { ...authHeaders(cfg.token), Accept: 'text/event-stream' }, + signal, + }); + if (!r.ok || !r.body) throw new Error(`ntfy subscribe ${r.status}`); + let carry = ''; + for await (const chunk of r.body) { + const { events, rest } = parseSseData(dec.decode(chunk, { stream: true }), carry); + carry = rest; + for (const ev of events) { + if (ev.id) since = ev.id; + const text = inboundText(ev); + if (!text) continue; + if (ev.id && seen.has(ev.id)) continue; + if (ev.id) { + seen.add(ev.id); + if (seen.size > 200) seen.delete(seen.values().next().value); + } + await onMessage(text, ev); + } + } + } catch (err) { + if (signal?.aborted) return; + console.warn('ntfy listen:', err?.message || err); + } + if (!signal?.aborted) await sleep(5000); + } +} diff --git a/web/web-ui/serve.mjs b/web/web-ui/serve.mjs index ce62320..311cfca 100644 --- a/web/web-ui/serve.mjs +++ b/web/web-ui/serve.mjs @@ -20,7 +20,8 @@ import path from 'node:path'; import crypto from 'node:crypto'; import { fileURLToPath } from 'node:url'; import OpenAI from 'openai'; -import { CASE_TOOLS, caseCall, caseToolPlan, runCaseTool, streamEventToNdjson, tracesFromOutput, chatAuth, resolveChatModel, histToAnthropicMessages, anthropicToolLoop, withRateRetry } from './case-tools.mjs'; +import { CASE_TOOLS, caseCall, caseToolPlan, runCaseTool, streamEventToNdjson, tracesFromOutput, chatAuth, envDriveAuth, resolveChatModel, histToAnthropicMessages, anthropicToolLoop, withRateRetry } from './case-tools.mjs'; +import * as ntfy from './ntfy.mjs'; const DIR = path.dirname(fileURLToPath(import.meta.url)); const PORT = Number(process.env.PORT || 4174); @@ -879,57 +880,34 @@ async function steer(req, res) { STEER.set(tid, q); return json(res, 200, { queued: true }); } -async function chat(req, res) { - const buf = await readBody(req, res); - if (!buf) return; - let body; - try { body = JSON.parse(buf.toString('utf8') || '{}'); } - catch { return send(res, 400, 'bad json'); } - const auth = chatAuth(req.headers); - if (!auth.key) return send(res, 401, 'missing key'); - const model = resolveChatModel(body.model, auth.provider); - const effort = ['none', 'low', 'medium', 'high', 'xhigh', 'max'].includes(body.effort) ? body.effort : 'medium'; - const inputText = String(body.input || '').slice(0, 32000); - const fileIds = Array.isArray(body.files) ? body.files.slice(0, ATTACH_MAX_N) : []; - const attaches = []; - for (const ref of fileIds) { - const rec = resolveAttach(typeof ref === 'string' ? ref : ref?.id); - if (!rec) return send(res, 400, 'attachment not found'); - attaches.push({ path: rec.path, name: rec.name, mime: rec.mime }); - } - if (!inputText && !attaches.length) return send(res, 400, 'empty input'); - const picked = String(body.computer_id || ''); - const thread = THREADS.get(String(body.thread_id || '')) || newThread(inputText || attaches[0].name, picked); - if (CHAT_BUSY.has(thread.id)) return json(res, 409, { error: 'this thread is still running a turn' }); - CHAT_BUSY.add(thread.id); - try { +export function phoneThread() { + let t = THREADS.get(ntfy.PHONE_THREAD_ID); + if (t) return t; + t = { + id: ntfy.PHONE_THREAD_ID, title: 'Phone', agent: '', items: [], + created: Date.now(), updated: Date.now(), + }; + THREADS.set(t.id, t); + saveThreads(); + return t; +} + +export async function runTurn({ + thread, inputText, attaches = [], auth, computerId = '', + model, effort = 'medium', emit, stopped = () => false, signal, disconnect, +}) { thread.items = histCloseOpenCalls(thread.items); // Route, don't gate: the thread's own agent wins, then the client's pick, then // the box's computer. Assignment is only *claimed* when a tool actually runs. - let id = thread.agent || String(body.computer_id || ''); + let id = thread.agent || computerId; if (!id) { try { id = await cid(); } catch { /* fall through */ } } - if (!id) { return json(res, 502, { error: 'no computer — create one first' }); } - res.writeHead(200, { - 'content-type': 'application/x-ndjson; charset=utf-8', - 'cache-control': 'no-store', 'x-accel-buffering': 'no', - }); - // STOP in the UI aborts the fetch; stop looping then, but keep the turn's - // history — tools already ran, and replaying them on "continue" double-acts. - const stopped = () => res.destroyed; - const emit = (obj) => { - if (stopped() || res.destroyed) return; - try { res.write(JSON.stringify(obj) + '\n'); } catch { /* client gone */ } - }; - // A long-poll tool (auth wait can block 4+ min) would hold CHAT_BUSY long after - // the client is gone. Race each tool against disconnect; once gone, later tools - // never start. The synthetic output keeps every function_call answered. - let clientGone; - const goneP = new Promise((r) => { clientGone = r; }); - // Tools are raced against disconnect, but a model round is not — and its HTTP - // request has to be torn down explicitly or the SDK keeps draining the stream - // into a dead socket, holding CHAT_BUSY (and billing) long after STOP. - const gone = new AbortController(); - res.on('close', () => { clientGone(); gone.abort(); }); + if (!id) { + const err = new Error('no computer — create one first'); + err.status = 502; + throw err; + } + const goneP = disconnect || new Promise(() => {}); + const gone = signal ? { signal } : new AbortController(); const toolOrStop = (start, act) => stopped() ? Promise.resolve({ ok: false, error: 'stopped by user', act }) : Promise.race([start(), goneP.then(() => ({ ok: false, error: 'stopped by user', act }))]); @@ -1011,8 +989,7 @@ async function chat(req, res) { thread.updated = Date.now(); saveThreads(); emit({ type: 'done', text, computer_id: id, thread_id: thread.id }); - res.end(); - return; + return { text, computerId: id, threadId: thread.id }; } const client = new OpenAI({ apiKey: auth.key }); let text = ''; @@ -1164,6 +1141,7 @@ async function chat(req, res) { + ` eff=${eff} out=${spend.out} rounds=${i}`); spend.eff = eff; emit({ type: 'done', text, computer_id: id, thread_id: thread.id, spend, rounds: i }); + return { text, computerId: id, threadId: thread.id }; } catch (err) { // Keep the turn even on provider errors: tools already ran, that work is // real. histCloseOpenCalls synthesizes outputs for any dangling @@ -1174,19 +1152,156 @@ async function chat(req, res) { thread.updated = Date.now(); saveThreads(); if (!stopped()) emit({ type: 'error', error: (err?.message || 'provider error') + ' — say continue, I pick up where I stopped.' }); - } - res.end(); + return { text: '', computerId: id, threadId: thread.id, error: err?.message || 'provider error' }; } finally { - // Last-round steers missed every drain. Land them in history before the - // inbox is forgotten so a reload or restart still has what the user typed. const leftover = takeSteers(thread.id); if (leftover.length) { pushSteerItems(thread.items, leftover); thread.updated = Date.now(); saveThreads(); } + } +} + +async function chat(req, res) { + const buf = await readBody(req, res); + if (!buf) return; + let body; + try { body = JSON.parse(buf.toString('utf8') || '{}'); } + catch { return send(res, 400, 'bad json'); } + const auth = chatAuth(req.headers); + if (!auth.key) return send(res, 401, 'missing key'); + const model = resolveChatModel(body.model, auth.provider); + const effort = ['none', 'low', 'medium', 'high', 'xhigh', 'max'].includes(body.effort) ? body.effort : 'medium'; + const inputText = String(body.input || '').slice(0, 32000); + const fileIds = Array.isArray(body.files) ? body.files.slice(0, ATTACH_MAX_N) : []; + const attaches = []; + for (const ref of fileIds) { + const rec = resolveAttach(typeof ref === 'string' ? ref : ref?.id); + if (!rec) return send(res, 400, 'attachment not found'); + attaches.push({ path: rec.path, name: rec.name, mime: rec.mime }); + } + if (!inputText && !attaches.length) return send(res, 400, 'empty input'); + const picked = String(body.computer_id || ''); + const thread = THREADS.get(String(body.thread_id || '')) || newThread(inputText || attaches[0].name, picked); + if (CHAT_BUSY.has(thread.id)) return json(res, 409, { error: 'this thread is still running a turn' }); + CHAT_BUSY.add(thread.id); + const stopped = () => res.destroyed; + const emit = (obj) => { + if (stopped() || res.destroyed) return; + if (!res.headersSent) { + res.writeHead(200, { + 'content-type': 'application/x-ndjson; charset=utf-8', + 'cache-control': 'no-store', 'x-accel-buffering': 'no', + }); + } + try { res.write(JSON.stringify(obj) + '\n'); } catch { /* client gone */ } + }; + let clientGone; + const goneP = new Promise((r) => { clientGone = r; }); + const gone = new AbortController(); + res.on('close', () => { clientGone(); gone.abort(); }); + try { + await runTurn({ + thread, inputText, attaches, auth, computerId: picked, model, effort, + emit, stopped, signal: gone.signal, disconnect: goneP, + }); + } catch (err) { + if (!res.headersSent) return json(res, err.status || 500, { error: err.message || 'internal' }); + if (!stopped()) emit({ type: 'error', error: err.message || 'internal' }); + } finally { CHAT_BUSY.delete(thread.id); + if (res.headersSent && !res.writableEnded) res.end(); + } +} + +async function pendingHandoffIds() { + const r = await api('GET', '/handoffs?status=pending', { timeoutMs: 8000 }); + return (r.json?.handoffs || []).map((h) => h.id).filter(Boolean); +} + +async function answerHandoff(hid, value) { + return api('POST', `/handoffs/${encodeURIComponent(hid)}/answer`, { + json: { value }, timeoutMs: 120000, + }); +} + +async function onPhoneMessage(cfg, auth, model, text) { + const thread = phoneThread(); + let pending = []; + try { pending = 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); + }); + if (decision.type === 'ignore') return; + if (decision.type === 'error') return say('[Case] 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('[Case] answered', `Answered ${decision.hid}`); + } catch (err) { + return say('[Case] 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'); + } + 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'); + CHAT_BUSY.add(thread.id); + await say('[Case] working', 'Working'); + let finalText = ''; + let errText = ''; + const emit = (obj) => { + if (obj?.type === 'done') finalText = obj.text || finalText; + if (obj?.type === 'text' && obj.text) finalText = obj.text; + if (obj?.type === 'error') errText = obj.error || 'provider error'; + }; + try { + const result = await runTurn({ + thread, inputText: decision.text, attaches: [], auth, computerId, + model, effort: 'medium', emit, stopped: () => false, + }); + if (result?.error) errText = result.error; + if (result?.text) finalText = result.text; + } catch (err) { + errText = err.message || 'turn failed'; + } finally { + CHAT_BUSY.delete(thread.id); + } + if (errText) return say('[Case] error', errText); + return say('[Case] done', finalText || 'done'); +} + +export function startPhoneNtfy(env = process.env) { + const cfg = ntfy.ntfyConfig(env); + if (!cfg.chat) return false; + if (!cfg.topic) { + console.warn('CASE_NTFY_CHAT=1 but CASE_NTFY_TOPIC unset'); + return false; + } + const auth = envDriveAuth(env); + if (!auth.key) { + console.warn('CASE_NTFY_CHAT=1 but CASE_DRIVE_API_KEY unset'); + return false; } + const model = resolveChatModel(env.CASE_DRIVE_MODEL || '', auth.provider); + ntfy.listen(cfg, (text) => onPhoneMessage(cfg, auth, model, text)); + console.log(`drive ntfy chat on ${cfg.url}`); + return true; } // ---------- noVNC proxy (compose network or host-mapped vnc_port) ---------- @@ -1307,5 +1422,6 @@ if (isMain) { server.on('clientError', (_e, s) => { try { s.destroy(); } catch { /* gone */ } }); 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(); }); } diff --git a/web/web-ui/test_ntfy.mjs b/web/web-ui/test_ntfy.mjs new file mode 100644 index 0000000..0688d50 --- /dev/null +++ b/web/web-ui/test_ntfy.mjs @@ -0,0 +1,173 @@ +#!/usr/bin/env node +// SPDX-License-Identifier: MIT +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, +} from './ntfy.mjs'; +import { startPhoneNtfy } from './serve.mjs'; + +assert.deepEqual(ntfyConfig({}), { + url: 'https://ntfy.sh', topic: '', token: '', chat: false, +}); +assert.deepEqual(ntfyConfig({ + CASE_NTFY_URL: 'https://ntfy.example/', CASE_NTFY_TOPIC: 'abc', + CASE_NTFY_TOKEN: 'tok', CASE_NTFY_CHAT: '1', +}), { url: 'https://ntfy.example', topic: 'abc', token: 'tok', chat: true }); +assert.equal(ntfyConfig({ CASE_NTFY_CHAT: 'true' }).chat, true); +assert.deepEqual(authHeaders('tok'), { Authorization: 'Bearer tok' }); +assert.deepEqual(authHeaders(''), {}); + +assert.deepEqual(envDriveAuth({}), { provider: '', key: '' }); +assert.deepEqual(envDriveAuth({ CASE_DRIVE_API_KEY: 'sk', CASE_DRIVE_PROVIDER: 'openai' }), + { provider: 'openai', key: 'sk' }); +assert.deepEqual(envDriveAuth({ CASE_DRIVE_API_KEY: 'sk', CASE_DRIVE_PROVIDER: 'anthropic' }), + { provider: 'anthropic', key: 'sk' }); +assert.deepEqual(envDriveAuth({ CASE_DRIVE_API_KEY: 'sk', CASE_DRIVE_PROVIDER: 'other' }), + { provider: '', key: '' }); + +assert.deepEqual(tagsOf({ tags: ['case-outbound', 'h_1'] }), ['case-outbound', 'h_1']); +assert.deepEqual(tagsOf({ tags: 'case-outbound,h_1' }), ['case-outbound', 'h_1']); +assert.ok(isOutbound({ tags: [OUTBOUND_TAG] })); +assert.equal(inboundText({ event: 'message', message: 'hi', tags: [OUTBOUND_TAG] }), ''); +assert.equal(inboundText({ event: 'keepalive', message: 'hi' }), ''); +assert.equal(inboundText({ event: 'message', message: ' hi ' }), 'hi'); + +{ + const { events, rest } = parseSseData( + 'data: {"id":"a","event":"message","message":"hi"}\n\n' + + 'data: {"event":"keepalive"}\n\n' + + 'data: {"id":"b","event":"message","message":"x"'); + assert.equal(events.length, 2); + assert.equal(events[0].message, 'hi'); + assert.equal(events[1].event, 'keepalive'); + 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'); + +{ + const calls = []; + const fetchImpl = async (url, opts) => { + calls.push({ url, opts }); + return { ok: true }; + }; + await publish( + { url: 'https://ntfy.sh', topic: 'top', token: 'secret' }, + { title: '[Case] done', message: 'ok' }, + fetchImpl, + ); + assert.equal(calls[0].url, 'https://ntfy.sh/top'); + assert.equal(calls[0].opts.method, 'POST'); + assert.equal(calls[0].opts.headers.Authorization, 'Bearer secret'); + assert.match(calls[0].opts.headers['X-Tags'], new RegExp(OUTBOUND_TAG)); + assert.equal(calls[0].opts.body, 'ok'); +} + +function sseBody(text) { + return new ReadableStream({ + start(c) { + c.enqueue(new TextEncoder().encode(text)); + c.close(); + }, + }); +} + +{ + const urls = []; + const ac = new AbortController(); + const fetchImpl = async (url) => { + urls.push(url); + throw new Error('stop'); + }; + try { + await listen( + { url: 'https://ntfy.sh', topic: 'top', token: 'tok' }, + () => {}, + { + fetchImpl, signal: ac.signal, now: () => 1700000000, + sleep: async () => { ac.abort(); }, + }, + ); + } catch { /* aborted */ } + assert.equal(urls[0], 'https://ntfy.sh/top/sse?since=1700000000'); +} + +{ + const got = []; + const urls = []; + const body = sseBody( + 'data: {"id":"1","event":"open"}\n\n' + + 'data: {"id":"2","event":"message","message":"self","tags":["case-outbound"]}\n\n' + + 'data: {"id":"3","event":"message","message":"check gmail","tags":[]}\n\n', + ); + let n = 0; + const ac = new AbortController(); + const fetchImpl = async (url, opts) => { + urls.push(url); + n += 1; + assert.equal(opts.headers.Authorization, 'Bearer tok'); + if (n === 1) return { ok: true, body }; + ac.abort(); + throw new Error('stop'); + }; + try { + await listen( + { url: 'https://ntfy.sh', topic: 'top', token: 'tok' }, + (text) => { got.push(text); }, + { fetchImpl, signal: ac.signal, now: () => 1700000000, sleep: async () => {} }, + ); + } catch { /* aborted */ } + assert.deepEqual(got, ['check gmail']); + assert.match(urls[0], /since=1700000000/); + assert.match(urls[1], /since=3/); +} + +{ + const got = []; + let n = 0; + const ac = new AbortController(); + const fetchImpl = async () => { + n += 1; + if (n > 2) { + ac.abort(); + throw new Error('stop'); + } + return { ok: true, body: sseBody('data: {"id":"3","event":"message","message":"check gmail","tags":[]}\n\n') }; + }; + try { + await listen( + { url: 'https://ntfy.sh', topic: 'top' }, + (text) => { got.push(text); }, + { fetchImpl, signal: ac.signal, now: () => 1700000000, sleep: async () => {} }, + ); + } catch { /* aborted */ } + assert.deepEqual(got, ['check gmail']); +} + +assert.equal(startPhoneNtfy({}), false); +assert.equal(startPhoneNtfy({ CASE_NTFY_CHAT: '1' }), false); +assert.equal(startPhoneNtfy({ CASE_NTFY_CHAT: '1', CASE_NTFY_TOPIC: 't' }), false); + +console.log('ok test_ntfy.mjs'); diff --git a/web/web-ui/test_serve.mjs b/web/web-ui/test_serve.mjs index 470d1da..0f36269 100644 --- a/web/web-ui/test_serve.mjs +++ b/web/web-ui/test_serve.mjs @@ -309,41 +309,45 @@ assert.equal(pageFile('/deploy.html'), '/deploy.html'); { const serveSrc = fs.readFileSync(fileURLToPath(new URL('./serve.mjs', import.meta.url)), 'utf8'); const caseToolsSrc = fs.readFileSync(fileURLToPath(new URL('./case-tools.mjs', import.meta.url)), 'utf8'); - const chatFn = serveSrc.slice(serveSrc.indexOf('async function chat('), serveSrc.indexOf('// ---------- noVNC')); + const runTurnFn = serveSrc.slice(serveSrc.indexOf('export async function runTurn('), serveSrc.indexOf('async function chat(')); + const chatFn = serveSrc.slice(serveSrc.indexOf('async function chat('), serveSrc.indexOf('async function pendingHandoffIds(')); + const loopFn = runTurnFn + chatFn; assert.match(chatFn, /if \(stopped\(\) \|\| res\.destroyed\) return/); - const outOfSteps = [...chatFn.matchAll(/if \(!finished(?: && !stopped\(\))?\)/g)]; + const outOfSteps = [...loopFn.matchAll(/if \(!finished(?: && !stopped\(\))?\)/g)]; assert.equal(outOfSteps.length, 2, 'Anthropic + OpenAI out-of-steps gates'); assert.ok(outOfSteps.every((m) => m[0].includes('!stopped()')), 'STOP is not out-of-rounds'); - assert.ok(!/turnStart/.test(chatFn), 'no turn rollback on provider error'); - assert.match(chatFn, /histCloseOpenCalls\(hist\.items\)/); - assert.match(chatFn, /withRateRetry\(round, emit, 5, gone\.signal\)/); + assert.ok(!/turnStart/.test(loopFn), 'no turn rollback on provider error'); + assert.match(loopFn, /histCloseOpenCalls\(hist\.items\)/); + assert.match(loopFn, /withRateRetry\(round, emit, 5, gone\.signal\)/); assert.match(chatFn, /res\.on\('close', \(\) => \{ clientGone\(\); gone\.abort\(\); \}\)/); - assert.match(chatFn, /responses\.create\(params, \{ signal: rc\.signal \}\)/); - assert.ok(!/responses\.create\(params\)/.test(chatFn), 'every round is abortable'); - assert.match(chatFn, /summary !== 'auto' && !gone\.signal\.aborted/, 'an abort never retries as a summary fallback'); - assert.match(chatFn, /gone\.signal\.removeEventListener\('abort', relay\)/, 'round listener is unlinked'); + assert.match(loopFn, /responses\.create\(params, \{ signal: rc\.signal \}\)/); + assert.ok(!/responses\.create\(params\)/.test(loopFn), 'every round is abortable'); + assert.match(loopFn, /summary !== 'auto' && !gone\.signal\.aborted/, 'an abort never retries as a summary fallback'); + assert.match(loopFn, /gone\.signal\.removeEventListener\('abort', relay\)/, 'round listener is unlinked'); assert.match(serveSrc, /takeSteers\(thread\.id\)/, 'steer inbox drained in the loop'); assert.match(serveSrc, /type: 'steer'/, 'steer emits to the stream'); assert.match(serveSrc, /beforeRound:/, 'anthropic loop drains steers each round'); assert.match(serveSrc, /pushSteerItems\(thread\.items, leftover\)/, 'last-round steers persist with the turn'); - assert.match(chatFn, /prompt_cache_key: thread\.id/); + assert.match(loopFn, /prompt_cache_key: thread\.id/); assert.match(serveSrc, /CASE_TURN_TOKENS/); - assert.match(chatFn, /tokenBudget: TURN_TOKEN_BUDGET/); - assert.match(chatFn, /signal: gone\.signal/); + assert.match(loopFn, /tokenBudget: TURN_TOKEN_BUDGET/); + assert.match(loopFn, /signal: gone\.signal/); assert.match(caseToolsSrc, /tokenBudget/); assert.match(caseToolsSrc, /stream\.abort\(\)/, 'Anthropic stream is canceled on disconnect'); assert.match(caseToolsSrc, /withRateRetry\(\(\) => round\(params\), emit, 5, signal\)/); - assert.match(chatFn, /if \(!stopped\(\)\) emit\(\{ type: 'error'/); + assert.match(loopFn, /if \(!stopped\(\)\) emit\(\{ type: 'error'/); assert.match(html, /\/api\/chat\/steer/); assert.match(html, /steerPrompt/); - assert.match(chatFn, /eff=\$\{eff\}/, 'turn log reports billed tokens, not nominal'); + assert.match(loopFn, /eff=\$\{eff\}/, 'turn log reports billed tokens, not nominal'); assert.match(serveSrc, /p === '\/api\/attach'/, 'user files land on disk, not in the chat body'); assert.match(fs.readFileSync(fileURLToPath(new URL('./case-tools.mjs', import.meta.url)), 'utf8'), /cache_control: \{ type: 'ephemeral' \}/, 'Anthropic path requests prompt cache'); - assert.match(chatFn, /hydrateShots\(hydrateAttaches\(hist\.items\)\)/); + assert.match(loopFn, /hydrateShots\(hydrateAttaches\(hist\.items\)\)/); assert.match(chatFn, /attachment not found/, 'a missing file is an error, not a silent drop'); - assert.ok(!/truncation:\s*['"]auto['"]/.test(chatFn), 'no truncation:auto'); + assert.ok(!/truncation:\s*['"]auto['"]/.test(loopFn), 'no truncation:auto'); assert.ok(!/compactHistory|SUMMARIZE_PROMPT|CASE_COMPACT_AT/.test(serveSrc), 'no compaction'); + assert.match(serveSrc, /try \{ computerId = await cid\(\); \}/); + assert.ok(!/drive ntfy chat on \$\{cfg\.url\}\/\$\{cfg\.topic\}/.test(serveSrc)); } {