import http from 'http'; import net from 'net'; import fs from 'fs'; import os from 'os'; import path from 'path'; import { WebSocketServer, WebSocket } from 'ws'; import { loadConfig, saveConfig } from '../shared/config.js'; import { createProvider, type AiProvider, type ChatMessage } from '../shared/ai.js'; import { paths } from '../shared/paths.js'; import { PKG_DIR, DATA_DIR, WORKSPACE_DIR } from '../shared/paths.js'; import { log } from '../shared/logger.js'; import { startTunnel, stopTunnel, isTunnelAlive, restartTunnel, startNamedTunnel, restartNamedTunnel } from './tunnel.js'; import { createWorkerApp } from '../worker/index.js'; import { closeDb, getSession, getSetting } from '../worker/db.js'; import { spawnBackend, stopBackend, restartBackend, getBackendPort, isBackendAlive, isBackendStopping, isBackendDead, readBackendLogTail, setBackendEnv, setBackendGiveUpHandler, probeBackendReady, backendJustSpawned } from './backend.js'; import { appendFrontendLog, tailFrontendLog, getFrontendLogCount, type FrontendLogKind } from './frontend-log.js'; import { handleAgentQuery, type AgentQueryRequest } from './agent-api.js'; import { updateTunnelUrl, startHeartbeat, stopHeartbeat, disconnect } from '../shared/relay.js'; import { startConversation, hasConversation, endConversation, endAllConversations, isConversationBusy, anyConversationBusy, anyOneShotActive, stopSubAgentTask, startBlobyAgentQuery, stopBlobyAgentQuery, warmUpForLiveConversation, type RecentMessage, } from './bloby-agent.js'; import { ensureFileDirs, saveAttachment, saveAudio, MAX_ATTACHMENTS_PER_MESSAGE, MAX_TOTAL_ATTACHMENT_BYTES, type SavedFile } from './file-saver.js'; import { approxBase64Bytes } from './harnesses/attachment-policy.js'; import { startViteDevServers, stopViteDevServers } from './vite-dev.js'; import { startScheduler, stopScheduler, readPulseConfig, readCronsConfig, nextRunISO, describeCron } from './scheduler.js'; import { execSync, spawn as cpSpawn } from 'child_process'; import crypto from 'crypto'; import zlib from 'zlib'; import { ChannelManager } from './channels/manager.js'; import { SHELL_HTML } from './shell.js'; // Last-resort process-level safety nets. The supervisor is the SINGLE process that keeps // chat alive (G1) and heals the backend (G3); a stray throw inside an emitter callback // (a WS frame, an fs.watch event, a read-stream error) or an unhandled rejection must NOT // take it down. Log loudly and stay up — never exit here. Per-site guards still handle // their own errors; this is defense-in-depth, not a substitute for them. process.on('uncaughtException', (err) => { log.error(`[supervisor] uncaughtException (kept alive): ${err instanceof Error ? err.stack || err.message : String(err)}`); }); process.on('unhandledRejection', (reason) => { log.error(`[supervisor] unhandledRejection (kept alive): ${reason instanceof Error ? reason.stack || reason.message : String(reason)}`); }); const DIST_BLOBY = path.join(PKG_DIR, 'dist-bloby'); const SUPERVISOR_PUBLIC = path.join(PKG_DIR, 'supervisor', 'public'); // Self-update coordination. The marker persists a queued update across a supervisor restart that // happens between the agent's request and the turn-complete flush (in-memory pendingUpdate alone // would be lost). attempts + a TTL bound the boot-resume retry so a persistently-failing update // can't loop on every boot. See queueUpdate/flushPendingUpdate/runDeferredUpdate. const UPDATE_MARKER = path.join(DATA_DIR, '.update-pending'); const UPDATE_MAX_ATTEMPTS = 2; const UPDATE_MARKER_TTL_MS = 30 * 60_000; // 30 min — a marker older than this is cleared, not retried /** True for the loopback, non-tunnel requests the local agent makes to the /__bloby/control/* and * channel-mutation endpoints. Identical trust model to the channel mutation guard: cloudflared * forwards over loopback so the IP check alone is a no-op behind the relay — we also reject any * request carrying cloudflared's cf-connecting-ip/cf-ray (tunnel-origin) headers. */ function isLoopbackAgentReq(req: http.IncomingMessage): boolean { const ip = req.socket.remoteAddress || ''; const isLoopback = ip === '127.0.0.1' || ip === '::1' || ip === '::ffff:127.0.0.1'; return isLoopback && !req.headers['cf-connecting-ip'] && !req.headers['cf-ray']; } // Proactive context recycling. The chat runs as one long-lived agent session per // conversation (so the user can keep talking while the agent works). That session's // context grows every turn and would eventually hit the wall. But continuity does NOT // live in that session — every fresh session re-injects the recent messages + memory // files — so when the context grows large AND the agent is idle, we end the session; // the next message starts a clean one. This keeps heavy/long chats off the wall // without lossy compaction, while preserving mid-turn responsiveness (we only recycle // between turns). When the harness reports the model's context window we recycle at // RECYCLE_FRACTION of it (adaptive to 200k vs 1M windows); otherwise we fall back to a // fixed token budget. Auto-compaction stays on as the in-turn safety net. const CONTEXT_RECYCLE_TOKENS = Number(process.env.BLOBY_CONTEXT_RECYCLE_TOKENS) || 140_000; const CONTEXT_RECYCLE_FRACTION = Number(process.env.BLOBY_CONTEXT_RECYCLE_FRACTION) || 0.7; // Platform assets that must survive workspace swaps — served directly by supervisor const PLATFORM_ASSETS = new Set([ '/spritesheet.webp', '/headphones_spritesheet.webp', '/morphy-icon-192.png', '/morphy-icon-512.png', '/morphy-badge.png', '/morphy-favicon.png', '/morphy_frame1.png', '/morphy.png', '/pi-logo.svg', '/codex.svg', '/manifest.json', '/morphy_sad.webm', '/morphy_sad.mov', ]); // Directory-prefix platform assets — anything under these is served from supervisor/public/. // Used for the Morphy animation set: drop new {clip}.png + {clip}.json into public/morphy/ // and they're automatically served without touching the allowlist. const PLATFORM_ASSET_DIRS = ['/morphy/']; // Ensure dist-bloby exists (postinstall may have failed silently) if (!fs.existsSync(DIST_BLOBY)) { log.info('Building bloby chat UI (first run)...'); try { execSync('npx vite build --config vite.bloby.config.ts', { cwd: PKG_DIR, stdio: 'ignore', }); log.ok('Bloby chat UI built'); } catch (err) { log.warn(`Failed to build bloby chat UI: ${err instanceof Error ? err.message : err}`); } } const MIME_TYPES: Record = { '.html': 'text/html', '.js': 'application/javascript', '.css': 'text/css', '.png': 'image/png', '.jpg': 'image/jpeg', '.svg': 'image/svg+xml', '.webm': 'video/webm', '.mov': 'video/mp4', '.mp4': 'video/mp4', '.woff2': 'font/woff2', '.json': 'application/json', '.webp': 'image/webp', }; // Service worker content — embedded here so it ships with supervisor/ (always updated) // Keep in sync with workspace/client/public/sw.js (prod builds) const SW_JS = `// Service worker — app-shell caching + push notifications // Caching strategy: // Hashed assets (/assets/*-AbCd12.js) → cache-first (immutable) // Navigation (HTML) → network-first (cache = offline fallback only) // Static assets / JS / CSS modules → network-first (cache = offline fallback only) // API, WebSocket, Vite internals → network-only (no cache) // Network-first (NOT stale-while-revalidate) on navigations + modules: Bloby is a LIVE-EDIT // tool, so the browser must always see the current app and its build errors — never a stale // cached shell that masks a broken (or just-fixed) frontend and produces the confusing // "normal refresh is broken but hard refresh works" split. Cache is a pure offline fallback. var CACHE = 'bloby-v25'; // Hash class includes -/_ — Vite's content hashes are base64url (e.g. bloby-Dfx1hOe-.js); // without them ~3 MB of hashed chat assets silently fell out of cache-first into network-first. var HASHED_RE = new RegExp('/assets/.+-[a-zA-Z0-9_-]{6,}[.](js|css)$'); // Precache the HTML shell on install so the cache is never empty. // Without this, the first navigation isn't intercepted (SW wasn't // controlling yet), so refresh would find an empty cache → white screen. self.addEventListener('install', function(e) { e.waitUntil( caches.open(CACHE) .then(function(c) { return c.add('/'); }) .then(function() { return self.skipWaiting(); }) .catch(function(err) { console.error('[SW] install failed:', err); throw err; }) ); }); self.addEventListener('activate', function(e) { e.waitUntil( caches.keys() .then(function(keys) { var old = keys.filter(function(k) { return k !== CACHE; }); return Promise.all(old.map(function(k) { return caches.delete(k); })); }) .then(function() { return self.clients.claim(); }) ); }); self.addEventListener('message', function(e) { if (e.data && e.data.type === 'SKIP_WAITING') self.skipWaiting(); }); self.addEventListener('fetch', function(event) { var request = event.request; var url = new URL(request.url); // Network-only: never cache these if ( request.method !== 'GET' || url.pathname.indexOf('/api/') === 0 || url.pathname.indexOf('/app/api/') === 0 || url.pathname.slice(-3) === '/ws' || url.pathname === '/sw.js' || url.pathname === '/bloby/sw.js' || url.pathname.indexOf('/@') === 0 || url.pathname.indexOf('/__') === 0 ) return; // Hashed assets (immutable, content-addressed) → cache-first if (HASHED_RE.test(url.pathname)) { event.respondWith(caches.open(CACHE).then(function(c) { return c.match(request).then(function(hit) { return hit || fetch(request).then(function(r) { if (r.status === 200) c.put(request, r.clone()); return r; }); }); })); return; } // Vite's optimized deps (/node_modules/.vite/deps/*?v=) are content-addressed and // served max-age=31536000,immutable — but fell into network-first below, re-crossing the // tunnel whenever the HTTP cache evicted them (common on mobile). Cache-first, pruning // older ?v= versions of the same module. Source modules (?t=, /@*) stay network-only. if (url.pathname.indexOf('/node_modules/.vite/deps/') === 0 && url.searchParams.has('v')) { event.respondWith(caches.open(CACHE).then(function(c) { return c.match(request).then(function(hit) { if (hit) return hit; return fetch(request).then(function(r) { if (r.status === 200) { // never 206 — Cache.put rejects partial (Range) responses c.put(request, r.clone()); c.keys().then(function(keys) { for (var i = 0; i < keys.length; i++) { var u = new URL(keys[i].url); if (u.pathname === url.pathname && u.search !== url.search) c.delete(keys[i]); } }); } return r; }); }); })); return; } // Navigation: only stale-while-revalidate the dashboard root (/). // /bloby/* is a separate app — let it go to network to avoid // caching the wrong HTML under the / key. if (request.mode === 'navigate') { // Workspace iframe documents (?__bloby_frame=1) share pathname '/' with the shell. // Network-only: never c.put() them (would overwrite the precached shell under the '/' // key) and never fall back to c.match('/') (would serve the shell INTO the iframe). if (url.searchParams.has('__bloby_frame')) return; if (url.pathname === '/' || url.pathname === '/index.html') { // Network-first: always fetch the live dashboard; only fall back to cache when offline. event.respondWith(caches.open(CACHE).then(function(c) { return fetch(request) .then(function(r) { if (r.status === 200) c.put('/', r.clone()); return r; }) .catch(function() { return c.match('/'); }); })); return; } // Other navigations (/bloby/*, etc.) — network only return; } // Everything else (JS/CSS modules, static assets) → network-first, cache as offline fallback. event.respondWith(caches.open(CACHE).then(function(c) { return fetch(request) .then(function(r) { if (r.status === 200) c.put(request, r.clone()); return r; }) .catch(function() { return c.match(request); }); })); }); // Push notifications self.addEventListener('push', function(event) { var data = { title: 'Bloby', body: 'New message' }; try { data = event.data.json(); } catch(e) {} event.waitUntil( self.registration.showNotification(data.title || 'Bloby', { body: data.body || '', icon: '/morphy-icon-192.png', badge: '/morphy-badge.png', vibrate: [100, 50, 100], tag: data.tag || 'bloby-default', data: { url: data.url || '/' }, }) ); }); // Notification click — focus or open app self.addEventListener('notificationclick', function(event) { event.notification.close(); var url = (event.notification.data && event.notification.data.url) || '/'; event.waitUntil( self.clients.matchAll({ type: 'window', includeUncontrolled: true }).then(function(clients) { for (var i = 0; i < clients.length; i++) { if (clients[i].url.indexOf('/bloby') !== -1 && 'focus' in clients[i]) return clients[i].focus(); } return self.clients.openWindow(url); }) ); }); `; // Shown when the dashboard's Vite dev server is briefly unreachable (startup, restart, or a // crash). This is a transient "reconnecting" state — no action needed — so it polls and reloads // itself the moment Vite is back (no blind 3s reload-into-the-same-error loop). Same branded // look as the backend-down interstitial; the chat widget is loaded so the user can still talk to // the agent if it doesn't recover on its own. const RECOVERING_HTML = ` Reconnecting · Bloby

Reconnecting…

Hang tight — your app is coming back online.

This usually happens right after an update or a restart. No action needed; this page refreshes itself the moment it's ready.

