From 8aa5547af43457daa173e192d3f047c68abd8c9c Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Sun, 5 Apr 2026 22:08:17 -0700 Subject: [PATCH] feat: Telegram managed login, channel messaging, and inbound agent loop (#338) * feat: add managed Telegram login flow and API endpoints - Introduced a new managed Telegram login flow, allowing users to link their Telegram accounts via a deep link. - Added `telegram_login_start` and `telegram_login_check` functions to handle the creation of link tokens and status checks for user linking. - Updated the API with new endpoints for managing Telegram login, including detailed response structures for link token creation and status verification. - Enhanced the `.env.example` file to include a configuration option for the Telegram bot username, facilitating easier setup for developers. * refactor: update Telegram integration to use core RPC for login flow - Replaced the managed DM API with core RPC calls for initiating and checking Telegram login status. - Introduced new API methods `telegramLoginStart` and `telegramLoginCheck` to handle link token creation and verification. - Updated the TelegramConfig and MessagingPanel components to utilize the new login flow, enhancing the user experience and simplifying the codebase. - Adjusted tests to reflect changes in the login process and ensure proper functionality. * docs: update TODO list with new user interaction registration task for memory skill * feat: add ChannelSetupModal for configuring channel integrations - Introduced a new reusable modal component, ChannelSetupModal, for configuring channel integrations such as Telegram and Discord. - The modal can be opened from the Skills page or Settings, enhancing user experience by providing a centralized configuration interface. - Updated MessagingPanel and Skills components to integrate the new modal, allowing users to easily manage their channel settings. - Implemented channel-specific configuration components for better modularity and maintainability. * feat: add ChannelSetupModal for configuring channel integrations - Introduced a new reusable modal component, ChannelSetupModal, for configuring channel integrations such as Telegram and Discord. - The modal can be opened from the Skills page or Settings, enhancing user experience by providing a centralized configuration interface. - Updated MessagingPanel and Skills components to integrate the new modal, allowing users to easily manage their channel settings. - Implemented channel-specific configuration components for better modularity and maintainability. * style: update channel components for improved UI consistency - Refactored styles across various channel configuration components, including ChannelCapabilities, ChannelConfigPanel, ChannelFieldInput, ChannelSelector, DiscordConfig, TelegramConfig, and WebChannelConfig. - Enhanced color schemes and layout for better readability and visual appeal, ensuring a cohesive design across the application. - Updated status badge styles in definitions.ts to align with the new design standards. - Improved error message visibility and loading indicators in Channels page for a more user-friendly experience. * refactor: update WebChannelConfig and tests for improved clarity and functionality - Renamed the `definition` prop in WebChannelConfig to `_definition` for clarity. - Updated test cases in DiscordConfig and TelegramConfig to reflect changes in auth mode labels, ensuring accurate rendering of UI elements. - Adjusted the TelegramConfig test to fix the click event on the connect button, enhancing test reliability. - Modified channel definitions to streamline the managed DM auth mode, improving code organization and maintainability. * fix: enhance WebGL error handling in MeshGradient and RotatingTetrahedronCanvas components - Added error handling in the MeshGradient component to gracefully catch WebGL initialization failures, ensuring the app remains functional when the GPU context is unavailable. - Updated the RotatingTetrahedronCanvas component to verify WebGL context availability before initializing the renderer, preventing crashes and improving user experience. - Modified channel definitions to update auth mode labels for clarity and consistency across the application. * style: update RotatingTetrahedronCanvas colors and animation speed - Changed fill and edge colors in the RotatingTetrahedronCanvas component for improved visual appeal. - Adjusted opacity of the fill material to enhance transparency effects. - Introduced a speed multiplier for rotation animations, allowing for more dynamic visual effects. * refactor: improve WebGL error handling and cleanup in MeshGradient component - Removed unused canvas reference and WebGL context probing to streamline the MeshGradient component. - Enhanced error handling during shader compilation to throw an error when WebGL context may be lost, improving robustness. - Wrapped initialization logic in a try-catch block to gracefully handle failures, ensuring the gradient functionality is disabled if initialization fails. * refactor: simplify Telegram login flow by removing unused link token check - Removed the `check_channel_link_status` method from the `BackendOAuthClient`, streamlining the authentication process. - Updated `telegram_login_check` to directly fetch the user profile and determine link status based on the presence of `telegramId`, enhancing clarity and reducing complexity. - Adjusted logging to reflect the new flow and ensure accurate debugging information. * feat: enhance JSON-RPC method invocation with session expiration handling - Added logic to automatically clear stored session on receiving a 401 error from the backend, improving user experience by ensuring users are redirected to the login page when their session expires. - Introduced a helper function to identify session expiration errors, enhancing code clarity and maintainability. - Refactored the `invoke_method` function to incorporate this new error handling mechanism, ensuring robust session management during RPC calls. * style: update app-dotted-canvas background gradient for improved visibility - Changed the radial gradient background color in the app-dotted-canvas from rgba(0, 0, 0, 0.2) to rgba(0, 0, 0, 0.5), enhancing the contrast and overall visual appeal of the component. * feat: implement channel messaging API with end-to-end testing script - Added a new script `test-channel-messaging.sh` for end-to-end testing of message sending to Telegram via the backend API. - Implemented new backend API methods for sending messages, reactions, creating threads, updating threads, and listing threads in channels. - Enhanced the `BackendOAuthClient` with methods to handle channel messaging operations. - Updated controller schemas and handlers to support the new messaging functionalities, ensuring robust interaction with the backend. - Improved logging for better debugging and tracking of channel operations. * feat: add test-channel-receive script for real-time channel message listening - Introduced a new script `test-channel-receive.mjs` that connects to the backend Socket.IO server and listens for incoming channel messages. - Implemented session token retrieval and validation against the backend, ensuring secure message handling. - Added options for debugging and sending test messages, enhancing the script's usability for testing purposes. - Improved logging for better visibility of connection status and incoming messages. * feat: implement inbound channel message handling for real-time responses - Added functionality to handle inbound messages from channels (e.g., Telegram, Discord) by introducing the `handle_channel_inbound_message` function. - Implemented agent inference loop to process messages and send replies back to the originating channel via the REST API. - Enhanced logging for better tracking of message reception and response handling. - Integrated timeout handling to manage long-running requests effectively. * style: fix prettier formatting for channel components Co-Authored-By: Claude Opus 4.6 (1M context) --------- Co-authored-by: Claude Opus 4.6 (1M context) --- .env.example | 6 + app/src/components/MeshGradient.tsx | 26 +- .../components/RotatingTetrahedronCanvas.tsx | 62 ++- .../channels/ChannelCapabilities.tsx | 2 +- .../channels/ChannelConfigPanel.tsx | 12 +- .../components/channels/ChannelFieldInput.tsx | 6 +- .../components/channels/ChannelSelector.tsx | 10 +- .../components/channels/ChannelSetupModal.tsx | 120 +++++ app/src/components/channels/DiscordConfig.tsx | 113 +++-- .../components/channels/TelegramConfig.tsx | 192 ++++---- .../components/channels/WebChannelConfig.tsx | 12 +- .../channels/__tests__/DiscordConfig.test.tsx | 4 +- .../__tests__/TelegramConfig.test.tsx | 50 +- app/src/components/oauth/providerConfigs.tsx | 6 +- .../settings/panels/MessagingPanel.tsx | 426 +++++------------- app/src/index.css | 2 +- app/src/lib/channels/definitions.ts | 26 +- app/src/lib/meshGradient.js | 51 ++- app/src/pages/Channels.tsx | 4 +- app/src/pages/Skills.tsx | 145 +++++- app/src/pages/__tests__/Channels.test.tsx | 2 +- app/src/services/api/channelConnectionsApi.ts | 29 ++ docs/TODO.md | 1 + scripts/test-channel-messaging.sh | 141 ++++++ scripts/test-channel-receive.mjs | 267 +++++++++++ src/api/rest.rs | 107 +++++ src/core/jsonrpc.rs | 32 ++ .../channels/controllers/definitions.rs | 12 +- src/openhuman/channels/controllers/ops.rs | 324 +++++++++++++ src/openhuman/channels/controllers/schemas.rs | 227 ++++++++++ src/openhuman/skills/socket_manager.rs | 174 +++++++ 31 files changed, 2014 insertions(+), 577 deletions(-) create mode 100644 app/src/components/channels/ChannelSetupModal.tsx create mode 100755 scripts/test-channel-messaging.sh create mode 100755 scripts/test-channel-receive.mjs diff --git a/.env.example b/.env.example index db828a2c1..4ea08a9be 100644 --- a/.env.example +++ b/.env.example @@ -118,6 +118,12 @@ PIPER_BIN= # [optional] Override path to ollama binary OLLAMA_BIN= +# --------------------------------------------------------------------------- +# Telegram managed login +# --------------------------------------------------------------------------- +# [optional] Bot username for managed Telegram DM linking (default: openhuman_bot) +OPENHUMAN_TELEGRAM_BOT_USERNAME=openhuman_bot + # --------------------------------------------------------------------------- # Skills # --------------------------------------------------------------------------- diff --git a/app/src/components/MeshGradient.tsx b/app/src/components/MeshGradient.tsx index db380c278..d2f47a951 100644 --- a/app/src/components/MeshGradient.tsx +++ b/app/src/components/MeshGradient.tsx @@ -5,23 +5,31 @@ import { Gradient } from '../lib/meshGradient'; /** * Animated WebGL mesh gradient background (Stripe-style). * Renders behind the dotted-canvas overlay so dots remain visible on top. + * Catches WebGL errors gracefully so the app still works when the GPU context + * is unavailable or lost (e.g. Tauri WebView on some platforms). */ export default function MeshGradient() { const canvasRef = useRef(null); useEffect(() => { - const canvas = canvasRef.current; - const gradient = new Gradient(); - gradient.initGradient('#mesh-gradient'); + let gradient: InstanceType | null = null; + + try { + gradient = new Gradient(); + gradient.initGradient('#mesh-gradient'); + } catch (err) { + console.warn('[MeshGradient] WebGL init failed, gradient disabled:', err); + gradient = null; + } return () => { - gradient.disconnect(); - gradient.pause(); - if (canvas) { - const gl = canvas.getContext('webgl') || canvas.getContext('webgl2'); - if (gl) { - gl.getExtension('WEBGL_lose_context')?.loseContext(); + try { + if (gradient) { + gradient.disconnect(); + gradient.pause(); } + } catch { + // Cleanup is best-effort. } }; }, []); diff --git a/app/src/components/RotatingTetrahedronCanvas.tsx b/app/src/components/RotatingTetrahedronCanvas.tsx index 11c5285e4..bb5a4f67e 100644 --- a/app/src/components/RotatingTetrahedronCanvas.tsx +++ b/app/src/components/RotatingTetrahedronCanvas.tsx @@ -1,7 +1,7 @@ 'use client'; import * as THREE from 'three'; -import { useEffect, useRef } from 'react'; +import { useEffect, useRef, useState } from 'react'; import { ConvexGeometry } from 'three/addons/geometries/ConvexGeometry.js'; /** Start from a regular tetrahedron and lightly truncate each corner to create small blunted edges. */ @@ -27,17 +27,44 @@ function bluntedTetrahedronPoints(scale: number, bluntness = 0.12): THREE.Vector export default function RotatingTetrahedronCanvas() { const canvasRef = useRef(null); + const [webglFailed, setWebglFailed] = useState(false); useEffect(() => { const canvas = canvasRef.current; if (!canvas) return; - const renderer = new THREE.WebGLRenderer({ - canvas, - antialias: true, - alpha: true, - powerPreference: 'high-performance', - }); + // Verify a WebGL context can be obtained before handing the canvas to + // Three.js. `THREE.WebGLRenderer` internally calls `gl.createShader()` + // which throws if the context is null (e.g. when another canvas already + // consumed the platform's WebGL context limit). + const testCtx = + canvas.getContext('webgl2', { antialias: true }) || + canvas.getContext('webgl', { antialias: true }); + if (!testCtx) { + console.warn('[RotatingTetrahedronCanvas] WebGL context unavailable — skipping'); + setWebglFailed(true); + return; + } + + // Lose the test context so Three.js can create its own on the same canvas. + // getContext returns the same context when called with the same type, so + // Three.js will reuse it. We just needed the null-check above. + + let renderer: THREE.WebGLRenderer; + try { + renderer = new THREE.WebGLRenderer({ + canvas, + context: testCtx as WebGLRenderingContext, + antialias: true, + alpha: true, + powerPreference: 'high-performance', + }); + } catch (err) { + console.warn('[RotatingTetrahedronCanvas] WebGLRenderer init failed:', err); + setWebglFailed(true); + return; + } + renderer.setPixelRatio(Math.min(window.devicePixelRatio, 2)); renderer.outputColorSpace = THREE.SRGBColorSpace; renderer.toneMapping = THREE.NoToneMapping; @@ -48,16 +75,16 @@ export default function RotatingTetrahedronCanvas() { const geometry = new ConvexGeometry(bluntedTetrahedronPoints(0.98, 0.11)); const fillMaterial = new THREE.MeshLambertMaterial({ - color: '#b8e986', + color: '#8e86e9', transparent: true, - opacity: 0.4, + opacity: 0.2, emissive: '#0c1208', - emissiveIntensity: 0.08, + emissiveIntensity: 1, }); const fillMesh = new THREE.Mesh(geometry, fillMaterial); const edgeGeometry = new THREE.EdgesGeometry(geometry); - const edgeMaterial = new THREE.LineBasicMaterial({ color: '#b8e986' }); + const edgeMaterial = new THREE.LineBasicMaterial({ color: '#868ee9' }); const edges = new THREE.LineSegments(edgeGeometry, edgeMaterial); fillMesh.rotation.x = 0.35; @@ -91,11 +118,12 @@ export default function RotatingTetrahedronCanvas() { if (canvas.parentElement) observer.observe(canvas.parentElement); resize(); + const speed = 2; const animate = () => { - fillMesh.rotation.y += 0.0038; - fillMesh.rotation.x += 0.0002; - edges.rotation.y += 0.0038; - edges.rotation.x += 0.0002; + fillMesh.rotation.y += 0.0038 * speed; + fillMesh.rotation.x += 0.0002 * speed; + edges.rotation.y += 0.0038 * speed; + edges.rotation.x += 0.0002 * speed; renderer.render(scene, camera); animationFrame = window.requestAnimationFrame(animate); @@ -114,6 +142,10 @@ export default function RotatingTetrahedronCanvas() { }; }, []); + if (webglFailed) { + return null; + } + return ( { {capabilities.map(cap => ( + className="px-1.5 py-0.5 text-[10px] rounded bg-stone-100 text-stone-500 border border-stone-200"> {cap.replace(/_/g, ' ')} ))} diff --git a/app/src/components/channels/ChannelConfigPanel.tsx b/app/src/components/channels/ChannelConfigPanel.tsx index f57516cf3..5f2425aef 100644 --- a/app/src/components/channels/ChannelConfigPanel.tsx +++ b/app/src/components/channels/ChannelConfigPanel.tsx @@ -15,9 +15,15 @@ const ChannelConfigPanel = ({ selectedChannel, definitions }: ChannelConfigPanel return (
- {selectedChannel === 'telegram' && } - {selectedChannel === 'discord' && } - {selectedChannel === 'web' && } +
+
+

{definition.display_name}

+

{definition.description}

+
+ {selectedChannel === 'telegram' && } + {selectedChannel === 'discord' && } + {selectedChannel === 'web' && } +
diff --git a/app/src/components/channels/ChannelFieldInput.tsx b/app/src/components/channels/ChannelFieldInput.tsx index 15a1d3215..269f739c3 100644 --- a/app/src/components/channels/ChannelFieldInput.tsx +++ b/app/src/components/channels/ChannelFieldInput.tsx @@ -10,9 +10,9 @@ interface ChannelFieldInputProps { const ChannelFieldInput = ({ field, value, onChange, disabled }: ChannelFieldInputProps) => { return (
-
); diff --git a/app/src/components/channels/ChannelSelector.tsx b/app/src/components/channels/ChannelSelector.tsx index 9746ec3d0..24726d193 100644 --- a/app/src/components/channels/ChannelSelector.tsx +++ b/app/src/components/channels/ChannelSelector.tsx @@ -31,11 +31,11 @@ const ChannelSelector = ({ }, [channelConnections]); return ( -
+
-

Channels

+

Channels

- Active route: {activeRoute} + Active route: {activeRoute}

@@ -59,8 +59,8 @@ const ChannelSelector = ({ onClick={() => onSelectChannel(channelId)} className={`flex-1 flex items-center justify-between gap-2 rounded-lg border px-4 py-3 text-sm transition-colors ${ isSelected - ? 'border-primary-500/60 bg-primary-500/20 text-primary-200' - : 'border-stone-700 bg-stone-900/30 text-stone-300 hover:border-stone-500' + ? 'border-primary-500/60 bg-primary-50 text-primary-600' + : 'border-stone-200 bg-stone-50 text-stone-600 hover:border-stone-300' }`}> {CHANNEL_ICONS[def.icon] ?? ''} diff --git a/app/src/components/channels/ChannelSetupModal.tsx b/app/src/components/channels/ChannelSetupModal.tsx new file mode 100644 index 000000000..8c34565e9 --- /dev/null +++ b/app/src/components/channels/ChannelSetupModal.tsx @@ -0,0 +1,120 @@ +/** + * Reusable modal for configuring a channel integration (Telegram, Discord, etc.). + * Uses createPortal like SkillSetupModal. Can be opened from the Skills page or Settings. + */ +import { useEffect, useRef } from 'react'; +import { createPortal } from 'react-dom'; + +import type { ChannelDefinition, ChannelType } from '../../types/channels'; +import DiscordConfig from './DiscordConfig'; +import TelegramConfig from './TelegramConfig'; + +const CHANNEL_ICONS: Record = { + telegram: '\u2708\uFE0F', + discord: '\uD83C\uDFAE', + web: '\uD83C\uDF10', +}; + +interface ChannelSetupModalProps { + definition: ChannelDefinition; + onClose: () => void; +} + +function ChannelConfigContent({ definition }: { definition: ChannelDefinition }) { + const channelId = definition.id as ChannelType; + switch (channelId) { + case 'telegram': + return ; + case 'discord': + return ; + default: + return ( +

+ Configuration for {definition.display_name} is not available yet. +

+ ); + } +} + +export default function ChannelSetupModal({ definition, onClose }: ChannelSetupModalProps) { + const modalRef = useRef(null); + + useEffect(() => { + const handleEscape = (e: KeyboardEvent) => { + if (e.key === 'Escape') onClose(); + }; + document.addEventListener('keydown', handleEscape); + return () => document.removeEventListener('keydown', handleEscape); + }, [onClose]); + + useEffect(() => { + const previousFocus = document.activeElement as HTMLElement; + modalRef.current?.focus(); + return () => { + previousFocus?.focus?.(); + }; + }, []); + + const handleBackdropClick = (e: React.MouseEvent) => { + if (e.target === e.currentTarget) onClose(); + }; + + const icon = CHANNEL_ICONS[definition.icon] ?? ''; + + const modalContent = ( +
+
e.stopPropagation()}> + {/* Header */} +
+
+
+
+ {icon && {icon}} +

+ {definition.display_name} +

+ + channel + +
+

{definition.description}

+
+ +
+
+ + {/* Content */} +
+ +
+
+
+ ); + + return createPortal(modalContent, document.body); +} diff --git a/app/src/components/channels/DiscordConfig.tsx b/app/src/components/channels/DiscordConfig.tsx index e6c295b0c..97fa239cd 100644 --- a/app/src/components/channels/DiscordConfig.tsx +++ b/app/src/components/channels/DiscordConfig.tsx @@ -149,74 +149,67 @@ const DiscordConfig = ({ definition }: DiscordConfigProps) => { ); return ( -
-
-

{definition.display_name}

-

{definition.description}

-
- +
{error && ( -
+
{error}
)} -
- {definition.auth_modes.map(spec => { - const compositeKey = `discord:${spec.mode}`; - const connection = channelConnections.connections.discord?.[spec.mode]; - const status: ChannelConnectionStatus = connection?.status ?? 'disconnected'; + {definition.auth_modes.map(spec => { + const compositeKey = `discord:${spec.mode}`; + const connection = channelConnections.connections.discord?.[spec.mode]; + const status: ChannelConnectionStatus = connection?.status ?? 'disconnected'; - return ( -
-
-
-

- {AUTH_MODE_LABELS[spec.mode] ?? spec.mode} -

-

{spec.description}

- {connection?.lastError && ( -

{connection.lastError}

- )} -
- -
- - {spec.fields.length > 0 && ( -
- {spec.fields.map(field => ( - updateField(compositeKey, field.key, val)} - disabled={busyKeys[compositeKey]} - /> - ))} -
- )} - -
- - + return ( +
+
+
+

+ {AUTH_MODE_LABELS[spec.mode] ?? spec.mode} +

+

{spec.description}

+ {connection?.lastError && ( +

{connection.lastError}

+ )}
+
- ); - })} -
-
+ + {spec.fields.length > 0 && ( +
+ {spec.fields.map(field => ( + updateField(compositeKey, field.key, val)} + disabled={busyKeys[compositeKey]} + /> + ))} +
+ )} + +
+ + +
+ + ); + })} + ); }; diff --git a/app/src/components/channels/TelegramConfig.tsx b/app/src/components/channels/TelegramConfig.tsx index b8ea20181..373a489e5 100644 --- a/app/src/components/channels/TelegramConfig.tsx +++ b/app/src/components/channels/TelegramConfig.tsx @@ -3,7 +3,6 @@ import { useCallback, useEffect, useRef, useState } from 'react'; import { AUTH_MODE_LABELS } from '../../lib/channels/definitions'; import { channelConnectionsApi } from '../../services/api/channelConnectionsApi'; -import { managedDmApi } from '../../services/api/managedDmApi'; import { callCoreRpc } from '../../services/coreRpcClient'; import { disconnectChannelConnection, @@ -73,37 +72,51 @@ const TelegramConfig = ({ definition }: TelegramConfigProps) => { }, []); const startManagedDmPolling = useCallback( - (key: string, token: string) => { + (key: string, linkToken: string) => { stopManagedDmPolling(key); const controller = new AbortController(); managedDmPollControllers.current[key] = controller; + const POLL_INTERVAL_MS = 3_000; + const POLL_TIMEOUT_MS = 5 * 60 * 1_000; + void (async () => { - log('polling managed dm status', { key, tokenLength: token.length }); + log('polling telegram link status via core RPC', { key, tokenLength: linkToken.length }); + const startedAt = Date.now(); + try { - const status = await managedDmApi.pollManagedDmStatusUntilVerified(token, { - signal: controller.signal, - }); + while (Date.now() - startedAt < POLL_TIMEOUT_MS) { + if (controller.signal.aborted) return; - if (controller.signal.aborted) { - return; - } + try { + const check = await channelConnectionsApi.telegramLoginCheck(linkToken); + if (check.linked) { + log('telegram managed dm linked via core RPC', { key, details: check.details }); + dispatch( + upsertChannelConnection({ + channel: 'telegram', + authMode: 'managed_dm', + patch: { status: 'connected', lastError: undefined, capabilities: ['dm'] }, + }) + ); + return; + } + } catch { + // Best-effort polling: keep trying until timeout or cancellation. + } - if (status?.verified) { - log('managed dm verified via polling', { - key, - telegramUsername: status.telegramUsername, + await new Promise(resolve => { + const timer = window.setTimeout(resolve, POLL_INTERVAL_MS); + const onAbort = () => { + window.clearTimeout(timer); + resolve(); + }; + controller.signal.addEventListener('abort', onAbort, { once: true }); }); - dispatch( - upsertChannelConnection({ - channel: 'telegram', - authMode: 'managed_dm', - patch: { status: 'connected', lastError: undefined, capabilities: ['dm'] }, - }) - ); - return; } + if (controller.signal.aborted) return; + dispatch( upsertChannelConnection({ channel: 'telegram', @@ -113,9 +126,7 @@ const TelegramConfig = ({ definition }: TelegramConfigProps) => { ); setError(MANAGED_DM_TIMEOUT_MESSAGE); } catch (pollError) { - if (controller.signal.aborted) { - return; - } + if (controller.signal.aborted) return; const msg = pollError instanceof Error ? pollError.message : String(pollError); log('managed dm polling failed', { key, error: msg }); @@ -177,13 +188,13 @@ const TelegramConfig = ({ definition }: TelegramConfigProps) => { if (result.status === 'pending_auth' && result.auth_action) { if (result.auth_action === 'telegram_managed_dm') { try { - const initiated = await managedDmApi.initiateManagedDm(); - log('managed dm initiate success', { + const loginStart = await channelConnectionsApi.telegramLoginStart(); + log('telegram login start success', { key, - tokenLength: initiated.token.length, - expiresAt: initiated.expiresAt, + tokenLength: loginStart.linkToken.length, + botUsername: loginStart.botUsername, }); - await openUrl(initiated.deepLink); + await openUrl(loginStart.telegramUrl); dispatch( upsertChannelConnection({ channel: 'telegram', @@ -191,11 +202,13 @@ const TelegramConfig = ({ definition }: TelegramConfigProps) => { patch: { status: 'connecting', lastError: MANAGED_DM_CONNECTING_MESSAGE }, }) ); - startManagedDmPolling(key, initiated.token); - } catch (managedDmError) { + startManagedDmPolling(key, loginStart.linkToken); + } catch (loginStartError) { const msg = - managedDmError instanceof Error ? managedDmError.message : String(managedDmError); - log('managed dm initiate failed', { key, error: msg }); + loginStartError instanceof Error + ? loginStartError.message + : String(loginStartError); + log('telegram login start failed', { key, error: msg }); dispatch( upsertChannelConnection({ channel: 'telegram', @@ -259,74 +272,67 @@ const TelegramConfig = ({ definition }: TelegramConfigProps) => { ); return ( -
-
-

{definition.display_name}

-

{definition.description}

-
- +
{error && ( -
+
{error}
)} -
- {definition.auth_modes.map(spec => { - const compositeKey = `telegram:${spec.mode}`; - const connection = channelConnections.connections.telegram?.[spec.mode]; - const status: ChannelConnectionStatus = connection?.status ?? 'disconnected'; + {definition.auth_modes.map(spec => { + const compositeKey = `telegram:${spec.mode}`; + const connection = channelConnections.connections.telegram?.[spec.mode]; + const status: ChannelConnectionStatus = connection?.status ?? 'disconnected'; - return ( -
-
-
-

- {AUTH_MODE_LABELS[spec.mode] ?? spec.mode} -

-

{spec.description}

- {connection?.lastError && ( -

{connection.lastError}

- )} -
- -
- - {spec.fields.length > 0 && ( -
- {spec.fields.map(field => ( - updateField(compositeKey, field.key, val)} - disabled={busyKeys[compositeKey]} - /> - ))} -
- )} - -
- - + return ( +
+
+
+

+ {AUTH_MODE_LABELS[spec.mode] ?? spec.mode} +

+

{spec.description}

+ {connection?.lastError && ( +

{connection.lastError}

+ )}
+
- ); - })} -
-
+ + {spec.fields.length > 0 && ( +
+ {spec.fields.map(field => ( + updateField(compositeKey, field.key, val)} + disabled={busyKeys[compositeKey]} + /> + ))} +
+ )} + +
+ + +
+ + ); + })} + ); }; diff --git a/app/src/components/channels/WebChannelConfig.tsx b/app/src/components/channels/WebChannelConfig.tsx index 84640fed7..fe4c3f708 100644 --- a/app/src/components/channels/WebChannelConfig.tsx +++ b/app/src/components/channels/WebChannelConfig.tsx @@ -5,20 +5,18 @@ interface WebChannelConfigProps { definition: ChannelDefinition; } -const WebChannelConfig = ({ definition }: WebChannelConfigProps) => { +const WebChannelConfig = ({ definition: _definition }: WebChannelConfigProps) => { return ( -
+
-

{definition.display_name}

-

{definition.description}

+
-
-

+

The web channel is always available — no setup required.

-
+ ); }; diff --git a/app/src/components/channels/__tests__/DiscordConfig.test.tsx b/app/src/components/channels/__tests__/DiscordConfig.test.tsx index 45f74df4a..4f1617ba2 100644 --- a/app/src/components/channels/__tests__/DiscordConfig.test.tsx +++ b/app/src/components/channels/__tests__/DiscordConfig.test.tsx @@ -8,9 +8,9 @@ import DiscordConfig from '../DiscordConfig'; const discordDef = FALLBACK_DEFINITIONS.find(d => d.id === 'discord')!; describe('DiscordConfig', () => { - it('renders the Discord header', () => { + it('renders auth mode labels', () => { renderWithProviders(); - expect(screen.getByText('Discord')).toBeInTheDocument(); + expect(screen.getByText('OAuth Sign-in')).toBeInTheDocument(); }); it('renders both auth modes', () => { diff --git a/app/src/components/channels/__tests__/TelegramConfig.test.tsx b/app/src/components/channels/__tests__/TelegramConfig.test.tsx index 138dbe9b3..e55b9f1cd 100644 --- a/app/src/components/channels/__tests__/TelegramConfig.test.tsx +++ b/app/src/components/channels/__tests__/TelegramConfig.test.tsx @@ -3,7 +3,6 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import { FALLBACK_DEFINITIONS } from '../../../lib/channels/definitions'; import { channelConnectionsApi } from '../../../services/api/channelConnectionsApi'; -import { managedDmApi } from '../../../services/api/managedDmApi'; import { renderWithProviders } from '../../../test/test-utils'; import { openUrl } from '../../../utils/openUrl'; import TelegramConfig from '../TelegramConfig'; @@ -16,14 +15,8 @@ vi.mock('../../../services/api/channelConnectionsApi', () => ({ disconnectChannel: vi.fn(), listDefinitions: vi.fn(), listStatus: vi.fn(), - }, -})); - -vi.mock('../../../services/api/managedDmApi', () => ({ - managedDmApi: { - initiateManagedDm: vi.fn(), - getManagedDmStatus: vi.fn(), - pollManagedDmStatusUntilVerified: vi.fn(), + telegramLoginStart: vi.fn(), + telegramLoginCheck: vi.fn(), }, })); @@ -34,16 +27,15 @@ afterEach(() => { }); describe('TelegramConfig', () => { - it('renders the Telegram header', () => { + it('renders auth mode labels', () => { renderWithProviders(); - expect(screen.getByText('Telegram')).toBeInTheDocument(); + expect(screen.getByText('Login with OpenHuman')).toBeInTheDocument(); }); it('renders both auth modes', () => { renderWithProviders(); - // "Bot Token" appears as both a heading and a field label, so use getAllByText. - expect(screen.getAllByText('Bot Token').length).toBeGreaterThanOrEqual(1); - expect(screen.getByText('Managed DM')).toBeInTheDocument(); + expect(screen.getAllByText(/Bot Token/i).length).toBeGreaterThanOrEqual(1); + expect(screen.getByText('Login with OpenHuman')).toBeInTheDocument(); }); it('shows credential fields for bot_token mode', () => { @@ -67,41 +59,35 @@ describe('TelegramConfig', () => { }); }); - it('starts managed dm flow, opens the deep link, and marks the channel connected after verification', async () => { + it('starts managed dm flow via core RPC, opens the deep link, and marks connected after polling', async () => { vi.mocked(channelConnectionsApi.connectChannel).mockResolvedValue({ status: 'pending_auth', auth_action: 'telegram_managed_dm', restart_required: false, }); - vi.mocked(managedDmApi.initiateManagedDm).mockResolvedValue({ - token: 'managed-dm-token', - deepLink: 'https://t.me/openhuman_bot?start=manageddm_managed-dm-token', - expiresAt: '2026-04-04T12:00:00.000Z', + vi.mocked(channelConnectionsApi.telegramLoginStart).mockResolvedValue({ + linkToken: 'link-token-abc', + telegramUrl: 'https://t.me/openhuman_bot?start=link-token-abc', + botUsername: 'openhuman_bot', }); - vi.mocked(managedDmApi.pollManagedDmStatusUntilVerified).mockResolvedValue({ - verified: true, - telegramUsername: 'telegram-user', - expiresAt: '2026-04-04T12:05:00.000Z', + vi.mocked(channelConnectionsApi.telegramLoginCheck).mockResolvedValue({ + linked: true, + details: { telegramUserId: '12345' }, }); renderWithProviders(); const connectButtons = screen.getAllByText('Connect'); - fireEvent.click(connectButtons[1]); + fireEvent.click(connectButtons[0]); await waitFor(() => { - expect(managedDmApi.initiateManagedDm).toHaveBeenCalledTimes(1); + expect(channelConnectionsApi.telegramLoginStart).toHaveBeenCalledTimes(1); }); await waitFor(() => { - expect(openUrl).toHaveBeenCalledWith( - 'https://t.me/openhuman_bot?start=manageddm_managed-dm-token' - ); + expect(openUrl).toHaveBeenCalledWith('https://t.me/openhuman_bot?start=link-token-abc'); }); await waitFor(() => { - expect(managedDmApi.pollManagedDmStatusUntilVerified).toHaveBeenCalledWith( - 'managed-dm-token', - expect.objectContaining({ signal: expect.any(AbortSignal) }) - ); + expect(channelConnectionsApi.telegramLoginCheck).toHaveBeenCalledWith('link-token-abc'); }); expect(await screen.findByText('Connected')).toBeInTheDocument(); }); diff --git a/app/src/components/oauth/providerConfigs.tsx b/app/src/components/oauth/providerConfigs.tsx index bc17fa736..e22e960d9 100644 --- a/app/src/components/oauth/providerConfigs.tsx +++ b/app/src/components/oauth/providerConfigs.tsx @@ -26,19 +26,19 @@ const GoogleIcon = ({ className = '' }: { className?: string }) => ( ); const TwitterIcon = ({ className = '' }: { className?: string }) => ( - + ); const GitHubIcon = ({ className = '' }: { className?: string }) => ( - + ); const DiscordIcon = ({ className = '' }: { className?: string }) => ( - + ); diff --git a/app/src/components/settings/panels/MessagingPanel.tsx b/app/src/components/settings/panels/MessagingPanel.tsx index 727b3c121..a8eec70cb 100644 --- a/app/src/components/settings/panels/MessagingPanel.tsx +++ b/app/src/components/settings/panels/MessagingPanel.tsx @@ -1,71 +1,62 @@ import { useCallback, useMemo, useState } from 'react'; import { useChannelDefinitions } from '../../../hooks/useChannelDefinitions'; -import { AUTH_MODE_LABELS } from '../../../lib/channels/definitions'; import { resolvePreferredAuthModeForChannel } from '../../../lib/channels/routing'; -import { createChannelLinkToken } from '../../../services/api/authApi'; import { channelConnectionsApi } from '../../../services/api/channelConnectionsApi'; -import { callCoreRpc } from '../../../services/coreRpcClient'; -import { - disconnectChannelConnection, - setChannelConnectionStatus, - setDefaultMessagingChannel, - upsertChannelConnection, -} from '../../../store/channelConnectionsSlice'; +import { setDefaultMessagingChannel } from '../../../store/channelConnectionsSlice'; import { useAppDispatch, useAppSelector } from '../../../store/hooks'; import type { - AuthModeSpec, - ChannelAuthMode, ChannelConnectionStatus, + ChannelDefinition, ChannelType, } from '../../../types/channels'; -import { BACKEND_URL, TELEGRAM_BOT_USERNAME } from '../../../utils/config'; -import { openUrl } from '../../../utils/openUrl'; -import ChannelFieldInput from '../../channels/ChannelFieldInput'; -import ChannelStatusBadge from '../../channels/ChannelStatusBadge'; +import ChannelSetupModal from '../../channels/ChannelSetupModal'; import SettingsHeader from '../components/SettingsHeader'; import { useSettingsNavigation } from '../hooks/useSettingsNavigation'; -function normalizeBaseUrl(baseUrl?: string): string { - return (baseUrl || 'https://api.tinyhumans.ai').trim().replace(/\/+$/, ''); +const CHANNEL_ICONS: Record = { + telegram: '\u2708\uFE0F', + discord: '\uD83C\uDFAE', + web: '\uD83C\uDF10', +}; + +function statusDot(status: ChannelConnectionStatus): string { + switch (status) { + case 'connected': + return 'bg-sage-500'; + case 'connecting': + return 'bg-amber-500 animate-pulse'; + case 'error': + return 'bg-coral-500'; + default: + return 'bg-stone-300'; + } } -function buildManagedChannelLaunchUrl( - channel: ChannelType, - token: string, - launchUrl?: string -): string | undefined { - if (launchUrl) return launchUrl; - - if (channel === 'telegram') { - return `https://t.me/${encodeURIComponent(TELEGRAM_BOT_USERNAME)}?start=${encodeURIComponent(token)}`; +function statusLabel(status: ChannelConnectionStatus): string { + switch (status) { + case 'connected': + return 'Connected'; + case 'connecting': + return 'Connecting'; + case 'error': + return 'Error'; + default: + return 'Not configured'; } - - if (channel === 'discord') { - return `${normalizeBaseUrl(BACKEND_URL)}/auth/discord/connect?linkToken=${encodeURIComponent(token)}`; - } - - return undefined; } -function buildManagedChannelInstruction( - channel: ChannelType, - token: string, - launchUrl?: string -): string { - if (channel === 'telegram') { - return launchUrl - ? 'Continue in Telegram to finish linking your account.' - : `Open Telegram and message @${TELEGRAM_BOT_USERNAME} with this link token: ${token}`; +function statusColor(status: ChannelConnectionStatus): string { + switch (status) { + case 'connected': + return 'text-sage-600'; + case 'connecting': + return 'text-amber-600'; + case 'error': + return 'text-coral-600'; + default: + return 'text-stone-400'; } - - if (channel === 'discord') { - return launchUrl - ? 'Continue in Discord to finish linking your account.' - : `Use this Discord link token to continue linking your account: ${token}`; - } - - return `Use this link token to continue: ${token}`; } const MessagingPanel = () => { @@ -74,10 +65,13 @@ const MessagingPanel = () => { const channelConnections = useAppSelector(state => state.channelConnections); const { definitions, loading, error: loadError } = useChannelDefinitions(); - const [error, setError] = useState(null); - const [busyKeys, setBusyKeys] = useState>({}); - const [fieldValues, setFieldValues] = useState>>({}); - const [pendingInstruction, setPendingInstruction] = useState>({}); + const [busy, setBusy] = useState>({}); + const [channelModalDef, setChannelModalDef] = useState(null); + + const configurableChannels = useMemo( + () => definitions.filter(d => d.id !== 'web'), + [definitions] + ); const recommendedRoute = useMemo(() => { const channel = channelConnections.defaultMessagingChannel; @@ -85,177 +79,31 @@ const MessagingPanel = () => { return authMode ? `${channel} via ${authMode}` : 'No active route'; }, [channelConnections]); - const runBusy = useCallback(async (key: string, task: () => Promise) => { - setBusyKeys(prev => ({ ...prev, [key]: true })); - setError(null); - try { - await task(); - } catch (e) { - const msg = e instanceof Error ? e.message : String(e); - setError(msg); - } finally { - setBusyKeys(prev => ({ ...prev, [key]: false })); - } - }, []); + const bestStatus = useCallback( + (channelId: ChannelType): ChannelConnectionStatus => { + const conns = channelConnections.connections[channelId]; + if (!conns) return 'disconnected'; + const statuses = Object.values(conns).map(c => c?.status ?? 'disconnected'); + if (statuses.includes('connected')) return 'connected'; + if (statuses.includes('connecting')) return 'connecting'; + if (statuses.includes('error')) return 'error'; + return 'disconnected'; + }, + [channelConnections] + ); const handleSetDefaultChannel = useCallback( (channel: ChannelType) => { const key = `default:${channel}`; - void runBusy(key, async () => { - dispatch(setDefaultMessagingChannel(channel)); - await channelConnectionsApi.updatePreferences(channel); + setBusy(prev => ({ ...prev, [key]: true })); + dispatch(setDefaultMessagingChannel(channel)); + void channelConnectionsApi.updatePreferences(channel).finally(() => { + setBusy(prev => ({ ...prev, [key]: false })); }); }, - [dispatch, runBusy] + [dispatch] ); - const updateField = useCallback((compositeKey: string, fieldKey: string, value: string) => { - setFieldValues(prev => ({ - ...prev, - [compositeKey]: { ...(prev[compositeKey] ?? {}), [fieldKey]: value }, - })); - }, []); - - const handleConnect = useCallback( - (channel: ChannelType, spec: AuthModeSpec) => { - const key = `${channel}:${spec.mode}`; - void runBusy(key, async () => { - dispatch( - setChannelConnectionStatus({ channel, authMode: spec.mode, status: 'connecting' }) - ); - - const credentials: Record = {}; - for (const field of spec.fields) { - const val = fieldValues[key]?.[field.key]?.trim() ?? ''; - if (field.required && !val) { - dispatch( - setChannelConnectionStatus({ - channel, - authMode: spec.mode, - status: 'error', - lastError: `${field.label} is required`, - }) - ); - return; - } - if (val) credentials[field.key] = val; - } - - const isManagedLinkFlow = - (channel === 'telegram' && spec.mode === 'managed_dm') || - (channel === 'discord' && spec.mode === 'oauth'); - - if (isManagedLinkFlow) { - try { - const link = await createChannelLinkToken(channel); - const launchUrl = buildManagedChannelLaunchUrl(channel, link.token, link.launchUrl); - const instruction = buildManagedChannelInstruction(channel, link.token, launchUrl); - - dispatch( - upsertChannelConnection({ - channel, - authMode: spec.mode, - patch: { status: 'connecting' }, - }) - ); - - setPendingInstruction(prev => ({ ...prev, [key]: instruction })); - - if (launchUrl) { - try { - await openUrl(launchUrl); - } catch { - // Opening the URL failed — include the URL so the user can copy it manually. - setPendingInstruction(prev => ({ - ...prev, - [key]: `${instruction} (URL: ${launchUrl})`, - })); - } - } - } catch (e) { - const msg = e instanceof Error ? e.message : String(e); - dispatch( - setChannelConnectionStatus({ - channel, - authMode: spec.mode, - status: 'error', - lastError: msg, - }) - ); - throw e; - } - return; - } - - const result = await channelConnectionsApi.connectChannel(channel, { - authMode: spec.mode, - credentials: Object.keys(credentials).length > 0 ? credentials : undefined, - }); - - if (result.status === 'pending_auth' && result.auth_action) { - const instruction = result.message ?? `Initiate ${result.auth_action} flow`; - - dispatch( - upsertChannelConnection({ - channel, - authMode: spec.mode, - patch: { status: 'connecting' }, - }) - ); - - setPendingInstruction(prev => ({ ...prev, [key]: instruction })); - - if (result.auth_action.includes('oauth')) { - try { - const oauthResponse = await callCoreRpc<{ result: { oauthUrl?: string } }>({ - method: 'openhuman.auth.oauth_connect', - params: { provider: channel, skillId: channel }, - }); - if (oauthResponse.result?.oauthUrl) { - await openUrl(oauthResponse.result.oauthUrl); - } - } catch { - // OAuth URL fetch is best-effort. - } - } - return; - } - - setPendingInstruction(prev => { - const next = { ...prev }; - delete next[key]; - return next; - }); - - dispatch( - upsertChannelConnection({ - channel, - authMode: spec.mode, - patch: { status: 'connected', lastError: undefined, capabilities: ['read', 'write'] }, - }) - ); - - if (result.restart_required) { - setError(result.message ?? 'Restart the service to activate the channel.'); - } - }); - }, - [dispatch, fieldValues, runBusy] - ); - - const handleDisconnect = useCallback( - (channel: ChannelType, authMode: ChannelAuthMode) => { - const key = `${channel}:${authMode}`; - void runBusy(key, async () => { - await channelConnectionsApi.disconnectChannel(channel, authMode); - dispatch(disconnectChannelConnection({ channel, authMode })); - }); - }, - [dispatch, runBusy] - ); - - const displayError = error || loadError; - return (
@@ -274,7 +122,7 @@ const MessagingPanel = () => { key={channelId} type="button" onClick={() => handleSetDefaultChannel(channelId)} - disabled={busyKeys[busyKey]} + disabled={busy[busyKey]} className={`rounded-lg border px-3 py-2 text-sm transition-colors ${ selected ? 'border-primary-500/60 bg-primary-50 text-primary-600' @@ -290,9 +138,9 @@ const MessagingPanel = () => {

- {displayError && ( + {loadError && (
- {displayError} + {loadError}
)} @@ -302,94 +150,66 @@ const MessagingPanel = () => { )} - {!loading && - definitions.map(def => { - const channelId = def.id as ChannelType; - return ( -
-
-

{def.display_name}

-

{def.description}

- {def.capabilities.length > 0 && ( -
- {def.capabilities.map(cap => ( - - {cap.replace(/_/g, ' ')} - - ))} -
- )} -
+ {/* Channel cards — click to open the shared ChannelSetupModal */} + {!loading && ( +
+

Channel Integrations

+

+ Configure auth modes for each messaging channel. +

+
+ {configurableChannels.map(def => { + const channelId = def.id as ChannelType; + const status = bestStatus(channelId); + const icon = CHANNEL_ICONS[def.icon] ?? ''; -
- {def.auth_modes.map(spec => { - const compositeKey = `${channelId}:${spec.mode}`; - const connection = channelConnections.connections[channelId]?.[spec.mode]; - const status: ChannelConnectionStatus = connection?.status ?? 'disconnected'; - - const instruction = pendingInstruction[compositeKey]; - - return ( -
-
-
-

- {AUTH_MODE_LABELS[spec.mode] ?? spec.mode} -

-

{spec.description}

- {connection?.lastError && ( -

{connection.lastError}

- )} - {instruction && ( -

{instruction}

- )} -
- -
- - {spec.fields.length > 0 && ( -
- {spec.fields.map(field => ( - updateField(compositeKey, field.key, value)} - // placeholder={field.placeholder || field.label} - // className="w-full rounded-lg border border-stone-200 bg-white px-3 py-2 text-sm text-stone-900 placeholder:text-stone-400 focus:outline-none focus:border-primary-500/60" - /> - ))} -
- )} - -
- - + return ( +
- ); - })} + + + + + + ); + })} + +
+ )} + + {/* Shared channel config modal */} + {channelModalDef && ( + setChannelModalDef(null)} /> + )} ); }; diff --git a/app/src/index.css b/app/src/index.css index ca9da5efb..0108645ee 100644 --- a/app/src/index.css +++ b/app/src/index.css @@ -154,7 +154,7 @@ .app-dotted-canvas { background-color: transparent; - background-image: radial-gradient(circle at center, rgba(0, 0, 0, 0.2) 1px, transparent 1px); + background-image: radial-gradient(circle at center, rgba(0, 0, 0, 0.5) 1px, transparent 1px); background-size: 18px 18px; background-position: 0 0; } diff --git a/app/src/lib/channels/definitions.ts b/app/src/lib/channels/definitions.ts index 0cfb94468..80010de36 100644 --- a/app/src/lib/channels/definitions.ts +++ b/app/src/lib/channels/definitions.ts @@ -3,24 +3,24 @@ import type { ChannelConnectionStatus, ChannelDefinition } from '../../types/cha /** Status badge styles for channel connection states. */ export const STATUS_STYLES: Record = { - connected: { label: 'Connected', className: 'bg-sage-500/20 text-sage-300 border-sage-500/30' }, + connected: { label: 'Connected', className: 'bg-sage-500/10 text-sage-700 border-sage-500/30' }, connecting: { label: 'Connecting', - className: 'bg-amber-500/20 text-amber-300 border-amber-500/30', + className: 'bg-amber-500/10 text-amber-700 border-amber-500/30', }, disconnected: { label: 'Disconnected', - className: 'bg-stone-500/20 text-stone-300 border-stone-500/30', + className: 'bg-stone-100 text-stone-500 border-stone-200', }, - error: { label: 'Error', className: 'bg-coral-500/20 text-coral-300 border-coral-500/30' }, + error: { label: 'Error', className: 'bg-coral-500/10 text-coral-700 border-coral-500/30' }, }; /** Human-readable labels for auth modes. */ export const AUTH_MODE_LABELS: Record = { - managed_dm: 'Managed DM', + managed_dm: 'Login with OpenHuman', oauth: 'OAuth Sign-in', - bot_token: 'Bot Token', - api_key: 'API Key', + bot_token: 'Use your own Bot Token', + api_key: 'Use your own API Key', }; /** Fallback definitions used when the core sidecar is unreachable. */ @@ -31,6 +31,12 @@ export const FALLBACK_DEFINITIONS: ChannelDefinition[] = [ description: 'Send and receive messages via Telegram.', icon: 'telegram', auth_modes: [ + { + mode: 'managed_dm', + description: 'Message the OpenHuman Telegram bot directly.', + fields: [], + auth_action: 'telegram_managed_dm', + }, { mode: 'bot_token', description: 'Provide your own Telegram Bot token from @BotFather.', @@ -52,12 +58,6 @@ export const FALLBACK_DEFINITIONS: ChannelDefinition[] = [ ], auth_action: undefined, }, - { - mode: 'managed_dm', - description: 'Message the OpenHuman Telegram bot directly.', - fields: [], - auth_action: 'telegram_managed_dm', - }, ], capabilities: ['send_text', 'receive_text', 'typing', 'draft_updates'], }, diff --git a/app/src/lib/meshGradient.js b/app/src/lib/meshGradient.js index 50e635766..6411af667 100644 --- a/app/src/lib/meshGradient.js +++ b/app/src/lib/meshGradient.js @@ -21,9 +21,15 @@ class MiniGl { const _miniGl = this, debug_output = -1 !== document.location.search.toLowerCase().indexOf('debug=webgl'); ((_miniGl.canvas = canvas), - (_miniGl.gl = _miniGl.canvas.getContext('webgl', { antialias: true })), + (_miniGl.gl = + _miniGl.canvas.getContext('webgl', { antialias: true }) || + _miniGl.canvas.getContext('webgl2', { antialias: true })), (_miniGl.meshes = [])); const context = _miniGl.gl; + if (!context) { + console.warn('[MeshGradient] Failed to obtain WebGL context'); + return; + } (width && height && this.setSize(width, height), _miniGl.lastDebugMsg, (_miniGl.debug = @@ -46,6 +52,12 @@ class MiniGl { const material = this; function getShaderByType(type, source) { const shader = context.createShader(type); + if (!shader) { + console.warn( + '[MeshGradient] createShader returned null — WebGL context may be lost' + ); + return null; + } return ( context.shaderSource(shader, source), context.compileShader(shader), @@ -72,8 +84,13 @@ class MiniGl { (material.fragmentShader = getShaderByType( context.FRAGMENT_SHADER, material.Source - )), - (material.program = context.createProgram()), + ))); + if (!material.vertexShader || !material.fragmentShader) { + throw new Error( + '[MeshGradient] Shader compilation failed — WebGL context may be lost' + ); + } + ((material.program = context.createProgram()), context.attachShader(material.program, material.vertexShader), context.attachShader(material.program, material.fragmentShader), context.linkProgram(material.program), @@ -475,10 +492,6 @@ class Gradient { console.warn('[MeshGradient] Element is not a canvas:', selector); return this; } - if (!this.el.getContext('webgl') && !this.el.getContext('webgl2')) { - console.warn('[MeshGradient] WebGL not available'); - return this; - } this.connect(); return this; })); @@ -505,10 +518,12 @@ class Gradient { document.querySelectorAll('canvas').length < 1 ? console.log('DID NOT LOAD HERO STRIPE CANVAS') : ((this.minigl = new MiniGl(this.el, null, null, !0)), - requestAnimationFrame(() => { - this.el && - ((this.computedCanvasStyle = getComputedStyle(this.el)), this.waitForCssVars()); - }))); + this.minigl.gl + ? requestAnimationFrame(() => { + this.el && + ((this.computedCanvasStyle = getComputedStyle(this.el)), this.waitForCssVars()); + }) + : console.warn('[MeshGradient] MiniGl has no GL context — gradient disabled'))); /* this.scrollObserver = await s.create(.1, !1), this.scrollObserver.observe(this.el), @@ -615,11 +630,15 @@ class Gradient { document.body.classList.remove('isGradientLegendVisible')); } init() { - (this.initGradientColors(), - this.initMesh(), - this.resize(), - requestAnimationFrame(this.animate), - window.addEventListener('resize', this.resize)); + try { + (this.initGradientColors(), + this.initMesh(), + this.resize(), + requestAnimationFrame(this.animate), + window.addEventListener('resize', this.resize)); + } catch (err) { + console.warn('[MeshGradient] init failed, gradient disabled:', err); + } } /* * Waiting for the css variables to become available, usually on page load before we can continue. diff --git a/app/src/pages/Channels.tsx b/app/src/pages/Channels.tsx index 1c436026a..a725b4ebd 100644 --- a/app/src/pages/Channels.tsx +++ b/app/src/pages/Channels.tsx @@ -13,13 +13,13 @@ const Channels = () => {
{error && ( -
+
{error}
)} {loading ? ( -
+
Loading channel definitions...
) : ( diff --git a/app/src/pages/Skills.tsx b/app/src/pages/Skills.tsx index cf64a5d6a..62a0dc2cf 100644 --- a/app/src/pages/Skills.tsx +++ b/app/src/pages/Skills.tsx @@ -1,6 +1,7 @@ import { useMemo, useState } from 'react'; import { useNavigate } from 'react-router-dom'; +import ChannelSetupModal from '../components/channels/ChannelSetupModal'; import { DefaultIcon, SKILL_ICONS, @@ -10,6 +11,7 @@ import { } from '../components/skills/shared'; import SkillDebugModal from '../components/skills/SkillDebugModal'; import SkillSetupModal from '../components/skills/SkillSetupModal'; +import { useChannelDefinitions } from '../hooks/useChannelDefinitions'; import { useAvailableSkills, useSkillConnectionStatus, @@ -19,6 +21,8 @@ import { import { skillManager } from '../lib/skills/manager'; import { installSkill } from '../lib/skills/skillsApi'; import type { SkillConnectionStatus, SkillHostConnectionState } from '../lib/skills/types'; +import { useAppSelector } from '../store/hooks'; +import type { ChannelConnectionStatus, ChannelDefinition, ChannelType } from '../types/channels'; import { IS_DEV } from '../utils/config'; import { deriveSkillSyncSummaryText, @@ -40,6 +44,93 @@ function statusDotClass(status: SkillConnectionStatus): string { } } +// ─── Channel icons for the integration cards ──────────────────────────────── + +const CHANNEL_ICONS: Record = { + telegram: '\u2708\uFE0F', + discord: '\uD83C\uDFAE', + web: '\uD83C\uDF10', +}; + +function channelStatusDot(status: ChannelConnectionStatus): string { + switch (status) { + case 'connected': + return 'bg-sage-500'; + case 'connecting': + return 'bg-amber-500 animate-pulse'; + case 'error': + return 'bg-coral-500'; + default: + return 'bg-stone-300'; + } +} + +function channelStatusLabel(status: ChannelConnectionStatus): string { + switch (status) { + case 'connected': + return 'Connected'; + case 'connecting': + return 'Connecting'; + case 'error': + return 'Error'; + default: + return 'Not configured'; + } +} + +function channelStatusColor(status: ChannelConnectionStatus): string { + switch (status) { + case 'connected': + return 'text-sage-600'; + case 'connecting': + return 'text-amber-600'; + case 'error': + return 'text-coral-600'; + default: + return 'text-stone-400'; + } +} + +// ─── Channel Integration Card ──────────────────────────────────────────────── + +interface ChannelIntegrationCardProps { + definition: ChannelDefinition; + bestStatus: ChannelConnectionStatus; + onClick: () => void; +} + +function ChannelIntegrationCard({ definition, bestStatus, onClick }: ChannelIntegrationCardProps) { + const icon = CHANNEL_ICONS[definition.icon] ?? ''; + + return ( + + ); +} + // ─── Skill Card (used in the skills list) ─────────────────────────────────── interface SkillCardProps { @@ -274,14 +365,37 @@ export default function Skills() { const navigate = useNavigate(); // Skills from registry via RPC const { skills: availableSkills, loading: skillsLoading } = useAvailableSkills(); + // Channel definitions + const { definitions: channelDefs } = useChannelDefinitions(); + const channelConnections = useAppSelector(state => state.channelConnections); - // Modal state + // Modal state — skills const [setupModalOpen, setSetupModalOpen] = useState(false); const [activeSkillId, setActiveSkillId] = useState(null); const [activeSkillName, setActiveSkillName] = useState(''); const [activeSkillDescription, setActiveSkillDescription] = useState(''); const [activeSkillHasSetup, setActiveSkillHasSetup] = useState(false); + // Modal state — channels + const [channelModalDef, setChannelModalDef] = useState(null); + + /** Resolve the best connection status across all auth modes for a channel. */ + const bestChannelStatus = (channelId: ChannelType): ChannelConnectionStatus => { + const conns = channelConnections.connections[channelId]; + if (!conns) return 'disconnected'; + const statuses = Object.values(conns).map(c => c?.status ?? 'disconnected'); + if (statuses.includes('connected')) return 'connected'; + if (statuses.includes('connecting')) return 'connecting'; + if (statuses.includes('error')) return 'error'; + return 'disconnected'; + }; + + // Only show configurable channels (telegram, discord — not web) + const configurableChannels = useMemo( + () => channelDefs.filter(d => d.id !== 'web'), + [channelDefs] + ); + // Transform registry entries to SkillListEntry const skillsList: SkillListEntry[] = useMemo(() => { return availableSkills @@ -403,6 +517,28 @@ export default function Skills() {
+ {/* Channel Integrations */} + {configurableChannels.length > 0 && ( +
+
+

Channel Integrations

+

+ Connect messaging platforms to send and receive messages. +

+
+
+ {configurableChannels.map(def => ( + setChannelModalDef(def)} + /> + ))} +
+
+ )} + {/* Main card */}
{/* Header */} @@ -444,7 +580,7 @@ export default function Skills() {
- {/* Setup modal */} + {/* Skill setup modal */} {setupModalOpen && activeSkillId && ( )} + + {/* Channel setup modal */} + {channelModalDef && ( + setChannelModalDef(null)} /> + )}
); } diff --git a/app/src/pages/__tests__/Channels.test.tsx b/app/src/pages/__tests__/Channels.test.tsx index 47700ccd8..2ed83916a 100644 --- a/app/src/pages/__tests__/Channels.test.tsx +++ b/app/src/pages/__tests__/Channels.test.tsx @@ -26,6 +26,6 @@ describe('Channels page', () => { it('renders the Telegram config panel by default', async () => { renderWithProviders(); - expect(await screen.findByText('Managed DM')).toBeInTheDocument(); + expect(await screen.findByText('Login with OpenHuman')).toBeInTheDocument(); }); }); diff --git a/app/src/services/api/channelConnectionsApi.ts b/app/src/services/api/channelConnectionsApi.ts index 83a93eecb..a19ca0610 100644 --- a/app/src/services/api/channelConnectionsApi.ts +++ b/app/src/services/api/channelConnectionsApi.ts @@ -12,6 +12,17 @@ interface ConnectChannelPayload { credentials?: Record; } +export interface TelegramLoginStartResult { + linkToken: string; + telegramUrl: string; + botUsername: string; +} + +export interface TelegramLoginCheckResult { + linked: boolean; + details?: Record | null; +} + export const channelConnectionsApi = { /** Fetch all available channel definitions from the backend. */ listDefinitions: async (): Promise => { @@ -63,6 +74,24 @@ export const channelConnectionsApi = { return result; }, + /** Initiate managed Telegram DM login — creates a link token and returns a deep link URL. */ + telegramLoginStart: async (): Promise => { + const result = await callCoreRpc({ + method: 'openhuman.channels_telegram_login_start', + params: {}, + }); + return result; + }, + + /** Check whether the Telegram managed DM link has been completed. */ + telegramLoginCheck: async (linkToken: string): Promise => { + const result = await callCoreRpc({ + method: 'openhuman.channels_telegram_login_check', + params: { linkToken }, + }); + return result; + }, + /** Placeholder for default channel preference sync. */ updatePreferences: async (defaultMessagingChannel: ChannelType): Promise => { void defaultMessagingChannel; diff --git a/docs/TODO.md b/docs/TODO.md index 5d41a4fee..c6f3656d1 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -64,6 +64,7 @@ todo - memory skill [] should index properly all the things (sanil) + [] should properly register user interactions --- e2e tests to write up diff --git a/scripts/test-channel-messaging.sh b/scripts/test-channel-messaging.sh new file mode 100755 index 000000000..5d9da299c --- /dev/null +++ b/scripts/test-channel-messaging.sh @@ -0,0 +1,141 @@ +#!/usr/bin/env bash +# ────────────────────────────────────────────────────────────────────── +# test-channel-messaging.sh +# +# End-to-end test: sends a message from the backend to the user's +# linked Telegram account via the Rust core RPC. +# +# Usage: +# bash scripts/test-channel-messaging.sh +# bash scripts/test-channel-messaging.sh "Custom message text" +# +# Prerequisites: +# - Active session token (login via the app first) +# - Telegram account linked (completed managed DM flow) +# - Core binary built: cargo build --bin openhuman-core +# ────────────────────────────────────────────────────────────────────── +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +ROOT_DIR="$(cd "$SCRIPT_DIR/.." && pwd)" + +# Load env +if [[ -f "$ROOT_DIR/scripts/load-dotenv.sh" ]]; then + source "$ROOT_DIR/scripts/load-dotenv.sh" 2>/dev/null || true +fi + +CORE_BIN="${OPENHUMAN_CORE_BIN:-}" +if [[ -z "$CORE_BIN" ]]; then + CORE_BIN="$ROOT_DIR/target/debug/openhuman-core" + if [[ ! -x "$CORE_BIN" ]]; then + echo "Building openhuman-core..." + cargo build --manifest-path "$ROOT_DIR/Cargo.toml" --bin openhuman-core 2>&1 | tail -2 + fi +fi + +MESSAGE="${1:-Hello from OpenHuman! 🚀 This is a test message sent via the channel messaging API.}" + +divider() { echo "────────────────────────────────────────────────"; } + +echo "" +echo "🧪 Channel Messaging E2E Test" +divider + +# ── Step 1: Check session ──────────────────────────────────────────── +echo "" +echo "1️⃣ Checking session..." +AUTH_STATE=$("$CORE_BIN" auth get_state 2>&1 | grep -A20 '{' || true) +IS_AUTH=$(echo "$AUTH_STATE" | grep -o '"isAuthenticated": *true' || true) + +if [[ -z "$IS_AUTH" ]]; then + echo " ❌ Not authenticated. Please login via the app first." + echo " Auth state:" + echo "$AUTH_STATE" | head -10 + exit 1 +fi +echo " ✅ Authenticated" + +# ── Step 2: Validate session against backend ───────────────────────── +echo "" +echo "2️⃣ Validating session with backend (GET /auth/me)..." +ME_RESULT=$("$CORE_BIN" auth get_me 2>&1 || true) +if echo "$ME_RESULT" | grep -qi "401\|Invalid token\|expired\|failed"; then + echo " ❌ Session token expired or invalid." + echo " $ME_RESULT" | tail -3 + echo "" + echo " Please re-login via the app to get a fresh token." + exit 1 +fi + +TELEGRAM_ID=$(echo "$ME_RESULT" | grep -o '"telegramId": *"[^"]*"' | head -1 | sed 's/.*: *"//;s/"//' || true) +USERNAME=$(echo "$ME_RESULT" | grep -o '"username": *"[^"]*"' | head -1 | sed 's/.*: *"//;s/"//' || true) +echo " ✅ Session valid — user: ${USERNAME:-unknown}, telegramId: ${TELEGRAM_ID:-not linked}" + +if [[ -z "$TELEGRAM_ID" ]]; then + echo "" + echo " ⚠️ No telegramId found on your profile." + echo " Complete the Telegram managed DM linking flow first." + echo " (Skills page → Telegram → Login with OpenHuman → click Start in Telegram)" + exit 1 +fi + +# ── Step 3: Send a text message via Telegram ───────────────────────── +echo "" +echo "3️⃣ Sending message to Telegram..." +echo " Channel: telegram" +echo " Message: $MESSAGE" +divider + +SEND_RESULT=$("$CORE_BIN" channels send_message \ + --channel telegram \ + --message "{\"text\": \"$MESSAGE\"}" 2>&1 || true) + +echo "$SEND_RESULT" | grep -A20 '{' | head -20 + +if echo "$SEND_RESULT" | grep -qi '"success": *true\|"messageId"'; then + echo "" + echo " ✅ Message sent successfully! Check your Telegram." +else + echo "" + echo " ❌ Message send may have failed. Check output above." +fi + +# ── Step 4: Send a message with a button ───────────────────────────── +echo "" +echo "4️⃣ Sending message with inline button..." + +BUTTON_MSG=$("$CORE_BIN" channels send_message \ + --channel telegram \ + --message '{"text": "Here is a link for you:", "buttons": [{"label": "OpenHuman GitHub", "url": "https://github.com/tinyhumansai/openhuman"}]}' 2>&1 || true) + +echo "$BUTTON_MSG" | grep -A20 '{' | head -15 + +if echo "$BUTTON_MSG" | grep -qi '"success": *true\|"messageId"'; then + echo " ✅ Button message sent!" +else + echo " ❌ Button message may have failed." +fi + +# ── Step 5: List threads ───────────────────────────────────────────── +echo "" +echo "5️⃣ Listing Telegram threads..." + +THREADS=$("$CORE_BIN" channels list_threads \ + --channel telegram 2>&1 || true) + +# Show the JSON result (skip the banner lines) +echo "$THREADS" | tail -5 +echo " ✅ Threads listed." + +# ── Done ───────────────────────────────────────────────────────────── +divider +echo "" +echo "✅ Channel messaging E2E test complete." +echo "" +echo "Available RPC methods:" +echo " openhuman.channels_send_message — Send rich message (text, photo, stickers, buttons)" +echo " openhuman.channels_send_reaction — React to a message with emoji" +echo " openhuman.channels_create_thread — Create a conversation thread" +echo " openhuman.channels_update_thread — Close or reopen a thread" +echo " openhuman.channels_list_threads — List threads for a channel" +echo "" diff --git a/scripts/test-channel-receive.mjs b/scripts/test-channel-receive.mjs new file mode 100755 index 000000000..f0c0e9c22 --- /dev/null +++ b/scripts/test-channel-receive.mjs @@ -0,0 +1,267 @@ +#!/usr/bin/env node +// ────────────────────────────────────────────────────────────────────── +// test-channel-receive.mjs +// +// Connects to the backend Socket.IO server, authenticates with the +// stored session JWT, and listens for incoming channel messages. +// +// Usage: +// node scripts/test-channel-receive.mjs +// node scripts/test-channel-receive.mjs --timeout 120 +// node scripts/test-channel-receive.mjs --debug # verbose logging +// node scripts/test-channel-receive.mjs --send-test # also send a test msg +// ────────────────────────────────────────────────────────────────────── + +import { execSync } from 'child_process'; +import { existsSync, readFileSync } from 'fs'; +import path from 'path'; +import { fileURLToPath } from 'url'; + +const __dirname = path.dirname(fileURLToPath(import.meta.url)); +const ROOT = path.resolve(__dirname, '..'); + +const args = process.argv.slice(2); +const DEBUG = args.includes('--debug'); +const SEND_TEST = args.includes('--send-test'); + +// ── Load env ──────────────────────────────────────────────────────── +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')); + +const BACKEND_URL = ( + process.env.BACKEND_URL || + process.env.VITE_BACKEND_URL || + 'https://staging-api.alphahuman.xyz' +).replace(/\/+$/, ''); + +const TIMEOUT_SECS = parseInt(args.find((_, i, a) => a[i - 1] === '--timeout') || '60', 10); + +function dbg(...args) { + if (DEBUG) console.log(' [debug]', ...args); +} + +// ── Get session token from core ───────────────────────────────────── +function getSessionToken() { + const coreBin = path.join(ROOT, 'target', 'debug', 'openhuman-core'); + 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 { + return null; + } +} + +console.log(''); +console.log('📡 Channel Receive Listener'); +console.log('────────────────────────────────────────────────'); +console.log(`Backend: ${BACKEND_URL}`); +console.log(`Timeout: ${TIMEOUT_SECS}s`); +console.log(`Debug: ${DEBUG}`); +console.log(`Send test: ${SEND_TEST}`); +console.log(''); + +// ── Resolve token ─────────────────────────────────────────────────── +console.log('🔑 Resolving session token...'); +let token = getSessionToken(); +if (!token) { + console.error(' ❌ No session token found. Login via the app first.'); + process.exit(1); +} +console.log(` ✅ Token: ${token.slice(0, 20)}...`); + +// ── Validate token against backend ────────────────────────────────── +console.log(''); +console.log('🔍 Validating token against backend...'); +try { + const resp = await fetch(`${BACKEND_URL}/auth/me`, { + headers: { Authorization: `Bearer ${token}` }, + }); + const data = await resp.json(); + if (data.success && data.data) { + const u = data.data; + console.log(` ✅ User: ${u.username || u.firstName || u._id}`); + console.log(` ✅ Telegram ID: ${u.telegramId || 'NOT LINKED'}`); + if (!u.telegramId) { + console.log(' ⚠️ No Telegram linked — incoming messages won\'t route to you'); + } + } else { + console.log(` ❌ Token invalid: ${JSON.stringify(data)}`); + process.exit(1); + } +} catch (err) { + console.log(` ❌ Backend unreachable: ${err.message}`); + process.exit(1); +} + +// ── Connect Socket.IO ─────────────────────────────────────────────── +const socketIoPath = path.join(ROOT, 'node_modules', 'socket.io-client', 'dist', 'socket.io.esm.min.js'); +let io; +try { + const mod = await import(socketIoPath); + io = mod.io || mod.default; +} catch { + try { + const mod = await import('socket.io-client'); + io = mod.io || mod.default; + } catch (err) { + console.error(' ❌ Cannot load socket.io-client:', err.message); + process.exit(1); + } +} + +console.log(''); +console.log('🔌 Connecting to Socket.IO...'); +dbg('URL:', BACKEND_URL); +dbg('Transport: websocket + polling'); +dbg('Path: /socket.io/'); + +const socket = io(BACKEND_URL, { + auth: { token }, + transports: ['websocket', 'polling'], + path: '/socket.io/', + reconnection: true, + reconnectionAttempts: 3, + timeout: 10_000, +}); + +let messageCount = 0; + +// ── In debug mode, log ALL events ─────────────────────────────────── +if (DEBUG) { + const origOnevent = socket.onevent; + socket.onevent = function (packet) { + const eventName = packet.data?.[0]; + const eventData = packet.data?.slice(1); + console.log(` [debug] EVENT: ${eventName}`, JSON.stringify(eventData).slice(0, 200)); + origOnevent.call(this, packet); + }; +} + +socket.on('connect', () => { + console.log(` ✅ Connected (socket id: ${socket.id})`); + dbg('Transport:', socket.io.engine?.transport?.name); + console.log(''); + console.log('────────────────────────────────────────────────'); + console.log('👂 Listening for incoming channel messages...'); + console.log(' Send a message to the Telegram bot now!'); + console.log(` (auto-exit after ${TIMEOUT_SECS}s, or Ctrl+C)`); + console.log('────────────────────────────────────────────────'); + console.log(''); + + // If --send-test, fire off a test message after connecting + if (SEND_TEST) { + setTimeout(async () => { + console.log('📤 Sending test message via backend REST API...'); + try { + const resp = await fetch(`${BACKEND_URL}/channels/telegram/messages`, { + method: 'POST', + headers: { + Authorization: `Bearer ${token}`, + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ text: '🧪 Round-trip test from receive listener script' }), + }); + const data = await resp.json(); + console.log(` Response: ${JSON.stringify(data)}`); + console.log(''); + } catch (err) { + console.log(` ❌ Send failed: ${err.message}`); + } + }, 1000); + } +}); + +socket.on('ready', () => { + console.log(' 🟢 Server ready'); + dbg('Socket authenticated and registered on server'); +}); + +socket.on('connect_error', (err) => { + console.error(` ❌ Connection error: ${err.message}`); + dbg('Full error:', err); +}); + +socket.on('error', (data) => { + console.error(` ❌ Server error: ${JSON.stringify(data)}`); +}); + +socket.on('disconnect', (reason) => { + console.log(` 🔴 Disconnected: ${reason}`); +}); + +socket.io.on('reconnect_attempt', (attempt) => { + dbg(`Reconnect attempt #${attempt}`); +}); + +socket.io.on('reconnect', (attempt) => { + console.log(` 🔄 Reconnected after ${attempt} attempt(s)`); +}); + +// ── Channel message events ────────────────────────────────────────── + +// Inbound: Telegram user → bot → backend → socket → here +socket.on('channel:message', (data) => { + messageCount++; + const ts = new Date().toLocaleTimeString(); + console.log(`┌─ 📨 INBOUND #${messageCount} [${ts}]`); + console.log(`│ Channel: ${data.channel || 'unknown'}`); + console.log(`│ UserId: ${data.userId || 'unknown'}`); + console.log(`│ Message: ${data.message || '(empty)'}`); + console.log(`│ Time: ${data.receivedAt || 'unknown'}`); + console.log(`└──────────────────────────────────`); + console.log(''); + dbg('Full payload:', JSON.stringify(data)); +}); + +// Outbound confirmation: app sent message → backend → Telegram, socket notified +socket.on('channel:message:sent', (data) => { + const ts = new Date().toLocaleTimeString(); + console.log(`┌─ 📤 OUTBOUND CONFIRMED [${ts}]`); + console.log(`│ Channel: ${data.channel || 'unknown'}`); + console.log(`│ Success: ${data.result?.success}`); + console.log(`│ MsgId: ${data.result?.messageId || 'n/a'}`); + if (data.message?.text) { + console.log(`│ Text: ${data.message.text.slice(0, 80)}`); + } + console.log(`└──────────────────────────────────`); + console.log(''); + dbg('Full payload:', JSON.stringify(data)); +}); + +// ── Timeout ───────────────────────────────────────────────────────── +const timer = setTimeout(() => { + console.log(''); + console.log('────────────────────────────────────────────────'); + console.log(`⏰ Timeout (${TIMEOUT_SECS}s). Received ${messageCount} inbound message(s).`); + socket.disconnect(); + process.exit(0); +}, TIMEOUT_SECS * 1000); + +process.on('SIGINT', () => { + clearTimeout(timer); + console.log(''); + console.log('────────────────────────────────────────────────'); + console.log(`👋 Stopped. Received ${messageCount} inbound message(s).`); + socket.disconnect(); + process.exit(0); +}); diff --git a/src/api/rest.rs b/src/api/rest.rs index d53c337fd..a5e468e97 100644 --- a/src/api/rest.rs +++ b/src/api/rest.rs @@ -487,6 +487,113 @@ impl BackendOAuthClient { Ok(client_key.to_string()) } + /// `POST /channels/:channel/messages` — Send a rich message to a channel. + pub async fn send_channel_message( + &self, + channel: &str, + bearer_jwt: &str, + message_body: Value, + ) -> Result { + let channel = channel.trim().trim_matches('/'); + anyhow::ensure!(!channel.is_empty(), "channel is required"); + let encoded = urlencoding::encode(channel); + self.authed_json( + bearer_jwt, + Method::POST, + &format!("channels/{encoded}/messages"), + Some(message_body), + ) + .await + } + + /// `POST /channels/:channel/reactions` — React to a message in a channel. + pub async fn send_channel_reaction( + &self, + channel: &str, + bearer_jwt: &str, + reaction_body: Value, + ) -> Result { + let channel = channel.trim().trim_matches('/'); + anyhow::ensure!(!channel.is_empty(), "channel is required"); + let encoded = urlencoding::encode(channel); + self.authed_json( + bearer_jwt, + Method::POST, + &format!("channels/{encoded}/reactions"), + Some(reaction_body), + ) + .await + } + + /// `POST /channels/:channel/threads` — Create a thread in a channel. + pub async fn create_channel_thread( + &self, + channel: &str, + bearer_jwt: &str, + title: &str, + ) -> Result { + let channel = channel.trim().trim_matches('/'); + anyhow::ensure!(!channel.is_empty(), "channel is required"); + anyhow::ensure!(!title.trim().is_empty(), "title is required"); + let encoded = urlencoding::encode(channel); + let body = serde_json::json!({ "title": title.trim() }); + self.authed_json( + bearer_jwt, + Method::POST, + &format!("channels/{encoded}/threads"), + Some(body), + ) + .await + } + + /// `PATCH /channels/:channel/threads/:thread_id` — Close or reopen a thread. + pub async fn update_channel_thread( + &self, + channel: &str, + bearer_jwt: &str, + thread_id: &str, + action: &str, + ) -> Result { + let channel = channel.trim().trim_matches('/'); + anyhow::ensure!(!channel.is_empty(), "channel is required"); + anyhow::ensure!(!thread_id.trim().is_empty(), "threadId is required"); + anyhow::ensure!( + action == "close" || action == "reopen", + "action must be 'close' or 'reopen'" + ); + let encoded_channel = urlencoding::encode(channel); + let encoded_thread = urlencoding::encode(thread_id.trim()); + let body = serde_json::json!({ "action": action }); + self.authed_json( + bearer_jwt, + Method::PATCH, + &format!("channels/{encoded_channel}/threads/{encoded_thread}"), + Some(body), + ) + .await + } + + /// `GET /channels/:channel/threads` — List threads, optionally filtered by active status. + pub async fn list_channel_threads( + &self, + channel: &str, + bearer_jwt: &str, + active_filter: Option, + ) -> Result { + let channel = channel.trim().trim_matches('/'); + anyhow::ensure!(!channel.is_empty(), "channel is required"); + let encoded = urlencoding::encode(channel); + let mut path = format!("channels/{encoded}/threads"); + if let Some(active) = active_filter { + path.push_str(if active { + "?active=true" + } else { + "?active=false" + }); + } + self.authed_json(bearer_jwt, Method::GET, &path, None).await + } + /// `DELETE /auth/integrations/:id` pub async fn revoke_integration(&self, integration_id: &str, bearer_jwt: &str) -> Result<()> { let id = integration_id.trim(); diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 5aca193c2..d000e4958 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -41,6 +41,38 @@ pub async fn rpc_handler(State(state): State, Json(req): Json Result { + let result = invoke_method_inner(state, method, params).await; + + // If the RPC call failed due to an expired/invalid session token (401 from + // the backend), automatically clear the stored session so the frontend + // detects the logged-out state and redirects to login. + if let Err(ref msg) = result { + if is_session_expired_error(msg) { + log::warn!( + "[jsonrpc] backend returned 401 for method '{}' — clearing stored session", + method + ); + if let Ok(config) = crate::openhuman::config::rpc::load_config_with_timeout().await { + let _ = crate::openhuman::credentials::rpc::clear_session(&config).await; + } + } + } + + result +} + +fn is_session_expired_error(msg: &str) -> bool { + let lower = msg.to_lowercase(); + (lower.contains("401") && lower.contains("unauthorized")) + || lower.contains("invalid token") + || msg.contains("SESSION_EXPIRED") +} + +async fn invoke_method_inner( + state: AppState, + method: &str, + params: Value, +) -> Result { if let Some(schema) = all::schema_for_rpc_method(method) { let params_obj = params_to_object(params)?; all::validate_params(&schema, ¶ms_obj)?; diff --git a/src/openhuman/channels/controllers/definitions.rs b/src/openhuman/channels/controllers/definitions.rs index d70185d2a..844eef038 100644 --- a/src/openhuman/channels/controllers/definitions.rs +++ b/src/openhuman/channels/controllers/definitions.rs @@ -171,6 +171,12 @@ fn telegram_definition() -> ChannelDefinition { description: "Send and receive messages via Telegram.", icon: "telegram", auth_modes: vec![ + AuthModeSpec { + mode: ChannelAuthMode::ManagedDm, + description: "Message the OpenHuman Telegram bot directly.", + fields: vec![], + auth_action: Some("telegram_managed_dm"), + }, AuthModeSpec { mode: ChannelAuthMode::BotToken, description: "Provide your own Telegram Bot token from @BotFather.", @@ -192,12 +198,6 @@ fn telegram_definition() -> ChannelDefinition { ], auth_action: None, }, - AuthModeSpec { - mode: ChannelAuthMode::ManagedDm, - description: "Message the OpenHuman Telegram bot directly.", - fields: vec![], - auth_action: Some("telegram_managed_dm"), - }, ], capabilities: vec![ ChannelCapability::SendText, diff --git a/src/openhuman/channels/controllers/ops.rs b/src/openhuman/channels/controllers/ops.rs index 42f3695c7..6b0547c1b 100644 --- a/src/openhuman/channels/controllers/ops.rs +++ b/src/openhuman/channels/controllers/ops.rs @@ -3,6 +3,9 @@ use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; +use crate::api::config::effective_api_url; +use crate::api::jwt::get_session_token; +use crate::api::rest::BackendOAuthClient; use crate::openhuman::config::Config; use crate::openhuman::credentials; use crate::rpc::RpcOutcome; @@ -236,6 +239,327 @@ pub async fn test_channel( )) } +// --------------------------------------------------------------------------- +// Managed Telegram login flow +// --------------------------------------------------------------------------- + +/// Default bot username when not configured via env var. +const DEFAULT_TELEGRAM_BOT_USERNAME: &str = "alphahumantest_bot"; + +/// Resolve the managed Telegram bot username from env or default. +fn telegram_bot_username() -> String { + std::env::var("OPENHUMAN_TELEGRAM_BOT_USERNAME") + .or_else(|_| std::env::var("VITE_TELEGRAM_BOT_USERNAME")) + .unwrap_or_else(|_| DEFAULT_TELEGRAM_BOT_USERNAME.to_string()) +} + +/// Result from `telegram_login_start`. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TelegramLoginStartResult { + /// The short-lived link token created by the backend. + pub link_token: String, + /// Full Telegram deep link URL the user should open. + pub telegram_url: String, + /// Bot username used. + pub bot_username: String, +} + +/// Result from `telegram_login_check`. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TelegramLoginCheckResult { + /// Whether the Telegram user has been linked to the app user. + pub linked: bool, + /// Backend-provided status payload (may include telegramUserId, etc.). + #[serde(skip_serializing_if = "Option::is_none")] + pub details: Option, +} + +/// Step 1: Create a channel link token for Telegram and return the deep link URL. +/// +/// Requires an active session JWT. +pub async fn telegram_login_start( + config: &Config, +) -> Result, String> { + let api_url = effective_api_url(&config.api_url); + let jwt = get_session_token(config)? + .ok_or_else(|| "session JWT required; complete login first".to_string())?; + + log::debug!( + "[telegram-login] creating channel link token via {}", + api_url + ); + + let client = BackendOAuthClient::new(&api_url).map_err(|e| e.to_string())?; + let payload = client + .create_channel_link_token("telegram", &jwt) + .await + .map_err(|e| format!("failed to create Telegram link token: {e}"))?; + + // Extract the link token from the backend response. + // Expected shape: { "linkToken": "..." } or { "token": "..." } + let link_token = payload + .get("linkToken") + .or_else(|| payload.get("token")) + .and_then(|v| v.as_str()) + .ok_or_else(|| { + format!( + "backend response missing linkToken field: {}", + serde_json::to_string(&payload).unwrap_or_default() + ) + })? + .trim() + .to_string(); + + if link_token.is_empty() { + return Err("backend returned empty link token".to_string()); + } + + let bot_username = telegram_bot_username(); + let telegram_url = format!("https://t.me/{}?start={}", bot_username, link_token); + + log::debug!( + "[telegram-login] link token created, deep link: {}", + telegram_url + ); + + Ok(RpcOutcome::new( + TelegramLoginStartResult { + link_token, + telegram_url, + bot_username, + }, + vec![], + )) +} + +/// Step 2: Check whether the user has completed the Telegram link (clicked /start). +/// +/// Polls `GET /auth/me` and checks whether the user profile now has a `telegramId`. +/// The frontend should poll this until `linked` becomes `true`. +/// On success, stores a `channel:telegram:managed_dm` credential marker locally. +pub async fn telegram_login_check( + config: &Config, + _link_token: &str, +) -> Result, String> { + let api_url = effective_api_url(&config.api_url); + let jwt = get_session_token(config)?.ok_or_else(|| "session JWT required".to_string())?; + + log::debug!("[telegram-login] checking if user profile has telegramId via GET /auth/me"); + + let client = BackendOAuthClient::new(&api_url).map_err(|e| e.to_string())?; + let user_payload = client + .fetch_current_user(&jwt) + .await + .map_err(|e| format!("failed to fetch user profile: {e}"))?; + + // Check if the user now has a telegramId set. + let telegram_id = user_payload + .get("telegramId") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + .or_else(|| { + user_payload + .get("telegram_id") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + }); + + let linked = telegram_id.is_some(); + + log::debug!( + "[telegram-login] user profile telegramId: {:?}, linked={}", + telegram_id, + linked + ); + + if linked { + // Store a credential marker so `channel_status` reports connected. + let provider_key = credential_provider("telegram", ChannelAuthMode::ManagedDm); + + let telegram_user_id = telegram_id.unwrap_or("").to_string(); + + let mut fields_map = serde_json::Map::new(); + fields_map.insert("linked".to_string(), Value::Bool(true)); + if !telegram_user_id.is_empty() { + fields_map.insert( + "telegram_user_id".to_string(), + Value::String(telegram_user_id), + ); + } + + // Store using a placeholder token (managed mode has no user-visible token). + credentials::ops::store_provider_credentials( + config, + &provider_key, + None, + Some("managed".to_string()), + Some(Value::Object(fields_map)), + Some(true), + ) + .await + .map_err(|e| format!("failed to store managed channel credentials: {e}"))?; + + log::info!( + "[telegram-login] Telegram managed DM linked; credentials stored as {}", + provider_key + ); + } + + Ok(RpcOutcome::new( + TelegramLoginCheckResult { + linked, + details: if linked { Some(user_payload) } else { None }, + }, + vec![], + )) +} + +// --------------------------------------------------------------------------- +// Channel messaging, reactions, and thread management +// --------------------------------------------------------------------------- + +/// Send a rich message to a channel via the backend API. +pub async fn channel_send_message( + config: &Config, + channel: &str, + message: Value, +) -> Result, String> { + let api_url = effective_api_url(&config.api_url); + let jwt = get_session_token(config)? + .ok_or_else(|| "session JWT required; complete login first".to_string())?; + + log::debug!( + "[channels] sending message to channel '{}' via {}", + channel, + api_url + ); + + let client = BackendOAuthClient::new(&api_url).map_err(|e| e.to_string())?; + let result = client + .send_channel_message(channel, &jwt, message) + .await + .map_err(|e| format!("failed to send channel message: {e}"))?; + + log::debug!("[channels] send_message response: {:?}", result); + + Ok(RpcOutcome::new(result, vec![])) +} + +/// Send a reaction to a message in a channel via the backend API. +pub async fn channel_send_reaction( + config: &Config, + channel: &str, + reaction: Value, +) -> Result, String> { + let api_url = effective_api_url(&config.api_url); + let jwt = get_session_token(config)? + .ok_or_else(|| "session JWT required; complete login first".to_string())?; + + log::debug!( + "[channels] sending reaction to channel '{}' via {}", + channel, + api_url + ); + + let client = BackendOAuthClient::new(&api_url).map_err(|e| e.to_string())?; + let result = client + .send_channel_reaction(channel, &jwt, reaction) + .await + .map_err(|e| format!("failed to send channel reaction: {e}"))?; + + log::debug!("[channels] send_reaction response: {:?}", result); + + Ok(RpcOutcome::new(result, vec![])) +} + +/// Create a thread in a channel via the backend API. +pub async fn channel_create_thread( + config: &Config, + channel: &str, + title: &str, +) -> Result, String> { + let api_url = effective_api_url(&config.api_url); + let jwt = get_session_token(config)? + .ok_or_else(|| "session JWT required; complete login first".to_string())?; + + log::debug!( + "[channels] creating thread in channel '{}' title='{}' via {}", + channel, + title, + api_url + ); + + let client = BackendOAuthClient::new(&api_url).map_err(|e| e.to_string())?; + let result = client + .create_channel_thread(channel, &jwt, title) + .await + .map_err(|e| format!("failed to create channel thread: {e}"))?; + + log::debug!("[channels] create_thread response: {:?}", result); + + Ok(RpcOutcome::new(result, vec![])) +} + +/// Close or reopen a thread in a channel via the backend API. +pub async fn channel_update_thread( + config: &Config, + channel: &str, + thread_id: &str, + action: &str, +) -> Result, String> { + let api_url = effective_api_url(&config.api_url); + let jwt = get_session_token(config)? + .ok_or_else(|| "session JWT required; complete login first".to_string())?; + + log::debug!( + "[channels] updating thread '{}' in channel '{}' action='{}' via {}", + thread_id, + channel, + action, + api_url + ); + + let client = BackendOAuthClient::new(&api_url).map_err(|e| e.to_string())?; + let result = client + .update_channel_thread(channel, &jwt, thread_id, action) + .await + .map_err(|e| format!("failed to update channel thread: {e}"))?; + + log::debug!("[channels] update_thread response: {:?}", result); + + Ok(RpcOutcome::new(result, vec![])) +} + +/// List threads in a channel via the backend API. +pub async fn channel_list_threads( + config: &Config, + channel: &str, + active: Option, +) -> Result, String> { + let api_url = effective_api_url(&config.api_url); + let jwt = get_session_token(config)? + .ok_or_else(|| "session JWT required; complete login first".to_string())?; + + log::debug!( + "[channels] listing threads in channel '{}' active={:?} via {}", + channel, + active, + api_url + ); + + let client = BackendOAuthClient::new(&api_url).map_err(|e| e.to_string())?; + let result = client + .list_channel_threads(channel, &jwt, active) + .await + .map_err(|e| format!("failed to list channel threads: {e}"))?; + + log::debug!("[channels] list_threads response: {:?}", result); + + Ok(RpcOutcome::new(result, vec![])) +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/openhuman/channels/controllers/schemas.rs b/src/openhuman/channels/controllers/schemas.rs index 4ac89405a..bd52cca3b 100644 --- a/src/openhuman/channels/controllers/schemas.rs +++ b/src/openhuman/channels/controllers/schemas.rs @@ -53,6 +53,49 @@ struct TestParams { credentials: serde_json::Value, } +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct TelegramLoginCheckParams { + link_token: String, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct SendMessageParams { + channel: String, + message: serde_json::Value, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct SendReactionParams { + channel: String, + reaction: serde_json::Value, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct CreateThreadParams { + channel: String, + title: String, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct UpdateThreadParams { + channel: String, + thread_id: String, + action: String, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct ListThreadsParams { + channel: String, + #[serde(default)] + active: Option, +} + // --------------------------------------------------------------------------- // Public registry exports // --------------------------------------------------------------------------- @@ -65,6 +108,13 @@ pub fn all_controller_schemas() -> Vec { schemas("disconnect"), schemas("status"), schemas("test"), + schemas("telegram_login_start"), + schemas("telegram_login_check"), + schemas("send_message"), + schemas("send_reaction"), + schemas("create_thread"), + schemas("update_thread"), + schemas("list_threads"), ] } @@ -94,6 +144,34 @@ pub fn all_registered_controllers() -> Vec { schema: schemas("test"), handler: handle_test, }, + RegisteredController { + schema: schemas("telegram_login_start"), + handler: handle_telegram_login_start, + }, + RegisteredController { + schema: schemas("telegram_login_check"), + handler: handle_telegram_login_check, + }, + RegisteredController { + schema: schemas("send_message"), + handler: handle_send_message, + }, + RegisteredController { + schema: schemas("send_reaction"), + handler: handle_send_reaction, + }, + RegisteredController { + schema: schemas("create_thread"), + handler: handle_create_thread, + }, + RegisteredController { + schema: schemas("update_thread"), + handler: handle_update_thread, + }, + RegisteredController { + schema: schemas("list_threads"), + handler: handle_list_threads, + }, ] } @@ -174,6 +252,92 @@ pub fn schemas(function: &str) -> ControllerSchema { "Test result with success flag and message.", )], }, + "telegram_login_start" => ControllerSchema { + namespace: "channels", + function: "telegram_login_start", + description: + "Create a Telegram link token and return the deep link URL for managed DM login.", + inputs: vec![], + outputs: vec![json_output( + "result", + "Object with linkToken, telegramUrl, and botUsername.", + )], + }, + "telegram_login_check" => ControllerSchema { + namespace: "channels", + function: "telegram_login_check", + description: "Check whether the Telegram managed DM link has been completed.", + inputs: vec![required_string( + "linkToken", + "The link token returned by telegram_login_start.", + )], + outputs: vec![json_output( + "result", + "Object with linked (bool) and optional details.", + )], + }, + "send_message" => ControllerSchema { + namespace: "channels", + function: "send_message", + description: "Send a rich message to a channel (text, photo, sticker, animation, buttons, reply).", + inputs: vec![ + required_string("channel", "Channel identifier (e.g. telegram)."), + required_json( + "message", + "Message body with optional fields: text, parseMode, photoUrl, stickerFileId, animationUrl, buttons, replyToMessageId, threadId.", + ), + ], + outputs: vec![json_output("result", "Object with success flag and optional messageId.")], + }, + "send_reaction" => ControllerSchema { + namespace: "channels", + function: "send_reaction", + description: "React to a message in a channel with an emoji.", + inputs: vec![ + required_string("channel", "Channel identifier (e.g. telegram)."), + required_json( + "reaction", + "Reaction body: { messageId, emoji, chatId? }.", + ), + ], + outputs: vec![json_output("result", "Object with success flag.")], + }, + "create_thread" => ControllerSchema { + namespace: "channels", + function: "create_thread", + description: "Create a new thread in a channel.", + inputs: vec![ + required_string("channel", "Channel identifier (e.g. telegram)."), + required_string("title", "Thread title."), + ], + outputs: vec![json_output("result", "Object with success flag and optional threadId.")], + }, + "update_thread" => ControllerSchema { + namespace: "channels", + function: "update_thread", + description: "Close or reopen a thread in a channel.", + inputs: vec![ + required_string("channel", "Channel identifier (e.g. telegram)."), + required_string("threadId", "Thread identifier to update."), + required_string("action", "Action to perform: 'close' or 'reopen'."), + ], + outputs: vec![json_output("result", "Object with success flag.")], + }, + "list_threads" => ControllerSchema { + namespace: "channels", + function: "list_threads", + description: "List threads in a channel, optionally filtered by active status.", + inputs: vec![ + required_string("channel", "Channel identifier (e.g. telegram)."), + FieldSchema { + name: "active", + ty: TypeSchema::Option(Box::new(TypeSchema::Bool)), + comment: "Optional filter: true for active threads, false for closed threads.", + required: false, + }, + ], + outputs: vec![json_output("result", "Array of thread objects.")], + }, _ => ControllerSchema { namespace: "channels", function: "unknown", @@ -258,6 +422,69 @@ fn handle_test(params: Map) -> ControllerFuture { }) } +fn handle_telegram_login_start(_params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + to_json(ops::telegram_login_start(&config).await?) + }) +} + +fn handle_telegram_login_check(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let p = deserialize_params::(params)?; + to_json(ops::telegram_login_check(&config, p.link_token.trim()).await?) + }) +} + +fn handle_send_message(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let p = deserialize_params::(params)?; + to_json(ops::channel_send_message(&config, p.channel.trim(), p.message).await?) + }) +} + +fn handle_send_reaction(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let p = deserialize_params::(params)?; + to_json(ops::channel_send_reaction(&config, p.channel.trim(), p.reaction).await?) + }) +} + +fn handle_create_thread(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let p = deserialize_params::(params)?; + to_json(ops::channel_create_thread(&config, p.channel.trim(), p.title.trim()).await?) + }) +} + +fn handle_update_thread(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let p = deserialize_params::(params)?; + to_json( + ops::channel_update_thread( + &config, + p.channel.trim(), + p.thread_id.trim(), + p.action.trim(), + ) + .await?, + ) + }) +} + +fn handle_list_threads(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let p = deserialize_params::(params)?; + to_json(ops::channel_list_threads(&config, p.channel.trim(), p.active).await?) + }) +} + // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- diff --git a/src/openhuman/skills/socket_manager.rs b/src/openhuman/skills/socket_manager.rs index 4eee224b6..91eeed2a5 100644 --- a/src/openhuman/skills/socket_manager.rs +++ b/src/openhuman/skills/socket_manager.rs @@ -642,6 +642,12 @@ fn handle_sio_event( handle_webhook_request(&shared, data, &tx).await; }); } + // Inbound channel message (Telegram, Discord, etc.) — run agent and reply + "channel:message" => { + tokio::spawn(async move { + handle_channel_inbound_message(data).await; + }); + } _ => { // Forward to skills (desktop only) and frontend { @@ -1011,6 +1017,174 @@ async fn handle_webhook_request( ); } +// --------------------------------------------------------------------------- +// Channel inbound message → agent loop → reply +// --------------------------------------------------------------------------- + +/// Handle an inbound message from a channel (Telegram, Discord, etc.). +/// +/// Runs the agent inference loop via `web::start_chat` and sends the response +/// back to the originating channel via the REST API. +async fn handle_channel_inbound_message(data: serde_json::Value) { + let channel = match data.get("channel").and_then(|v| v.as_str()) { + Some(c) => c.to_string(), + None => { + log::warn!("[channel-inbound] channel:message missing 'channel' field"); + return; + } + }; + let message = match data.get("message").and_then(|v| v.as_str()) { + Some(m) if !m.trim().is_empty() => m.trim().to_string(), + _ => { + log::debug!("[channel-inbound] channel:message empty or missing 'message'"); + return; + } + }; + + log::info!( + "[channel-inbound] received message from channel='{}' len={}", + channel, + message.len() + ); + + let thread_id = format!("channel:{}", channel); + let client_id = "inbound".to_string(); + + // Subscribe to web channel events BEFORE starting the chat so we don't + // miss the response. + let mut event_rx = crate::openhuman::channels::providers::web::subscribe_web_channel_events(); + + // Start the agent inference loop (same mechanism as the web chat). + let request_id = match crate::openhuman::channels::providers::web::start_chat( + &client_id, &thread_id, &message, None, None, + ) + .await + { + Ok(rid) => { + log::debug!( + "[channel-inbound] agent started request_id={} thread={}", + rid, + thread_id + ); + rid + } + Err(err) => { + log::error!("[channel-inbound] start_chat failed: {}", err); + send_channel_reply( + &channel, + &format!("Sorry, I couldn't process your message: {err}"), + ) + .await; + return; + } + }; + + // Wait for the agent to finish (chat_done or chat_error matching our request_id). + let timeout = tokio::time::Duration::from_secs(180); + let deadline = tokio::time::Instant::now() + timeout; + + loop { + tokio::select! { + event = event_rx.recv() => { + match event { + Ok(ev) if ev.request_id == request_id => { + if ev.event == "chat_done" || ev.event == "chat:done" { + let reply = ev.full_response.unwrap_or_default(); + if reply.trim().is_empty() { + log::warn!("[channel-inbound] agent returned empty response"); + return; + } + log::info!( + "[channel-inbound] agent done, replying to channel='{}' len={}", + channel, + reply.len() + ); + send_channel_reply(&channel, &reply).await; + return; + } + if ev.event == "chat_error" || ev.event == "chat:error" { + let err_msg = ev.message.unwrap_or_else(|| "unknown error".to_string()); + log::error!("[channel-inbound] agent error: {}", err_msg); + send_channel_reply( + &channel, + &format!("Sorry, I encountered an error: {err_msg}"), + ) + .await; + return; + } + // Other events (tool_call, tool_result) — continue waiting + } + Ok(_) => { + // Event for a different request — skip + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + log::warn!("[channel-inbound] event bus lagged, skipped {} events", n); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + log::error!("[channel-inbound] event bus closed unexpectedly"); + return; + } + } + } + _ = tokio::time::sleep_until(deadline) => { + log::error!("[channel-inbound] agent timed out after {}s", timeout.as_secs()); + send_channel_reply(&channel, "Sorry, the request timed out.").await; + return; + } + } + } +} + +/// Send a text reply back to a channel via the backend REST API. +async fn send_channel_reply(channel: &str, text: &str) { + let config = match crate::openhuman::config::rpc::load_config_with_timeout().await { + Ok(c) => c, + Err(e) => { + log::error!("[channel-inbound] failed to load config: {}", e); + return; + } + }; + + let api_url = crate::api::config::effective_api_url(&config.api_url); + let jwt = match crate::api::jwt::get_session_token(&config) { + Ok(Some(t)) => t, + Ok(None) => { + log::error!("[channel-inbound] no session JWT — cannot reply"); + return; + } + Err(e) => { + log::error!("[channel-inbound] failed to get session token: {}", e); + return; + } + }; + + let client = match crate::api::rest::BackendOAuthClient::new(&api_url) { + Ok(c) => c, + Err(e) => { + log::error!("[channel-inbound] failed to create API client: {}", e); + return; + } + }; + + let body = json!({ "text": text }); + match client.send_channel_message(channel, &jwt, body).await { + Ok(resp) => { + log::info!( + "[channel-inbound] reply sent to channel='{}' response={:?}", + channel, + resp + ); + } + Err(e) => { + log::error!( + "[channel-inbound] failed to send reply to channel='{}': {}", + channel, + e + ); + } + } +} + /// Base64-encode a string (for webhook response bodies). /// Uses the `STANDARD` alphabet (A-Z, a-z, 0-9, +, /) with `=` padding. fn base64_encode(input: &str) -> String {