#!/usr/bin/env node // ────────────────────────────────────────────────────────────────────── // debug-composio-trigger.mjs // // Composio trigger — Socket.IO live listener. // // Rust-side counterpart to // backend-1/src/scripts/live-test-composio-trigger.ts // // Opens a socket.io client against the openhuman backend (same endpoint // the Rust core's SocketManager hits), authenticates with a JWT, and // waits for `composio:trigger` events to land on the socket when the // backend's POST /webhooks/composio receives and HMAC-verifies an // incoming Composio webhook. // // End-to-end path under test: // // [Gmail event] // └─► Composio fires webhook // └─► POST /webhooks/composio (HMAC verified) // └─► handleWebhook.ts → emit('composio:trigger', …) // └─► this script (or the Rust core) receives the event // // Prerequisites: // 1. The backend is reachable at BACKEND_URL. // 2. The backend is publicly addressable at the URL you configured in // Composio's dashboard for the webhook (usually via ngrok for local // dev). If Composio can't POST to /webhooks/composio, no events // will ever land on the socket — no amount of listening will help. // 3. The test user already has an ACTIVE gmail connection — run // `bash scripts/debug-composio-login.sh` first to set one up. // 4. A trigger instance exists for the user. Unlike the backend // script, this one does NOT create the trigger — we don't have // the Composio API key on the client. Create it once via the // backend team's `src/scripts/live-test-composio-trigger.ts` // (with CLEANUP=keep) or via the Composio dashboard, then run // this script as many times as you like. // // Usage: // node scripts/debug-composio-trigger.mjs // node scripts/debug-composio-trigger.mjs --timeout 600 // node scripts/debug-composio-trigger.mjs --debug // node scripts/debug-composio-trigger.mjs --trigger GMAIL_NEW_GMAIL_MESSAGE // node scripts/debug-composio-trigger.mjs --max-events 3 // node scripts/debug-composio-trigger.mjs --send-test # (placeholder — see below) // // Env vars (loaded from .env + app/.env.local): // BACKEND_URL / VITE_BACKEND_URL — backend API base // JWT_TOKEN — bearer JWT (optional, overrides // the `openhuman-core auth get_session_token` // fallback) // TRIGGER_SLUG — override via CLI flag `--trigger` // ────────────────────────────────────────────────────────────────────── import { execSync } from 'child_process'; import { existsSync, readFileSync } from 'fs'; import path from 'path'; import { fileURLToPath, pathToFileURL } from 'url'; const __dirname = path.dirname(fileURLToPath(import.meta.url)); const ROOT = path.resolve(__dirname, '..'); // ── Env loader (matches test-channel-receive.mjs) ─────────────────── // // Declared + invoked BEFORE the CLI constants below read `process.env`, // otherwise values defined in `.env` / `app/.env.local` (like // TRIGGER_SLUG) would be ignored on the first run of the script. function loadEnv(filepath) { if (!existsSync(filepath)) return; const lines = readFileSync(filepath, 'utf-8').split('\n'); for (const line of lines) { const trimmed = line.trim(); if (!trimmed || trimmed.startsWith('#')) continue; const eqIdx = trimmed.indexOf('='); if (eqIdx < 0) continue; const key = trimmed.slice(0, eqIdx).trim(); const val = trimmed.slice(eqIdx + 1).trim(); if (!process.env[key]) process.env[key] = val; } } loadEnv(path.join(ROOT, '.env')); loadEnv(path.join(ROOT, 'app', '.env.local')); // ── CLI argument parsing ──────────────────────────────────────────── const args = process.argv.slice(2); const flag = (name) => args.includes(name); const valueOf = (name, fallback) => { const idx = args.indexOf(name); if (idx === -1 || idx === args.length - 1) return fallback; return args[idx + 1]; }; const DEBUG = flag('--debug'); const TIMEOUT_SECS = parseInt(valueOf('--timeout', '0'), 10); // 0 = forever const MAX_EVENTS = parseInt(valueOf('--max-events', '0'), 10); // 0 = unlimited const TRIGGER_SLUG = (valueOf('--trigger', process.env.TRIGGER_SLUG || 'GMAIL_NEW_GMAIL_MESSAGE')).trim(); const TOOLKIT_OVERRIDE = (valueOf('--toolkit', process.env.COMPOSIO_TOOLKIT || '')).trim(); function dbg(...a) { if (DEBUG) console.log(' [debug]', ...a); } const BACKEND_URL = ( process.env.BACKEND_URL || process.env.VITE_BACKEND_URL || 'https://staging-api.alphahuman.xyz' ).replace(/\/+$/, ''); // ── Pretty-print helpers ──────────────────────────────────────────── const C = { reset: '\x1b[0m', bold: '\x1b[1m', dim: '\x1b[2m', green: '\x1b[32m', red: '\x1b[31m', yellow: '\x1b[33m', cyan: '\x1b[36m', blue: '\x1b[34m', magenta: '\x1b[35m', }; function header(text) { console.log(`\n${C.cyan}${'─'.repeat(60)}${C.reset}`); console.log(`${C.cyan} ${text}${C.reset}`); console.log(`${C.cyan}${'─'.repeat(60)}${C.reset}`); } function ok(detail) { console.log(`${C.green} ✓${C.reset}${detail ? ` ${C.dim}${detail}${C.reset}` : ''}`); } function fail(detail) { console.log(`${C.red} ✗ ${detail}${C.reset}`); } function info(label, value) { console.log(`${C.dim} ${label}: ${C.reset}${value}`); } function ts() { return new Date().toISOString().replace('T', ' ').slice(0, 19); } // ── Banner ────────────────────────────────────────────────────────── console.log(''); console.log(`${C.bold}📡 Composio Trigger — Socket.IO Live Listener${C.reset}`); console.log(''); info('Backend', BACKEND_URL); info('Trigger', TRIGGER_SLUG); info('Timeout', TIMEOUT_SECS > 0 ? `${TIMEOUT_SECS}s` : 'forever (Ctrl+C to stop)'); info('Max events', MAX_EVENTS > 0 ? MAX_EVENTS : 'unlimited'); info('Debug', DEBUG); // ── Resolve JWT ───────────────────────────────────────────────────── header('1. Authentication'); function getSessionTokenFromCore() { const coreBin = path.join(ROOT, 'target', 'debug', 'openhuman-core'); if (!existsSync(coreBin)) { if (DEBUG) console.debug(`[debug] core binary not found at ${coreBin}`); return null; } try { const output = execSync(`"${coreBin}" auth get_session_token`, { cwd: ROOT, encoding: 'utf-8', timeout: 10_000, stdio: ['pipe', 'pipe', 'pipe'], }); const match = output.match(/"token":\s*"([^"]+)"/); return match?.[1] || null; } catch (err) { if (DEBUG) { const msg = err instanceof Error ? err.message : String(err); console.debug(`[debug] ${coreBin} auth get_session_token failed: ${msg}`); } return null; } } let token = process.env.JWT_TOKEN?.trim() || null; if (token) { info('Source', 'JWT_TOKEN env'); } else { token = getSessionTokenFromCore(); if (token) info('Source', 'target/debug/openhuman-core auth get_session_token'); } if (!token) { fail('No JWT_TOKEN in env and no session token available from the core binary.'); console.log( `${C.dim} Set JWT_TOKEN in .env, or run the app once to log in so the core has a session.${C.reset}`, ); process.exit(1); } ok(`token=${token.slice(0, 20)}…`); // ── Verify JWT against /auth/me ───────────────────────────────────── header('2. Verify Token'); // We hold onto these so we can print them alongside every dropped-event // diagnostic below. If the trigger was registered under a different // user, the ids printed here will NOT match whatever the backend's // `getSocketsByUserId(verified.payload.userId)` uses, and the emit is // silently dropped. let authedUserId = null; let authedUserLabel = '(unknown)'; try { const resp = await fetch(`${BACKEND_URL}/auth/me`, { headers: { Authorization: `Bearer ${token}` }, }); const data = await resp.json(); if (resp.ok && data.success && data.data) { const u = data.data; authedUserId = u._id || u.id || null; authedUserLabel = u.username || u.firstName || authedUserId || '(unknown)'; ok(`user=${authedUserLabel}`); // Print the exact socket-map key the backend will use for this // connection. Compare against the trigger's registered userId if // you're seeing "backend captured, socket didn't". info('mongo _id', authedUserId ?? '(missing from /auth/me)'); if (u.telegramId != null) info('telegramId', u.telegramId); } else { fail(`token invalid: ${JSON.stringify(data).slice(0, 300)}`); process.exit(1); } } catch (err) { fail(`backend unreachable: ${err.message}`); process.exit(1); } // ── Verify target toolkit has a connection ───────────────────────── // // The "target toolkit" is explicit if the caller passed `--toolkit`; // otherwise we derive it from the trigger slug prefix (e.g. // GMAIL_NEW_GMAIL_MESSAGE → "gmail"). We no longer hard-exit for any // toolkit other than gmail — missing non-gmail connections drop to a // warning so a broader socket listener can keep running. const desiredToolkit = ( TOOLKIT_OVERRIDE || (TRIGGER_SLUG.includes('_') ? TRIGGER_SLUG.split('_')[0] : '') || 'gmail' ) .toLowerCase() .trim(); header(`3. Verify ${desiredToolkit || 'target'} connection`); try { const resp = await fetch(`${BACKEND_URL}/agent-integrations/composio/connections`, { headers: { Authorization: `Bearer ${token}` }, }); const data = await resp.json(); if (!resp.ok || !data.success) { fail(`GET /connections → HTTP ${resp.status}: ${data?.error || 'unknown error'}`); process.exit(1); } const connections = data.data?.connections ?? []; const match = connections.find( (c) => (c.toolkit || '').toLowerCase() === desiredToolkit, ); if (!match) { // Only gmail is considered a blocking prerequisite — the legacy // script behaviour — because `debug-composio-login.sh` defaults to // connecting gmail. For every other toolkit, warn and keep going // so the listener can still receive events from whatever IS // connected. if (desiredToolkit === 'gmail') { fail('no gmail connection found for this user'); console.log( `${C.yellow} Run: bash scripts/debug-composio-login.sh first to connect gmail.${C.reset}`, ); process.exit(1); } console.log( `${C.yellow} ⚠ no ${desiredToolkit} connection found — listener will still run,\n` + ` but no ${TRIGGER_SLUG} events will arrive until one is connected.${C.reset}`, ); } else { const status = String(match.status).toUpperCase(); if (status === 'ACTIVE' || status === 'CONNECTED') { ok(`${desiredToolkit} ACTIVE (id=${match.id})`); } else { console.log( `${C.yellow} ⚠ ${desiredToolkit} connection status is "${match.status}" — trigger may not fire.${C.reset}`, ); info('connection id', match.id); } } } catch (err) { fail(`connection check failed: ${err.message}`); process.exit(1); } // ── Trigger-create reminder ───────────────────────────────────────── // // We deliberately don't create the Composio trigger from this script: // the Composio SDK needs COMPOSIO_API_KEY, which is a backend secret. // The backend team's live-test-composio-trigger.ts already handles // trigger creation — run it once with CLEANUP=keep and then this // listener will see every fired event until you explicitly delete // the trigger. header('4. Trigger Setup Reminder'); console.log( `${C.dim} This script only LISTENS. It does not create/delete${C.reset}\n` + `${C.dim} Composio triggers (that requires the backend's COMPOSIO_API_KEY).${C.reset}\n\n` + `${C.dim} To create the trigger once, run the backend team's tool:${C.reset}\n` + `${C.dim} CLEANUP=keep TRIGGER_SLUG=${TRIGGER_SLUG} \\${C.reset}\n` + `${C.dim} npx ts-node -r tsconfig-paths/register \\${C.reset}\n` + `${C.dim} src/scripts/live-test-composio-trigger.ts${C.reset}\n\n` + `${C.dim} Then run THIS script from the openhuman-2 repo to verify${C.reset}\n` + `${C.dim} that trigger events reach the client socket.${C.reset}`, ); // ── Load socket.io-client ─────────────────────────────────────────── header('5. Connect Socket.IO'); const socketIoPath = path.join( ROOT, 'app', 'node_modules', 'socket.io-client', 'dist', 'socket.io.esm.min.js', ); let io; try { // Prefer the version co-located with the app workspace — same binary // the React client uses at runtime. Convert the filesystem path to a // `file://` URL via `pathToFileURL` so Node ESM accepts it on Windows // (bare OS paths work on macOS/Linux but fail on Windows). if (existsSync(socketIoPath)) { const mod = await import(pathToFileURL(socketIoPath).href); io = mod.io || mod.default; dbg('loaded socket.io-client from', socketIoPath); } else { const mod = await import('socket.io-client'); io = mod.io || mod.default; dbg('loaded socket.io-client from node resolution'); } } catch (err) { fail(`cannot load socket.io-client: ${err.message}`); console.log( `${C.dim} Try: cd app && yarn install${C.reset}`, ); process.exit(1); } const socket = io(BACKEND_URL, { auth: { token }, transports: ['websocket', 'polling'], path: '/socket.io/', reconnection: true, reconnectionAttempts: 5, timeout: 10_000, }); // ── Catchall: log every event the server sends us ───────────────── // // This is on regardless of --debug because the #1 reason "the backend // logged a webhook but my socket didn't" is that the emit targeted a // different userId. If this catchall is silent during the wait, your // socket is NOT receiving any server traffic at all — either because // the server thinks the socket is dead, or because your auth maps to // a user the backend isn't emitting to. // // `onAny` is the socket.io v4 blessed API for catchall listeners. // Normal named `.on('composio:trigger', …)` handlers still fire after // this runs. socket.onAny((eventName, ...payload) => { const first = payload[0]; const preview = first === undefined ? '' : ` ${JSON.stringify(first).slice(0, 160)}`; console.log( `${C.dim} [${ts()}] ⇢ ${eventName}${preview}${C.reset}`, ); }); // Low-frequency heartbeat so silent disconnects are obvious. If the // transport dies without firing `disconnect` on the client, this will // let us notice by the absence of any traffic over 30s. let lastEventAt = Date.now(); socket.onAny(() => { lastEventAt = Date.now(); }); const heartbeatTimer = setInterval(() => { const idleSecs = Math.round((Date.now() - lastEventAt) / 1000); if (idleSecs >= 30 && socket.connected) { console.log( `${C.dim} [${ts()}] idle ${idleSecs}s (socket.connected=${socket.connected}, transport=${socket.io.engine?.transport?.name})${C.reset}`, ); } }, 15_000); // ── Connection lifecycle ──────────────────────────────────────────── let eventCount = 0; socket.on('connect', () => { ok(`connected (socket.id=${socket.id})`); dbg('transport:', socket.io.engine?.transport?.name); }); socket.on('ready', () => { console.log(`${C.green} 🟢 server ready — auth accepted${C.reset}`); header('6. Listening for composio:trigger'); console.log( `${C.magenta} Take the action that fires ${TRIGGER_SLUG}${C.reset}`, ); console.log( `${C.dim} (e.g. for GMAIL_NEW_GMAIL_MESSAGE, send yourself an email).${C.reset}`, ); console.log(`${C.dim} Press Ctrl+C to stop.${C.reset}\n`); }); socket.on('connect_error', (err) => { fail(`connect_error: ${err.message}`); dbg('full error:', err); }); socket.on('error', (data) => { fail(`server error: ${JSON.stringify(data)}`); }); socket.on('disconnect', (reason) => { console.log(`${C.yellow} 🔴 disconnected: ${reason}${C.reset}`); }); socket.io.on('reconnect', (attempt) => { console.log(`${C.yellow} 🔄 reconnected after ${attempt} attempt(s)${C.reset}`); }); // ── The event under test ──────────────────────────────────────────── function printEvent(event) { eventCount++; const label = event?.trigger || '(unknown)'; const toolkit = event?.toolkit || '(unknown)'; console.log( `${C.green} [${ts()}] #${eventCount} composio:trigger${C.reset} ` + `${C.bold}${label}${C.reset} ${C.dim}(${toolkit})${C.reset}`, ); if (event?.metadata) { info('metadata', JSON.stringify(event.metadata)); } const payload = event?.payload ?? {}; const preview = JSON.stringify(payload, null, 2) || '{}'; const lines = preview.split('\n'); const limited = lines.slice(0, 20).join('\n'); console.log( `${C.dim} payload:\n${limited .split('\n') .map((l) => ` ${l}`) .join('\n')}${C.reset}`, ); if (lines.length > 20) { console.log(`${C.dim} …(${lines.length - 20} more lines)${C.reset}`); } console.log(''); } const done = new Promise((resolve) => { socket.on('composio:trigger', (event) => { try { // Filter on slug when the caller passed --trigger so you can // keep a broad listener running while debugging one hook at a // time. `event.trigger` is the slug emitted by the backend. if ( TRIGGER_SLUG && event?.trigger && event.trigger.toUpperCase() !== TRIGGER_SLUG.toUpperCase() ) { dbg(`ignoring ${event.trigger} (filter=${TRIGGER_SLUG})`); } else { printEvent(event); } } catch (err) { fail(`error printing event: ${err.message}`); } if (MAX_EVENTS > 0 && eventCount >= MAX_EVENTS) { console.log( `${C.yellow} Reached --max-events=${MAX_EVENTS} — shutting down.${C.reset}`, ); resolve(); } }); // Optional hard timeout. if (TIMEOUT_SECS > 0) { setTimeout(() => { console.log( `${C.yellow} --timeout=${TIMEOUT_SECS}s elapsed — shutting down.${C.reset}`, ); resolve(); }, TIMEOUT_SECS * 1_000); } // Ctrl+C. process.on('SIGINT', () => { console.log(`\n${C.yellow} SIGINT received — shutting down.${C.reset}`); resolve(); }); }); await done; // ── Cleanup ───────────────────────────────────────────────────────── header('7. Cleanup'); clearInterval(heartbeatTimer); try { socket.close(); ok('socket closed'); } catch { fail('failed to close socket'); } if (eventCount === 0) { // Surface the three usual suspects. This is the script talking, not // the backend — the backend drops mismatched emits silently, so it // won't tell us which one we're hitting. console.log(''); console.log( `${C.yellow}${C.bold} No composio:trigger events received.${C.reset}`, ); console.log(''); console.log(`${C.dim} Checklist:${C.reset}`); console.log( `${C.dim} 1. Trigger userId vs your user:${C.reset}\n` + `${C.dim} Your socket is authed as _id=${C.reset}${C.bold}${authedUserId ?? '(unknown)'}${C.reset}${C.dim}.${C.reset}\n` + `${C.dim} The trigger must have been CREATED with that exact userId on${C.reset}\n` + `${C.dim} the Composio side. If you used the backend's${C.reset}\n` + `${C.dim} live-test-composio-trigger.ts, it registers with user.id${C.reset}\n` + `${C.dim} from whichever TELEGRAM_ID/USER_ID/JWT_TOKEN you ran it with.${C.reset}\n` + `${C.dim} If those don't match, the backend's getSocketsByUserId${C.reset}\n` + `${C.dim} returns [] and the emit is dropped silently.${C.reset}`, ); console.log(''); console.log( `${C.dim} 2. Did the catchall above log ANY inbound events?${C.reset}\n` + `${C.dim} - If yes (e.g. "ready", "toast", heartbeat), your socket is${C.reset}\n` + `${C.dim} alive and the issue is targeting (#1).${C.reset}\n` + `${C.dim} - If no, the backend thinks your socket is gone. Check${C.reset}\n` + `${C.dim} whether another client (the Rust core running locally?)${C.reset}\n` + `${C.dim} is holding the slot, or that your JWT hasn't expired${C.reset}\n` + `${C.dim} mid-run.${C.reset}`, ); console.log(''); console.log( `${C.dim} 3. Add one-line logging to the backend's${C.reset}\n` + `${C.dim} src/controllers/agentIntegrations/composio/handleWebhook.ts${C.reset}\n` + `${C.dim} right after the getSocketsByUserId call so the backend tells${C.reset}\n` + `${C.dim} you how many sockets it matched for which userId. That is${C.reset}\n` + `${C.dim} the only truly authoritative signal — everything else is a${C.reset}\n` + `${C.dim} guess.${C.reset}`, ); console.log(''); process.exit(2); } console.log( `\n${C.green}${C.bold} Received ${eventCount} composio:trigger event(s).${C.reset}\n`, ); process.exit(0);