Reconnecting…
Powered by Bloby
`; /** Interstitial shown (by the supervisor, not the workspace) when the workspace backend has * crash-looped and given up. Replaces proxying the dashboard SPA to Vite — which would 503 on * every /app/api call and, for the common workspace-lock template, misread "no backend" as * "no password set" and show the lock-setup screen. Embeds the Bloby chat widget so the user * can ask the agent to fix it inline, a "copy logs" button (last 100 backend log lines baked in * at render time), and a poll that reloads into the real dashboard once the backend is back. */ function backendDownPage(logTail: string): string { // Embed logs as a JS string literal; escape `<` so a stray `` in the logs can't break out. const logs = JSON.stringify(logTail && logTail.length ? logTail : '(no backend logs were captured)').replace(/ Backend down · Bloby

Your app's backend is down

The workspace server crashed and couldn't restart on its own.

Ask your agent to fix it — the chat is right here in the corner. Tap below to copy the logs so it can debug faster.

Watching for recovery…
Powered by Bloby
`; } /** Kill any stale process holding a port. Ensures clean startup after crashes/updates. */ function killPort(port: number): void { try { const pids = execSync(`lsof -ti :${port} 2>/dev/null`, { encoding: 'utf-8' }).trim(); if (pids) { const pidList = pids.split('\n').filter(Boolean); const ownPid = process.pid.toString(); const toKill = pidList.filter((p) => p !== ownPid); if (toKill.length) { log.info(`[startup] Killing stale process(es) on port ${port}: ${toKill.join(', ')}`); execSync(`kill -9 ${toKill.join(' ')} 2>/dev/null`); } } } catch { // No process on port, or kill failed (already dead) — fine } } export async function startSupervisor() { const config = loadConfig(); const backendPort = getBackendPort(config.port); const internalSecret = crypto.randomBytes(16).toString('hex'); const agentSecret = crypto.randomBytes(32).toString('hex'); // Expose the supervisor's own HTTP port to EVERY child subprocess via our own process.env — // notably the agent harness (claude/codex/pi all spread ...process.env), whose Bash tool curls // the /__bloby/control/* surface as http://127.0.0.1:$SUPERVISOR_PORT/... . Previously this was // injected ONLY into the backend (setBackendEnv below), so the agent had no reliable port var. process.env.SUPERVISOR_PORT = String(config.port); // Inject agent secret + supervisor port into workspace backend env setBackendEnv({ BLOBY_AGENT_SECRET: agentSecret, SUPERVISOR_PORT: String(config.port), }); // Kill any stale processes from previous crashes/updates log.info(`[startup] Clearing ports ${config.port}, ${config.port + 2}, ${backendPort}...`); killPort(config.port); // supervisor killPort(config.port + 2); // vite killPort(backendPort); // backend // Create HTTP server first (Vite needs it for HMR WebSocket) // The request handler is set up later via server.on('request') const server = http.createServer(); // cloudflared keeps a pool of ~100 origin connections with a 90s idle timeout; Node's // default keepAliveTimeout is 5s, so after any pause every request burst re-paid TCP setup — // and the 5s-vs-90s mismatch is the classic reused-socket race behind sporadic tunnel 502s // ("error opening a stream to the origin"). Outlive cloudflared's pool instead. server.keepAliveTimeout = 120_000; server.headersTimeout = 125_000; // must exceed keepAliveTimeout (Node guards header floods with it) // Start Vite dev server — pass supervisor server so Vite attaches HMR WebSocket directly. // A Vite boot failure must NOT take down the supervisor (G1): chat is served independently and // is the lifeline the user needs to ask the agent to fix things. On failure we fall back to a // sentinel port — the dashboard proxy then serves the branded "Reconnecting" page (which polls // and carries the chat widget) while chat, the worker API, channels, and the tunnel all still // come up. Previously this throw reached the top-level catch → process.exit before chat listened. console.log('[supervisor] Starting Vite dev server...'); let vitePorts: { dashboard: number }; try { vitePorts = await startViteDevServers(config.port, server); console.log(`[supervisor] Vite ready — dashboard :${vitePorts.dashboard}`); } catch (err) { log.error(`Vite dev server failed to start — dashboard degraded, chat still available: ${err instanceof Error ? err.message : err}`); vitePorts = { dashboard: -1 }; // sentinel → dashboard proxy serves RECOVERING_HTML } // Ensure file storage dirs exist ensureFileDirs(); // Initialize worker routes in-process (no separate child process) const workerApp = createWorkerApp(); // Bloby's AI brain let ai: AiProvider | null = null; if (config.ai.provider && (config.ai.apiKey || config.ai.provider === 'ollama')) { ai = createProvider(config.ai.provider, config.ai.apiKey, config.ai.baseUrl); } // Bloby chat conversations (in-memory for now) const conversations = new Map(); // Track active DB conversation per WS client const clientConvs = new Map(); // Track current stream state so reconnecting clients get caught up let currentStreamConvId: string | null = null; let currentStreamBuffer = ''; // Workspace channel: SSE subscribers receive every chat event the dashboard widget WS // clients receive. A subscriber is just a held-open HTTP response we write `data: ...\n\n` // chunks to. Registered by `GET /api/agent/chat/stream`, used by `broadcastBloby` below. interface ChatSubscriber { id: string; clientId?: string; send: (type: string, data: any) => void; close: () => void; } const chatSubscribers = new Set(); /** Call worker API endpoints (in-process via supervisor's own HTTP server). * `timeoutMs` is opt-in: pass it only for calls that must not hang the caller * (e.g. the bot:response persist that gates the chat reply broadcast). Leave it * unset for calls whose duration is legitimately long (e.g. Whisper transcription), * so a generous timeout here never spuriously fails them. */ async function workerApi(apiPath: string, method = 'GET', body?: any, timeoutMs?: number) { const opts: RequestInit = { method, headers: { 'Content-Type': 'application/json', 'x-internal': internalSecret } }; if (body) opts.body = JSON.stringify(body); if (timeoutMs) opts.signal = AbortSignal.timeout(timeoutMs); const res = await fetch(`http://127.0.0.1:${config.port}${apiPath}`, opts); return res.json(); } /** Broadcast to all bloby chat surfaces EXCEPT the sender WS. Workspace SSE subscribers * always receive (sender is a WS, so no SSE collision). */ function broadcastBlobyExcept(sender: WebSocket, type: string, data: any) { const msg = JSON.stringify({ type, data }); for (const client of blobyWss.clients) { if (client !== sender && client.readyState === WebSocket.OPEN) client.send(msg); } for (const sub of chatSubscribers) { try { sub.send(type, data); } catch {} } } // ── Auth middleware ── const tokenCache = new Map(); // token → expiry timestamp const TOKEN_CACHE_TTL = 60_000; // 60s async function validateToken(token: string): Promise { const cached = tokenCache.get(token); if (cached && cached > Date.now()) return true; try { const session = getSession(token); if (session) { tokenCache.set(token, Date.now() + TOKEN_CACHE_TTL); return true; } tokenCache.delete(token); return false; } catch { return false; } } let authRequiredCache: { value: boolean; expires: number } | null = null; async function isAuthRequired(): Promise { if (authRequiredCache && authRequiredCache.expires > Date.now()) return authRequiredCache.value; try { const required = !!getSetting('portal_pass'); authRequiredCache = { value: required, expires: Date.now() + 30_000 }; return required; } catch { return false; } } // SECURITY MODEL — public allowlist (secure by default). When a portal password is set, EVERY // /api/* route requires a valid Bearer token EXCEPT the ones listed here (the pre-login surface: // login/onboarding/health/non-secret config). A new /api route is therefore GATED by default — // it can only leak if someone explicitly adds it here. (Previously the gate skipped ALL GET/HEAD, // so every data read — conversations, context, wallet, devices — was readable with no token.) // Note: /api/agent/* (agent secret) and /api/channels/* are intercepted + returned BEFORE this // gate, so they're unaffected; the channel entries below are belt-and-suspenders. const PUBLIC_PRELOGIN_ROUTES = [ 'GET /api/health', 'GET /api/onboard/status', 'GET /api/settings', // secrets already stripped (worker denylist); widget + onboard read flags pre-login 'GET /api/push/vapid-public-key', 'GET /api/portal/login', 'POST /api/portal/login', 'GET /api/portal/validate-token', 'POST /api/portal/validate-token', 'GET /api/portal/login/totp', 'GET /api/portal/totp/status', 'POST /api/portal/totp/setup', // self-protected in-handler (Bearer OR password OR first-run) 'POST /api/portal/totp/verify-setup', // self-protected in-handler 'POST /api/portal/totp/disable', // self-protected (requires password + valid code) 'POST /api/portal/verify-password', // verifies the password itself — cannot require a token // NOTE: 'POST /api/onboard' is NOT public — gated below: open only on genuine first run // (no portal_pass yet), token-required afterward. Dashboard re-onboard uses the x-internal WS path. // Channel onboarding (also intercepted earlier; kept for completeness/safety): 'GET /api/channels/status', 'GET /api/channels/whatsapp/qr', 'GET /api/channels/whatsapp/qr-page', 'POST /api/channels/whatsapp/connect', 'POST /api/channels/whatsapp/disconnect', 'POST /api/channels/whatsapp/logout', 'POST /api/channels/whatsapp/configure', 'POST /api/channels/whatsapp/pairing-code', 'POST /api/channels/whatsapp/react', 'POST /api/channels/send', 'POST /api/channels/alexa/handle', // NOTE: 'POST /api/channels/telegram/connect' is NOT public — it sets the bot token, so it is // gated in-handler (first run open, portal token required once a password is set), like /api/onboard. 'GET /api/channels/telegram/pair-page', 'GET /api/channels/telegram/status', 'POST /api/channels/telegram/configure', 'POST /api/channels/telegram/disconnect', 'POST /api/channels/telegram/reconnect', 'POST /api/channels/telegram/logout', ]; // Method-specific public PREFIXES — onboarding namespaces with sub-paths / params that carry no // private chat data: provider OAuth setup/status (all of /api/auth/*), and handle availability // (GET /api/handle/* only — handle register/change are POSTs and stay gated). const PUBLIC_PRELOGIN_PREFIXES = [ 'POST /api/auth/', 'GET /api/auth/', 'DELETE /api/auth/', 'GET /api/handle/', ]; function isPublicRoute(method: string, url: string): boolean { const m = method === 'HEAD' ? 'GET' : method; // a HEAD to a public GET route is public const path = url.split('?')[0]; if (PUBLIC_PRELOGIN_ROUTES.includes(`${m} ${path}`)) return true; return PUBLIC_PRELOGIN_PREFIXES.some((r) => { const sp = r.indexOf(' '); return m === r.slice(0, sp) && path.startsWith(r.slice(sp + 1)); }); } // ── Cached static serving: in-memory buffers + content ETags + 304s + gzip ────────────── // Everything the supervisor serves on the refresh path used to be fs.readFileSync per request // with no validator — nothing could ever 304, so ~90 KB of identical bytes crossed the tunnel // on every refresh (plus sync SD-card reads on Pi-class hardware). Entries are stat-validated // by mtime, so editing a file (dev) or `bloby update` + restart (prod) is always picked up; // `no-cache` semantics (revalidate every load) are preserved — clients just get 304s now. type CachedAsset = { buf: Buffer; gz: Buffer | null; etag: string; mtimeMs: number }; const staticCache = new Map(); function buildCachedAsset(buf: Buffer, mtimeMs: number): CachedAsset { const etag = '"' + crypto.createHash('sha1').update(buf).digest('base64url').slice(0, 16) + '"'; const gz = buf.length > 1024 ? zlib.gzipSync(buf, { level: 6 }) : null; // Keep gzip only when it actually pays for the Content-Encoding overhead. return { buf, gz: gz && gz.length < buf.length * 0.92 ? gz : null, etag, mtimeMs }; } function serveCachedText( req: http.IncomingMessage, res: http.ServerResponse, key: string, src: { file?: string; content?: string }, contentType: string, cacheControl: string, extra: Record = {}, ): void { let entry = staticCache.get(key); if (src.file !== undefined) { let mtimeMs = 0; try { mtimeMs = fs.statSync(src.file).mtimeMs; } catch { res.writeHead(404); res.end('Not found'); return; } if (!entry || entry.mtimeMs !== mtimeMs) { if (staticCache.size > 200) staticCache.clear(); // runaway guard; rebuilt on demand try { entry = buildCachedAsset(fs.readFileSync(src.file), mtimeMs); staticCache.set(key, entry); } catch { res.writeHead(404); res.end('Not found'); return; } } } else if (!entry) { entry = buildCachedAsset(Buffer.from(src.content || '', 'utf-8'), -1); staticCache.set(key, entry); } if (req.headers['if-none-match'] === entry.etag) { res.writeHead(304, { ETag: entry.etag, 'Cache-Control': cacheControl, ...extra }); res.end(); return; } const gzOk = entry.gz !== null && String(req.headers['accept-encoding'] || '').includes('gzip'); res.writeHead(200, { 'Content-Type': contentType, 'Cache-Control': cacheControl, ETag: entry.etag, Vary: 'Accept-Encoding', ...(gzOk ? { 'Content-Encoding': 'gzip' } : {}), ...extra, }); res.end(gzOk ? entry.gz : entry.buf); } // Read once per process — the version can only change via an update, which restarts us. let pkgVersion = 'unknown'; try { pkgVersion = JSON.parse(fs.readFileSync(path.join(PKG_DIR, 'package.json'), 'utf-8')).version || 'unknown'; } catch {} // Keep-alive agent for the loopback proxies (Vite dev server + workspace backend). Without it // (on Node 18, the install floor) every proxied module request opens a fresh TCP connection — // dozens of handshakes per refresh that a Pi actually feels. const loopbackAgent = new http.Agent({ keepAlive: true, maxSockets: 64 }); const HOP_BY_HOP = ['connection', 'keep-alive', 'proxy-authenticate', 'proxy-authorization', 'te', 'trailer', 'transfer-encoding', 'upgrade']; function stripHopByHop(h: http.IncomingHttpHeaders): http.IncomingHttpHeaders { const out: http.IncomingHttpHeaders = {}; for (const k in h) if (!HOP_BY_HOP.includes(k)) out[k] = h[k]; return out; // forwarding 'connection: close' would defeat the keep-alive pool } // ── Refresh-performance instrumentation ───────────────────────────────────────────────── // Server side: a ring of recent request timings (duration, bytes out, encoding, whether the // socket was reused). Client side: the shell assembles marks from itself + widget + the // workspace iframe into one beacon per page load and POSTs it here. Read both via // GET /__bloby/perf — LOCALHOST ONLY (request paths/timings stay off the tunnel); remote // debugging uses the console print behind localStorage.bloby_perf='1' in the shell. const perfRequests: object[] = []; const perfBeacons: object[] = []; function isLoopback(req: http.IncomingMessage): boolean { const a = req.socket.remoteAddress || ''; return a === '127.0.0.1' || a === '::1' || a === '::ffff:127.0.0.1'; } // HTTP request handler — proxies to Vite dev servers + worker API server.on('request', async (req, res) => { // Every response carries the agent-origin stamp. The relay treats it as authoritative // proof the bytes came from the agent (never substitutes a branded page) AND skips its // 4 KB/1.5 s error-body sniff buffer on 4xx/5xx — without this, every legit 404/500 // through the tunnel paid the sniff. Proxied responses keep it unless upstream overrides. res.setHeader('X-Bloby-Origin', 'supervisor'); // Request timing sample — attached before any routing so every branch is covered. { const t0 = Date.now(); const sock = req.socket as any; const reused = !!sock.__blobySeen; sock.__blobySeen = true; const w0 = sock.bytesWritten || 0; let sampled = false; res.on('close', () => { if (sampled) return; sampled = true; perfRequests.push({ ts: t0, m: req.method, p: (req.url || '').split('?')[0].slice(0, 120), s: res.statusCode, ms: Date.now() - t0, out: Math.max(0, (req.socket.bytesWritten || 0) - w0), enc: String(res.getHeader('content-encoding') || ''), reused, }); if (perfRequests.length > 400) perfRequests.splice(0, perfRequests.length - 400); }); } // Client perf beacon — one compact JSON per page load, posted by the shell. if (req.url === '/__bloby/perf' && req.method === 'POST') { const chunks: Buffer[] = []; let size = 0; req.on('data', (c: Buffer) => { size += c.length; if (size <= 32_768) chunks.push(c); }); req.on('end', () => { try { if (size <= 32_768) { const beacon = JSON.parse(Buffer.concat(chunks).toString('utf-8')); perfBeacons.push({ ts: Date.now(), ...beacon }); if (perfBeacons.length > 20) perfBeacons.splice(0, perfBeacons.length - 20); } } catch {} res.writeHead(204); res.end(); }); return; } if (req.url === '/__bloby/perf' && req.method === 'GET') { if (!isLoopback(req)) { res.writeHead(403); res.end('localhost only'); return; } res.writeHead(200, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' }); res.end(JSON.stringify({ beacons: perfBeacons, requests: perfRequests })); return; } // Bloby widget — served directly (not part of Vite build) if (req.url === '/bloby/widget.js') { serveCachedText(req, res, 'widget', { file: paths.widgetJs }, 'application/javascript', 'no-cache'); return; } // App WS client — served directly (proxies /app/api calls through WebSocket) if (req.url === '/bloby/app-ws.js') { serveCachedText(req, res, 'app-ws', { file: path.join(PKG_DIR, 'supervisor', 'app-ws.js') }, 'application/javascript', 'no-cache'); return; } // Workspace guard — injected by the supervisor into the dashboard HTML (see the Vite proxy // below). Auto-reloads into the "backend down" interstitial when the backend gives up, and // replaces Vite's raw error overlay with a friendly one. Served here so it's always current. if (req.url === '/bloby/workspace-guard.js') { serveCachedText(req, res, 'guard', { file: path.join(PKG_DIR, 'supervisor', 'workspace-guard.js') }, 'application/javascript', 'no-cache'); return; } // Service worker — served from embedded constant (supervisor/ is always updated) if (req.url === '/sw.js' || req.url === '/bloby/sw.js') { serveCachedText(req, res, 'sw', { content: SW_JS }, 'application/javascript', 'no-cache'); return; } // Backend liveness for the "backend down" interstitial's recovery poll. Supervisor-served // (not proxied) so it answers even when the workspace backend is dead, and independent of // whatever routes the user's backend happens to define. if (req.url === '/__bloby/backend-status') { res.writeHead(200, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' }); res.end(JSON.stringify({ alive: isBackendAlive(), dead: isBackendDead() })); return; } // Running platform version — the chat compares this against its baked-in build version on // every reconnect. A mismatch means the immortal shell survived a self-update running // pre-update code; the chat then triggers one full shell reload to pick up the new bundle. if (req.url === '/__bloby/version') { res.writeHead(200, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' }); res.end(JSON.stringify({ version: pkgVersion })); return; } // ── Agent control surface (/__bloby/control/*) ────────────────────────────────────────────── // The Bloby agent drives backend restarts, self-update, and log tails through these endpoints // instead of the old lossy fs.watch trigger files (.restart/.update). Every call returns a // SYNCHRONOUS JSON ack — that explicit acknowledgment is the reliability fix (no silent drops). // The agent curls http://127.0.0.1:$SUPERVISOR_PORT/__bloby/control/... (SUPERVISOR_PORT is // injected into its env). All routes are loopback-only (same cf-reject guard as the channel // mutations) so they are NEVER reachable over the public tunnel — EXCEPT fe-log, which the // user's browser posts to (write-only, capped). Served here, before auth and the Vite catch-all, // so they answer even when the backend/Vite are down. if (req.url?.startsWith('/__bloby/control/')) { const ctlPath = req.url.split('?')[0]; const ctlQuery = new URLSearchParams(req.url.split('?')[1] || ''); // POST /__bloby/control/fe-log — browser → supervisor frontend-error ingest. NOT loopback- // gated (the workspace-guard posts from the user's browser, possibly over the tunnel). // Write-only, size-capped, no read-back, no side effects → worst case is capped log spam. if (ctlPath === '/__bloby/control/fe-log' && req.method === 'POST') { let feBody = ''; let feTooBig = false; // Cap by DROPPING the payload once oversize (not req.destroy(), which fires neither 'end' nor // 'error' → the response would never be sent). Keep reading to a clean 'end' and 204 always. req.on('data', (chunk: Buffer) => { if (feTooBig) return; feBody += chunk.toString(); if (feBody.length > 16_384) { feTooBig = true; feBody = ''; } }); req.on('end', () => { if (!feTooBig) { try { const parsed = JSON.parse(feBody); const entries = Array.isArray(parsed?.entries) ? parsed.entries.slice(-40) : []; const allowed = ['error', 'unhandledrejection', 'console.error', 'console.warn', 'vite-overlay']; for (const e of entries) { if (e && typeof e.text === 'string') { const kind = (allowed.includes(e.kind) ? e.kind : 'error') as FrontendLogKind; appendFrontendLog(kind, e.text, typeof e.stack === 'string' ? e.stack : undefined); } } } catch {} } try { res.writeHead(204); res.end(); } catch {} }); req.on('error', () => { try { res.writeHead(204); res.end(); } catch {} }); return; } // Every other control route is loopback-only (agent-driven, can restart/update the instance). if (!isLoopbackAgentReq(req)) { res.writeHead(403, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ ok: false, error: 'This control endpoint is localhost-only.' })); return; } res.setHeader('Cache-Control', 'no-store'); // GET /__bloby/control/logs/backend?lines=N[&prev=1] — tail the current (or last-crashed) run. if (ctlPath === '/__bloby/control/logs/backend' && req.method === 'GET') { const lines = Math.max(1, Math.min(1000, parseInt(ctlQuery.get('lines') || '100', 10) || 100)); const prev = ctlQuery.get('prev') === '1'; res.writeHead(200, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ ok: true, lines, prev, justRestarted: backendJustSpawned(), log: readBackendLogTail(lines, prev) })); return; } // GET /__bloby/control/logs/frontend?lines=N — runtime + console + Vite-compile frontend errors. if (ctlPath === '/__bloby/control/logs/frontend' && req.method === 'GET') { const lines = Math.max(1, Math.min(1000, parseInt(ctlQuery.get('lines') || '100', 10) || 100)); res.writeHead(200, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ ok: true, lines, entries: getFrontendLogCount(), log: tailFrontendLog(lines) })); return; } // GET /__bloby/control/update-status — is a queued update running / did it fail? if (ctlPath === '/__bloby/control/update-status' && req.method === 'GET') { res.writeHead(200, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ ok: true, ...getUpdateStatus() })); return; } // POST /__bloby/control/update — queue a self-update (acknowledged, idempotent, deferred). if (ctlPath === '/__bloby/control/update' && req.method === 'POST') { const r = queueUpdate(); res.writeHead(200, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ ok: true, queued: r.queued || r.alreadyQueued, alreadyQueued: r.alreadyQueued, retrying: r.retrying, deferred: r.deferred, message: r.alreadyQueued ? 'An update is already queued or running.' : r.retrying ? 'A previous update attempt failed — re-queued; it retries after your turn ends. Check update-status (state:failed exposes the prior error in logTail).' : 'Update queued — it runs after your turn ends. You will NOT die mid-turn; finish your turn normally. The page is unresponsive ~1–2 min while Bloby restarts on the new version. Check update-status after.', })); return; } // POST /__bloby/control/restart-backend { wait?:bool=true, timeoutMs?:num=15000, logLines?:num=60 } // Restarts the backend through the existing serialized doRestart() funnel and (when wait) blocks // until the backend's PORT is listening — so the agent can restart-and-verify WITHIN its turn. if (ctlPath === '/__bloby/control/restart-backend' && req.method === 'POST') { let rbBody = ''; let rbTooBig = false; // Cap by dropping the payload (not req.destroy(), which would fire neither 'end' nor 'error' // → no JSON ack ever sent, violating the endpoint contract). Always answer from 'end'. req.on('data', (chunk: Buffer) => { if (rbTooBig) return; rbBody += chunk.toString(); if (rbBody.length > 4096) { rbTooBig = true; rbBody = ''; } }); req.on('end', async () => { if (rbTooBig) { try { res.writeHead(413, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ ok: false, error: 'Request body too large.' })); } catch {} return; } let rbOpts: any = {}; try { rbOpts = rbBody ? JSON.parse(rbBody) : {}; } catch {} const wait = rbOpts.wait !== false; // default true const timeoutMs = Math.max(1000, Math.min(30_000, Number(rbOpts.timeoutMs) || 15_000)); const logLines = Math.max(0, Math.min(400, Number(rbOpts.logLines) || 60)); const wasDead = isBackendDead(); // had crash-looped & given up BEFORE this explicit restart const started = Date.now(); try { await doRestart(); // resetBackendRestarts + serialized stop→spawn; preserves all invariants } catch (err: any) { try { res.writeHead(200, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ ok: false, restarted: false, error: String(err?.message || err) })); } catch {} return; } const listening = wait ? await probeBackendReady(backendPort, timeoutMs) : isBackendAlive(); const gaveUp = isBackendDead(); try { res.writeHead(200, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ ok: true, restarted: true, healthy: listening && !gaveUp, listening, gaveUp, wasDead, // If it gave up AGAIN, restarting won't help — tell the agent to fix the code, not re-restart. hint: gaveUp ? 'Backend crash-looped and gave up again — restarting will not fix it. Read the logs and fix the code.' : undefined, waitedMs: Date.now() - started, logs: logLines ? readBackendLogTail(logLines) : undefined, })); } catch {} }); req.on('error', () => { try { if (!res.headersSent) { res.writeHead(500); res.end(); } } catch {} }); return; } res.writeHead(404, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ ok: false, error: 'Unknown control endpoint.' })); return; } // App API routes → proxy to user's backend server if (req.url?.startsWith('/app/api')) { const backendPath = req.url.replace(/^\/app/, ''); if (!isBackendAlive()) { res.writeHead(503, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ error: 'Backend is starting...' })); return; } const proxy = http.request( { host: '127.0.0.1', port: backendPort, path: backendPath, method: req.method, headers: stripHopByHop(req.headers), agent: loopbackAgent }, (proxyRes) => { const ct = String(proxyRes.headers['content-type'] || ''); const isSse = ct.includes('text/event-stream'); res.writeHead(proxyRes.statusCode!, proxyRes.headers); // SSE needs Nagle off so chat tokens flush immediately instead of in batched frames. if (isSse) { try { res.socket?.setNoDelay(true); } catch {} } proxyRes.pipe(res); }, ); proxy.on('error', (e) => { console.error(`[supervisor] Backend proxy error: ${req.url}`, e.message); res.writeHead(503, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ error: 'Backend unavailable' })); }); req.pipe(proxy); return; } // ── Channel API routes (handled by supervisor, not worker) ── if (req.url?.startsWith('/api/channels')) { const channelPath = req.url.split('?')[0]; res.setHeader('Content-Type', 'application/json'); res.setHeader('Cache-Control', 'no-store, no-cache, must-revalidate'); // ── Loopback-only guard for channel MUTATION endpoints ── // These are only ever called by the local agent over loopback (curl localhost:7400), // never legitimately from the public Cloudflare tunnel. Without this, an unauthenticated // remote request could set mode/admins or flip allowOthersToTrigger and seize the agent // (it can run Bash/edit files). Same guard the Agent API uses: cloudflared forwards over // loopback so the IP check alone is a no-op behind the relay — we also reject any request // carrying cloudflared's cf-connecting-ip/cf-ray (tunnel-origin) headers. Reads (status, // qr, qr-page) and alexa/handle (relay-origin, secret-gated) deliberately stay public. const WA_MUTATION_ROUTES = new Set([ 'POST /api/channels/whatsapp/configure', 'POST /api/channels/whatsapp/connect', 'POST /api/channels/whatsapp/disconnect', 'POST /api/channels/whatsapp/logout', 'POST /api/channels/whatsapp/pairing-code', 'POST /api/channels/send', // Telegram mutation endpoints that seize/alter the agent — agent-driven over loopback only. 'POST /api/channels/telegram/configure', 'POST /api/channels/telegram/disconnect', 'POST /api/channels/telegram/reconnect', 'POST /api/channels/telegram/logout', ]); if (WA_MUTATION_ROUTES.has(`${req.method} ${channelPath}`)) { const remoteIp = req.socket.remoteAddress || ''; const isLoopback = remoteIp === '127.0.0.1' || remoteIp === '::1' || remoteIp === '::ffff:127.0.0.1'; if (!isLoopback || req.headers['cf-connecting-ip'] || req.headers['cf-ray']) { res.writeHead(403); res.end(JSON.stringify({ ok: false, error: 'This channel endpoint is localhost-only.' })); return; } } // GET /api/channels/status — all channel statuses if (req.method === 'GET' && channelPath === '/api/channels/status') { res.writeHead(200); res.end(JSON.stringify(channelManager.getStatuses())); return; } // GET /api/channels/whatsapp/qr — raw QR SVG data if (req.method === 'GET' && channelPath === '/api/channels/whatsapp/qr') { const qr = channelManager.getQrCode('whatsapp'); res.writeHead(200); res.end(JSON.stringify({ qr })); return; } // GET /api/channels/whatsapp/qr-page — standalone QR scanning page if (req.method === 'GET' && channelPath === '/api/channels/whatsapp/qr-page') { res.setHeader('Content-Type', 'text/html'); const qr = channelManager.getQrCode('whatsapp'); const status = channelManager.getStatus('whatsapp'); const connected = status?.connected || false; res.writeHead(200); const confettiHTML = Array.from({ length: 30 }, (_, i) => { const colors = ['#0166FF', '#009AFE', '#4AEEFF', '#4ade80', '#facc15', '#818cf8']; const color = colors[Math.floor(Math.random() * colors.length)]; const left = Math.random() * 100; const delay = i * 0.04; const drift = (Math.random() - 0.5) * 120; const duration = 1.8 + Math.random() * 0.8; return `
`; }).join(''); res.end(` WhatsApp QR
${connected ? `
${confettiHTML}
Connected!

WhatsApp is linked. You can close this page.

` : qr ? `
${qr}

Scan with WhatsApp to link

or

Digits only, with country code. No + or dashes.

Enter this code in WhatsApp:

  1. Open WhatsApp on your phone
  2. Go to Settings > Linked Devices
  3. Tap Link a Device
  4. Tap Link with phone number instead
  5. Enter the code above
` : '

Starting WhatsApp...

'}
${!connected ? `` : ''} `); return; } // POST /api/channels/whatsapp/pairing-code — request phone-number pairing code (mobile alternative to QR) if (req.method === 'POST' && channelPath === '/api/channels/whatsapp/pairing-code') { let body = ''; req.on('data', (chunk: Buffer) => { body += chunk.toString(); }); req.on('end', async () => { try { const { phoneNumber } = JSON.parse(body); if (!phoneNumber) { res.writeHead(400); res.end(JSON.stringify({ error: 'Missing phoneNumber' })); return; } const code = await channelManager.requestWhatsAppPairingCode(phoneNumber); res.writeHead(200); res.end(JSON.stringify({ ok: true, code })); } catch (err: any) { const status = err.message?.includes('rate') ? 429 : 500; res.writeHead(status); res.end(JSON.stringify({ error: err.message })); } }); return; } // POST /api/channels/whatsapp/connect — start WhatsApp connection (triggers QR) if (req.method === 'POST' && channelPath === '/api/channels/whatsapp/connect') { (async () => { try { // Enable WhatsApp in config const cfg = loadConfig(); if (!cfg.channels) cfg.channels = {}; if (!cfg.channels.whatsapp) cfg.channels.whatsapp = { enabled: true, mode: 'channel' }; cfg.channels.whatsapp.enabled = true; saveConfig(cfg); await channelManager.connectWhatsApp(); res.writeHead(200); res.end(JSON.stringify({ ok: true })); } catch (err: any) { res.writeHead(500); res.end(JSON.stringify({ error: err.message })); } })(); return; } // POST /api/channels/whatsapp/disconnect — disconnect WhatsApp if (req.method === 'POST' && channelPath === '/api/channels/whatsapp/disconnect') { (async () => { try { await channelManager.disconnectChannel('whatsapp'); const cfg = loadConfig(); if (cfg.channels?.whatsapp) cfg.channels.whatsapp.enabled = false; saveConfig(cfg); res.writeHead(200); res.end(JSON.stringify({ ok: true })); } catch (err: any) { res.writeHead(500); res.end(JSON.stringify({ error: err.message })); } })(); return; } // POST /api/channels/whatsapp/logout — disconnect + delete credentials if (req.method === 'POST' && channelPath === '/api/channels/whatsapp/logout') { (async () => { try { await channelManager.logoutWhatsApp(); const cfg = loadConfig(); if (cfg.channels?.whatsapp) cfg.channels.whatsapp.enabled = false; saveConfig(cfg); res.writeHead(200); res.end(JSON.stringify({ ok: true })); } catch (err: any) { res.writeHead(500); res.end(JSON.stringify({ error: err.message })); } })(); return; } // POST /api/channels/whatsapp/configure — set mode + admins if (req.method === 'POST' && channelPath === '/api/channels/whatsapp/configure') { let body = ''; req.on('data', (chunk: Buffer) => { body += chunk.toString(); }); req.on('end', () => { try { const data = JSON.parse(body); const cfg = loadConfig(); if (!cfg.channels) cfg.channels = {}; if (!cfg.channels.whatsapp) cfg.channels.whatsapp = { enabled: true, mode: 'channel' }; if (data.mode) cfg.channels.whatsapp.mode = data.mode; if (data.admins !== undefined) cfg.channels.whatsapp.admins = data.admins; if (data.skill !== undefined) cfg.channels.whatsapp.skill = data.skill; if (data.allowGroups !== undefined) cfg.channels.whatsapp.allowGroups = !!data.allowGroups; if (data.allowOthersToTrigger !== undefined) cfg.channels.whatsapp.allowOthersToTrigger = !!data.allowOthersToTrigger; saveConfig(cfg); res.writeHead(200); res.end(JSON.stringify({ ok: true, config: cfg.channels.whatsapp })); } catch (err: any) { res.writeHead(400); res.end(JSON.stringify({ error: err.message })); } }); return; } // POST /api/channels/whatsapp/react — send (or remove) an emoji reaction on a WhatsApp message if (req.method === 'POST' && channelPath === '/api/channels/whatsapp/react') { let body = ''; req.on('data', (chunk: Buffer) => { body += chunk.toString(); }); req.on('end', async () => { try { const { chatJid, messageId, fromMe, participant, emoji } = JSON.parse(body) as { chatJid?: string; messageId?: string; fromMe?: boolean; participant?: string; emoji?: string; }; if (!chatJid || !messageId) { res.writeHead(400); res.end(JSON.stringify({ error: 'chatJid and messageId are required' })); return; } // emoji='' is intentionally supported — it removes a previous reaction (Baileys convention). const key = { remoteJid: chatJid, id: messageId, fromMe: !!fromMe, participant } as any; await channelManager.reactToMessage('whatsapp', chatJid, key, emoji ?? '👍'); res.writeHead(200); res.end(JSON.stringify({ ok: true })); } catch (err: any) { res.writeHead(500); res.end(JSON.stringify({ error: err.message })); } }); return; } // POST /api/channels/send — send a message (and/or media) via any channel if (req.method === 'POST' && channelPath === '/api/channels/send') { let body = ''; req.on('data', (chunk: Buffer) => { body += chunk.toString(); }); req.on('end', async () => { try { const { channel, to, text, media } = JSON.parse(body); if (!channel || !to) { res.writeHead(400); res.end(JSON.stringify({ error: 'Missing channel or to' })); return; } if (!text && !media) { res.writeHead(400); res.end(JSON.stringify({ error: 'Missing text or media' })); return; } if (media) { if (!media.type || !media.path) { res.writeHead(400); res.end(JSON.stringify({ error: 'media requires { type, path }' })); return; } await channelManager.sendMedia(channel, to, media, text); } else { await channelManager.sendMessage(channel, to, text); } res.writeHead(200); res.end(JSON.stringify({ ok: true })); } catch (err: any) { res.writeHead(500); res.end(JSON.stringify({ error: err.message })); } }); return; } // POST /api/channels/alexa/handle — relay-forwarded Alexa utterance. // Authed via x-bloby-alexa-secret (per-user shared secret minted at pair time). // Returns plain text the relay then formats as Alexa SSML. if (req.method === 'POST' && channelPath === '/api/channels/alexa/handle') { let body = ''; req.on('data', (chunk: Buffer) => { body += chunk.toString(); }); req.on('end', async () => { try { const cfg = loadConfig(); const expected = cfg.channels?.alexa?.sharedSecret; const provided = req.headers['x-bloby-alexa-secret']; if (!expected || !provided || provided !== expected) { res.writeHead(401); res.end(JSON.stringify({ ok: false, error: 'alexa-auth-failed' })); return; } if (!cfg.channels?.alexa?.enabled) { res.writeHead(503); res.end(JSON.stringify({ ok: false, error: 'alexa-disabled' })); return; } const { text, alexaUserId, sessionId, deviceId, locale, kind, apiEndpoint, apiAccessToken, requestId } = JSON.parse(body || '{}') as { text?: string; alexaUserId?: string; sessionId?: string; deviceId?: string; locale?: string; kind?: string; apiEndpoint?: string; apiAccessToken?: string; requestId?: string; }; // Launch / Stop / Help intents don't need to hit the agent — the relay // can short-circuit, but we accept them here too for symmetry. if (kind === 'launch') { res.writeHead(200); res.end(JSON.stringify({ ok: true, reply: 'Morphy here, what can I help with?', endSession: false })); return; } if (kind === 'stop' || kind === 'cancel') { res.writeHead(200); res.end(JSON.stringify({ ok: true, reply: 'Goodbye.', endSession: true })); return; } if (!text || !alexaUserId) { res.writeHead(400); res.end(JSON.stringify({ ok: false, error: 'text and alexaUserId required' })); return; } try { const reply = await channelManager.handleAlexaInbound({ text, alexaUserId, alexaSessionId: sessionId, deviceId, locale, apiEndpoint, apiAccessToken, requestId, }); res.writeHead(200); res.end(JSON.stringify({ ok: true, reply, endSession: false })); } catch (err: any) { // Timeout or aborted turn — agent will still finish and the result lands // in the shared conversation. The agent is responsible for any // out-of-band notification (HA announce, chat, WhatsApp) — that's // a skill concern, not a supervisor concern. res.writeHead(200); res.end(JSON.stringify({ ok: true, reply: "I'll reply in your chat when ready.", endSession: false, overflow: true, })); } } catch (err: any) { res.writeHead(500); res.end(JSON.stringify({ ok: false, error: err.message })); } }); return; } // POST /api/channels/alexa/pair — dashboard mints a pairing code via the relay. // Body: optional { ttlSeconds }. Returns: { code, expiresAt, sharedSecret }. // We persist sharedSecret locally so future /handle calls can authenticate. if (req.method === 'POST' && channelPath === '/api/channels/alexa/pair') { (async () => { try { const cfg = loadConfig(); if (!cfg.relay?.token) { res.writeHead(400); res.end(JSON.stringify({ ok: false, error: 'no-relay-token' })); return; } // The relay API always lives at api.bloby.bot — `cfg.relay.url` is the // bot's public profile URL (e.g. https://bloby.bot/), not the API. const resp = await fetch('https://api.bloby.bot/api/alexa/pair', { method: 'POST', headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${cfg.relay.token}`, }, body: '{}', }); const data = await resp.json() as { code?: string; expiresAt?: string; sharedSecret?: string; error?: string }; if (!resp.ok || !data.code || !data.sharedSecret) { res.writeHead(resp.status || 500); res.end(JSON.stringify({ ok: false, error: data.error || 'pair-failed' })); return; } // Persist the shared secret + mark channel enabled so handleInbound is gated. if (!cfg.channels) cfg.channels = {}; cfg.channels.alexa = { ...(cfg.channels.alexa || {}), enabled: true, sharedSecret: data.sharedSecret, }; saveConfig(cfg); // Initialize the channel if it wasn't already running. await channelManager.init().catch(() => {}); res.writeHead(200); res.end(JSON.stringify({ ok: true, code: data.code, expiresAt: data.expiresAt })); } catch (err: any) { log.warn(`[alexa/pair] ${err.message}`); res.writeHead(500); res.end(JSON.stringify({ ok: false, error: err.message })); } })(); return; } // GET /api/channels/alexa/pair-page — inline HTML page showing the code with countdown. // The chat surface auto-converts URLs ending in /pair-page into a button (see MessageBubble). if (req.method === 'GET' && channelPath === '/api/channels/alexa/pair-page') { res.setHeader('Content-Type', 'text/html'); const alexaStatus = channelManager.getStatus('alexa'); const alreadyLinked = !!(alexaStatus?.info as any)?.linked; const confettiHTML = Array.from({ length: 30 }, (_, i) => { const colors = ['#0166FF', '#009AFE', '#4AEEFF', '#4ade80', '#facc15', '#818cf8']; const color = colors[Math.floor(Math.random() * colors.length)]; const left = Math.random() * 100; const delay = i * 0.04; const drift = (Math.random() - 0.5) * 120; const duration = 1.8 + Math.random() * 0.8; return `
`; }).join(''); res.writeHead(200); res.end(`Connect Alexa
${alreadyLinked ? `
${confettiHTML}
Connected!

Alexa is linked. Say "Alexa, open Morphy Agent" to start a conversation, or "Alexa, tell Morphy Agent <command>" for one-shots.

You can close this page.

` : `
Amazon Alexa
Link your Alexa

Open the Alexa app and enable the Morphy Agent skill, then say the phrase below.

Pairing code
— — — — — —
Generating code…
Say to Alexa
"Alexa, ask Morphy Agent to link with code ———"
How to link
1
Open the Alexa app on your phone
2
Find and enable the Morphy Agent skill
3
Say the phrase above to your Alexa device

`}
`); return; } // GET /api/channels/alexa/status — current alexa channel status if (req.method === 'GET' && channelPath === '/api/channels/alexa/status') { const status = channelManager.getStatus('alexa'); res.writeHead(200); res.end(JSON.stringify(status || { channel: 'alexa', connected: false })); return; } // ── Telegram (BYO bot — the user pastes a @BotFather token; we validate it and // long-poll Telegram DIRECTLY. NO relay anywhere in the pairing or message path) ── // POST /api/channels/telegram/connect — { botToken }. Sets the agent's bot token, so it is // GATED like /api/onboard: open only on genuine first run (no portal_pass yet); once a portal // password is set it requires a valid Bearer token (the pair-page sends the dashboard's // bloby_token). Without this gate, any unauthenticated caller over the tunnel could inject // their own bot token and hijack the channel (and, via trust-on-first-use, the agent). if (req.method === 'POST' && channelPath === '/api/channels/telegram/connect') { let body = ''; let tooLarge = false; req.on('data', (chunk: Buffer) => { body += chunk.toString(); if (body.length > 8_000) { tooLarge = true; req.destroy(); } }); req.on('error', () => { if (res.headersSent) return; res.writeHead(400); res.end(JSON.stringify({ ok: false, error: 'read-error' })); }); req.on('end', () => { (async () => { try { if (tooLarge) { res.writeHead(413); res.end(JSON.stringify({ ok: false, error: 'body-too-large' })); return; } // Auth gate (matches /api/onboard): internal calls bypass; otherwise require a valid // portal token whenever a password is set. First run (no password) stays open. if (req.headers['x-internal'] !== internalSecret && await isAuthRequired()) { const auth = req.headers.authorization; const tok = auth?.startsWith('Bearer ') ? auth.slice(7) : null; if (!tok || !(await validateToken(tok))) { res.writeHead(401); res.end(JSON.stringify({ ok: false, error: 'auth-required' })); return; } } const data = JSON.parse(body || '{}'); const botToken = typeof data.botToken === 'string' ? data.botToken.trim() : ''; if (!botToken || !/^\d{6,}:[A-Za-z0-9_-]{30,}$/.test(botToken)) { res.writeHead(400); res.end(JSON.stringify({ ok: false, error: 'invalid-token-format' })); return; } // Authoritative validation: ask Telegram who this token belongs to (with a timeout // so a hung TLS connection can't wedge the request / the pair modal). let me: any; try { const r = await fetch(`https://api.telegram.org/bot${encodeURIComponent(botToken)}/getMe`, { signal: AbortSignal.timeout(8000) }); me = await r.json().catch(() => ({})); } catch { res.writeHead(502); res.end(JSON.stringify({ ok: false, error: 'telegram-unreachable' })); return; } if (!me || me.ok !== true || !me.result || me.result.is_bot !== true) { res.writeHead(400); res.end(JSON.stringify({ ok: false, error: 'invalid-bot-token' })); return; } const botUsername = me.result.username || undefined; const cfg = loadConfig(); if (!cfg.channels) cfg.channels = {}; // On a token CHANGE (not a same-token reconnect), drop identity-bound fields so // trust-on-first-use re-adopts the owner against the NEW bot — otherwise a stale // ownerUserId/admins would silently lock out the new operator. const prev = cfg.channels.telegram; const tokenChanged = !!prev?.botToken && prev.botToken !== botToken; cfg.channels.telegram = { ...(prev || {}), enabled: true, mode: prev?.mode || 'channel', botToken, botUsername, ...(tokenChanged ? { ownerUserId: undefined, admins: undefined } : {}), }; saveConfig(cfg); // Re-connecting with a new token: tear down any existing provider so init() // reconnects with the new token (init() is a no-op when a provider already exists). await channelManager.disconnectChannel('telegram').catch(() => {}); await channelManager.init().catch(() => {}); res.writeHead(200); res.end(JSON.stringify({ ok: true, botUsername })); } catch (err: any) { log.warn(`[telegram/connect] ${err.message}`); if (!res.headersSent) { res.writeHead(500); res.end(JSON.stringify({ ok: false, error: err.message })); } } })(); }); return; } // POST /api/channels/telegram/configure — set mode + admins + skill (loopback-only). if (req.method === 'POST' && channelPath === '/api/channels/telegram/configure') { let body = ''; req.on('data', (chunk: Buffer) => { body += chunk.toString(); }); req.on('end', () => { try { const data = JSON.parse(body || '{}'); const cfg = loadConfig(); if (!cfg.channels) cfg.channels = {}; if (!cfg.channels.telegram) cfg.channels.telegram = { enabled: false, mode: 'channel' }; if (data.mode) cfg.channels.telegram.mode = data.mode; if (data.admins !== undefined) cfg.channels.telegram.admins = data.admins; if (data.skill !== undefined) cfg.channels.telegram.skill = data.skill; if (data.allowGroups !== undefined) cfg.channels.telegram.allowGroups = !!data.allowGroups; if (data.allowOthersToTrigger !== undefined) cfg.channels.telegram.allowOthersToTrigger = !!data.allowOthersToTrigger; saveConfig(cfg); res.writeHead(200); res.end(JSON.stringify({ ok: true, config: { ...cfg.channels.telegram, botToken: cfg.channels.telegram.botToken ? '***' : undefined } })); } catch (err: any) { res.writeHead(400); res.end(JSON.stringify({ error: err.message })); } }); return; } // POST /api/channels/telegram/disconnect — stop polling, KEEP the token (loopback-only). if (req.method === 'POST' && channelPath === '/api/channels/telegram/disconnect') { (async () => { try { await channelManager.disconnectChannel('telegram'); res.writeHead(200); res.end(JSON.stringify({ ok: true })); } catch (err: any) { res.writeHead(500); res.end(JSON.stringify({ ok: false, error: err.message })); } })(); return; } // POST /api/channels/telegram/logout — stop polling + forget the token (loopback-only). if (req.method === 'POST' && channelPath === '/api/channels/telegram/logout') { (async () => { try { await channelManager.disconnectChannel('telegram'); const cfg = loadConfig(); if (cfg.channels?.telegram) { // "Forget the token" also forgets the identity bound to it, so a future bot re-adopts // its owner cleanly via trust-on-first-use. delete cfg.channels.telegram.botToken; delete cfg.channels.telegram.ownerUserId; delete cfg.channels.telegram.admins; cfg.channels.telegram.enabled = false; saveConfig(cfg); } res.writeHead(200); res.end(JSON.stringify({ ok: true })); } catch (err: any) { res.writeHead(500); res.end(JSON.stringify({ ok: false, error: err.message })); } })(); return; } // POST /api/channels/telegram/reconnect — resume polling an already-connected bot // after a disconnect, without re-pairing (loopback-only). if (req.method === 'POST' && channelPath === '/api/channels/telegram/reconnect') { (async () => { try { const cfg = loadConfig(); if (!cfg.channels?.telegram?.botToken) { res.writeHead(400); res.end(JSON.stringify({ ok: false, error: 'no-bot-token' })); return; } if (cfg.channels.telegram.enabled !== true) { cfg.channels.telegram.enabled = true; saveConfig(cfg); } await channelManager.init().catch(() => {}); res.writeHead(200); res.end(JSON.stringify({ ok: true })); } catch (err: any) { res.writeHead(500); res.end(JSON.stringify({ ok: false, error: err.message })); } })(); return; } // GET /api/channels/telegram/status — current telegram channel status if (req.method === 'GET' && channelPath === '/api/channels/telegram/status') { const status = channelManager.getStatus('telegram'); res.writeHead(200); res.end(JSON.stringify(status || { channel: 'telegram', connected: false })); return; } // GET /api/channels/telegram/pair-page — BYO BotFather token flow: button opens a modal explaining // how to mint a token in @BotFather, takes the pasted token, POSTs {botToken} to /connect. if (req.method === 'GET' && channelPath === '/api/channels/telegram/pair-page') { res.setHeader('Content-Type', 'text/html'); const tgStatus = channelManager.getStatus('telegram'); const alreadyLinked = !!(tgStatus?.info as any)?.linked; const linkedUsername = (tgStatus?.info as any)?.botUsername || ''; const confettiHTML = Array.from({ length: 30 }, (_, i) => { const colors = ['#0166FF', '#009AFE', '#4AEEFF', '#4ade80', '#facc15', '#818cf8']; const color = colors[Math.floor(Math.random() * colors.length)]; const left = Math.random() * 100; const delay = i * 0.04; const drift = (Math.random() - 0.5) * 120; const duration = 1.8 + Math.random() * 0.8; return `
`; }).join(''); res.writeHead(200); res.end(`Connect Telegram
${alreadyLinked ? `
${confettiHTML}
Connected!

Telegram is linked${linkedUsername ? ` to @${linkedUsername}` : ''}. Open Telegram and message your bot to start chatting.

You can close this page.

` : `
Telegram
Connect Telegram

Bring your own bot. Create one for free in @BotFather, paste its token, and you're live.

It takes about a minute and keeps your bot fully private to you.

`}
${alreadyLinked ? '' : `
`} `); return; } // Fallback for unknown channel routes res.writeHead(404); res.end(JSON.stringify({ error: 'Not found' })); return; } // GET /api/env — read workspace .env back, grouped by `# Section` comment headers. // Used by the Settings → Environment Variables screen. Reading the workspace .env (which // holds the user's secrets) is privileged: gate it exactly like the POST below — portal // token when a password is set, x-internal for internal supervisor calls. if (req.method === 'GET' && (req.url === '/api/env' || req.url?.startsWith('/api/env?'))) { if (req.headers['x-internal'] !== internalSecret && await isAuthRequired()) { const authHeader = req.headers['authorization']; const token = authHeader?.startsWith('Bearer ') ? authHeader.slice(7) : null; if (!token || !(await validateToken(token))) { res.writeHead(401, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ error: 'Unauthorized' })); return; } } try { const envPath = path.join(WORKSPACE_DIR, '.env'); const groups: { title: string; vars: { name: string; value: string }[] }[] = []; if (fs.existsSync(envPath)) { const indexByTitle = new Map(); const seen = new Set(); // first occurrence wins — mirrors the POST writer + backend loader let currentTitle = ''; let pendingTitle = ''; let adjacentComment = false; // was the immediately-preceding non-blank line a comment? for (const line of fs.readFileSync(envPath, 'utf-8').split('\n')) { const trimmed = line.trim(); if (!trimmed) { adjacentComment = false; continue; } if (trimmed.startsWith('#')) { // Remember this comment, but only adopt it as the section title when a variable // sits directly under it — so multi-line file-header comment blocks (separated // from the vars by a blank line) don't become a bogus section heading. pendingTitle = trimmed.replace(/^#+/, '').trim(); adjacentComment = true; continue; } const eq = trimmed.indexOf('='); if (eq === -1) { adjacentComment = false; continue; } const name = trimmed.slice(0, eq).trim(); if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(name)) { adjacentComment = false; continue; } if (adjacentComment) currentTitle = pendingTitle; adjacentComment = false; if (seen.has(name)) continue; seen.add(name); let value = trimmed.slice(eq + 1).trim(); // Only unwrap a fully-balanced matching quote pair. The POST writer stores values // verbatim (no quoting), so a lone trailing/leading quote is part of the secret — // never strip it (that would silently corrupt the value by one character). if (value.length >= 2 && ((value[0] === '"' && value[value.length - 1] === '"') || (value[0] === "'" && value[value.length - 1] === "'"))) { value = value.slice(1, -1); } let gi = indexByTitle.get(currentTitle); if (gi === undefined) { gi = groups.length; indexByTitle.set(currentTitle, gi); groups.push({ title: currentTitle, vars: [] }); } groups[gi].vars.push({ name, value }); } } res.writeHead(200, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ groups })); } catch (err: any) { res.writeHead(500, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ error: err.message })); } return; } // POST /api/env — write env vars to workspace .env (used by chat EnvForm). if (req.method === 'POST' && req.url === '/api/env') { // This route is intercepted before the /api worker gate, so gate it here: writing the // workspace .env (which the backend loads) is a privileged action — require the portal token // when a password is set. Internal supervisor calls (x-internal) bypass. The chat EnvForm // sends the token via authFetch. if (req.headers['x-internal'] !== internalSecret && await isAuthRequired()) { const authHeader = req.headers['authorization']; const token = authHeader?.startsWith('Bearer ') ? authHeader.slice(7) : null; if (!token || !(await validateToken(token))) { res.writeHead(401, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ error: 'Unauthorized' })); return; } } let body = ''; let bodyBytes = 0; let tooLarge = false; req.on('data', (chunk: Buffer) => { bodyBytes += chunk.length; if (bodyBytes > 1_000_000) { tooLarge = true; req.destroy(); return; } // .env is tiny; 1MB is generous body += chunk.toString(); }); req.on('end', () => { if (tooLarge) return; try { const { vars, remove } = JSON.parse(body) as { vars?: Record; remove?: string[] }; if ((!vars || typeof vars !== 'object') && !Array.isArray(remove)) { res.writeHead(400, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ error: 'Missing vars object' })); return; } const envPath = path.join(WORKSPACE_DIR, '.env'); let lines: string[] = []; if (fs.existsSync(envPath)) { lines = fs.readFileSync(envPath, 'utf-8').split('\n'); } // Delete requested keys first (so removing then re-adding the same key in one request // behaves predictably). A key line is `KEY=...` / `KEY =...`. if (Array.isArray(remove)) { for (const rawKey of remove) { const key = String(rawKey).trim(); if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(key)) continue; lines = lines.filter((l) => { const t = l.trim(); return !(t.startsWith(`${key}=`) || t.startsWith(`${key} =`)); }); } } for (const [rawKey, rawValue] of Object.entries(vars || {})) { const key = rawKey.trim(); // Validate the key as a real env var name; reject anything that could inject extra // lines or break .env parsing. if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(key)) { res.writeHead(400, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ error: `Invalid env var name: ${rawKey}` })); return; } const value = typeof rawValue === 'string' ? rawValue.trim() : rawValue; // Reject embedded newlines/CR in the value — they'd inject arbitrary extra .env lines. if (typeof value === 'string' && /[\r\n]/.test(value)) { res.writeHead(400, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ error: `Invalid value for ${key}: must not contain newlines` })); return; } // Find existing line for this key (supports KEY=val, KEY="val", KEY='val') const idx = lines.findIndex((l) => { const trimmed = l.trim(); return trimmed.startsWith(`${key}=`) || trimmed.startsWith(`${key} =`); }); const newLine = `${key}=${value}`; if (idx >= 0) { lines[idx] = newLine; // Update existing } else { lines.push(newLine); // Append new } } // Remove trailing empty lines, ensure final newline while (lines.length > 0 && lines[lines.length - 1].trim() === '') lines.pop(); fs.writeFileSync(envPath, lines.join('\n') + '\n', 'utf-8'); res.writeHead(200, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ ok: true })); } catch (err: any) { res.writeHead(500, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ error: err.message })); } }); return; } // ── Pulse & Cron schedule (Settings → Pulse & Crons) ── // Reads/writes workspace PULSE.json + CRONS.json (+ tasks/{id}.md). Same privileged gate as // /api/env — portal token when a password is set, x-internal for internal supervisor calls. // These routes are intercepted before the /api worker gate, so the gate is load-bearing // (CRONS.json + task files hold the agent's instructions). `id` is validated against a slug // regex before any path.join to block path traversal (e.g. id=../../.env). if (req.url === '/api/schedule' || req.url?.startsWith('/api/schedule?') || req.url === '/api/pulse' || req.url === '/api/crons/pause' || req.url === '/api/crons/delete' || req.url?.startsWith('/api/crons/task')) { const scheduleAuthed = async (): Promise => { if (req.headers['x-internal'] === internalSecret) return true; if (!(await isAuthRequired())) return true; const authHeader = req.headers['authorization']; const token = authHeader?.startsWith('Bearer ') ? authHeader.slice(7) : null; return !!token && await validateToken(token); }; if (!(await scheduleAuthed())) { res.writeHead(401, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ error: 'Unauthorized' })); return; } const CRON_ID_RE = /^[A-Za-z0-9_-]+$/; const TIME_RE = /^([01]\d|2[0-3]):[0-5]\d$/; const PULSE_PATH = path.join(WORKSPACE_DIR, 'PULSE.json'); const CRONS_PATH = path.join(WORKSPACE_DIR, 'CRONS.json'); const sendJson = (code: number, obj: any) => { res.writeHead(code, { 'Content-Type': 'application/json' }); res.end(JSON.stringify(obj)); }; const readBody = () => new Promise((resolve, reject) => { let body = ''; let bytes = 0; req.on('data', (chunk: Buffer) => { bytes += chunk.length; if (bytes > 200_000) { req.destroy(); reject(new Error('Body too large')); return; } body += chunk.toString(); }); req.on('end', () => resolve(body)); req.on('error', reject); }); // GET /api/schedule — pulse config + crons (humanized schedule, next run, hasTaskFile). if (req.method === 'GET' && (req.url === '/api/schedule' || req.url.startsWith('/api/schedule?'))) { try { const pulse = readPulseConfig(); const crons = readCronsConfig().map((c) => ({ id: c.id, schedule: c.schedule, task: typeof c.task === 'string' ? c.task : '', oneShot: !!c.oneShot, paused: c.paused === true, hasTaskFile: CRON_ID_RE.test(c.id || '') && fs.existsSync(path.join(WORKSPACE_DIR, 'tasks', `${c.id}.md`)), nextRun: nextRunISO(c.schedule), description: describeCron(c.schedule), })); sendJson(200, { pulse, crons }); } catch (err: any) { sendJson(500, { error: err.message }); } return; } // POST /api/pulse — write PULSE.json (validated). Does NOT restart the backend; the // scheduler re-reads PULSE.json on its next 60s tick. if (req.method === 'POST' && req.url === '/api/pulse') { try { const { enabled, intervalMinutes, quietHours } = JSON.parse(await readBody()); const mins = Math.floor(Number(intervalMinutes)); // Reject NaN/out-of-range — a non-positive interval makes the pulse fire every tick. if (!Number.isFinite(mins) || mins < 1 || mins > 1440) { sendJson(400, { error: 'intervalMinutes must be a whole number between 1 and 1440' }); return; } // Reject malformed quiet hours — isInQuietHours can't validate, so a bad value would // silently disable quiet hours entirely. if (!quietHours || typeof quietHours !== 'object' || !TIME_RE.test(quietHours.start) || !TIME_RE.test(quietHours.end)) { sendJson(400, { error: 'quietHours.start and .end must be HH:MM (00:00–23:59)' }); return; } const config = { enabled: !!enabled, intervalMinutes: mins, quietHours: { start: quietHours.start, end: quietHours.end } }; fs.writeFileSync(PULSE_PATH, JSON.stringify(config, null, 2) + '\n', 'utf-8'); sendJson(200, { ok: true }); } catch (err: any) { sendJson(500, { error: err.message }); } return; } // POST /api/crons/pause — read-modify-write the `paused` flag of one cron, preserving // every other field (never reconstruct the entry — that would clobber enabled/oneShot/task). if (req.method === 'POST' && req.url === '/api/crons/pause') { try { const { id, paused } = JSON.parse(await readBody()); if (typeof id !== 'string' || !CRON_ID_RE.test(id)) { sendJson(400, { error: 'Invalid cron id' }); return; } const crons = readCronsConfig(); const idx = crons.findIndex((c) => c.id === id); if (idx < 0) { sendJson(404, { error: 'Cron not found' }); return; } crons[idx] = { ...crons[idx], paused: !!paused }; fs.writeFileSync(CRONS_PATH, JSON.stringify(crons, null, 2) + '\n', 'utf-8'); sendJson(200, { ok: true }); } catch (err: any) { sendJson(500, { error: err.message }); } return; } // POST /api/crons/delete — remove the cron and its task file (idempotent). if (req.method === 'POST' && req.url === '/api/crons/delete') { try { const { id } = JSON.parse(await readBody()); if (typeof id !== 'string' || !CRON_ID_RE.test(id)) { sendJson(400, { error: 'Invalid cron id' }); return; } const crons = readCronsConfig(); const remaining = crons.filter((c) => c.id !== id); fs.writeFileSync(CRONS_PATH, JSON.stringify(remaining, null, 2) + '\n', 'utf-8'); try { fs.unlinkSync(path.join(WORKSPACE_DIR, 'tasks', `${id}.md`)); } catch {} sendJson(200, { ok: true }); } catch (err: any) { sendJson(500, { error: err.message }); } return; } // GET /api/crons/task?id= — download the cron's task file (authed → blob on the client). if (req.method === 'GET' && req.url.startsWith('/api/crons/task')) { try { const id = new URL(req.url, `http://${req.headers.host}`).searchParams.get('id') || ''; if (!CRON_ID_RE.test(id)) { sendJson(400, { error: 'Invalid cron id' }); return; } const taskPath = path.join(WORKSPACE_DIR, 'tasks', `${id}.md`); if (!fs.existsSync(taskPath)) { sendJson(404, { error: 'Task file not found' }); return; } const content = fs.readFileSync(taskPath, 'utf-8'); res.writeHead(200, { 'Content-Type': 'text/markdown; charset=utf-8', 'Content-Disposition': `attachment; filename="${id}.md"`, }); res.end(content); } catch (err: any) { sendJson(500, { error: err.message }); } return; } sendJson(404, { error: 'Not found' }); return; } // ── Agent API — SDK gateway for workspace code ── if (req.url?.startsWith('/api/agent/')) { const agentPath = req.url.split('?')[0]; // Localhost-only guard. NOTE: cloudflared connects over loopback, so a tunneled request also // arrives as 127.0.0.1 — this IP check alone is a no-op behind the relay. We additionally // reject any request carrying cloudflared's cf-connecting-ip header (tunnel-origin traffic); // the real defense is the 256-bit agent secret below. The Agent API is only ever called by // the local workspace backend (loopback), never legitimately from the public tunnel. const remoteIp = req.socket.remoteAddress || ''; const isLoopback = remoteIp === '127.0.0.1' || remoteIp === '::1' || remoteIp === '::ffff:127.0.0.1'; if (!isLoopback || req.headers['cf-connecting-ip'] || req.headers['cf-ray']) { res.setHeader('Content-Type', 'application/json'); res.writeHead(403); res.end(JSON.stringify({ ok: false, error: 'Agent API is localhost-only.' })); return; } // Auth: x-agent-secret header (query param fallback for EventSource which cannot set headers). // Constant-time compare (length-guarded) so the secret can't be recovered via timing. const urlObj = new URL(req.url, `http://${req.headers.host}`); const headerSecret = req.headers['x-agent-secret']; const querySecret = urlObj.searchParams.get('secret'); const presented = typeof headerSecret === 'string' ? headerSecret : (querySecret || ''); const secretOk = presented.length === agentSecret.length && crypto.timingSafeEqual(Buffer.from(presented), Buffer.from(agentSecret)); if (!secretOk) { res.setHeader('Content-Type', 'application/json'); res.writeHead(401); res.end(JSON.stringify({ ok: false, error: 'Invalid or missing x-agent-secret.' })); return; } const readJsonBody = () => new Promise((resolve, reject) => { let body = ''; req.on('data', (chunk: Buffer) => { body += chunk.toString(); }); req.on('end', () => { try { resolve(body ? JSON.parse(body) : {}); } catch (err: any) { reject(err); } }); req.on('error', reject); }); // POST /api/agent/query — one-shot agent query (legacy AGENT-API.md endpoint) if (req.method === 'POST' && agentPath === '/api/agent/query') { res.setHeader('Content-Type', 'application/json'); res.setHeader('Cache-Control', 'no-store, no-cache, must-revalidate'); let body = ''; req.on('data', (chunk: Buffer) => { body += chunk.toString(); }); req.on('end', async () => { try { const parsed = JSON.parse(body) as AgentQueryRequest; const result = await handleAgentQuery(parsed); if (result.usedFileTools) { void doRestart(); broadcastBloby('app:hmr-update', {}); } res.writeHead(result.ok ? 200 : 400); res.end(JSON.stringify(result)); } catch (err: any) { res.writeHead(500); res.end(JSON.stringify({ ok: false, error: err.message })); } }); return; } // ── Workspace channel: live-conversation mirror of the Bloby chat ── // The workspace is a chat surface like the dashboard widget. Everything users // see in the widget + WhatsApp is mirrored here; messages sent from workspace // run through the SAME orchestrator with the SAME memory/agents/recent history. // GET /api/agent/chat/stream — Server-Sent Events, every chat event (bot:*, chat:*) if (req.method === 'GET' && agentPath === '/api/agent/chat/stream') { const clientId = urlObj.searchParams.get('clientId') || undefined; const subId = crypto.randomBytes(8).toString('hex'); res.setHeader('Content-Type', 'text/event-stream'); res.setHeader('Cache-Control', 'no-store, no-cache, must-revalidate'); res.setHeader('Connection', 'keep-alive'); res.setHeader('X-Accel-Buffering', 'no'); res.writeHead(200); try { res.socket?.setNoDelay(true); } catch {} res.write(': connected\n\n'); const sub: ChatSubscriber = { id: subId, clientId, send: (type, data) => { if (res.writableEnded) return; res.write(`event: ${type}\ndata: ${JSON.stringify(data)}\n\n`); }, close: () => { try { res.end(); } catch {} }, }; chatSubscribers.add(sub); if (agentQueryActive && currentStreamConvId) { sub.send('chat:state', { streaming: true, conversationId: currentStreamConvId, buffer: currentStreamBuffer, }); } const keepAlive = setInterval(() => { if (res.writableEnded) return; res.write(': ping\n\n'); }, 25_000); req.on('close', () => { clearInterval(keepAlive); chatSubscribers.delete(sub); }); // Keep an error listener so a client reset can't surface as an unhandled 'error'. res.on('error', () => {}); return; } // GET /api/agent/chat/state — snapshot: convId + bot/user names + recent messages if (req.method === 'GET' && agentPath === '/api/agent/chat/state') { res.setHeader('Content-Type', 'application/json'); (async () => { try { const ctx = await workerApi('/api/context/current'); const convId: string | undefined = ctx.conversationId; const [status, recentRaw] = await Promise.all([ workerApi('/api/onboard/status'), convId ? workerApi(`/api/conversations/${convId}/messages/recent?limit=50`) : Promise.resolve([]), ]); const messages = Array.isArray(recentRaw) ? recentRaw .filter((m: any) => m.role === 'user' || m.role === 'assistant') .map((m: any) => ({ role: m.role, content: m.content, timestamp: m.created_at })) : []; res.writeHead(200); res.end(JSON.stringify({ ok: true, conversationId: convId || null, agentName: status?.agentName || 'Bloby', userName: status?.userName || 'Human', streaming: agentQueryActive, streamConversationId: currentStreamConvId, streamBuffer: currentStreamBuffer, messages, })); } catch (err: any) { res.writeHead(500); res.end(JSON.stringify({ ok: false, error: err.message })); } })(); return; } // POST /api/agent/chat/message — push a user message into the shared conversation. // Mirrors the chat-WS `user:message` flow: ensures live conversation, persists, pushes. if (req.method === 'POST' && agentPath === '/api/agent/chat/message') { res.setHeader('Content-Type', 'application/json'); (async () => { try { const body = await readJsonBody(); const content: string = body.content; const clientId: string | undefined = body.clientId; if (!content || typeof content !== 'string') { res.writeHead(400); res.end(JSON.stringify({ ok: false, error: 'Missing "content".' })); return; } const freshConfig = loadConfig(); const provider = freshConfig.ai.provider; if (provider !== 'anthropic' && provider !== 'openai' && provider !== 'pi') { res.writeHead(400); res.end(JSON.stringify({ ok: false, error: `Workspace chat requires a harnessed provider (got "${provider}").` })); return; } let savedFiles: SavedFile[] = []; // Bounded subset that saved within the caps — handed to the harness so the // model sees exactly what's persisted/shown (parity with the PWA path). const acceptedAttachments: any[] = []; if (Array.isArray(body.attachments) && body.attachments.length) { let totalBytes = 0; for (const att of body.attachments.slice(0, MAX_ATTACHMENTS_PER_MESSAGE)) { totalBytes += approxBase64Bytes(att?.data || ''); if (totalBytes > MAX_TOTAL_ATTACHMENT_BYTES) { log.warn(`[workspace-chat] attachment total exceeds cap — dropping remaining`); break; } try { savedFiles.push(saveAttachment(att)); acceptedAttachments.push(att); } catch (err: any) { log.warn(`[workspace-chat] attachment save: ${err.message}`); } } } let convId: string; const ctx = await workerApi('/api/context/current'); if (ctx.conversationId) { convId = ctx.conversationId; } else { const conv = await workerApi('/api/conversations', 'POST', { title: content.slice(0, 80), model: freshConfig.ai.model }); convId = conv.id; await workerApi('/api/context/set', 'POST', { conversationId: convId }); } const meta: any = { model: freshConfig.ai.model, channel: 'workspace' }; if (savedFiles.length) { meta.attachments = JSON.stringify(savedFiles.map((f) => ({ type: f.type, name: f.name, mediaType: f.mediaType, filePath: f.relPath, }))); } // Prepend the channel tag so the UI can show the workspace icon. const taggedWorkspaceContent = `[workspace]\n${content}`; try { await workerApi(`/api/conversations/${convId}/messages`, 'POST', { role: 'user', content: taggedWorkspaceContent, meta, }); } catch (err: any) { log.warn(`[workspace-chat] DB persist error: ${err.message}`); broadcastBloby('chat:persist-error', { conversationId: convId, role: 'user', error: err.message }); } // Echo user message to every OTHER surface (dashboard WS + other SSE subscribers). // The originating workspace subscriber renders its own message optimistically. broadcastBlobyExceptSubscriber(clientId, 'chat:sync', { conversationId: convId, message: { role: 'user', content: taggedWorkspaceContent, timestamp: new Date().toISOString(), attachments: savedFiles.length ? savedFiles.map((f) => ({ type: f.type, name: f.name, mediaType: f.mediaType, filePath: f.relPath })) : undefined, }, }); let botName = 'Bloby', humanName = 'Human'; let recentMessages: RecentMessage[] = []; try { const [status, recentRaw] = await Promise.all([ workerApi('/api/onboard/status'), workerApi(`/api/conversations/${convId}/messages/recent?limit=30`), ]); botName = status.agentName || 'Bloby'; humanName = status.userName || 'Human'; if (Array.isArray(recentRaw)) { const filtered = recentRaw.filter((m: any) => m.role === 'user' || m.role === 'assistant'); if (filtered.length > 0) { recentMessages = filtered.slice(0, -1).map((m: any) => ({ role: m.role as 'user' | 'assistant', content: m.content, })); } } } catch (err: any) { log.warn(`[workspace-chat] recent/status fetch failed (seeding without history): ${err?.message || err}`); } if (!hasConversation(convId)) { log.info(`[workspace-chat] Starting new live conversation: ${convId}`); const waState = channelManager.createWaStreamState(); await startConversation( convId, freshConfig.ai.model, createSharedChatOnMessage(convId, freshConfig.ai.model, botName, waState), { botName, humanName }, recentMessages, ); } const agentAttachments = acceptedAttachments.length ? acceptedAttachments : undefined; // Mirror to WhatsApp self-chat if connected (same behaviour as the widget). const waStatus = channelManager.getStatus('whatsapp'); const ownPhone = waStatus?.connected ? (waStatus.info?.phoneNumber as string | undefined) : undefined; const waMirrorTo = ownPhone ? `${ownPhone}@s.whatsapp.net` : undefined; // Channel tag so the agent knows the message came from the dashboard workspace // (parallel to [WhatsApp ...] and [Alexa ...] tags). Already prepended above. channelManager.pushWithRouting( convId, { surface: 'workspace', waSendTo: waMirrorTo, isSelfChat: true }, taggedWorkspaceContent, agentAttachments, savedFiles, ); res.writeHead(200); res.end(JSON.stringify({ ok: true, conversationId: convId })); } catch (err: any) { log.warn(`[workspace-chat] message error: ${err.message}`); res.writeHead(500); res.end(JSON.stringify({ ok: false, error: err.message })); } })(); return; } // POST /api/agent/chat/stop — interrupt the current turn (mirrors user:stop) if (req.method === 'POST' && agentPath === '/api/agent/chat/stop') { res.setHeader('Content-Type', 'application/json'); (async () => { try { const ctx = await workerApi('/api/context/current'); const convId: string | undefined = ctx.conversationId; if (convId && hasConversation(convId)) { log.info(`[workspace-chat] stop — ending live conversation ${convId}`); endConversation(convId); } res.writeHead(200); res.end(JSON.stringify({ ok: true })); } catch (err: any) { res.writeHead(500); res.end(JSON.stringify({ ok: false, error: err.message })); } })(); return; } // POST /api/agent/chat/clear — clear context (mirrors user:clear-context) if (req.method === 'POST' && agentPath === '/api/agent/chat/clear') { res.setHeader('Content-Type', 'application/json'); (async () => { try { const ctx = await workerApi('/api/context/current'); const convId: string | undefined = ctx.conversationId; if (convId && hasConversation(convId)) endConversation(convId); await workerApi('/api/context/clear', 'POST'); broadcastBloby('chat:cleared', {}); res.writeHead(200); res.end(JSON.stringify({ ok: true })); } catch (err: any) { res.writeHead(500); res.end(JSON.stringify({ ok: false, error: err.message })); } })(); return; } // POST /api/agent/chat/whisper — transcribe audio via the same Whisper the widget uses if (req.method === 'POST' && agentPath === '/api/agent/chat/whisper') { res.setHeader('Content-Type', 'application/json'); (async () => { try { const body = await readJsonBody(); if (!body.audio || typeof body.audio !== 'string') { res.writeHead(400); res.end(JSON.stringify({ ok: false, error: 'Missing "audio" (base64).' })); return; } const result = await workerApi('/api/whisper/transcribe', 'POST', { audio: body.audio }); res.writeHead(200); res.end(JSON.stringify({ ok: !result.error, ...result })); } catch (err: any) { res.writeHead(500); res.end(JSON.stringify({ ok: false, error: err.message })); } })(); return; } res.setHeader('Content-Type', 'application/json'); res.writeHead(404); res.end(JSON.stringify({ ok: false, error: 'Unknown agent endpoint.' })); return; } // API routes → handled in-process by worker Express app if (req.url?.startsWith('/api')) { // Internal supervisor calls (workerApi) bypass auth — they carry a per-process secret const isInternal = req.headers['x-internal'] === internalSecret; if (!isInternal) { // Require a token for EVERY /api route except the public pre-login allowlist. This now // covers GET data reads (conversations, context, wallet, devices, push status) that // previously skipped auth and leaked over the public relay. const method = req.method || 'GET'; if (!isPublicRoute(method, req.url || '')) { // POST /api/onboard is open only on genuine first run (no portal_pass yet). Read the // setting DIRECTLY rather than the 30s-cached isAuthRequired() so the gate closes the // instant onboarding sets a password — no stale-cache window for a takeover. The // dashboard's own re-onboard goes through the internal x-internal WS path (isInternal // above), so it never reaches here. All other routes keep using the cached check. const isOnboard = method === 'POST' && (req.url || '').split('?')[0] === '/api/onboard'; const needsAuth = isOnboard ? !!getSetting('portal_pass') : await isAuthRequired(); if (needsAuth) { const authHeader = req.headers['authorization']; const token = authHeader?.startsWith('Bearer ') ? authHeader.slice(7) : null; if (!token || !(await validateToken(token))) { res.writeHead(401, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ error: 'Unauthorized' })); return; } } } } // After successful Claude re-auth, end live conversations so they restart with a fresh token if (req.method === 'POST' && req.url === '/api/auth/claude/exchange') { const origEnd = res.end.bind(res); (res as any).end = function (this: typeof res, ...args: any[]) { try { const body = typeof args[0] === 'string' ? args[0] : args[0]?.toString(); if (body) { const json = JSON.parse(body); if (json.success) { log.info('[orchestrator] Claude re-auth succeeded — restarting conversations with fresh token'); endAllConversations(); } } } catch {} return origEnd(...args); }; } // Same for Codex OAuth: the app-server reads ~/.codex/auth.json at spawn, so a // running subprocess only adopts a new identity on re-spawn. End live conversations // after a successful exchange so the next message cold-starts with the fresh token. // (Wraps only the one-shot /exchange; the device-code /status route latches // success on every poll and must not re-fire teardown — handled at re-spawn instead.) if (req.method === 'POST' && req.url === '/api/auth/codex/exchange') { const origEnd = res.end.bind(res); (res as any).end = function (this: typeof res, ...args: any[]) { try { const body = typeof args[0] === 'string' ? args[0] : args[0]?.toString(); if (body) { const json = JSON.parse(body); if (json.success) { log.info('[orchestrator] Codex re-auth succeeded — restarting conversations with fresh token'); endAllConversations(); } } } catch {} return origEnd(...args); }; } workerApp(req, res); return; } // Bloby routes → serve pre-built static files from dist-bloby/ // Note: must check '/bloby/' (with slash) so the route only claims the chat UI under // /bloby/, never a root-served asset that merely starts with "bloby". if (req.url === '/bloby' || req.url?.startsWith('/bloby/')) { // Strip /bloby prefix, then query strings, then resolve file path let filePath = req.url!.replace(/^\/bloby\/?/, '').split('?')[0] || 'bloby.html'; const fullPath = path.join(DIST_BLOBY, filePath); // Security: prevent directory traversal if (!fullPath.startsWith(DIST_BLOBY)) { res.writeHead(403); res.end('Forbidden'); return; } try { const stat = fs.statSync(fullPath); if (stat.isFile()) { const ext = path.extname(fullPath); const mime = MIME_TYPES[ext] || 'application/octet-stream'; // HTML files: no-cache so rebuilds are picked up immediately // Hashed assets (.js, .css): immutable caching const cacheControl = ext === '.html' ? 'no-cache' : 'public, max-age=31536000, immutable'; // Text assets go through the in-memory cache: gzip (the 750 KB chat entry → ~230 KB // over the tunnel) + ETag/304. The mtime check inside handles rebuilds. Binary types // (images, fonts, webm) stream as before — gzip wouldn't help them. if (['.js', '.css', '.html', '.svg', '.json'].includes(ext) && stat.size < 5_000_000) { serveCachedText(req, res, 'bloby:' + filePath, { file: fullPath }, mime, cacheControl); return; } res.writeHead(200, { 'Content-Type': mime, 'Cache-Control': cacheControl }); // A file removed/replaced mid-stream (common during a workspace rebuild) emits an // async 'error' on the stream — without this listener it crashes the supervisor (G1). const rs = fs.createReadStream(fullPath); rs.on('error', () => { if (!res.headersSent) { res.writeHead(404); res.end('Not found'); } else res.destroy(); }); rs.pipe(res); } else { res.writeHead(404); res.end('Not found'); } } catch { res.writeHead(404); res.end('Not found'); } return; } // Platform assets — served from supervisor/public/ so they survive workspace swaps const cleanUrl = (req.url || '').split('?')[0]; const inAssetDir = PLATFORM_ASSET_DIRS.some((d) => cleanUrl.startsWith(d) && !cleanUrl.includes('..')); if (PLATFORM_ASSETS.has(cleanUrl) || inAssetDir) { const assetPath = path.join(SUPERVISOR_PUBLIC, cleanUrl); try { const stat = fs.statSync(assetPath); if (stat.isFile()) { const ext = path.extname(assetPath); const mime = MIME_TYPES[ext] || 'application/octet-stream'; // ETag + 304 via the memory cache (these had no validator, so every >24h-stale // client re-downloaded the sprite sheets in full). Binary types won't gzip below // the keep threshold — the helper drops unhelpful gzip automatically. Very large // files (videos) keep streaming. if (stat.size < 5_000_000) { serveCachedText(req, res, 'pub:' + cleanUrl, { file: assetPath }, mime, 'public, max-age=86400'); return; } res.writeHead(200, { 'Content-Type': mime, 'Cache-Control': 'public, max-age=86400' }); const rs = fs.createReadStream(assetPath); rs.on('error', () => { if (!res.headersSent) { res.writeHead(404); res.end('Not found'); } else res.destroy(); }); rs.pipe(res); return; } } catch { /* fall through to Vite */ } } // Document-ish request? Used by the backend-down interstitial (NOT the shell branch, // which needs top-level-only gating). sec-fetch-dest 'document' covers top-level // navigations; sec-fetch-mode 'navigate' also matches iframe documents (dest 'iframe', // wanted here — the interstitial must show inside the workspace frame too); the accept // fallback covers old clients that send no sec-fetch headers at all. const wantsHtml = req.method === 'GET' && ( req.headers['sec-fetch-dest'] === 'document' || req.headers['sec-fetch-mode'] === 'navigate' || (!req.headers['sec-fetch-dest'] && String(req.headers['accept'] || '').includes('text/html')) ); // SHELL INVERSION: TOP-LEVEL document navigations get the immortal shell (bubble + // chat + a same-origin iframe that hosts the actual workspace app). The iframe // re-requests the same URL with ?__bloby_frame=1, which skips this branch and falls // through to the interstitials / Vite proxy below — so Vite reloads, rebuilds, and // crash pages all happen INSIDE the iframe while the chat chrome survives. (The // Lovable/Bolt pattern.) The __bloby_frame param is lost on real in-frame navigations // ( links, location.href redirects, 302s), so the frame is also excluded by // sec-fetch-dest — those requests carry dest 'iframe' and must fall through to Vite, // never get a nested shell. wantsHtml is intentionally NOT reused here: its // mode === 'navigate' arm matches iframe documents (fine for the interstitial below, // wrong for the shell). pathname === '/' is special-cased because the service worker's // install-time precache fetch of '/' may carry no sec-fetch/accept headers yet must // still cache the shell. // Kill switch: BLOBY_NO_SHELL=1 restores legacy behavior (workspace served top-level; // widget.js injects the bubble into the workspace document as before). const fetchDest = req.headers['sec-fetch-dest']; const isSubframe = fetchDest === 'iframe' || fetchDest === 'frame' || fetchDest === 'embed' || fetchDest === 'object'; if ( req.method === 'GET' && process.env.BLOBY_NO_SHELL !== '1' && !(req.url || '').includes('__bloby_frame=1') && !isSubframe && (cleanUrl === '/' || fetchDest === 'document' || (!fetchDest && String(req.headers['accept'] || '').includes('text/html'))) ) { serveCachedText(req, res, 'shell', { content: SHELL_HTML }, 'text/html', 'no-cache', { 'X-Bloby-Origin': 'supervisor' }); return; } // Workspace backend has crash-looped and given up → serve the "backend down" interstitial // for dashboard DOCUMENT navigations, instead of proxying to Vite (which serves the user's // SPA that then 503s on every /app/api call and, for the common workspace-lock template, // misreads the dead backend as "no password set" and shows the lock-setup screen). Scoped to // top-level navigations only (not assets/HMR/XHR) and only when the backend has truly given // up — never during a normal 1–2s restart. The chat PWA (/bloby/*) is served earlier and is // unaffected. if (wantsHtml && isBackendDead()) { // X-Bloby-Origin marks this as the agent's OWN branded page so the relay passes it through // (and never mistakes it for a Cloudflare tunnel error to be replaced). res.writeHead(503, { 'Content-Type': 'text/html', 'X-Bloby-Origin': 'supervisor', 'Cache-Control': 'no-store, no-cache, must-revalidate' }); res.end(backendDownPage(readBackendLogTail(100))); return; } // Vite failed to boot (sentinel port) → serve the recovering page directly instead of // proxying to a dead port. Chat (/bloby/*) is served earlier, so the lifeline stays up. if (vitePorts.dashboard < 0) { res.writeHead(503, { 'Content-Type': 'text/html', 'X-Bloby-Origin': 'supervisor', 'Cache-Control': 'no-store, no-cache, must-revalidate' }); res.end(RECOVERING_HTML); return; } // Everything else → proxy to dashboard Vite dev server const GUARD_TAG = ''; // Vite dev serves everything identity-encoded and nothing else on the origin side // compresses, so the residential UPLINK — the slowest hop — carried every module raw. // Gzip text responses on the fly (level 4: ~4x smaller for ~nothing on a Pi, per response). // Streams (SSE), pre-encoded bodies, non-text types, and 304s pass through untouched. const clientAcceptsGzip = String(req.headers['accept-encoding'] || '').includes('gzip'); const GZIPPABLE_RE = /^(text\/(?!event-stream)|application\/(javascript|json|manifest\+json)|image\/svg)/; const proxy = http.request( { host: '127.0.0.1', port: vitePorts.dashboard, path: req.url, method: req.method, headers: stripHopByHop(req.headers), agent: loopbackAgent }, (proxyRes) => { const ct = String(proxyRes.headers['content-type'] || ''); const enc = String(proxyRes.headers['content-encoding'] || ''); // Inject the workspace guard into dashboard HTML *documents* only (and only when // uncompressed — Vite dev serves plain HTML). Assets, HMR, JSON, etc. stream through // untouched. The guard auto-reloads into the "backend down" interstitial and replaces // Vite's raw error overlay — supervisor-side so it reaches every workspace, no edits needed. if (!ct.includes('text/html') || enc) { const len = Number(proxyRes.headers['content-length'] || NaN); const gzip = clientAcceptsGzip && !enc && req.method === 'GET' && proxyRes.statusCode === 200 && GZIPPABLE_RE.test(ct) && !proxyRes.headers['content-range'] && !(len < 1024); if (!gzip) { res.writeHead(proxyRes.statusCode!, proxyRes.headers); proxyRes.pipe(res); return; } const headers = { ...proxyRes.headers, 'content-encoding': 'gzip', vary: 'Accept-Encoding' }; delete headers['content-length']; // length changes; let Node chunk it res.writeHead(proxyRes.statusCode!, headers); const gz = zlib.createGzip({ level: 4 }); gz.on('error', () => { try { res.destroy(); } catch {} }); proxyRes.pipe(gz).pipe(res); return; } const chunks: Buffer[] = []; proxyRes.on('data', (c: Buffer) => chunks.push(c)); proxyRes.on('end', () => { let html = Buffer.concat(chunks).toString('utf-8'); if (!html.includes('workspace-guard.js')) { html = html.includes('') ? html.replace('', GUARD_TAG + '') : GUARD_TAG + html; } const headers = { ...proxyRes.headers }; // Body length changed — drop both framing headers and let Node set content-length // from the single res.end() write. Vite's validators describe the UNMODIFIED body: // forwarding them would let a future conditional request 304 against bytes we changed. delete headers['content-length']; delete headers['transfer-encoding']; delete headers['etag']; delete headers['last-modified']; if (clientAcceptsGzip && proxyRes.statusCode === 200) { headers['content-encoding'] = 'gzip'; headers['vary'] = 'Accept-Encoding'; res.writeHead(proxyRes.statusCode!, headers); res.end(zlib.gzipSync(Buffer.from(html, 'utf-8'), { level: 4 })); return; } res.writeHead(proxyRes.statusCode!, headers); res.end(html); }); proxyRes.on('error', () => { try { res.destroy(); } catch {} }); }, ); proxy.on('error', (e) => { console.error(`[supervisor] Dashboard Vite proxy error: ${req.url}`, e.message); res.writeHead(503, { 'Content-Type': 'text/html', 'X-Bloby-Origin': 'supervisor' }); res.end(RECOVERING_HTML); }); req.pipe(proxy); }); // Bound WS frames so a single message can't stream an unbounded blob into memory // (ws default is ~100MiB). Sized to comfortably hold the per-message attachment caps // (MAX_TOTAL_ATTACHMENT_BYTES of decoded bytes ≈ 1.33× as base64) plus JSON overhead. const WS_MAX_PAYLOAD = 80 * 1024 * 1024; // WebSocket: Bloby chat + proxy worker WS const blobyWss = new WebSocketServer({ noServer: true, maxPayload: WS_MAX_PAYLOAD }); // WebSocket: App API proxy (routes /app/api calls through WS to avoid tunnel POST issues) const appWss = new WebSocketServer({ noServer: true, maxPayload: WS_MAX_PAYLOAD }); appWss.on('connection', (ws) => { // An 'error' event with no listener is rethrown by Node as an uncaught exception, // which would crash the whole supervisor. ws still tears down + fires 'close'. ws.on('error', () => {}); // Liveness: a half-open socket (mobile/Wi-Fi drop behind the tunnel) never fires 'close', so // its chat subscription + maps would leak and broadcastBloby would keep writing to it. The // heartbeat below pings; a peer that misses a pong is terminated (which fires 'close' → cleanup). (ws as any).isAlive = true; ws.on('pong', () => { (ws as any).isAlive = true; }); // Per-WS chat subscription: when the client opts in, this WS joins chatSubscribers // and receives every bot:* / chat:* event the dashboard widget does. SSE through the // Cloudflare tunnel buffers chunks; WebSocket frames flow through reliably (same // reason /app/api fetch is routed through this WS to begin with). let chatSub: ChatSubscriber | null = null; ws.on('message', (raw) => { const rawStr = raw.toString(); if (rawStr === 'ping') { if (ws.readyState === WebSocket.OPEN) ws.send('pong'); return; } let msg: any; try { msg = JSON.parse(rawStr); } catch { return; } if (msg.type === 'chat:subscribe') { if (chatSub) chatSubscribers.delete(chatSub); const clientId = msg.data?.clientId; const subId = crypto.randomBytes(8).toString('hex'); chatSub = { id: subId, clientId, send: (type, data) => { if (ws.readyState !== WebSocket.OPEN) return; ws.send(JSON.stringify({ type: 'chat:event', data: { eventType: type, eventData: data } })); }, close: () => {}, }; chatSubscribers.add(chatSub); if (agentQueryActive && currentStreamConvId) { chatSub.send('chat:state', { streaming: true, conversationId: currentStreamConvId, buffer: currentStreamBuffer, }); } if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'chat:subscribed', data: { clientId, subId } })); } return; } if (msg.type === 'chat:unsubscribe') { if (chatSub) { chatSubscribers.delete(chatSub); chatSub = null; } return; } if (msg.type !== 'app:api' || !msg.data) return; const { id, method, path: reqPath, headers: reqHeaders, body } = msg.data; const backendPath = (reqPath || '').replace(/^\/app/, ''); if (!isBackendAlive()) { if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'app:api:response', data: { id, status: 503, headers: { 'content-type': 'application/json' }, body: JSON.stringify({ error: 'Backend is starting...' }) }, })); } return; } const proxyHeaders: Record = { ...(reqHeaders || {}) }; if (body && !proxyHeaders['content-type']) { proxyHeaders['content-type'] = 'application/json'; } const proxyReq = http.request( { host: '127.0.0.1', port: backendPort, path: backendPath, method, headers: proxyHeaders }, (proxyRes) => { const chunks: Buffer[] = []; proxyRes.on('data', (chunk: Buffer) => chunks.push(chunk)); proxyRes.on('end', () => { const responseBody = Buffer.concat(chunks).toString('utf-8'); const resHeaders: Record = {}; for (const [k, v] of Object.entries(proxyRes.headers)) { if (typeof v === 'string') resHeaders[k] = v; else if (Array.isArray(v)) resHeaders[k] = v.join(', '); } if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'app:api:response', data: { id, status: proxyRes.statusCode, headers: resHeaders, body: responseBody }, })); } }); }, ); proxyReq.on('error', (e) => { console.error(`[supervisor] App WS backend error: ${backendPath}`, e.message); if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'app:api:response', data: { id, status: 503, headers: { 'content-type': 'application/json' }, body: JSON.stringify({ error: 'Backend unavailable' }) }, })); } }); if (body) proxyReq.write(body); proxyReq.end(); }); ws.on('close', () => { if (chatSub) { chatSubscribers.delete(chatSub); chatSub = null; } }); }); /** Send a message to every chat surface — dashboard widget WS clients AND workspace SSE * subscribers. The workspace mirror sees the exact same event stream the widget does. */ function broadcastBloby(type: string, data: any = {}) { const msg = JSON.stringify({ type, data }); for (const client of blobyWss.clients) { if (client.readyState === WebSocket.OPEN) client.send(msg); } for (const sub of chatSubscribers) { try { sub.send(type, data); } catch {} } } /** Like broadcastBloby but skips the workspace SSE subscriber identified by clientId. * Used when the workspace itself originated the event — that subscriber renders the * event optimistically and shouldn't receive its own echo. */ function broadcastBlobyExceptSubscriber(excludeClientId: string | undefined, type: string, data: any) { const msg = JSON.stringify({ type, data }); for (const client of blobyWss.clients) { if (client.readyState === WebSocket.OPEN) client.send(msg); } for (const sub of chatSubscribers) { if (excludeClientId && sub.clientId === excludeClientId) continue; try { sub.send(type, data); } catch {} } } /** Build the onMessage handler used by EVERY surface that pushes into the shared live * conversation. Chat-WS, the workspace SSE channel, and (in spirit) the channel manager * all want the same wiring: track stream buffer for late joiners, route streaming text * via the manager's per-conversation routing FIFO, persist responses, defer backend * restarts, and broadcast every event to all chat surfaces. * * Factored so adding a new surface (here: workspace) cannot drift from the chat WS * behaviour — same buffer, same persistence, same restart timing. */ const CHAT_TURN_EVENTS = new Set(['bot:token', 'bot:response', 'bot:tool', 'bot:task-created', 'bot:task-progress', 'bot:task-done']); function createSharedChatOnMessage( convId: string, model: string, botName: string, waState: ReturnType, ) { return async (type: string, eventData: any) => { // Capture surface BEFORE routeWaStreamEvent consumes the routing target on bot:response. // Used below to suppress chat-bubble broadcasts for non-dashboard turns. // Synthetic turns (background-task continuations — see routeWaStreamEvent's // guard) never own a routing target; the FIFO head, if any, belongs to a // QUEUED channel message, so peeking it would misclassify the turn. // They are always dashboard turns. const triggerSurface = channelManager.peekCurrentSurface(convId); const isDashboardTurn = eventData?.synthetic === true || !triggerSurface || triggerSurface === 'workspace' || triggerSurface === 'chat'; if (type === 'bot:typing') { currentStreamConvId = convId; currentStreamBuffer = ''; agentQueryActive = true; } if (type === 'bot:token' && eventData.token && isDashboardTurn) { currentStreamBuffer += eventData.token; } channelManager.routeWaStreamEvent(waState, type, eventData, botName); if (type === 'bot:turn-complete') { log.info(`[orchestrator] ──── TURN COMPLETE ────`); log.info(`[orchestrator] File tools used: ${eventData.usedFileTools}`); agentQueryActive = false; currentStreamConvId = null; currentStreamBuffer = ''; if (eventData.usedFileTools || pendingBackendRestart) { log.info('[orchestrator] Restarting backend (file tools used / pending watcher change)'); void doRestart(); } flushPendingUpdate(); // run a queued self-update now that this dashboard turn has ended // Proactive session recycling (see CONTEXT_RECYCLE_TOKENS). Only when the // harness reports the session idle (no queued message) — and this handler runs // synchronously from the harness callback, so nothing can be pushed between the // idle check and endConversation. Recycling here therefore never drops a queued // message; the next user message re-injects recent history + memory. if (eventData.idle && hasConversation(convId)) { const used = typeof eventData.contextTokens === 'number' ? eventData.contextTokens : 0; const window = typeof eventData.contextWindow === 'number' ? eventData.contextWindow : 0; const recycleAt = window > 0 ? Math.floor(window * CONTEXT_RECYCLE_FRACTION) : CONTEXT_RECYCLE_TOKENS; if (used > recycleAt) { log.info(`[orchestrator] Context ~${used} tok > ${recycleAt} (window=${window || 'n/a'}) — recycling session ${convId}; next message re-injects recent + memory.`); endConversation(convId); } } broadcastBloby('bot:idle', { conversationId: convId }); return; } if (type === 'bot:conversation-ended') { log.info(`[orchestrator] Conversation ended: ${convId}`); agentQueryActive = false; currentStreamConvId = null; currentStreamBuffer = ''; channelManager.clearRoutes(convId); // A turn that ended by exception/recycle (not a clean bot:turn-complete) must still flush a // queued self-update — otherwise it'd wait for the next turn/reboot. Self-defers + idempotent. flushPendingUpdate(); return; } if (type === 'bot:response') { currentStreamBuffer = ''; try { // 15s timeout: if this in-process write ever hangs (vs. rejects), the reply // broadcast below would never fire and the user's answer would silently vanish. // The timeout converts a hang into the already-handled catch → chat:persist-error, // and the reply still broadcasts. A message INSERT is sub-ms, so 15s is generous. await workerApi(`/api/conversations/${convId}/messages`, 'POST', { role: 'assistant', content: eventData.content, meta: { model }, }, 15000); } catch (err: any) { log.warn(`[bloby] DB persist bot response error: ${err.message}`); broadcastBloby('chat:persist-error', { conversationId: convId, role: 'assistant', error: err.message }); } } // Suppress agent-turn events from non-dashboard surfaces. WhatsApp/Alexa replies // are already delivered via the routing FIFO; broadcasting them would bleed // content into chat-bubble clients that weren't part of that conversation. if (CHAT_TURN_EVENTS.has(type) && !isDashboardTurn) return; broadcastBloby(type, eventData); }; } blobyWss.on('connection', (ws) => { log.info('Bloby chat connected'); // See appWss above: a listener-less 'error' event would crash the supervisor and kill // chat for everyone (G1). ws still fires 'close' afterward, so map cleanup still runs. ws.on('error', (err: any) => log.warn(`[bloby-ws] socket error: ${err?.message || err}`)); (ws as any).isAlive = true; ws.on('pong', () => { (ws as any).isAlive = true; }); let convId = Math.random().toString(36).slice(2) + Date.now().toString(36); conversations.set(ws, []); // Send current streaming state so reconnecting clients can catch up if (agentQueryActive && currentStreamConvId) { ws.send(JSON.stringify({ type: 'chat:state', data: { streaming: true, conversationId: currentStreamConvId, buffer: currentStreamBuffer, }, })); } ws.on('message', (raw) => { const rawStr = raw.toString(); // Heartbeat if (rawStr === 'ping') { if (ws.readyState === WebSocket.OPEN) ws.send('pong'); return; } // Guarded parse — a single malformed (non-'ping') text frame must never throw out // of this handler and crash the supervisor (G1). Mirrors the appWss handler above. let msg: any; try { msg = JSON.parse(rawStr); } catch { return; } if (!msg || typeof msg !== 'object') return; // Whisper transcription via WebSocket (bypasses relay POST issues) if (msg.type === 'whisper:transcribe') { (async () => { try { const result = await workerApi('/api/whisper/transcribe', 'POST', { audio: msg.data.audio }); if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'whisper:result', data: result })); } } catch (err: any) { if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'whisper:result', data: { error: err.message } })); } } })(); return; } // Push subscribe via WebSocket (bypasses relay POST issues) if (msg.type === 'push:subscribe') { (async () => { try { const result = await workerApi('/api/push/subscribe', 'POST', msg.data); if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'push:subscribed', data: result })); } } catch (err: any) { if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'push:subscribe-error', data: { error: err.message } })); } } })(); return; } // Push unsubscribe via WebSocket (bypasses relay POST issues) if (msg.type === 'push:unsubscribe') { (async () => { try { const result = await workerApi('/api/push/unsubscribe', 'DELETE', msg.data); if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'push:unsubscribed', data: result })); } } catch (err: any) { if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'push:unsubscribe-error', data: { error: err.message } })); } } })(); return; } // Save settings via WebSocket (bypasses relay POST issues) if (msg.type === 'settings:save') { (async () => { try { const result = await workerApi('/api/onboard', 'POST', msg.data); // Settings change may affect model/provider/token — restart conversations log.info('[orchestrator] Settings saved — restarting conversations'); endAllConversations(); if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'settings:saved', data: result })); } } catch (err: any) { log.error(`[bloby] settings:save failed: ${err.message}`); if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'settings:save-error', data: { error: err.message } })); } } })(); return; } // Switch tunnel mode at runtime (off ↔ quick) if (msg.type === 'tunnel:switch') { (async () => { try { const newMode = msg.data?.mode as 'off' | 'quick'; const cfg = loadConfig(); if (newMode === 'off') { // Stop tunnel, heartbeat, relay, watchdog stopHeartbeat(); if (cfg.relay?.token) { try { await disconnect(cfg.relay.token); } catch {} } stopTunnel(); if (watchdogInterval) { clearInterval(watchdogInterval); watchdogInterval = null; } tunnelUrl = null; cfg.tunnel.mode = 'off'; delete cfg.tunnelUrl; saveConfig(cfg); log.ok('[tunnel:switch] Switched to off — tunnel stopped'); if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'tunnel:switched', data: { mode: 'off' } })); } } else if (newMode === 'quick') { // Start tunnel const newUrl = await startTunnel(cfg.port); tunnelUrl = newUrl; cfg.tunnel.mode = 'quick'; cfg.tunnelUrl = newUrl; saveConfig(cfg); log.ok(`[tunnel:switch] Tunnel started: ${newUrl}`); // Wait for local server readiness for (let i = 0; i < 15; i++) { try { const res = await fetch(`http://127.0.0.1:${cfg.port}/api/health?_cb=${Date.now()}`, { signal: AbortSignal.timeout(3000), }); if (res.ok) break; } catch {} await new Promise(r => setTimeout(r, 1000)); } // Reconnect relay if handle exists if (cfg.relay?.token) { try { await updateTunnelUrl(cfg.relay.token, newUrl); startHeartbeat(cfg.relay.token, newUrl); log.ok('[tunnel:switch] Relay reconnected'); } catch (err) { log.warn(`[tunnel:switch] Relay reconnect: ${err instanceof Error ? err.message : err}`); } } // Re-create watchdog if (watchdogInterval) clearInterval(watchdogInterval); let lastTick = Date.now(); let healthCounter = 0; watchdogInterval = setInterval(async () => { const now = Date.now(); const wakeGap = now - lastTick > 60_000; const periodicCheck = ++healthCounter % 10 === 0; lastTick = now; if (wakeGap || periodicCheck) { const alive = await isTunnelAlive(tunnelUrl!, cfg.port); if (!alive) { log.warn('Tunnel dead, restarting...'); try { const restartedUrl = await restartTunnel(cfg.port); await new Promise(r => setTimeout(r, 3000)); tunnelUrl = restartedUrl; const latestCfg = loadConfig(); latestCfg.tunnelUrl = restartedUrl; saveConfig(latestCfg); if (latestCfg.relay?.token) { stopHeartbeat(); startHeartbeat(latestCfg.relay.token, restartedUrl); await updateTunnelUrl(latestCfg.relay.token, restartedUrl); } log.ok(`Tunnel restored: ${restartedUrl}`); } catch (err) { log.error(`Tunnel restart failed: ${err instanceof Error ? err.message : err}`); } } } }, 30_000); if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'tunnel:switched', data: { mode: 'quick', tunnelUrl: newUrl } })); } } } catch (err: any) { log.error(`[tunnel:switch] Error: ${err.message}`); if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'tunnel:switch-error', data: { error: err.message } })); } } })(); return; } // New protocol: { type: 'user:message', data: { content, conversationId? } } if (msg.type === 'user:message') { const data = msg.data || {}; const hasAtts = Array.isArray(data.attachments) && data.attachments.length > 0; const hasAudio = typeof data.audioData === 'string' && data.audioData.length > 0; // Allow attachment-only / voice-only turns (a pasted image with no caption, or a // voice note whose transcription came back empty) — previously these were dropped, // leaving a ghost optimistic bubble. Fall back to a placeholder so title/prompt work. let content: string = typeof data.content === 'string' ? data.content : ''; if (!content.trim() && !hasAtts && !hasAudio) return; if (!content.trim()) content = hasAtts ? '(attached files)' : '(voice message)'; // Note: we intentionally ignore data.conversationId from the client. // The server is the authority on which DB conversation this WS belongs to — // honoring a client-supplied id let stale browser state drive messages into // an orphan conv whose row had been deleted, causing FK failures on every // INSERT. Server resolution below (clientConvs → context.current → create). // Re-read config on each message so post-onboard changes are picked up const freshConfig = loadConfig(); const freshAi = (freshConfig.ai.provider && (freshConfig.ai.apiKey || freshConfig.ai.provider === 'ollama')) ? createProvider(freshConfig.ai.provider, freshConfig.ai.apiKey, freshConfig.ai.baseUrl) : null; log.info(`[bloby] provider=${freshConfig.ai.provider}, model=${freshConfig.ai.model}`); // Route through the agent harness for any provider that has one // (Anthropic → Claude SDK, OpenAI → Codex app-server, Bloby/pi → pi // harness). The dispatcher in bloby-agent.ts picks the right harness; // credentials live next to each harness (claude.json, codex auth.json, // pi-auth.json) — not in config.ai.apiKey. if (freshConfig.ai.provider === 'anthropic' || freshConfig.ai.provider === 'openai' || freshConfig.ai.provider === 'pi') { // Server-side persistence: create or reuse DB conversation, save user message (async () => { // Save attachments to disk (before try so it's accessible in startBlobyAgentQuery below). // saveAttachment enforces a per-file cap + content sniff; we additionally bound count + total. let savedFiles: SavedFile[] = []; // The bounded subset of raw attachments that actually saved within the caps — // this (not the full client array) is what the harness inlines, so the model // sees exactly what gets persisted + shown in chat (no over-cap divergence). const acceptedAttachments: any[] = []; if (hasAtts) { let totalBytes = 0; for (const att of data.attachments.slice(0, MAX_ATTACHMENTS_PER_MESSAGE)) { totalBytes += approxBase64Bytes(att?.data || ''); if (totalBytes > MAX_TOTAL_ATTACHMENT_BYTES) { log.warn(`[bloby] attachment total exceeds cap — dropping remaining`); break; } try { savedFiles.push(saveAttachment(att)); acceptedAttachments.push(att); } catch (err: any) { log.warn(`[bloby] File save error: ${err.message}`); } } } // Persist the voice clip (if any) so the chat can replay it after a refresh. let savedAudioPath: string | undefined; if (hasAudio) { try { savedAudioPath = saveAudio(data.audioData).relPath; } catch (err: any) { log.warn(`[bloby] Audio save error: ${err.message}`); } } try { // Check if we have an existing conversation for this client let dbConvId = clientConvs.get(ws); if (!dbConvId) { // Check if there's a current conversation set in settings const ctx = await workerApi('/api/context/current'); if (ctx.conversationId) { dbConvId = ctx.conversationId; } else { // Create a new conversation const conv = await workerApi('/api/conversations', 'POST', { title: content.slice(0, 80), model: freshConfig.ai.model }); dbConvId = conv.id; await workerApi('/api/context/set', 'POST', { conversationId: dbConvId }); } clientConvs.set(ws, dbConvId!); // Notify client of the conversation ID if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'chat:conversation-created', data: { conversationId: dbConvId } })); } } convId = dbConvId!; // Save user message to DB (include attachment + audio metadata as JSON string) const meta: any = { model: freshConfig.ai.model }; if (savedFiles.length) { meta.attachments = JSON.stringify(savedFiles.map((f) => ({ type: f.type, name: f.name, mediaType: f.mediaType, filePath: f.relPath, }))); } if (savedAudioPath) meta.audio_data = savedAudioPath; await workerApi(`/api/conversations/${convId}/messages`, 'POST', { role: 'user', content, meta, }); // Broadcast user message to other clients (include saved attachment + audio metadata) broadcastBlobyExcept(ws, 'chat:sync', { conversationId: convId, message: { role: 'user', content, timestamp: new Date().toISOString(), attachments: savedFiles.length ? savedFiles.map((f) => ({ type: f.type, name: f.name, mediaType: f.mediaType, filePath: f.relPath })) : undefined, // snake_case to match the client reader (useChat chat:sync handler), // the persisted meta.audio_data, and the DB-reload loader — one canonical name. audio_data: savedAudioPath, }, }); } catch (err: any) { log.warn(`[bloby] DB persist error: ${err.message}`); // Surface to all clients so they can flag the missing user bubble // instead of pretending it's saved. addMessage() in worker/db.ts is // self-healing for orphan convIds, so this should now be rare. broadcastBloby('chat:persist-error', { conversationId: convId, role: 'user', error: err.message }); } // Fetch agent/user names and recent messages in parallel let botName = 'Bloby', humanName = 'Human'; let recentMessages: RecentMessage[] = []; try { const [status, recentRaw] = await Promise.all([ workerApi('/api/onboard/status') as Promise, workerApi(`/api/conversations/${convId}/messages/recent?limit=30`) as Promise, ]); botName = status.agentName || 'Bloby'; humanName = status.userName || 'Human'; // Filter to user/assistant only, exclude the last entry (current message already sent as SDK prompt) if (Array.isArray(recentRaw)) { const filtered = recentRaw .filter((m: any) => m.role === 'user' || m.role === 'assistant'); // Slice off the last entry — it's the current user message if (filtered.length > 0) { recentMessages = filtered.slice(0, -1).map((m: any) => ({ role: m.role as 'user' | 'assistant', content: m.content, })); } } } catch (err: any) { log.warn(`[bloby] recent/status fetch failed (seeding without history): ${err?.message || err}`); } log.info(`[orchestrator] ──── USER MESSAGE ────`); log.info(`[orchestrator] Content: "${content.slice(0, 100)}..."`); log.info(`[orchestrator] Conv: ${convId}`); log.info(`[orchestrator] Live conversation exists: ${hasConversation(convId)}`); // Start a live conversation if one doesn't exist. The onMessage handler is // shared with the workspace channel — both surfaces push into the same conv // and observe the same events. if (!hasConversation(convId)) { log.info(`[orchestrator] Starting new live conversation...`); const waState = channelManager.createWaStreamState(); await startConversation( convId, freshConfig.ai.model, createSharedChatOnMessage(convId, freshConfig.ai.model, botName, waState), { botName, humanName }, recentMessages, ); } // Push the user message into the live conversation with a pinned routing // target. Chat-bubble responses are broadcast to all WS clients regardless; // the WhatsApp self-chat mirror (if connected) is the optional secondary // destination, baked in at push time so it cannot drift to a different chat. const waStatus = channelManager.getStatus('whatsapp'); const ownPhone = waStatus?.connected ? (waStatus.info?.phoneNumber as string | undefined) : undefined; const waMirrorTo = ownPhone ? `${ownPhone}@s.whatsapp.net` : undefined; log.info(`[orchestrator] Pushing message into live conversation (waMirror=${waMirrorTo || 'none'})`); // Don't prepend [PWA] if content already carries its own channel tag (e.g. [Mac] from Morphy) const alreadyTagged = /^\[[^\]]+\]\n/.test(content); channelManager.pushWithRouting( convId, { surface: 'chat', waSendTo: waMirrorTo, isSelfChat: true }, alreadyTagged ? content : `[PWA]\n${content}`, acceptedAttachments.length ? acceptedAttachments : undefined, savedFiles, ); })(); return; } // Other providers: use ai.chat() with conversation history const history = conversations.get(ws) || []; history.push({ role: 'user', content }); if (!freshAi) { if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'bot:error', data: { error: 'AI not configured. Set up your provider first.' } })); } return; } if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'bot:typing', data: { conversationId: convId } })); } freshAi.chat( [{ role: 'system', content: 'You are Bloby, a helpful AI assistant. You help users manage and customize their self-hosted bot.' }, ...history], freshConfig.ai.model, (token) => { if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ type: 'bot:token', data: { token, conversationId: convId } })); }, (full) => { history.push({ role: 'assistant', content: full }); if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ type: 'bot:response', data: { conversationId: convId, content: full } })); }, (err) => { if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ type: 'bot:error', data: { error: err.message } })); }, ); return; } if (msg.type === 'user:stop') { // End the live conversation (if any) or stop a one-shot query if (hasConversation(convId)) { log.info(`[orchestrator] user:stop — ending live conversation ${convId}`); endConversation(convId); } else { stopBlobyAgentQuery(convId); } return; } if (msg.type === 'user:stop-task') { const taskId = (msg as any).data?.taskId; if (taskId) { log.info(`[orchestrator] Stopping sub-agent task: ${taskId}`); stopSubAgentTask(convId, taskId).catch((err) => { log.warn(`[orchestrator] Failed to stop task ${taskId}: ${err.message}`); }); } return; } if (msg.type === 'user:clear-context') { (async () => { try { // End the live conversation if (hasConversation(convId)) { log.info(`[orchestrator] clear-context — ending live conversation ${convId}`); endConversation(convId); } clientConvs.delete(ws); await workerApi('/api/context/clear', 'POST'); } catch (err: any) { log.warn(`[bloby] Clear context error: ${err.message}`); } // Broadcast clear to ALL clients broadcastBloby('chat:cleared'); })(); return; } }); ws.on('close', () => { conversations.delete(ws); clientConvs.delete(ws); }); }); // Bloby chat WebSocket — Vite HMR is handled automatically (hmr.server = this server) server.on('upgrade', async (req, socket: net.Socket, head) => { // Strip the query string: /bloby/ws?token=<7-day session token> must not be logged. // App API WebSocket — no auth (backend handles its own auth) if (req.url?.startsWith('/app/ws')) { appWss.handleUpgrade(req, socket, head, (ws) => appWss.emit('connection', ws, req)); return; } if (!req.url?.startsWith('/bloby/ws')) { return; } // Auth check for WebSocket const needsAuth = await isAuthRequired(); if (needsAuth) { const urlObj = new URL(req.url, `http://${req.headers.host}`); const token = urlObj.searchParams.get('token'); if (!token || !(await validateToken(token))) { console.log('[supervisor] WS auth failed — rejecting'); socket.write('HTTP/1.1 401 Unauthorized\r\n\r\n'); socket.destroy(); return; } } blobyWss.handleUpgrade(req, socket, head, (ws) => blobyWss.emit('connection', ws, req)); }); // Start server.on('error', (err: NodeJS.ErrnoException) => { if (err.code === 'EADDRINUSE') { log.error(`Port ${config.port} is already in use. Stop the other process or change the port in ${paths.config}`); } else { log.error(`Server error: ${err.message}`); } process.exit(1); }); server.listen(config.port, () => { log.ok(`Supervisor on http://localhost:${config.port}`); log.ok(`Bloby chat at http://localhost:${config.port}/bloby`); writeRuntimeFile(); if (config.tunnel.mode === 'off') { console.log('__READY__'); } }); // ~/.bloby/supervisor.json — the CLI's single source of truth for "is Bloby // running and which process is it". Written here (not by the CLI) because the // supervisor has at least four different parents (cli foreground, launchd, // systemd, docker) and only the supervisor itself is present in all of them. // The CLI validates pid liveness + command line, so a stale file after kill -9 // is harmless. function writeRuntimeFile() { try { let version = 'unknown'; try { version = JSON.parse(fs.readFileSync(path.join(PKG_DIR, 'package.json'), 'utf-8')).version || 'unknown'; } catch {} fs.writeFileSync(path.join(DATA_DIR, 'supervisor.json'), JSON.stringify({ pid: process.pid, startedAt: Date.now(), version, port: config.port, parent: process.env.INVOCATION_ID ? 'systemd' : (process.env.XPC_SERVICE_NAME?.includes('bloby') ? 'launchd' : 'other'), }, null, 2)); } catch { /* best-effort — the CLI falls back to launchd/systemd queries */ } } function removeRuntimeFile() { try { const p = path.join(DATA_DIR, 'supervisor.json'); const cur = JSON.parse(fs.readFileSync(p, 'utf-8')); if (cur.pid === process.pid) fs.unlinkSync(p); // never delete a newer instance's file } catch {} } // Track whether an agent is actively processing — file watcher defers restarts during active turns let agentQueryActive = false; let pendingBackendRestart = false; // Set when file watcher fires during agent turn let pendingUpdate = false; // An update is queued; runs at the next turn-complete (flushPendingUpdate) let updateInProgress = false; // The update child has actually been spawned — idempotency guard // Note: with live conversations, agentQueryActive is true while the agent processes a message // and false when it's idle (waiting for next message). The live conversation stays alive between messages. // ── Self-update marker (persists a queued update across a supervisor restart in the request→flush // window) ────────────────────────────────────────────────────────────────────────────────── function readUpdateMarker(): { queuedAt: number; attempts: number } | null { try { const m = JSON.parse(fs.readFileSync(UPDATE_MARKER, 'utf-8')); if (m && typeof m.queuedAt === 'number') return { queuedAt: m.queuedAt, attempts: Number(m.attempts) || 0 }; } catch {} return null; } function writeUpdateMarker(m: { queuedAt: number; attempts: number }): void { try { fs.writeFileSync(UPDATE_MARKER, JSON.stringify(m)); } catch {} } function clearUpdateMarker(): void { try { fs.unlinkSync(UPDATE_MARKER); } catch {} } /** Queue a self-update. Acknowledged + idempotent (the core fix vs the old fire-and-forget * `touch .update`). The update RUNS at the next turn-complete so the agent's current turn finishes * first (it does NOT die mid-turn). When truly idle it flushes on the next tick. */ function queueUpdate(): { queued: boolean; alreadyQueued: boolean; deferred: boolean; retrying: boolean } { if (updateInProgress) return { queued: false, alreadyQueued: true, deferred: false, retrying: false }; const marker = readUpdateMarker(); const alreadyQueued = pendingUpdate; // genuinely already waiting to run const retrying = !pendingUpdate && !!marker && marker.attempts > 0; // a prior attempt failed; re-queue it pendingUpdate = true; if (!marker) writeUpdateMarker({ queuedAt: Date.now(), attempts: 0 }); flushPendingUpdate(); // self-defers — runs now only if nothing is mid-turn return { queued: true, alreadyQueued, deferred: aTurnIsActive(), retrying }; } /** Run the queued update once NO turn is active on any surface. Deferred one tick so the * just-completed turn's in-flight flags (agentQueryActive / conv.busy / activeQueries) have * cleared first: that lets the completing turn's OWN queued update fire, while still never tearing * down a concurrent dashboard / channel / one-shot turn. If something else is still active it * stays pending and re-fires at the next turn-complete or boot-resume (idempotent, marker-backed). */ function flushPendingUpdate(): void { if (!pendingUpdate || updateInProgress) return; setImmediate(() => { if (!pendingUpdate || updateInProgress || aTurnIsActive()) return; pendingUpdate = false; try { for (const cid of Array.from(clientConvs.values())) if (hasConversation(cid)) endConversation(cid); } catch {} runDeferredUpdate(); }); } /** Status for GET /__bloby/control/update-status — lets the agent confirm a queued update actually * ran / read update.log on failure (a successful update ends in process.exit + daemon restart, so * the agent sees a connection drop then a new version on reconnect). */ function getUpdateStatus(): { state: 'idle' | 'queued' | 'running' | 'failed'; attempts: number; logTail: string } { const marker = readUpdateMarker(); let state: 'idle' | 'queued' | 'running' | 'failed'; if (updateInProgress) state = 'running'; else if (pendingUpdate) state = 'queued'; else if (marker && marker.attempts > 0) state = 'failed'; // a prior attempt failed; retries at next turn/boot else if (marker) state = 'queued'; else state = 'idle'; let logTail = ''; try { logTail = fs.readFileSync(path.join(DATA_DIR, 'update.log'), 'utf-8').split('\n').slice(-60).join('\n').trim(); } catch {} return { state, attempts: marker?.attempts ?? 0, logTail }; } // Run bloby update as a child process. BLOBY_SELF_UPDATE=1 tells bin/cli.js to skip daemon // stop/restart — the supervisor exits after the update finishes, and systemd (Restart=on-failure) // or launchd (KeepAlive.SuccessfulExit=false) restarts us with the new code. The marker's attempts // counter bounds retries (the TTL is enforced only on the boot-resume path so it can't strand a // legit update queued early in a >TTL-long turn). function runDeferredUpdate() { if (updateInProgress) { log.info('Update already in progress — skipping duplicate trigger'); return; } const marker = readUpdateMarker() || { queuedAt: Date.now(), attempts: 0 }; if (marker.attempts >= UPDATE_MAX_ATTEMPTS) { log.error(`Self-update failed ${marker.attempts}× — giving up. Run \`bloby update\` manually or check ${path.join(DATA_DIR, 'update.log')}`); clearUpdateMarker(); try { broadcastBloby('backend:failed', { message: 'Self-update failed repeatedly. Ask your human to run `bloby update`.' }); } catch {} return; } updateInProgress = true; writeUpdateMarker({ queuedAt: marker.queuedAt, attempts: marker.attempts + 1 }); // Tell connected chats BEFORE the npm install starts: they flip the header to "Updating…" // and block input, so the imminent WS drop reads as an update, not a mystery "Offline". try { broadcastBloby('agent:updating', {}); } catch {} const cliPath = path.join(PKG_DIR, 'bin', 'cli.js'); const updateLog = path.join(DATA_DIR, 'update.log'); log.info('Deferred update triggered — running bloby update...'); try { const logFd = fs.openSync(updateLog, 'w'); const child = cpSpawn(process.execPath, [cliPath, 'update'], { stdio: ['ignore', logFd, logFd], env: { ...process.env, BLOBY_SELF_UPDATE: '1' }, }); child.on('exit', (code) => { try { fs.closeSync(logFd); } catch {} if (code === 0) { clearUpdateMarker(); // success (updated or already-latest) — don't re-run on the next boot log.ok('Update completed — relaunching daemon onto the new version...'); relaunchSupervisor(); // NO process.exit here: the old reliance on launchd KeepAlive after exit(1) does NOT fire // when the supervisor runs in the foreground or the launchd job isn't loaded (the user hit // exactly this). relaunchSupervisor() actively reloads the daemon (mirroring the battle- // tested manual `bloby update` path); its `launchctl unload` / the new daemon's killPort // terminates THIS old process. If the daemon isn't installed at all (pure foreground dev), // the relaunch no-ops and we stay up on the old code rather than going down unrecoverably. } else { // Leave the marker so the next boot retries (bounded by attempts); allow another flush now. updateInProgress = false; log.error(`Update process exited with code ${code} — see ${updateLog}. Will retry on next restart (attempt ${marker.attempts + 1}/${UPDATE_MAX_ATTEMPTS}).`); } }); child.on('error', (err) => { try { fs.closeSync(logFd); } catch {} updateInProgress = false; log.error(`Update process failed to start: ${err.message}`); }); } catch (err) { updateInProgress = false; log.error(`Deferred update failed: ${err instanceof Error ? err.message : err}`); } } /** Relaunch the supervisor onto freshly-updated code by actively (re)loading the daemon — the same * `launchctl unload/load` (macOS) / `systemctl restart` (Linux) that the battle-tested manual * `bloby update` uses. Spawned DETACHED + unref'd so it OUTLIVES this process when the relaunch's * unload terminates us. `sleep 1` lets the final HTTP ack flush first. If the daemon isn't * installed (foreground dev with no plist), `bloby daemon restart` no-ops — we then stay up on the * current code instead of going down with nothing to bring us back. Replaces the previous reliance * on launchd KeepAlive after process.exit, which never fires in the foreground / when unloaded. */ function relaunchSupervisor(): void { // Under systemd we ARE the unit's main process: a non-zero exit triggers // Restart=on-failure and systemd respawns us onto the new code in ~5s. The // old detached `bloby daemon restart` path is a dead end here — it needs // sudo for systemctl and there is no TTY to ask for a password. if (process.env.INVOCATION_ID) { log.ok('Updated — exiting so systemd restarts the unit onto the new version...'); try { removeRuntimeFile(); } catch {} try { const cfg = loadConfig(); delete cfg.tunnelUrl; saveConfig(cfg); } catch {} setTimeout(() => process.exit(1), 1000); // let the final HTTP ack flush return; } // No service installed (pure foreground dev, no plist/unit): `bloby restart` // now consolidates foreground instances INTO the daemon, so spawning it here // would kill this process and silently install a service nobody asked for. // Stay up on the old code instead — the human restarts when they choose. const plistPath = path.join(os.homedir(), 'Library', 'LaunchAgents', 'com.bloby.bot.plist'); const unitPath = '/etc/systemd/system/bloby.service'; if (!fs.existsSync(process.platform === 'darwin' ? plistPath : unitPath)) { log.warn('Updated, but no background service is installed — restart Bloby manually to load the new version.'); return; } const cliPath = path.join(PKG_DIR, 'bin', 'cli.js'); try { // `bloby daemon restart` boots the launchd job out and bootstraps it again; the // bootout terminates THIS process, so the relauncher must be detached + unref'd // to outlive it. const relauncher = cpSpawn(process.execPath, [cliPath, 'daemon', 'restart'], { detached: true, stdio: 'ignore', env: { ...process.env }, }); relauncher.unref(); } catch (err) { log.error(`Relaunch failed to spawn: ${err instanceof Error ? err.message : err} — run \`bloby restart\` manually.`); } } /** On boot, resume an update that was queued but never ran (supervisor died in the request→flush * window). Safe to auto-run: bin/cli.js update version-checks and no-ops if already latest, and * the marker's TTL + attempts cap prevent a restart loop. */ function resumePendingUpdateOnBoot(): void { const marker = readUpdateMarker(); if (!marker) return; if (Date.now() - marker.queuedAt > UPDATE_MARKER_TTL_MS) { clearUpdateMarker(); return; } log.info('Found a pending update from before restart — resuming...'); pendingUpdate = true; flushPendingUpdate(); // no active turn at boot } // Tell the live chat when the backend gives up — the dashboard interstitial covers page loads, // but an already-open chat client gets an explicit event it can surface ("ask me to fix the backend"). setBackendGiveUpHandler(() => { broadcastBloby('backend:failed', { message: 'The workspace backend crashed and could not restart. Ask Bloby to fix it.' }); }); // Spawn backend (worker runs in-process) spawnBackend(backendPort); // Start pulse/cron scheduler startScheduler({ broadcastBloby, workerApi, restartBackend: () => doRestart(), getModel: () => loadConfig().ai.model, onTurnComplete: () => { if (pendingBackendRestart) void doRestart(); flushPendingUpdate(); }, // flush a deferred backend restart + queued self-update after a pulse/cron turn }); // Initialize channel manager (WhatsApp, Telegram, etc.) const channelManager = new ChannelManager({ broadcastBloby, workerApi, restartBackend: () => doRestart(), getModel: () => loadConfig().ai.model, onTurnComplete: () => { if (pendingBackendRestart) void doRestart(); flushPendingUpdate(); }, // flush a deferred backend restart + queued self-update after a channel turn }); // Broadcast channel status changes to all connected chat clients channelManager.onStatusChange((status) => { broadcastBloby('channel:status', status); // Also broadcast QR code updates if (status.info?.hasQr) { const qr = channelManager.getQrCode(status.channel); if (qr) broadcastBloby('channel:qr', { channel: status.channel, qr }); } }); // Auto-init channels (will connect WhatsApp if previously configured) channelManager.init().catch((err) => { log.warn(`[channels] Init failed: ${err.message}`); }); // Pre-warm the Claude CLI subprocess for the next live conversation so // the first user message doesn't wait on subprocess spawn + init. // Fire-and-forget: failures are logged but don't block boot. const prewarmCfg = loadConfig(); if (prewarmCfg.ai.model) { warmUpForLiveConversation(prewarmCfg.ai.model); } // Watch workspace files for changes — auto-restart backend // Catches edits from VS Code, CLI, or any external tool. // During agent turns, defers to bot:done (avoids mid-turn restarts). const workspaceDir = WORKSPACE_DIR; const backendDir = path.join(workspaceDir, 'backend'); let backendRestartTimer: ReturnType | null = null; /** Single funnel for every DELIBERATE backend restart (file watcher, turn-complete, agent-api * one-shot, scheduler pulse, channel manager). Clears the deferred-restart flag and the * debounce timer, then delegates to backend.ts's serialized + coalescing restartBackend so * concurrent triggers can never double-spawn onto the contended port. */ function doRestart(): Promise { pendingBackendRestart = false; if (backendRestartTimer) { clearTimeout(backendRestartTimer); backendRestartTimer = null; } return restartBackend(backendPort); } /** True while any surface is mid-turn. Dashboard chat sets agentQueryActive; WhatsApp/Alexa live * turns set the harness conv.busy; pulse/cron + customer-WhatsApp ONE-SHOT turns set neither * (they live in the harness activeQueries map) — so we check all three. Otherwise an agent * editing the backend over any of these surfaces, or queuing a self-update from one, would get * the backend restarted / the supervisor exited out from under it mid-turn. */ const aTurnIsActive = () => agentQueryActive || anyConversationBusy() || anyOneShotActive(); function scheduleBackendRestart(reason: string) { if (aTurnIsActive()) { // A turn is working — don't restart now; flush at turn-complete (createSharedChatOnMessage) // or via the channel manager's own post-turn restart. pendingBackendRestart = true; return; } // Skip if a stop/restart is already in progress (that restart owns the spawn). if (isBackendStopping()) return; if (backendRestartTimer) clearTimeout(backendRestartTimer); backendRestartTimer = setTimeout(() => { backendRestartTimer = null; // Re-check at fire time: a turn may have started during the 1s debounce window. if (aTurnIsActive()) { pendingBackendRestart = true; return; } if (isBackendStopping()) return; log.info(`[watcher] ${reason} — restarting backend...`); void doRestart(); }, 1000); } // Self-healing file watchers. Two failure modes the audit flagged, both of which would hurt G3 // (auto-heal) or G1 (chat): (a) fs.watch throws synchronously if its target is missing — at boot // that reached the top-level catch → process.exit before chat listened; (b) a watcher 'error' // event (EMFILE under load, the watched inode removed during a workspace swap) has no listener, // so it crashes the supervisor AND leaves a silently-dead watcher (auto-heal stops with no // signal). Fix: ensure the dir exists, attach an 'error' listener, and re-arm with backoff. let backendWatcher: fs.FSWatcher | null = null; let workspaceWatcher: fs.FSWatcher | null = null; function armBackendWatcher() { try { fs.mkdirSync(backendDir, { recursive: true }); // fs.watch throws if the target is missing const w = fs.watch(backendDir, { recursive: true }, (_event, filename) => { if (!filename || !filename.toString().match(/\.(ts|js|json)$/)) return; scheduleBackendRestart(`Backend file changed: ${filename}`); }); w.on('error', (err: any) => { log.warn(`[watcher] backend watcher error: ${err?.message || err} — re-arming in 2s`); try { w.close(); } catch {} backendWatcher = null; setTimeout(armBackendWatcher, 2000); }); backendWatcher = w; } catch (err: any) { log.warn(`[watcher] backend watcher failed to arm: ${err?.message || err} — retry in 5s`); setTimeout(armBackendWatcher, 5000); } } function onWorkspaceChange(_event: fs.WatchEventType, filename: string | Buffer | null) { if (!filename) return; if (filename === '.env') { scheduleBackendRestart('.env changed'); } if (filename === 'package.json' || filename === 'package-lock.json') { // The agent ran `npm install` to add/fix a backend dependency. Neither watcher otherwise // covers workspace-root deps (backendWatcher only watches backend/; node_modules is huge // and intentionally unwatched). Without this, an install done to fix an ENOENT crash — where // the import already exists so no Write tool fires and usedFileTools stays false — never // restarts the backend, leaving it broken until some unrelated edit. npm install runs inside // the agent's turn, so this defers (like every trigger) and lands at turn-complete, after // the install has fully written package.json + node_modules. scheduleBackendRestart(`workspace dependencies changed (${filename})`); } if (filename === '.restart') { // DEPRECATED fallback — agents now use POST /__bloby/control/restart-backend (synchronous ack, // no lossy fs.watch). Kept so a human/external script touching .restart still works. try { fs.unlinkSync(path.join(workspaceDir, '.restart')); } catch {} scheduleBackendRestart('.restart trigger (deprecated)'); } if (filename === '.update') { // DEPRECATED fallback — agents now use POST /__bloby/control/update (acknowledged + idempotent). // Route through queueUpdate(), which carries every fix the old inline path lacked: the // idempotency guard (the watcher's own unlink re-fires this event → double-spawn), the // aTurnIsActive() gate (was agentQueryActive-only → fired mid-turn on pulse/channel turns), the // persisted marker, and the all-surface turn-complete flush. try { fs.unlinkSync(path.join(workspaceDir, '.update')); } catch {} queueUpdate(); } } function armWorkspaceWatcher() { try { const w = fs.watch(workspaceDir, onWorkspaceChange); w.on('error', (err: any) => { log.warn(`[watcher] workspace watcher error: ${err?.message || err} — re-arming in 2s`); try { w.close(); } catch {} workspaceWatcher = null; setTimeout(armWorkspaceWatcher, 2000); }); workspaceWatcher = w; } catch (err: any) { log.warn(`[watcher] workspace watcher failed to arm: ${err?.message || err} — retry in 5s`); setTimeout(armWorkspaceWatcher, 5000); } } armBackendWatcher(); armWorkspaceWatcher(); // Resume a self-update that was queued but never ran (supervisor died in the request→flush window). resumePendingUpdateOnBoot(); // WebSocket liveness heartbeat — ping the app + chat WS clients every 30s and terminate any // that missed the previous pong (half-open sockets that never fired 'close'). Terminating fires // 'close', which runs the existing map/subscription cleanup. Scoped to our two WSS only (Vite's // HMR socket is separate and managed by Vite). Cleared in shutdown(). const wsHeartbeat = setInterval(() => { for (const wss of [blobyWss, appWss]) { for (const ws of wss.clients) { if ((ws as any).isAlive === false) { try { ws.terminate(); } catch {} continue; } (ws as any).isAlive = false; try { ws.ping(); } catch {} } } }, 30_000); // Tunnel let tunnelUrl: string | null = null; if (config.tunnel.mode === 'quick') { try { tunnelUrl = await startTunnel(config.port); log.ok(`Tunnel: ${tunnelUrl}`); console.log(`__TUNNEL_URL__=${tunnelUrl}`); // Persist tunnel URL so the worker can read it after handle registration config.tunnelUrl = tunnelUrl; saveConfig(config); // Wait for local server to be reachable before telling the relay. // Probes localhost directly — avoids macOS issue where server can't // reach itself through the Cloudflare tunnel URL. let tunnelReady = false; log.info('Readiness probe: waiting for local server...'); for (let i = 0; i < 30; i++) { try { const res = await fetch(`http://127.0.0.1:${config.port}/api/health?_cb=${Date.now()}`, { signal: AbortSignal.timeout(3000), }); log.info(`Readiness probe #${i + 1}: ${res.status}`); if (res.ok) { tunnelReady = true; break; } } catch (err) { log.info(`Readiness probe #${i + 1}: ${err instanceof Error ? err.message : 'error'}`); } await new Promise(r => setTimeout(r, 1000)); } if (!tunnelReady) { log.warn('Local server readiness probe timed out — updating relay anyway'); } // Now register tunnel URL with relay and start heartbeats if (config.relay?.token) { try { await updateTunnelUrl(config.relay.token, tunnelUrl); startHeartbeat(config.relay.token, tunnelUrl); if (config.relay.url) { log.ok(`Relay: ${config.relay.url}`); console.log(`__RELAY_URL__=${config.relay.url}`); } } catch (err) { log.warn(`Relay: ${err instanceof Error ? err.message : err}`); } } console.log('__READY__'); } catch (err) { log.warn(`Tunnel: ${err instanceof Error ? err.message : err}`); console.log('__TUNNEL_FAILED__'); } } if (config.tunnel.mode === 'named') { try { await startNamedTunnel(config.tunnel.configPath!, config.tunnel.name!); tunnelUrl = `https://${config.tunnel.domain}`; log.ok(`Named tunnel: ${tunnelUrl}`); console.log(`__TUNNEL_URL__=${tunnelUrl}`); config.tunnelUrl = tunnelUrl; saveConfig(config); // Readiness probe — check local server (not tunnel URL, avoids macOS self-reach issue) let tunnelReady = false; log.info('Readiness probe: waiting for local server...'); for (let i = 0; i < 30; i++) { try { const res = await fetch(`http://127.0.0.1:${config.port}/api/health?_cb=${Date.now()}`, { signal: AbortSignal.timeout(3000), }); log.info(`Readiness probe #${i + 1}: ${res.status}`); if (res.ok) { tunnelReady = true; break; } } catch (err) { log.info(`Readiness probe #${i + 1}: ${err instanceof Error ? err.message : 'error'}`); } await new Promise(r => setTimeout(r, 1000)); } if (!tunnelReady) { log.warn('Local server readiness probe timed out'); } // Register the (stable) named-tunnel URL with the relay and start heartbeats — // mirrors the quick branch. Without this, a bot that holds a relay handle but runs // a named tunnel is never marked online: its .bloby.bot handle goes stale and // 503s on both page loads and WS, even though the named domain itself works. if (config.relay?.token) { try { await updateTunnelUrl(config.relay.token, tunnelUrl); startHeartbeat(config.relay.token, tunnelUrl); if (config.relay.url) { log.ok(`Relay: ${config.relay.url}`); console.log(`__RELAY_URL__=${config.relay.url}`); } } catch (err) { log.warn(`Relay: ${err instanceof Error ? err.message : err}`); } } console.log('__READY__'); } catch (err) { log.warn(`Named tunnel: ${err instanceof Error ? err.message : err}`); console.log('__TUNNEL_FAILED__'); } } // Tunnel watchdog — detects sleep/wake + periodic health checks let watchdogInterval: ReturnType | null = null; if (tunnelUrl) { let lastTick = Date.now(); let healthCounter = 0; watchdogInterval = setInterval(async () => { const now = Date.now(); const wakeGap = now - lastTick > 60_000; const periodicCheck = ++healthCounter % 10 === 0; lastTick = now; // Update immediately so concurrent ticks don't see a stale gap if (wakeGap || periodicCheck) { // A wake-gap (a tick delayed >60s) means the host was suspended — laptop/Mac slept. // For a QUICK tunnel, cloudflared almost always survives suspension as a live process // while its *.trycloudflare.com edge hostname has already been reclaimed. isTunnelAlive() // only checks process liveness + localhost health (never the public URL), so it returns // a false positive and the bot would stay unreachable with no self-heal. So on a wake-gap // we force a fresh quick-tunnel rotation instead of trusting the local probe. Named // tunnels keep a stable URL and auto-reconnect, so they take the normal liveness path; // the routine periodicCheck also stays on isTunnelAlive (don't churn the URL every 5min). const forceQuickRotation = wakeGap && config.tunnel.mode === 'quick'; const alive = forceQuickRotation ? false : await isTunnelAlive(tunnelUrl!, config.port); if (!alive) { log.warn(forceQuickRotation ? 'Wake detected — rotating quick tunnel (edge hostname likely stale)...' : 'Tunnel dead, restarting...'); try { if (config.tunnel.mode === 'named') { // Named tunnel: restart process, URL doesn't change await restartNamedTunnel(config.tunnel.configPath!, config.tunnel.name!); log.ok(`Named tunnel restored: ${tunnelUrl}`); // Re-assert online with the relay after the restart (URL is stable; the // heartbeat keeps running, but this immediately clears any stale offline flag). const latestCfg = loadConfig(); if (latestCfg.relay?.token) { await updateTunnelUrl(latestCfg.relay.token, tunnelUrl!); } } else { // Quick tunnel: restart and get new URL const newUrl = await restartTunnel(config.port); // Brief pause to let cloudflared establish the tunnel await new Promise(r => setTimeout(r, 3000)); tunnelUrl = newUrl; // Re-read config from disk — relay token may have been added during onboarding const latestCfg = loadConfig(); latestCfg.tunnelUrl = newUrl; saveConfig(latestCfg); if (latestCfg.relay?.token) { stopHeartbeat(); startHeartbeat(latestCfg.relay.token, newUrl); await updateTunnelUrl(latestCfg.relay.token, newUrl); log.ok(`Relay updated with new tunnel URL`); } log.ok(`Tunnel restored: ${newUrl}`); } } catch (err) { log.error(`Tunnel restart failed: ${err instanceof Error ? err.message : err}`); } } } }, 30_000); } // Shutdown let shuttingDown = false; const shutdown = async () => { if (shuttingDown) return; // both SIGINT and SIGTERM (or a repeat) call this shuttingDown = true; // Hard-exit deadline: never let a hung teardown step (e.g. the relay `disconnect` // fetch on a dead network during sleep/wake) block process exit. Without this the // daemon manager SIGKILLs us, orphaning the backend child + cloudflared. .unref() so // the timer itself can never keep the loop alive past a clean, fast shutdown. setTimeout(() => process.exit(0), 5000).unref(); log.info('Shutting down...'); await channelManager.disconnectAll(); stopScheduler(); backendWatcher?.close(); workspaceWatcher?.close(); clearInterval(wsHeartbeat); if (backendRestartTimer) clearTimeout(backendRestartTimer); if (watchdogInterval) clearInterval(watchdogInterval); stopHeartbeat(); const latestConfig = loadConfig(); if (latestConfig.relay?.token) { await disconnect(latestConfig.relay.token); } // Clear persisted tunnel URL so stale values aren't reused delete latestConfig.tunnelUrl; saveConfig(latestConfig); closeDb(); await stopBackend(); stopTunnel(); await stopViteDevServers(); server.close(); removeRuntimeFile(); process.exit(0); }; process.on('SIGINT', () => shutdown()); process.on('SIGTERM', () => shutdown()); return tunnelUrl; } startSupervisor().catch((err) => { log.error('Fatal', err); process.exit(1); });