diff --git a/.github/ISSUE_TEMPLATE/bug.md b/.github/ISSUE_TEMPLATE/bug.md index 0fde91665..85c720929 100644 --- a/.github/ISSUE_TEMPLATE/bug.md +++ b/.github/ISSUE_TEMPLATE/bug.md @@ -1,9 +1,10 @@ --- name: Bug -description: Report incorrect behavior, regressions, or broken flows +about: Used for bug reports title: "[Bug] " -labels: - - bug +labels: bug, enhancement +assignees: '' + --- ## Summary diff --git a/.github/ISSUE_TEMPLATE/feature.md b/.github/ISSUE_TEMPLATE/feature.md index 68957eac4..c9732c413 100644 --- a/.github/ISSUE_TEMPLATE/feature.md +++ b/.github/ISSUE_TEMPLATE/feature.md @@ -1,9 +1,10 @@ --- name: Feature -description: Propose a new capability with tests and code documentation +about: Used for new features or suggestions title: "[Feature] " -labels: - - enhancement +labels: enhancement +assignees: '' + --- ## Summary diff --git a/app/src/AppRoutes.tsx b/app/src/AppRoutes.tsx index 6f2b8b666..4f0c33b46 100644 --- a/app/src/AppRoutes.tsx +++ b/app/src/AppRoutes.tsx @@ -10,6 +10,7 @@ import Intelligence from './pages/Intelligence'; import Invites from './pages/Invites'; import Settings from './pages/Settings'; import Skills from './pages/Skills'; +import Webhooks from './pages/Webhooks'; import Welcome from './pages/Welcome'; const AppRoutes = () => { @@ -88,6 +89,15 @@ const AppRoutes = () => { } /> + + + + } + /> + { const [apiUrl, setApiUrl] = useState(''); const [defaultModel, setDefaultModel] = useState(''); const [defaultTemp, setDefaultTemp] = useState('0.7'); - const [tunnelProvider, setTunnelProvider] = useState('none'); - const [cloudflareToken, setCloudflareToken] = useState(''); - const [ngrokToken, setNgrokToken] = useState(''); - const [tailscaleHostname, setTailscaleHostname] = useState(''); - const [customCommand, setCustomCommand] = useState(''); const [memoryBackend, setMemoryBackend] = useState('sqlite'); const [memoryAutoSave, setMemoryAutoSave] = useState(true); const [embeddingProvider, setEmbeddingProvider] = useState('none'); @@ -277,14 +270,6 @@ const TauriCommandsPanel = () => { setConfigLoaded(true); // Load other configuration sections - const tunnel = (config.tunnel as Record) ?? {}; - setTunnelProvider((tunnel.provider as string) ?? 'none'); - setCloudflareToken(((tunnel.cloudflare as Record)?.token as string) ?? ''); - setNgrokToken(((tunnel.ngrok as Record)?.auth_token as string) ?? ''); - setTailscaleHostname( - ((tunnel.tailscale as Record)?.hostname as string) ?? '' - ); - setCustomCommand(((tunnel.custom as Record)?.start_command as string) ?? ''); const memory = (config.memory as Record) ?? {}; setMemoryBackend((memory.backend as string) ?? 'sqlite'); @@ -344,22 +329,6 @@ const TauriCommandsPanel = () => { [daemonShowTray, tauriAvailable] ); - const buildTunnelConfig = (): TunnelConfig => { - if (tunnelProvider === 'cloudflare') { - return { provider: 'cloudflare', cloudflare: { token: cloudflareToken } }; - } - if (tunnelProvider === 'ngrok') { - return { provider: 'ngrok', ngrok: { auth_token: ngrokToken } }; - } - if (tunnelProvider === 'tailscale') { - return { provider: 'tailscale', tailscale: { hostname: tailscaleHostname || null } }; - } - if (tunnelProvider === 'custom') { - return { provider: 'custom', custom: { start_command: customCommand } }; - } - return { provider: 'none' }; - }; - const saveModelSettings = async () => { // Pre-save validation if (!performValidation()) { @@ -470,9 +439,6 @@ const TauriCommandsPanel = () => { } }; - const saveTunnelSettings = () => - run(() => openhumanUpdateTunnelSettings(buildTunnelConfig()), 'saveTunnelSettings'); - const saveMemorySettings = () => run( () => @@ -1023,66 +989,7 @@ const TauriCommandsPanel = () => { collapsible={true} defaultExpanded={!isCollapsed('network-infrastructure')} hasChanges={false} - loading={operationLoading?.includes('Tunnel') || operationLoading?.includes('Memory')}> -
- - - - - {tunnelProvider === 'cloudflare' && ( - - setCloudflareToken(event.target.value)} - /> - - )} - {tunnelProvider === 'ngrok' && ( - - setNgrokToken(event.target.value)} - /> - - )} - {tunnelProvider === 'tailscale' && ( - - setTailscaleHostname(event.target.value)} - /> - - )} - {tunnelProvider === 'custom' && ( - - setCustomCommand(event.target.value)} - /> - - )} - -
- + loading={operationLoading?.includes('Memory')}> @@ -1135,11 +1042,6 @@ const TauriCommandsPanel = () => { - - Save Tunnel Settings - diff --git a/app/src/components/webhooks/TunnelList.tsx b/app/src/components/webhooks/TunnelList.tsx new file mode 100644 index 000000000..37cc136e9 --- /dev/null +++ b/app/src/components/webhooks/TunnelList.tsx @@ -0,0 +1,219 @@ +import { useState } from 'react'; + +import type { Tunnel } from '../../services/api/tunnelsApi'; +import type { TunnelRegistration } from '../../store/webhooksSlice'; +import { BACKEND_URL } from '../../utils/config'; + +interface TunnelListProps { + tunnels: Tunnel[]; + registrations: TunnelRegistration[]; + loading: boolean; + onCreateTunnel: (name: string, description?: string) => Promise; + onDeleteTunnel: (id: string) => Promise; + onRefresh: () => Promise; +} + +export default function TunnelList({ + tunnels, + registrations, + loading, + onCreateTunnel, + onDeleteTunnel, + onRefresh, +}: TunnelListProps) { + const [showCreate, setShowCreate] = useState(false); + const [newName, setNewName] = useState(''); + const [newDesc, setNewDesc] = useState(''); + const [creating, setCreating] = useState(false); + const [actionError, setActionError] = useState(null); + + const handleCreate = async () => { + if (!newName.trim()) return; + setCreating(true); + setActionError(null); + try { + await onCreateTunnel(newName.trim(), newDesc.trim() || undefined); + setNewName(''); + setNewDesc(''); + setShowCreate(false); + } catch (err) { + setActionError(err instanceof Error ? err.message : 'Failed to create tunnel'); + } finally { + setCreating(false); + } + }; + + const getRegistration = (uuid: string) => registrations.find(r => r.tunnel_uuid === uuid); + + const webhookUrl = (uuid: string) => + `${BACKEND_URL || 'https://api.tinyhumans.ai'}/webhooks/${uuid}`; + + return ( +
+ {/* Header */} +
+

Webhook Tunnels

+
+ + +
+
+ + {/* Create form */} + {showCreate && ( +
+ setNewName(e.target.value)} + className="w-full px-3 py-2 text-sm border border-stone-200 rounded-lg focus:outline-none focus:ring-2 focus:ring-primary-500/30 focus:border-primary-500" + autoFocus + /> + setNewDesc(e.target.value)} + className="w-full px-3 py-2 text-sm border border-stone-200 rounded-lg focus:outline-none focus:ring-2 focus:ring-primary-500/30 focus:border-primary-500" + /> +
+ + +
+
+ )} + + {/* Error display */} + {actionError && ( +
+ {actionError} + +
+ )} + + {/* Tunnel list */} + {tunnels.length === 0 && !loading && ( +

+ No tunnels yet. Create one to receive webhook events. +

+ )} + +
+ {tunnels.map(tunnel => { + const reg = getRegistration(tunnel.uuid); + return ( + onDeleteTunnel(tunnel.id)} + onError={setActionError} + /> + ); + })} +
+
+ ); +} + +// ── Tunnel Card ─────────────────────────────────────────────────────────────── + +interface TunnelCardProps { + tunnel: Tunnel; + registration?: TunnelRegistration; + webhookUrl: string; + onDelete: () => Promise; + onError: (msg: string) => void; +} + +function TunnelCard({ tunnel, registration, webhookUrl, onDelete, onError }: TunnelCardProps) { + const [copied, setCopied] = useState(false); + const [deleting, setDeleting] = useState(false); + + const handleCopy = async () => { + try { + await navigator.clipboard.writeText(webhookUrl); + setCopied(true); + setTimeout(() => setCopied(false), 2000); + } catch { + // Clipboard may not be available in Tauri WebView + } + }; + + const handleDelete = async () => { + setDeleting(true); + try { + await onDelete(); + } catch (err) { + onError(err instanceof Error ? err.message : 'Failed to delete tunnel'); + } finally { + setDeleting(false); + } + }; + + return ( +
+
+
+
+

{tunnel.name}

+ + {tunnel.isActive ? 'Active' : 'Inactive'} + + {registration && ( + + {registration.skill_id} + + )} +
+ {tunnel.description && ( +

{tunnel.description}

+ )} +
+ + {webhookUrl} + + +
+
+ +
+
+ ); +} diff --git a/app/src/components/webhooks/WebhookActivity.tsx b/app/src/components/webhooks/WebhookActivity.tsx new file mode 100644 index 000000000..6a1cc918c --- /dev/null +++ b/app/src/components/webhooks/WebhookActivity.tsx @@ -0,0 +1,83 @@ +import type { WebhookActivityEntry } from '../../store/webhooksSlice'; + +interface WebhookActivityProps { + activity: WebhookActivityEntry[]; +} + +const METHOD_COLORS: Record = { + GET: 'text-primary-600 bg-primary-50', + POST: 'text-sage-700 bg-sage-50', + PUT: 'text-amber-700 bg-amber-50', + PATCH: 'text-amber-700 bg-amber-50', + DELETE: 'text-coral-700 bg-coral-50', +}; + +function statusColor(code: number | null): string { + if (code === null) return 'text-stone-400'; + if (code >= 200 && code < 300) return 'text-sage-600'; + if (code >= 400 && code < 500) return 'text-amber-600'; + if (code >= 500) return 'text-coral-600'; + return 'text-stone-600'; +} + +function formatTime(ts: number): string { + return new Date(ts).toLocaleTimeString(undefined, { + hour: '2-digit', + minute: '2-digit', + second: '2-digit', + }); +} + +export default function WebhookActivity({ activity }: WebhookActivityProps) { + if (activity.length === 0) { + return ( +
+

Recent Activity

+

+ No webhook activity yet. Events will appear here when webhooks are received. +

+
+ ); + } + + return ( +
+

+ Recent Activity{' '} + ({activity.length}) +

+
+ {activity.map(entry => ( +
+ + {formatTime(entry.timestamp)} + + + {entry.method} + + + {entry.path || '/'} + + + {entry.status_code ?? '---'} + + {entry.skill_id && ( + + {entry.skill_id} + + )} + + {entry.tunnel_name} + +
+ ))} +
+
+ ); +} diff --git a/app/src/hooks/useWebhooks.ts b/app/src/hooks/useWebhooks.ts new file mode 100644 index 000000000..e284306cb --- /dev/null +++ b/app/src/hooks/useWebhooks.ts @@ -0,0 +1,93 @@ +import debug from 'debug'; +import { useCallback, useEffect } from 'react'; + +import { tunnelsApi } from '../services/api/tunnelsApi'; +import { useAppDispatch, useAppSelector } from '../store/hooks'; +import { addTunnel, removeTunnel, setError, setLoading, setTunnels } from '../store/webhooksSlice'; + +const log = debug('webhooks'); + +/** + * Hook for managing webhook tunnels. + * Fetches tunnels from the backend API and provides CRUD operations. + */ +export function useWebhooks() { + const dispatch = useAppDispatch(); + const { tunnels, registrations, activity, loading, error } = useAppSelector( + state => state.webhooks + ); + const token = useAppSelector(state => state.auth.token); + + // Fetch tunnels on mount (when authenticated) + useEffect(() => { + if (!token) return; + + const fetchTunnels = async () => { + dispatch(setLoading(true)); + try { + const data = await tunnelsApi.getTunnels(); + dispatch(setTunnels(data)); + log('Fetched %d tunnels', data.length); + } catch (err) { + const msg = err instanceof Error ? err.message : 'Failed to fetch tunnels'; + dispatch(setError(msg)); + log('Error fetching tunnels: %s', msg); + } + }; + + fetchTunnels(); + }, [token, dispatch]); + + const createTunnel = useCallback( + async (name: string, description?: string) => { + try { + const tunnel = await tunnelsApi.createTunnel({ name, description }); + dispatch(addTunnel(tunnel)); + log('Created tunnel: %s (%s)', tunnel.name, tunnel.uuid); + return tunnel; + } catch (err) { + const msg = err instanceof Error ? err.message : 'Failed to create tunnel'; + dispatch(setError(msg)); + throw err; + } + }, + [dispatch] + ); + + const deleteTunnel = useCallback( + async (id: string) => { + try { + await tunnelsApi.deleteTunnel(id); + dispatch(removeTunnel(id)); + log('Deleted tunnel: %s', id); + } catch (err) { + const msg = err instanceof Error ? err.message : 'Failed to delete tunnel'; + dispatch(setError(msg)); + throw err; + } + }, + [dispatch] + ); + + const refreshTunnels = useCallback(async () => { + dispatch(setLoading(true)); + try { + const data = await tunnelsApi.getTunnels(); + dispatch(setTunnels(data)); + } catch (err) { + const msg = err instanceof Error ? err.message : 'Failed to refresh tunnels'; + dispatch(setError(msg)); + } + }, [dispatch]); + + return { + tunnels, + registrations, + activity, + loading, + error, + createTunnel, + deleteTunnel, + refreshTunnels, + }; +} diff --git a/app/src/pages/Webhooks.tsx b/app/src/pages/Webhooks.tsx new file mode 100644 index 000000000..4ee30e166 --- /dev/null +++ b/app/src/pages/Webhooks.tsx @@ -0,0 +1,48 @@ +import TunnelList from '../components/webhooks/TunnelList'; +import WebhookActivity from '../components/webhooks/WebhookActivity'; +import { useWebhooks } from '../hooks/useWebhooks'; + +export default function Webhooks() { + const { + tunnels, + registrations, + activity, + loading, + error, + createTunnel, + deleteTunnel, + refreshTunnels, + } = useWebhooks(); + + if (loading && tunnels.length === 0) { + return ( +
+
+
+
+ Loading webhooks… +
+
+
+ ); + } + + return ( +
+
+ {error &&
{error}
} + + + + +
+
+ ); +} diff --git a/app/src/services/api/tunnelsApi.ts b/app/src/services/api/tunnelsApi.ts index 548c35028..218ad4743 100644 --- a/app/src/services/api/tunnelsApi.ts +++ b/app/src/services/api/tunnelsApi.ts @@ -4,7 +4,10 @@ import { apiClient } from '../apiClient'; // ── Types ───────────────────────────────────────────────────────────────────── export interface Tunnel { + /** Internal backend ID (used for CRUD endpoints: GET/PATCH/DELETE /tunnels/:id). */ id: string; + /** External UUID used for webhook routing (appears in webhook URLs and local registrations). */ + uuid: string; name: string; description?: string; isActive: boolean; @@ -51,20 +54,20 @@ export const tunnelsApi = { return response.data; }, - /** GET /tunnels/:id — get a specific webhook tunnel */ - getTunnel: async (id: string): Promise => { - const response = await apiClient.get>(`/tunnels/${id}`); + /** GET /tunnels/:tunnelId — get a specific webhook tunnel by its internal ID. */ + getTunnel: async (tunnelId: string): Promise => { + const response = await apiClient.get>(`/tunnels/${tunnelId}`); return response.data; }, - /** PATCH /tunnels/:id — update a webhook tunnel */ - updateTunnel: async (id: string, body: UpdateTunnelRequest): Promise => { - const response = await apiClient.patch>(`/tunnels/${id}`, body); + /** PATCH /tunnels/:tunnelId — update a webhook tunnel by its internal ID. */ + updateTunnel: async (tunnelId: string, body: UpdateTunnelRequest): Promise => { + const response = await apiClient.patch>(`/tunnels/${tunnelId}`, body); return response.data; }, - /** DELETE /tunnels/:id — delete a webhook tunnel */ - deleteTunnel: async (id: string): Promise => { - await apiClient.delete>(`/tunnels/${id}`); + /** DELETE /tunnels/:tunnelId — delete a webhook tunnel by its internal ID. */ + deleteTunnel: async (tunnelId: string): Promise => { + await apiClient.delete>(`/tunnels/${tunnelId}`); }, }; diff --git a/app/src/services/coreRpcClient.ts b/app/src/services/coreRpcClient.ts index 80693e899..a1873c301 100644 --- a/app/src/services/coreRpcClient.ts +++ b/app/src/services/coreRpcClient.ts @@ -41,7 +41,6 @@ const LEGACY_METHOD_ALIASES: Record = { 'openhuman.update_runtime_settings': 'openhuman.config_update_runtime_settings', 'openhuman.update_screen_intelligence_settings': 'openhuman.config_update_screen_intelligence_settings', - 'openhuman.update_tunnel_settings': 'openhuman.config_update_tunnel_settings', 'openhuman.workspace_onboarding_flag_exists': 'openhuman.config_workspace_onboarding_flag_exists', 'openhuman.workspace_onboarding_flag_set': 'openhuman.config_workspace_onboarding_flag_set', }; diff --git a/app/src/store/index.ts b/app/src/store/index.ts index ca3fa9def..d07c6213b 100644 --- a/app/src/store/index.ts +++ b/app/src/store/index.ts @@ -29,6 +29,7 @@ import socketReducer from './socketSlice'; import teamReducer from './teamSlice'; import threadReducer from './threadSlice'; import userReducer from './userSlice'; +import webhooksReducer from './webhooksSlice'; // Persist config for auth only const authPersistConfig = { @@ -132,6 +133,7 @@ export const store = configureStore({ invite: inviteReducer, accessibility: accessibilityReducer, channelConnections: persistedChannelConnectionsReducer, + webhooks: webhooksReducer, }, middleware: getDefaultMiddleware => { const middleware = getDefaultMiddleware({ diff --git a/app/src/store/webhooksSlice.ts b/app/src/store/webhooksSlice.ts new file mode 100644 index 000000000..306446b24 --- /dev/null +++ b/app/src/store/webhooksSlice.ts @@ -0,0 +1,92 @@ +import { createSlice, type PayloadAction } from '@reduxjs/toolkit'; + +import type { Tunnel } from '../services/api/tunnelsApi'; + +// ── Types ───────────────────────────────────────────────────────────────────── + +/** Local tunnel-to-skill registration (from the Rust core WebhookRouter). */ +export interface TunnelRegistration { + tunnel_uuid: string; + skill_id: string; + tunnel_name: string | null; + backend_tunnel_id: string | null; +} + +/** Entry in the webhook activity log. */ +export interface WebhookActivityEntry { + correlation_id: string; + tunnel_name: string; + method: string; + path: string; + status_code: number | null; + skill_id: string | null; + timestamp: number; +} + +interface WebhooksState { + /** Tunnels from the backend API. */ + tunnels: Tunnel[]; + /** Local tunnel-to-skill registrations from the Rust core. */ + registrations: TunnelRegistration[]; + /** Recent webhook activity (ring buffer, newest first). */ + activity: WebhookActivityEntry[]; + loading: boolean; + error: string | null; +} + +// ── Slice ───────────────────────────────────────────────────────────────────── + +const MAX_ACTIVITY_ENTRIES = 100; + +const initialState: WebhooksState = { + tunnels: [], + registrations: [], + activity: [], + loading: false, + error: null, +}; + +const webhooksSlice = createSlice({ + name: 'webhooks', + initialState, + reducers: { + setTunnels: (state, action: PayloadAction) => { + state.tunnels = action.payload; + state.loading = false; + state.error = null; + }, + addTunnel: (state, action: PayloadAction) => { + state.tunnels.push(action.payload); + }, + removeTunnel: (state, action: PayloadAction) => { + state.tunnels = state.tunnels.filter(t => t.id !== action.payload); + }, + setRegistrations: (state, action: PayloadAction) => { + state.registrations = action.payload; + }, + addActivity: (state, action: PayloadAction) => { + state.activity.unshift(action.payload); + if (state.activity.length > MAX_ACTIVITY_ENTRIES) { + state.activity.splice(MAX_ACTIVITY_ENTRIES); + } + }, + setLoading: (state, action: PayloadAction) => { + state.loading = action.payload; + }, + setError: (state, action: PayloadAction) => { + state.error = action.payload; + state.loading = false; + }, + }, +}); + +export const { + setTunnels, + addTunnel, + removeTunnel, + setRegistrations, + addActivity, + setLoading, + setError, +} = webhooksSlice.actions; +export default webhooksSlice.reducer; diff --git a/app/src/utils/tauriCommands.ts b/app/src/utils/tauriCommands.ts index 9c37891ae..b7d32d4d0 100644 --- a/app/src/utils/tauriCommands.ts +++ b/app/src/utils/tauriCommands.ts @@ -984,18 +984,6 @@ function tauriErrorMessage(err: unknown): string { return 'Unknown Tauri invoke error'; } -export interface TunnelConfig { - provider: string; - cloudflare?: { token: string } | null; - tailscale?: { funnel?: boolean; hostname?: string | null } | null; - ngrok?: { auth_token: string; domain?: string | null } | null; - custom?: { - start_command: string; - health_url?: string | null; - url_pattern?: string | null; - } | null; -} - export async function openhumanGetConfig(): Promise> { if (!isTauri()) { throw new Error('Not running in Tauri'); @@ -1027,18 +1015,6 @@ export async function openhumanUpdateMemorySettings( }); } -export async function openhumanUpdateTunnelSettings( - tunnel: TunnelConfig -): Promise> { - if (!isTauri()) { - throw new Error('Not running in Tauri'); - } - return await callCoreRpc>({ - method: 'openhuman.update_tunnel_settings', - params: tunnel, - }); -} - export async function openhumanUpdateRuntimeSettings( update: RuntimeSettingsUpdate ): Promise> { diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 0f86fe9fe..70f3c918c 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -87,6 +87,176 @@ pub fn default_state() -> AppState { // --- HTTP server (Axum) ---------------------------------------------------- +#[derive(Debug, serde::Deserialize)] +struct TelegramAuthQuery { + token: Option, +} + +fn success_html() -> String { + r#" + + + + + OpenHuman — Connected + + + +
+
✔
+

Connected!

+

Your Telegram account has been connected to OpenHuman. You can close this tab.

+
+ +"# + .to_string() +} + +fn escape_html(s: &str) -> String { + s.replace('&', "&") + .replace('<', "<") + .replace('>', ">") + .replace('"', """) + .replace('\'', "'") +} + +fn error_html(message: &str) -> String { + let escaped_message = escape_html(message); + format!( + r#" + + + + + OpenHuman — Error + + + +
+
⚠
+

Something went wrong

+

{escaped_message}

+
+ +"# + ) +} + +async fn telegram_auth_handler(Query(query): Query) -> impl IntoResponse { + let html_response = |status: StatusCode, body: String| -> Response { + ( + status, + [(header::CONTENT_TYPE, "text/html; charset=utf-8")], + body, + ) + .into_response() + }; + + let token = match query + .token + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + { + Some(t) => t.to_string(), + None => { + return html_response( + StatusCode::BAD_REQUEST, + error_html("Missing token parameter. Send /start register to the bot again."), + ) + } + }; + + log::info!("[auth:telegram] Received registration callback with token"); + + let config = match crate::openhuman::config::Config::load_or_init().await { + Ok(c) => c, + Err(e) => { + log::error!("[auth:telegram] Failed to load config: {e}"); + return html_response( + StatusCode::INTERNAL_SERVER_ERROR, + error_html("Internal error. Please try again."), + ); + } + }; + + let api_url = crate::api::config::effective_api_url(&config.api_url); + + let client = match crate::api::rest::BackendOAuthClient::new(&api_url) { + Ok(c) => c, + Err(e) => { + log::error!("[auth:telegram] Failed to create API client: {e}"); + return html_response( + StatusCode::INTERNAL_SERVER_ERROR, + error_html("Internal error. Please try again."), + ); + } + }; + + let jwt_token = match client.consume_login_token(&token).await { + Ok(jwt) => jwt, + Err(e) => { + let error_str = e.to_string(); + // Check if this is a client-side error (token validation) or server-side error + let is_client_error = error_str.contains("expired") + || error_str.contains("invalid") + || error_str.contains("not found") + || error_str.contains("already used") + || error_str.contains("401") + || error_str.contains("400") + || error_str.contains("404"); + + if is_client_error { + log::warn!("[auth:telegram] Token consumption failed (client error): {e}"); + return html_response( + StatusCode::BAD_REQUEST, + error_html( + "This link has expired or was already used. Send /start register to the bot again.", + ), + ); + } else { + log::error!("[auth:telegram] Token consumption failed (server error): {e}"); + return html_response( + StatusCode::INTERNAL_SERVER_ERROR, + error_html("Internal server error, please try again later."), + ); + } + } + }; + + match crate::openhuman::credentials::ops::store_session(&config, &jwt_token, None, None).await { + Ok(outcome) => { + for msg in &outcome.logs { + log::info!("[auth:telegram] {msg}"); + } + log::info!("[auth:telegram] Session stored successfully"); + } + Err(e) => { + log::error!("[auth:telegram] Failed to store session: {e}"); + return html_response( + StatusCode::INTERNAL_SERVER_ERROR, + error_html("Connected to Telegram but failed to save session. Please try again."), + ); + } + } + + html_response(StatusCode::OK, success_html()) +} + pub fn build_core_http_router(socketio_enabled: bool) -> Router { let router = Router::new() .route("/", get(root_handler)) @@ -94,6 +264,7 @@ pub fn build_core_http_router(socketio_enabled: bool) -> Router { .route("/schema", get(schema_handler)) .route("/events", get(events_handler)) .route("/rpc", post(rpc_handler)) + .route("/auth/telegram", get(telegram_auth_handler)) .fallback(not_found_handler) .layer(middleware::from_fn(http_request_log_middleware)) .layer(middleware::from_fn(cors_middleware)) diff --git a/src/openhuman/agent/agent.rs b/src/openhuman/agent/agent.rs index 942874ad5..b850bf207 100644 --- a/src/openhuman/agent/agent.rs +++ b/src/openhuman/agent/agent.rs @@ -1,6 +1,7 @@ use super::dispatcher::{ NativeToolDispatcher, ParsedToolCall, ToolDispatcher, ToolExecutionResult, XmlToolDispatcher, }; +use super::hooks::{self, sanitize_tool_output, PostTurnHook, ToolCallRecord, TurnContext}; use super::memory_loader::{DefaultMemoryLoader, MemoryLoader}; use super::prompt::{PromptContext, SystemPromptBuilder}; use crate::openhuman::agent::host_runtime; @@ -34,6 +35,8 @@ pub struct Agent { history: Vec, classification_config: crate::openhuman::config::QueryClassificationConfig, available_hints: Vec, + post_turn_hooks: Vec>, + learning_enabled: bool, } pub struct AgentBuilder { @@ -52,6 +55,8 @@ pub struct AgentBuilder { auto_save: Option, classification_config: Option, available_hints: Option>, + post_turn_hooks: Vec>, + learning_enabled: bool, } impl Default for AgentBuilder { @@ -78,6 +83,8 @@ impl AgentBuilder { auto_save: None, classification_config: None, available_hints: None, + post_turn_hooks: Vec::new(), + learning_enabled: false, } } @@ -162,6 +169,16 @@ impl AgentBuilder { self } + pub fn post_turn_hooks(mut self, hooks: Vec>) -> Self { + self.post_turn_hooks = hooks; + self + } + + pub fn learning_enabled(mut self, enabled: bool) -> Self { + self.learning_enabled = enabled; + self + } + pub fn build(self) -> Result { let tools = self .tools @@ -198,6 +215,8 @@ impl AgentBuilder { history: Vec::new(), classification_config: self.classification_config.unwrap_or_default(), available_hints: self.available_hints.unwrap_or_default(), + post_turn_hooks: self.post_turn_hooks, + learning_enabled: self.learning_enabled, }) } } @@ -317,6 +336,70 @@ impl Agent { let available_hints: Vec = config.model_routes.iter().map(|r| r.hint.clone()).collect(); + // Build prompt builder, optionally with learning sections + let mut prompt_builder = SystemPromptBuilder::with_defaults(); + if config.learning.enabled { + prompt_builder = prompt_builder + .add_section(Box::new( + crate::openhuman::learning::LearnedContextSection::new(memory.clone()), + )) + .add_section(Box::new( + crate::openhuman::learning::UserProfileSection::new(memory.clone()), + )); + log::info!("[learning] prompt sections registered (learned_context, user_profile)"); + } + + // Build post-turn hooks when learning is enabled + let mut post_turn_hooks: Vec> = Vec::new(); + if config.learning.enabled { + let full_config = Arc::new(config.clone()); + + if config.learning.reflection_enabled { + // For cloud reflection, wrap the provider in an Arc. + // For local, no provider needed. + let reflection_provider: Option> = + if config.learning.reflection_source + == crate::openhuman::config::ReflectionSource::Cloud + { + Some(Arc::from(providers::create_routed_provider( + config.api_key.as_deref(), + config.api_url.as_deref(), + &config.reliability, + &config.model_routes, + &model_name, + )?)) + } else { + None + }; + post_turn_hooks.push(Arc::new(crate::openhuman::learning::ReflectionHook::new( + config.learning.clone(), + full_config.clone(), + memory.clone(), + reflection_provider, + ))); + log::info!( + "[learning] reflection hook registered (source={:?})", + config.learning.reflection_source + ); + } + + if config.learning.user_profile_enabled { + post_turn_hooks.push(Arc::new(crate::openhuman::learning::UserProfileHook::new( + config.learning.clone(), + memory.clone(), + ))); + log::info!("[learning] user_profile hook registered"); + } + + if config.learning.tool_tracking_enabled { + post_turn_hooks.push(Arc::new(crate::openhuman::learning::ToolTrackerHook::new( + config.learning.clone(), + memory.clone(), + ))); + log::info!("[learning] tool_tracker hook registered"); + } + } + Agent::builder() .provider(provider) .tools(tools) @@ -326,7 +409,7 @@ impl Agent { 5, config.memory.min_relevance_score, ))) - .prompt_builder(SystemPromptBuilder::with_defaults()) + .prompt_builder(prompt_builder) .config(config.agent.clone()) .model_name(model_name) .temperature(config.default_temperature) @@ -336,6 +419,8 @@ impl Agent { .identity_config(config.identity.clone()) .skills(crate::openhuman::skills::load_skills(&config.workspace_dir)) .auto_save(config.memory.auto_save) + .post_turn_hooks(post_turn_hooks) + .learning_enabled(config.learning.enabled) .build() } @@ -366,7 +451,62 @@ impl Agent { self.history.extend(other_messages); } - fn build_system_prompt(&self) -> Result { + /// Pre-fetch learned context data from memory (async, non-blocking). + async fn fetch_learned_context(&self) -> crate::openhuman::agent::prompt::LearnedContextData { + use crate::openhuman::agent::prompt::LearnedContextData; + + if !self.learning_enabled { + return LearnedContextData::default(); + } + + let obs_entries = self + .memory + .list( + Some(&MemoryCategory::Custom("learning_observations".into())), + None, + ) + .await + .unwrap_or_default(); + + let pat_entries = self + .memory + .list( + Some(&MemoryCategory::Custom("learning_patterns".into())), + None, + ) + .await + .unwrap_or_default(); + + let profile_entries = self + .memory + .list(Some(&MemoryCategory::Custom("user_profile".into())), None) + .await + .unwrap_or_default(); + + LearnedContextData { + observations: obs_entries + .iter() + .rev() + .take(5) + .map(|e| sanitize_learned_entry(&e.content)) + .collect(), + patterns: pat_entries + .iter() + .take(3) + .map(|e| sanitize_learned_entry(&e.content)) + .collect(), + user_profile: profile_entries + .iter() + .take(20) + .map(|e| sanitize_learned_entry(&e.content)) + .collect(), + } + } + + fn build_system_prompt( + &self, + learned: crate::openhuman::agent::prompt::LearnedContextData, + ) -> Result { let instructions = self.tool_dispatcher.prompt_instructions(&self.tools); let ctx = PromptContext { workspace_dir: &self.workspace_dir, @@ -375,58 +515,103 @@ impl Agent { skills: &self.skills, identity_config: Some(&self.identity_config), dispatcher_instructions: &instructions, + learned, }; self.prompt_builder.build(&ctx) } - async fn execute_tool_call(&self, call: &ParsedToolCall) -> ToolExecutionResult { - let started = std::time::Instant::now(); - log::info!("[agent_loop] tool start name={}", call.name); - let result = if let Some(tool) = self.tools.iter().find(|t| t.name() == call.name) { - match tool.execute(call.arguments.clone()).await { - Ok(r) => { - if r.success { - r.output - } else { - format!("Error: {}", r.error.unwrap_or(r.output)) - } - } - Err(e) => { - format!("Error executing {}: {e}", call.name) - } - } + /// Sanitize tool output to prevent PII/secrets in learning data. + /// Returns a safe metadata string: tool type, status, and error class. + fn sanitize_tool_output(raw_output: &str, tool_name: &str, success: bool) -> String { + if success { + // For successful calls, return a structured summary without raw data + let char_count = raw_output.chars().count(); + format!( + "tool={} status=success output_length={}", + tool_name, char_count + ) } else { - format!("Unknown tool: {}", call.name) - }; - log::info!( - "[agent_loop] tool finish name={} elapsed_ms={} output_chars={}", - call.name, - started.elapsed().as_millis(), - result.chars().count() - ); - - ToolExecutionResult { - name: call.name.clone(), - output: result, - success: true, - tool_call_id: call.tool_call_id.clone(), + // For errors, classify the error type without exposing details + let error_class = if raw_output.contains("permission") || raw_output.contains("denied") + { + "permission_error" + } else if raw_output.contains("not found") || raw_output.contains("404") { + "not_found" + } else if raw_output.contains("timeout") || raw_output.contains("timed out") { + "timeout" + } else if raw_output.contains("network") || raw_output.contains("connection") { + "network_error" + } else if raw_output.contains("invalid") || raw_output.contains("parse") { + "validation_error" + } else { + "unknown_error" + }; + format!("tool={} status=error class={}", tool_name, error_class) } } - async fn execute_tools(&self, calls: &[ParsedToolCall]) -> Vec { - if !self.config.parallel_tools { - let mut results = Vec::with_capacity(calls.len()); - for call in calls { - results.push(self.execute_tool_call(call).await); - } - return results; - } + async fn execute_tool_call( + &self, + call: &ParsedToolCall, + ) -> (ToolExecutionResult, ToolCallRecord) { + let started = std::time::Instant::now(); + log::info!("[agent_loop] tool start name={}", call.name); + let (result, success) = + if let Some(tool) = self.tools.iter().find(|t| t.name() == call.name) { + match tool.execute(call.arguments.clone()).await { + Ok(r) => { + if r.success { + (r.output, true) + } else { + (format!("Error: {}", r.error.unwrap_or(r.output)), false) + } + } + Err(e) => (format!("Error executing {}: {e}", call.name), false), + } + } else { + (format!("Unknown tool: {}", call.name), false) + }; + let elapsed_ms = started.elapsed().as_millis() as u64; + log::info!( + "[agent_loop] tool finish name={} elapsed_ms={} output_chars={} success={}", + call.name, + elapsed_ms, + result.chars().count(), + success + ); + let output_summary = sanitize_tool_output(&result, &call.name, success); + + let record = ToolCallRecord { + name: call.name.clone(), + arguments: call.arguments.clone(), + success, + output_summary, + duration_ms: elapsed_ms, + }; + + let exec_result = ToolExecutionResult { + name: call.name.clone(), + output: result, + success, + tool_call_id: call.tool_call_id.clone(), + }; + + (exec_result, record) + } + + async fn execute_tools( + &self, + calls: &[ParsedToolCall], + ) -> (Vec, Vec) { let mut results = Vec::with_capacity(calls.len()); + let mut records = Vec::with_capacity(calls.len()); for call in calls { - results.push(self.execute_tool_call(call).await); + let (exec_result, record) = self.execute_tool_call(call).await; + results.push(exec_result); + records.push(record); } - results + (results, records) } fn classify_model(&self, user_message: &str) -> String { @@ -440,14 +625,18 @@ impl Agent { } pub async fn turn(&mut self, user_message: &str) -> Result { + let turn_started = std::time::Instant::now(); log::info!( "[agent_loop] turn start message_chars={} history_len={} max_tool_iterations={}", user_message.chars().count(), self.history.len(), self.config.max_tool_iterations ); + // Pre-fetch learned context async before building the system prompt + let learned = self.fetch_learned_context().await; + if self.history.is_empty() { - let system_prompt = self.build_system_prompt()?; + let system_prompt = self.build_system_prompt(learned)?; log::info!( "[agent_loop] system prompt built chars={} content=\n{}", system_prompt.chars().count(), @@ -457,6 +646,15 @@ impl Agent { .push(ConversationMessage::Chat(ChatMessage::system( system_prompt, ))); + } else if self.learning_enabled { + // Rebuild system prompt on subsequent turns to include newly learned context + let system_prompt = self.build_system_prompt(learned)?; + if let Some(pos) = self.history.iter().position( + |msg| matches!(msg, ConversationMessage::Chat(chat) if chat.role == "system"), + ) { + self.history[pos] = ConversationMessage::Chat(ChatMessage::system(system_prompt)); + log::debug!("[agent_loop] system prompt refreshed with learned context"); + } } if self.auto_save { @@ -484,6 +682,9 @@ impl Agent { let effective_model = self.classify_model(user_message); log::info!("[agent_loop] model selected model={}", effective_model); + // Collect tool call records across all iterations for post-turn hooks + let mut all_tool_records: Vec = Vec::new(); + for iteration in 0..self.config.max_tool_iterations { log::info!( "[agent_loop] iteration start i={} history_len={}", @@ -562,6 +763,19 @@ impl Agent { .await; } + // Fire post-turn hooks (non-blocking) + if !self.post_turn_hooks.is_empty() { + let ctx = TurnContext { + user_message: user_message.to_string(), + assistant_response: final_text.clone(), + tool_calls: all_tool_records, + turn_duration_ms: turn_started.elapsed().as_millis() as u64, + session_id: None, + iteration_count: iteration + 1, + }; + hooks::fire_hooks(&self.post_turn_hooks, ctx); + } + return Ok(final_text); } @@ -601,7 +815,8 @@ impl Agent { tool_calls: persisted_tool_calls, }); - let results = self.execute_tools(&calls).await; + let (results, records) = self.execute_tools(&calls).await; + all_tool_records.extend(records); log::info!( "[agent_loop] tool results complete i={} result_count={}", iteration + 1, @@ -682,6 +897,27 @@ pub async fn run( Ok(()) } +/// Sanitize a learned memory entry before injecting into the system prompt. +/// Strips raw data, limits length, and removes potential secrets. +fn sanitize_learned_entry(content: &str) -> String { + let trimmed = content.trim(); + if trimmed.is_empty() { + return String::new(); + } + // Truncate to a safe length + let max_len = 200; + let sanitized: String = trimmed.chars().take(max_len).collect(); + // Strip anything that looks like a secret/token + if sanitized.contains("Bearer ") + || sanitized.contains("sk-") + || sanitized.contains("ghp_") + || sanitized.contains("-----BEGIN") + { + return "[redacted: potential secret]".to_string(); + } + sanitized +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/openhuman/agent/hooks.rs b/src/openhuman/agent/hooks.rs new file mode 100644 index 000000000..71bfe47e8 --- /dev/null +++ b/src/openhuman/agent/hooks.rs @@ -0,0 +1,116 @@ +//! Post-turn hook infrastructure for agent self-learning. +//! +//! Hooks fire asynchronously after a turn completes, receiving a snapshot of +//! what happened (user message, assistant response, tool calls with outcomes). +//! The agent does not wait for hooks — they run in the background via `tokio::spawn`. + +use async_trait::async_trait; +use serde::{Deserialize, Serialize}; +use std::sync::Arc; + +/// Snapshot of a completed agent turn, passed to every registered hook. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TurnContext { + pub user_message: String, + pub assistant_response: String, + pub tool_calls: Vec, + pub turn_duration_ms: u64, + pub session_id: Option, + pub iteration_count: usize, +} + +/// Record of a single tool invocation within a turn. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ToolCallRecord { + pub name: String, + pub arguments: serde_json::Value, + pub success: bool, + /// Sanitized, non-sensitive summary (tool type, status/error class, safe message). + /// Never contains raw tool output or PII. + pub output_summary: String, + pub duration_ms: u64, +} + +/// Produce a safe, non-sensitive summary of a tool result for learning records. +/// +/// Strips raw payloads, file contents, API responses, and credentials — returns +/// only the tool name, status, error class (if failed), and a short length hint. +pub fn sanitize_tool_output(output: &str, tool_name: &str, success: bool) -> String { + if success { + let char_count = output.chars().count(); + return format!("{tool_name}: ok ({char_count} chars)"); + } + + // For failures, extract a safe error class without raw payload + let lower = output.to_lowercase(); + let error_class = if lower.contains("timeout") { + "timeout" + } else if lower.contains("not found") || lower.contains("no such file") { + "not_found" + } else if lower.contains("permission") || lower.contains("denied") { + "permission_denied" + } else if lower.contains("connection") || lower.contains("network") { + "connection_error" + } else if lower.contains("parse") || lower.contains("invalid") || lower.contains("syntax") { + "parse_error" + } else if lower.contains("unknown tool") { + "unknown_tool" + } else { + "error" + }; + + format!("{tool_name}: failed ({error_class})") +} + +/// Trait for post-turn hooks that react to completed turns. +/// +/// Implementations must be cheap to clone (wrapped in `Arc`) and safe to call +/// concurrently from multiple `tokio::spawn` tasks. +#[async_trait] +pub trait PostTurnHook: Send + Sync { + /// Human-readable name for logging. + fn name(&self) -> &str; + + /// Called after the agent produces a final response. + /// Errors are logged but do not propagate to the caller. + async fn on_turn_complete(&self, ctx: &TurnContext) -> anyhow::Result<()>; +} + +/// Fire all hooks in parallel, logging errors without blocking the caller. +pub fn fire_hooks(hooks: &[Arc], ctx: TurnContext) { + log::debug!( + "[learning] dispatching {} post-turn hook(s) (tool_calls={}, response_chars={})", + hooks.len(), + ctx.tool_calls.len(), + ctx.assistant_response.chars().count() + ); + for (idx, hook) in hooks.iter().enumerate() { + let hook = Arc::clone(hook); + let ctx = ctx.clone(); + log::trace!( + "[learning] scheduling hook {}/{}: '{}'", + idx + 1, + hooks.len(), + hook.name() + ); + tokio::spawn(async move { + let started = std::time::Instant::now(); + match hook.on_turn_complete(&ctx).await { + Ok(()) => { + log::debug!( + "[learning] hook '{}' completed in {}ms", + hook.name(), + started.elapsed().as_millis() + ); + } + Err(e) => { + log::warn!( + "[learning] hook '{}' failed after {}ms: {e:#}", + hook.name(), + started.elapsed().as_millis() + ); + } + } + }); + } +} diff --git a/src/openhuman/agent/mod.rs b/src/openhuman/agent/mod.rs index 98d0807f3..9854ba10f 100644 --- a/src/openhuman/agent/mod.rs +++ b/src/openhuman/agent/mod.rs @@ -2,6 +2,7 @@ pub mod agent; pub mod classifier; pub mod dispatcher; +pub mod hooks; pub mod host_runtime; pub mod identity; pub mod loop_; diff --git a/src/openhuman/agent/prompt.rs b/src/openhuman/agent/prompt.rs index 06283b098..cd3156615 100644 --- a/src/openhuman/agent/prompt.rs +++ b/src/openhuman/agent/prompt.rs @@ -9,6 +9,17 @@ use std::path::Path; const BOOTSTRAP_MAX_CHARS: usize = 20_000; +/// Pre-fetched learned context data for prompt sections (avoids blocking the runtime). +#[derive(Debug, Clone, Default)] +pub struct LearnedContextData { + /// Recent observations from the learning subsystem. + pub observations: Vec, + /// Recognized patterns. + pub patterns: Vec, + /// Learned user profile entries. + pub user_profile: Vec, +} + pub struct PromptContext<'a> { pub workspace_dir: &'a Path, pub model_name: &'a str, @@ -16,6 +27,8 @@ pub struct PromptContext<'a> { pub skills: &'a [Skill], pub identity_config: Option<&'a IdentityConfig>, pub dispatcher_instructions: &'a str, + /// Pre-fetched learned context (empty when learning is disabled). + pub learned: LearnedContextData, } pub trait PromptSection: Send + Sync { @@ -339,6 +352,7 @@ mod tests { skills: &[], identity_config: Some(&identity_config), dispatcher_instructions: "", + learned: LearnedContextData::default(), }; let section = IdentitySection; @@ -366,6 +380,7 @@ mod tests { skills: &[], identity_config: None, dispatcher_instructions: "instr", + learned: LearnedContextData::default(), }; let prompt = SystemPromptBuilder::with_defaults().build(&ctx).unwrap(); assert!(prompt.contains("## Tools")); @@ -387,6 +402,7 @@ mod tests { skills: &[], identity_config: None, dispatcher_instructions: "", + learned: LearnedContextData::default(), }; let section = IdentitySection; @@ -426,6 +442,7 @@ mod tests { skills: &[], identity_config: None, dispatcher_instructions: "instr", + learned: LearnedContextData::default(), }; let rendered = DateTimeSection.build(&ctx).unwrap(); diff --git a/src/openhuman/config/mod.rs b/src/openhuman/config/mod.rs index 9f5145d03..75b04c174 100644 --- a/src/openhuman/config/mod.rs +++ b/src/openhuman/config/mod.rs @@ -14,16 +14,16 @@ pub use schema::{ apply_runtime_proxy_to_builder, build_runtime_proxy_client, build_runtime_proxy_client_with_timeouts, runtime_proxy_config, set_runtime_proxy_config, AgentConfig, AuditConfig, AutocompleteConfig, AutonomyConfig, BrowserComputerUseConfig, - BrowserConfig, ChannelsConfig, ClassificationRule, CloudflareTunnelConfig, ComposioConfig, - Config, CostConfig, CronConfig, CustomTunnelConfig, DelegateAgentConfig, DiscordConfig, - DockerRuntimeConfig, EmbeddingRouteConfig, HardwareConfig, HardwareTransport, HeartbeatConfig, - HttpRequestConfig, IMessageConfig, IdentityConfig, LarkConfig, LocalAiConfig, MatrixConfig, - MemoryConfig, ModelRouteConfig, MultimodalConfig, NgrokTunnelConfig, ObservabilityConfig, - PeripheralBoardConfig, PeripheralsConfig, ProxyConfig, ProxyScope, QueryClassificationConfig, + BrowserConfig, ChannelsConfig, ClassificationRule, ComposioConfig, Config, CostConfig, + CronConfig, DelegateAgentConfig, DiscordConfig, DockerRuntimeConfig, EmbeddingRouteConfig, + HardwareConfig, HardwareTransport, HeartbeatConfig, HttpRequestConfig, IMessageConfig, + IdentityConfig, LarkConfig, LearningConfig, LocalAiConfig, MatrixConfig, MemoryConfig, + ModelRouteConfig, MultimodalConfig, ObservabilityConfig, PeripheralBoardConfig, + PeripheralsConfig, ProxyConfig, ProxyScope, QueryClassificationConfig, ReflectionSource, ReliabilityConfig, ResourceLimitsConfig, RuntimeConfig, SandboxBackend, SandboxConfig, SchedulerConfig, ScreenIntelligenceConfig, SecretsConfig, SecurityConfig, SlackConfig, - StorageConfig, StorageProviderConfig, StorageProviderSection, StreamMode, - TailscaleTunnelConfig, TelegramConfig, TunnelConfig, WebSearchConfig, WebhookConfig, + StorageConfig, StorageProviderConfig, StorageProviderSection, StreamMode, TelegramConfig, + WebSearchConfig, WebhookConfig, }; pub use schemas::{ all_controller_schemas as all_config_controller_schemas, diff --git a/src/openhuman/config/ops.rs b/src/openhuman/config/ops.rs index 21dbacb58..83d17c3a7 100644 --- a/src/openhuman/config/ops.rs +++ b/src/openhuman/config/ops.rs @@ -5,7 +5,7 @@ use std::path::PathBuf; use serde::Serialize; use serde_json::json; -use crate::openhuman::config::{Config, TunnelConfig}; +use crate::openhuman::config::Config; use crate::openhuman::screen_intelligence; use crate::rpc::RpcOutcome; @@ -223,22 +223,6 @@ pub async fn apply_screen_intelligence_settings( )) } -pub async fn apply_tunnel_settings( - config: &mut Config, - tunnel: TunnelConfig, -) -> Result, String> { - config.tunnel = tunnel; - config.save().await.map_err(|e| e.to_string())?; - let snapshot = snapshot_config_json(config)?; - Ok(RpcOutcome::new( - snapshot, - vec![format!( - "tunnel settings saved to {}", - config.config_path.display() - )], - )) -} - pub async fn apply_runtime_settings( config: &mut Config, update: RuntimeSettingsPatch, @@ -304,13 +288,6 @@ pub async fn load_and_apply_screen_intelligence_settings( apply_screen_intelligence_settings(&mut config, update).await } -pub async fn load_and_apply_tunnel_settings( - tunnel: TunnelConfig, -) -> Result, String> { - let mut config = load_config_with_timeout().await?; - apply_tunnel_settings(&mut config, tunnel).await -} - pub async fn load_and_apply_runtime_settings( update: RuntimeSettingsPatch, ) -> Result, String> { diff --git a/src/openhuman/config/schema/learning.rs b/src/openhuman/config/schema/learning.rs new file mode 100644 index 000000000..4fa130076 --- /dev/null +++ b/src/openhuman/config/schema/learning.rs @@ -0,0 +1,84 @@ +//! Self-learning configuration — reflection, user profiling, tool tracking. + +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +/// Which LLM to use for reflection inference. +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ReflectionSource { + /// Use the local Ollama model via `LocalAiService::prompt()`. + /// Model is determined by `config.local_ai.chat_model_id`. + Local, + /// Use the cloud reasoning model via `Provider::simple_chat("hint:reasoning")`. + Cloud, +} + +impl Default for ReflectionSource { + fn default() -> Self { + Self::Local + } +} + +/// Configuration for the agent self-learning subsystem. +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct LearningConfig { + /// Master switch. Default: false. + #[serde(default)] + pub enabled: bool, + + /// Enable post-turn reflection (observation extraction). Default: true when learning is enabled. + #[serde(default = "default_true")] + pub reflection_enabled: bool, + + /// Enable automatic user profile extraction. Default: true when learning is enabled. + #[serde(default = "default_true")] + pub user_profile_enabled: bool, + + /// Enable tool effectiveness tracking. Default: true when learning is enabled. + #[serde(default = "default_true")] + pub tool_tracking_enabled: bool, + + /// Enable autonomous skill creation from experience. Default: false (Phase 5). + #[serde(default)] + pub skill_creation_enabled: bool, + + /// Which LLM to use for reflection. Default: local (Ollama). + #[serde(default)] + pub reflection_source: ReflectionSource, + + /// Maximum reflections per session before throttling. Default: 20. + #[serde(default = "default_max_reflections")] + pub max_reflections_per_session: usize, + + /// Minimum tool calls in a turn to trigger reflection. Default: 1. + #[serde(default = "default_min_turn_complexity")] + pub min_turn_complexity: usize, +} + +fn default_true() -> bool { + true +} + +fn default_max_reflections() -> usize { + 20 +} + +fn default_min_turn_complexity() -> usize { + 1 +} + +impl Default for LearningConfig { + fn default() -> Self { + Self { + enabled: false, + reflection_enabled: default_true(), + user_profile_enabled: default_true(), + tool_tracking_enabled: default_true(), + skill_creation_enabled: false, + reflection_source: ReflectionSource::default(), + max_reflections_per_session: default_max_reflections(), + min_turn_complexity: default_min_turn_complexity(), + } + } +} diff --git a/src/openhuman/config/schema/load.rs b/src/openhuman/config/schema/load.rs index 92dd2e0bf..5e17abc9e 100644 --- a/src/openhuman/config/schema/load.rs +++ b/src/openhuman/config/schema/load.rs @@ -611,11 +611,156 @@ impl Config { } } + // Learning subsystem overrides + if let Ok(flag) = std::env::var("OPENHUMAN_LEARNING_ENABLED") { + let normalized = flag.trim().to_ascii_lowercase(); + match normalized.as_str() { + "1" | "true" | "yes" | "on" => self.learning.enabled = true, + "0" | "false" | "no" | "off" => self.learning.enabled = false, + _ => {} + } + } + if let Ok(flag) = std::env::var("OPENHUMAN_LEARNING_REFLECTION_ENABLED") { + let normalized = flag.trim().to_ascii_lowercase(); + match normalized.as_str() { + "1" | "true" | "yes" | "on" => self.learning.reflection_enabled = true, + "0" | "false" | "no" | "off" => self.learning.reflection_enabled = false, + _ => {} + } + } + if let Ok(flag) = std::env::var("OPENHUMAN_LEARNING_USER_PROFILE_ENABLED") { + let normalized = flag.trim().to_ascii_lowercase(); + match normalized.as_str() { + "1" | "true" | "yes" | "on" => self.learning.user_profile_enabled = true, + "0" | "false" | "no" | "off" => self.learning.user_profile_enabled = false, + _ => {} + } + } + if let Ok(flag) = std::env::var("OPENHUMAN_LEARNING_TOOL_TRACKING_ENABLED") { + let normalized = flag.trim().to_ascii_lowercase(); + match normalized.as_str() { + "1" | "true" | "yes" | "on" => self.learning.tool_tracking_enabled = true, + "0" | "false" | "no" | "off" => self.learning.tool_tracking_enabled = false, + _ => {} + } + } + if let Ok(flag) = std::env::var("OPENHUMAN_LEARNING_SKILL_CREATION_ENABLED") { + let normalized = flag.trim().to_ascii_lowercase(); + match normalized.as_str() { + "1" | "true" | "yes" | "on" => self.learning.skill_creation_enabled = true, + "0" | "false" | "no" | "off" => self.learning.skill_creation_enabled = false, + _ => {} + } + } + if let Ok(source) = std::env::var("OPENHUMAN_LEARNING_REFLECTION_SOURCE") { + let normalized = source.trim().to_ascii_lowercase(); + match normalized.as_str() { + "local" => { + self.learning.reflection_source = + crate::openhuman::config::ReflectionSource::Local + } + "cloud" => { + self.learning.reflection_source = + crate::openhuman::config::ReflectionSource::Cloud + } + _ => { + tracing::warn!( + source = %source, + "ignoring invalid OPENHUMAN_LEARNING_REFLECTION_SOURCE (valid: local, cloud)" + ); + } + } + } + if let Ok(val) = std::env::var("OPENHUMAN_LEARNING_MAX_REFLECTIONS_PER_SESSION") { + if let Ok(max) = val.trim().parse::() { + self.learning.max_reflections_per_session = max; + } + } + if let Ok(val) = std::env::var("OPENHUMAN_LEARNING_MIN_TURN_COMPLEXITY") { + if let Ok(min) = val.trim().parse::() { + self.learning.min_turn_complexity = min; + } + } + if self.proxy.enabled && self.proxy.scope == ProxyScope::Environment { self.proxy.apply_to_process_env(); } set_runtime_proxy_config(self.proxy.clone()); + + // Learning subsystem configuration + if let Ok(flag) = std::env::var("OPENHUMAN_LEARNING_ENABLED") { + let normalized = flag.trim().to_ascii_lowercase(); + match normalized.as_str() { + "1" | "true" | "yes" | "on" => self.learning.enabled = true, + "0" | "false" | "no" | "off" => self.learning.enabled = false, + _ => {} + } + } + + if let Ok(flag) = std::env::var("OPENHUMAN_LEARNING_REFLECTION_ENABLED") { + let normalized = flag.trim().to_ascii_lowercase(); + match normalized.as_str() { + "1" | "true" | "yes" | "on" => self.learning.reflection_enabled = true, + "0" | "false" | "no" | "off" => self.learning.reflection_enabled = false, + _ => {} + } + } + + if let Ok(flag) = std::env::var("OPENHUMAN_LEARNING_USER_PROFILE_ENABLED") { + let normalized = flag.trim().to_ascii_lowercase(); + match normalized.as_str() { + "1" | "true" | "yes" | "on" => self.learning.user_profile_enabled = true, + "0" | "false" | "no" | "off" => self.learning.user_profile_enabled = false, + _ => {} + } + } + + if let Ok(flag) = std::env::var("OPENHUMAN_LEARNING_TOOL_TRACKING_ENABLED") { + let normalized = flag.trim().to_ascii_lowercase(); + match normalized.as_str() { + "1" | "true" | "yes" | "on" => self.learning.tool_tracking_enabled = true, + "0" | "false" | "no" | "off" => self.learning.tool_tracking_enabled = false, + _ => {} + } + } + + if let Ok(flag) = std::env::var("OPENHUMAN_LEARNING_SKILL_CREATION_ENABLED") { + let normalized = flag.trim().to_ascii_lowercase(); + match normalized.as_str() { + "1" | "true" | "yes" | "on" => self.learning.skill_creation_enabled = true, + "0" | "false" | "no" | "off" => self.learning.skill_creation_enabled = false, + _ => {} + } + } + + if let Ok(source_str) = std::env::var("OPENHUMAN_LEARNING_REFLECTION_SOURCE") { + let normalized = source_str.trim().to_ascii_lowercase(); + match normalized.as_str() { + "local" => self.learning.reflection_source = super::ReflectionSource::Local, + "cloud" => self.learning.reflection_source = super::ReflectionSource::Cloud, + _ => { + tracing::warn!( + source = %source_str, + "Ignoring invalid OPENHUMAN_LEARNING_REFLECTION_SOURCE (valid: local, cloud)" + ); + } + } + } + + if let Ok(max_str) = std::env::var("OPENHUMAN_LEARNING_MAX_REFLECTIONS_PER_SESSION") { + if let Ok(max_val) = max_str.parse::() { + if max_val > 0 { + self.learning.max_reflections_per_session = max_val; + } + } + } + + if let Ok(min_str) = std::env::var("OPENHUMAN_LEARNING_MIN_TURN_COMPLEXITY") { + if let Ok(min_val) = min_str.parse::() { + self.learning.min_turn_complexity = min_val; + } + } } pub async fn save(&self) -> Result<()> { diff --git a/src/openhuman/config/schema/mod.rs b/src/openhuman/config/schema/mod.rs index 2b5a8993b..67ca31304 100644 --- a/src/openhuman/config/schema/mod.rs +++ b/src/openhuman/config/schema/mod.rs @@ -11,6 +11,7 @@ mod defaults; mod hardware; mod heartbeat_cron; mod identity_cost; +mod learning; mod load; mod local_ai; mod observability; @@ -19,7 +20,6 @@ mod routes; mod runtime; mod storage_memory; mod tools; -mod tunnel; pub use accessibility::ScreenIntelligenceConfig; pub use agent::{AgentConfig, DelegateAgentConfig}; @@ -36,6 +36,7 @@ pub use heartbeat_cron::{CronConfig, HeartbeatConfig}; pub use identity_cost::{ CostConfig, IdentityConfig, ModelPricing, PeripheralBoardConfig, PeripheralsConfig, }; +pub use learning::{LearningConfig, ReflectionSource}; pub use local_ai::LocalAiConfig; pub use observability::ObservabilityConfig; pub use proxy::{ @@ -54,10 +55,5 @@ pub use tools::{ BrowserComputerUseConfig, BrowserConfig, ComposioConfig, HttpRequestConfig, MultimodalConfig, SecretsConfig, WebSearchConfig, }; -pub use tunnel::{ - CloudflareTunnelConfig, CustomTunnelConfig, NgrokTunnelConfig, TailscaleTunnelConfig, - TunnelConfig, -}; - mod types; pub use types::*; diff --git a/src/openhuman/config/schema/tunnel.rs b/src/openhuman/config/schema/tunnel.rs deleted file mode 100644 index 9bc890899..000000000 --- a/src/openhuman/config/schema/tunnel.rs +++ /dev/null @@ -1,54 +0,0 @@ -//! Tunnel (Cloudflare, Tailscale, ngrok, custom) configuration. - -use schemars::JsonSchema; -use serde::{Deserialize, Serialize}; - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct TunnelConfig { - pub provider: String, - #[serde(default)] - pub cloudflare: Option, - #[serde(default)] - pub tailscale: Option, - #[serde(default)] - pub ngrok: Option, - #[serde(default)] - pub custom: Option, -} - -impl Default for TunnelConfig { - fn default() -> Self { - Self { - provider: "none".into(), - cloudflare: None, - tailscale: None, - ngrok: None, - custom: None, - } - } -} - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct CloudflareTunnelConfig { - pub token: String, -} - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct TailscaleTunnelConfig { - #[serde(default)] - pub funnel: bool, - pub hostname: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct NgrokTunnelConfig { - pub auth_token: String, - pub domain: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct CustomTunnelConfig { - pub start_command: String, - pub health_url: Option, - pub url_pattern: Option, -} diff --git a/src/openhuman/config/schema/types.rs b/src/openhuman/config/schema/types.rs index e953ca309..eeeb73d8a 100644 --- a/src/openhuman/config/schema/types.rs +++ b/src/openhuman/config/schema/types.rs @@ -66,9 +66,6 @@ pub struct Config { #[serde(default)] pub storage: StorageConfig, - #[serde(default)] - pub tunnel: TunnelConfig, - #[serde(default)] pub composio: ComposioConfig, @@ -107,6 +104,9 @@ pub struct Config { #[serde(default)] pub local_ai: LocalAiConfig, + + #[serde(default)] + pub learning: LearningConfig, } impl Default for Config { @@ -137,7 +137,6 @@ impl Default for Config { channels_config: ChannelsConfig::default(), memory: MemoryConfig::default(), storage: StorageConfig::default(), - tunnel: TunnelConfig::default(), composio: ComposioConfig::default(), secrets: SecretsConfig::default(), browser: BrowserConfig::default(), @@ -152,6 +151,7 @@ impl Default for Config { hardware: HardwareConfig::default(), local_ai: LocalAiConfig::default(), query_classification: QueryClassificationConfig::default(), + learning: LearningConfig::default(), } } } diff --git a/src/openhuman/config/schemas.rs b/src/openhuman/config/schemas.rs index dc3de2b80..e5d6f72b8 100644 --- a/src/openhuman/config/schemas.rs +++ b/src/openhuman/config/schemas.rs @@ -5,7 +5,6 @@ use serde_json::{Map, Value}; use crate::core::all::{ControllerFuture, RegisteredController}; use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; use crate::openhuman::config::rpc as config_rpc; -use crate::openhuman::config::TunnelConfig; use crate::rpc::RpcOutcome; const DEFAULT_ONBOARDING_FLAG_NAME: &str = ".skip_onboarding"; @@ -77,7 +76,6 @@ pub fn all_controller_schemas() -> Vec { schemas("update_model_settings"), schemas("update_memory_settings"), schemas("update_screen_intelligence_settings"), - schemas("update_tunnel_settings"), schemas("update_runtime_settings"), schemas("update_browser_settings"), schemas("resolve_api_url"), @@ -109,10 +107,6 @@ pub fn all_registered_controllers() -> Vec { schema: schemas("update_screen_intelligence_settings"), handler: handle_update_screen_intelligence_settings, }, - RegisteredController { - schema: schemas("update_tunnel_settings"), - handler: handle_update_tunnel_settings, - }, RegisteredController { schema: schemas("update_runtime_settings"), handler: handle_update_runtime_settings, @@ -245,39 +239,6 @@ pub fn schemas(function: &str) -> ControllerSchema { ], outputs: vec![json_output("snapshot", "Updated config snapshot.")], }, - "update_tunnel_settings" => ControllerSchema { - namespace: "config", - function: "update_tunnel_settings", - description: "Replace tunnel settings with provided config payload.", - inputs: vec![ - required_string("provider", "Tunnel provider id."), - FieldSchema { - name: "cloudflare", - ty: TypeSchema::Option(Box::new(TypeSchema::Ref("CloudflareTunnelConfig"))), - comment: "Cloudflare tunnel settings.", - required: false, - }, - FieldSchema { - name: "tailscale", - ty: TypeSchema::Option(Box::new(TypeSchema::Ref("TailscaleTunnelConfig"))), - comment: "Tailscale tunnel settings.", - required: false, - }, - FieldSchema { - name: "ngrok", - ty: TypeSchema::Option(Box::new(TypeSchema::Ref("NgrokTunnelConfig"))), - comment: "ngrok tunnel settings.", - required: false, - }, - FieldSchema { - name: "custom", - ty: TypeSchema::Option(Box::new(TypeSchema::Ref("CustomTunnelConfig"))), - comment: "Custom tunnel settings.", - required: false, - }, - ], - outputs: vec![json_output("snapshot", "Updated config snapshot.")], - }, "update_runtime_settings" => ControllerSchema { namespace: "config", function: "update_runtime_settings", @@ -470,13 +431,6 @@ fn handle_update_screen_intelligence_settings(params: Map) -> Con }) } -fn handle_update_tunnel_settings(params: Map) -> ControllerFuture { - Box::pin(async move { - let tunnel = deserialize_params::(params)?; - to_json(config_rpc::load_and_apply_tunnel_settings(tunnel).await?) - }) -} - fn handle_update_runtime_settings(params: Map) -> ControllerFuture { Box::pin(async move { let update = deserialize_params::(params)?; diff --git a/src/openhuman/config/settings_cli.rs b/src/openhuman/config/settings_cli.rs index c352eaa39..d4588f92e 100644 --- a/src/openhuman/config/settings_cli.rs +++ b/src/openhuman/config/settings_cli.rs @@ -28,10 +28,6 @@ pub fn settings_section_json( .get("memory") .cloned() .unwrap_or(serde_json::Value::Null), - "tunnel" => cfg - .get("tunnel") - .cloned() - .unwrap_or(serde_json::Value::Null), "runtime" => cfg .get("runtime") .cloned() diff --git a/src/openhuman/learning/mod.rs b/src/openhuman/learning/mod.rs new file mode 100644 index 000000000..5cb9290b2 --- /dev/null +++ b/src/openhuman/learning/mod.rs @@ -0,0 +1,14 @@ +//! Agent self-learning subsystem. +//! +//! Post-turn hooks that reflect on completed turns, extract user preferences, +//! track tool effectiveness, and store learnings in the Memory backend. + +pub mod prompt_sections; +pub mod reflection; +pub mod tool_tracker; +pub mod user_profile; + +pub use prompt_sections::{LearnedContextSection, UserProfileSection}; +pub use reflection::ReflectionHook; +pub use tool_tracker::ToolTrackerHook; +pub use user_profile::UserProfileHook; diff --git a/src/openhuman/learning/prompt_sections.rs b/src/openhuman/learning/prompt_sections.rs new file mode 100644 index 000000000..61719644e --- /dev/null +++ b/src/openhuman/learning/prompt_sections.rs @@ -0,0 +1,83 @@ +//! Prompt sections that inject learned context into the agent's system prompt. +//! +//! These sections read pre-fetched data from `PromptContext.learned` — no async +//! or blocking I/O happens during prompt building. + +use crate::openhuman::agent::prompt::{PromptContext, PromptSection}; +use anyhow::Result; + +/// Injects recent observations and patterns from the learning subsystem. +pub struct LearnedContextSection; + +impl LearnedContextSection { + pub fn new(_memory: std::sync::Arc) -> Self { + // Memory parameter kept for API compatibility but data comes from PromptContext.learned + Self + } +} + +impl PromptSection for LearnedContextSection { + fn name(&self) -> &str { + "learned_context" + } + + fn build(&self, ctx: &PromptContext<'_>) -> Result { + if ctx.learned.observations.is_empty() && ctx.learned.patterns.is_empty() { + return Ok(String::new()); + } + + let mut out = String::from("## Learned Context\n\n"); + + if !ctx.learned.observations.is_empty() { + out.push_str("### Recent Observations\n"); + for obs in &ctx.learned.observations { + out.push_str("- "); + out.push_str(obs); + out.push('\n'); + } + out.push('\n'); + } + + if !ctx.learned.patterns.is_empty() { + out.push_str("### Recognized Patterns\n"); + for pat in &ctx.learned.patterns { + out.push_str("- "); + out.push_str(pat); + out.push('\n'); + } + out.push('\n'); + } + + Ok(out) + } +} + +/// Injects the learned user profile into the system prompt. +pub struct UserProfileSection; + +impl UserProfileSection { + pub fn new(_memory: std::sync::Arc) -> Self { + Self + } +} + +impl PromptSection for UserProfileSection { + fn name(&self) -> &str { + "user_profile" + } + + fn build(&self, ctx: &PromptContext<'_>) -> Result { + if ctx.learned.user_profile.is_empty() { + return Ok(String::new()); + } + + let mut out = String::from("## User Profile (Learned)\n\n"); + for entry in &ctx.learned.user_profile { + out.push_str("- "); + out.push_str(entry); + out.push('\n'); + } + out.push('\n'); + Ok(out) + } +} diff --git a/src/openhuman/learning/reflection.rs b/src/openhuman/learning/reflection.rs new file mode 100644 index 000000000..7f7a34496 --- /dev/null +++ b/src/openhuman/learning/reflection.rs @@ -0,0 +1,341 @@ +//! Post-turn reflection engine. +//! +//! After each qualifying turn, builds a reflection prompt, sends it to the +//! configured LLM (local Ollama or cloud reasoning model), parses structured +//! JSON output, and stores observations in memory. + +use crate::openhuman::agent::hooks::{PostTurnHook, TurnContext}; +use crate::openhuman::config::{Config, LearningConfig, ReflectionSource}; +use crate::openhuman::memory::{Memory, MemoryCategory}; +use async_trait::async_trait; +use parking_lot::Mutex; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::sync::Arc; + +/// Structured output expected from the reflection LLM call. +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct ReflectionOutput { + #[serde(default)] + pub observations: Vec, + #[serde(default)] + pub patterns: Vec, + #[serde(default)] + pub user_preferences: Vec, +} + +/// Post-turn hook that reflects on completed turns and stores observations. +pub struct ReflectionHook { + config: LearningConfig, + full_config: Arc, + memory: Arc, + provider: Option>, + /// Per-session reflection counts for throttling. Key is session_id (or "__global__"). + session_counts: Mutex>, +} + +impl ReflectionHook { + pub fn new( + config: LearningConfig, + full_config: Arc, + memory: Arc, + provider: Option>, + ) -> Self { + Self { + config, + full_config, + memory, + provider, + session_counts: Mutex::new(HashMap::new()), + } + } + + fn session_key(ctx: &TurnContext) -> String { + ctx.session_id + .clone() + .unwrap_or_else(|| "__global__".to_string()) + } + + /// Attempt to increment the session counter. Returns true if under the limit. + fn try_increment(&self, session_key: &str) -> bool { + let mut counts = self.session_counts.lock(); + let count = counts.entry(session_key.to_string()).or_insert(0); + if *count >= self.config.max_reflections_per_session { + log::debug!( + "[learning] reflection throttled for session {session_key}: {count} >= {}", + self.config.max_reflections_per_session + ); + return false; + } + *count += 1; + true + } + + /// Rollback the session counter (e.g. on reflection failure). + fn rollback_increment(&self, session_key: &str) { + let mut counts = self.session_counts.lock(); + if let Some(count) = counts.get_mut(session_key) { + *count = count.saturating_sub(1); + } + } + + /// Check if this turn warrants reflection (complexity check only). + fn should_reflect(&self, ctx: &TurnContext) -> bool { + if !self.config.enabled || !self.config.reflection_enabled { + return false; + } + + // Check minimum complexity + let tool_count = ctx.tool_calls.len(); + let response_long = ctx.assistant_response.chars().count() > 500; + tool_count >= self.config.min_turn_complexity || response_long + } + + /// Build the reflection prompt from turn context. + fn build_reflection_prompt(&self, ctx: &TurnContext) -> String { + let mut prompt = String::from( + "Analyze this completed agent turn and extract learnings.\n\ + Return a JSON object with these fields:\n\ + - \"observations\": array of strings — what worked, what failed, notable patterns\n\ + - \"patterns\": array of strings — recurring patterns worth remembering\n\ + - \"user_preferences\": array of strings — any user preferences detected\n\n\ + Keep each entry concise (one sentence). Return ONLY valid JSON, no markdown.\n\n", + ); + + prompt.push_str(&format!( + "## User Message\n{}\n\n", + truncate(&ctx.user_message, 500) + )); + prompt.push_str(&format!( + "## Assistant Response\n{}\n\n", + truncate(&ctx.assistant_response, 500) + )); + + if !ctx.tool_calls.is_empty() { + prompt.push_str("## Tool Calls\n"); + for tc in &ctx.tool_calls { + prompt.push_str(&format!( + "- {} (success={}, duration={}ms): {}\n", + tc.name, + tc.success, + tc.duration_ms, + truncate(&tc.output_summary, 100) + )); + } + prompt.push('\n'); + } + + prompt.push_str(&format!( + "Turn took {}ms across {} iteration(s).\n", + ctx.turn_duration_ms, ctx.iteration_count + )); + + prompt + } + + /// Call the configured LLM for reflection. + async fn run_reflection(&self, prompt: &str) -> anyhow::Result { + match self.config.reflection_source { + ReflectionSource::Local => { + let service = crate::openhuman::local_ai::global(&self.full_config); + service + .prompt(&self.full_config, prompt, Some(512), true) + .await + .map_err(|e| anyhow::anyhow!("local reflection failed: {e}")) + } + ReflectionSource::Cloud => { + let provider = self.provider.as_ref().ok_or_else(|| { + anyhow::anyhow!("no cloud provider configured for reflection") + })?; + provider.simple_chat(prompt, "hint:reasoning", 0.3).await + } + } + } + + /// Parse the LLM response into structured reflection output. + fn parse_reflection(raw: &str) -> ReflectionOutput { + // Try to extract JSON from the response (may have surrounding text) + let trimmed = raw.trim(); + let json_str = if let Some(start) = trimmed.find('{') { + if let Some(end) = trimmed.rfind('}') { + &trimmed[start..=end] + } else { + trimmed + } + } else { + trimmed + }; + + serde_json::from_str(json_str).unwrap_or_else(|_| { + log::debug!( + "[learning] could not parse reflection JSON, using raw text as observation" + ); + ReflectionOutput { + observations: vec![trimmed.to_string()], + patterns: Vec::new(), + user_preferences: Vec::new(), + } + }) + } + + /// Store reflection output in memory. + async fn store_reflection(&self, output: &ReflectionOutput) -> anyhow::Result<()> { + let date = chrono::Local::now().format("%Y-%m-%d").to_string(); + let hash = &uuid::Uuid::new_v4().to_string()[..8]; + + if !output.observations.is_empty() { + let content = output.observations.join("\n"); + let key = format!("obs/{date}/{hash}"); + self.memory + .store( + &key, + &content, + MemoryCategory::Custom("learning_observations".into()), + None, + ) + .await?; + log::debug!( + "[learning] stored {} observation(s) at {key}", + output.observations.len() + ); + } + + for pattern in &output.patterns { + let slug = slugify(pattern); + let key = format!("pat/{slug}"); + self.memory + .store( + &key, + pattern, + MemoryCategory::Custom("learning_patterns".into()), + None, + ) + .await?; + } + + // User preferences are handled by UserProfileHook, but store raw if present + for pref in &output.user_preferences { + let slug = slugify(pref); + let key = format!("pref/{slug}"); + self.memory + .store( + &key, + pref, + MemoryCategory::Custom("user_profile".into()), + None, + ) + .await?; + } + + Ok(()) + } +} + +#[async_trait] +impl PostTurnHook for ReflectionHook { + fn name(&self) -> &str { + "reflection" + } + + async fn on_turn_complete(&self, ctx: &TurnContext) -> anyhow::Result<()> { + if !self.should_reflect(ctx) { + return Ok(()); + } + + let session_key = Self::session_key(ctx); + if !self.try_increment(&session_key) { + return Ok(()); + } + + log::debug!("[learning] starting reflection for session={session_key}",); + + let prompt = self.build_reflection_prompt(ctx); + let result = self.run_reflection(&prompt).await; + + let raw = match result { + Ok(raw) => raw, + Err(e) => { + // Rollback the counter so failures don't consume quota + self.rollback_increment(&session_key); + return Err(e); + } + }; + + let output = Self::parse_reflection(&raw); + + log::info!( + "[learning] reflection complete: observations={} patterns={} prefs={}", + output.observations.len(), + output.patterns.len(), + output.user_preferences.len() + ); + + if let Err(e) = self.store_reflection(&output).await { + self.rollback_increment(&session_key); + return Err(e); + } + + Ok(()) + } +} + +fn truncate(s: &str, max: usize) -> String { + if s.chars().count() <= max { + s.to_string() + } else { + let truncated: String = s.chars().take(max).collect(); + format!("{truncated}...") + } +} + +fn slugify(s: &str) -> String { + s.chars() + .filter_map(|c| { + if c.is_alphanumeric() { + Some(c.to_ascii_lowercase()) + } else if c == ' ' || c == '-' || c == '_' { + Some('_') + } else { + None + } + }) + .take(40) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_reflection_valid_json() { + let raw = r#"{"observations":["Tool A was effective"],"patterns":["User prefers concise output"],"user_preferences":["timezone: PST"]}"#; + let output = ReflectionHook::parse_reflection(raw); + assert_eq!(output.observations.len(), 1); + assert_eq!(output.patterns.len(), 1); + assert_eq!(output.user_preferences.len(), 1); + } + + #[test] + fn parse_reflection_with_surrounding_text() { + let raw = r#"Here is the analysis: +{"observations":["worked well"],"patterns":[],"user_preferences":[]} +That's my assessment."#; + let output = ReflectionHook::parse_reflection(raw); + assert_eq!(output.observations, vec!["worked well"]); + } + + #[test] + fn parse_reflection_invalid_json_falls_back() { + let raw = "This is not JSON at all"; + let output = ReflectionHook::parse_reflection(raw); + assert_eq!(output.observations.len(), 1); + assert!(output.observations[0].contains("not JSON")); + } + + #[test] + fn slugify_produces_clean_keys() { + assert_eq!(slugify("User prefers Rust"), "user_prefers_rust"); + assert_eq!(slugify("hello-world_test"), "hello_world_test"); + } +} diff --git a/src/openhuman/learning/tool_tracker.rs b/src/openhuman/learning/tool_tracker.rs new file mode 100644 index 000000000..e9af73450 --- /dev/null +++ b/src/openhuman/learning/tool_tracker.rs @@ -0,0 +1,209 @@ +//! Tool effectiveness tracking hook. +//! +//! For each tool call in a completed turn, updates running tallies of +//! total calls, successes, failures, and average duration. Stored in the +//! `tool_effectiveness` memory category keyed by `tool/{name}`. + +use crate::openhuman::agent::hooks::{PostTurnHook, TurnContext}; +use crate::openhuman::config::LearningConfig; +use crate::openhuman::memory::{Memory, MemoryCategory}; +use async_trait::async_trait; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::Mutex; + +/// Per-tool effectiveness stats stored in memory. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ToolStats { + pub total_calls: u64, + pub successes: u64, + pub failures: u64, + pub avg_duration_ms: f64, + #[serde(default)] + pub common_error_patterns: Vec, +} + +impl Default for ToolStats { + fn default() -> Self { + Self { + total_calls: 0, + successes: 0, + failures: 0, + avg_duration_ms: 0.0, + common_error_patterns: Vec::new(), + } + } +} + +impl ToolStats { + /// Update stats with a new tool call outcome. + pub fn record_call(&mut self, success: bool, duration_ms: u64, error_snippet: Option<&str>) { + self.total_calls += 1; + if success { + self.successes += 1; + } else { + self.failures += 1; + if let Some(err) = error_snippet { + let pattern = err.chars().take(80).collect::(); + if !self.common_error_patterns.contains(&pattern) { + self.common_error_patterns.push(pattern); + // Keep only recent error patterns + if self.common_error_patterns.len() > 5 { + self.common_error_patterns.remove(0); + } + } + } + } + // Running average + let prev_total = self.total_calls - 1; + self.avg_duration_ms = (self.avg_duration_ms * prev_total as f64 + duration_ms as f64) + / self.total_calls as f64; + } + + /// Format stats for display. + pub fn summary(&self) -> String { + let success_rate = if self.total_calls > 0 { + (self.successes as f64 / self.total_calls as f64) * 100.0 + } else { + 0.0 + }; + format!( + "calls={} success_rate={:.0}% avg_duration={:.0}ms failures={}", + self.total_calls, success_rate, self.avg_duration_ms, self.failures + ) + } +} + +/// Post-turn hook that tracks tool effectiveness. +pub struct ToolTrackerHook { + config: LearningConfig, + memory: Arc, + /// Per-tool lock to serialize read-modify-write cycles. + tool_locks: Mutex>>>, +} + +impl ToolTrackerHook { + pub fn new(config: LearningConfig, memory: Arc) -> Self { + Self { + config, + memory, + tool_locks: Mutex::new(HashMap::new()), + } + } + + /// Get or create a per-tool lock. + async fn tool_lock(&self, tool_name: &str) -> Arc> { + let mut locks = self.tool_locks.lock().await; + locks + .entry(tool_name.to_string()) + .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))) + .clone() + } + + /// Atomically load, update, and save stats for a single tool under a lock. + async fn update_stats( + &self, + tool_name: &str, + success: bool, + duration_ms: u64, + error_summary: Option<&str>, + ) -> anyhow::Result<()> { + let lock = self.tool_lock(tool_name).await; + let _guard = lock.lock().await; + + let key = format!("tool/{tool_name}"); + let mut stats: ToolStats = match self.memory.get(&key).await { + Ok(Some(entry)) => serde_json::from_str(&entry.content).unwrap_or_default(), + _ => ToolStats::default(), + }; + + stats.record_call(success, duration_ms, error_summary); + + let content = serde_json::to_string(&stats)?; + self.memory + .store( + &key, + &content, + MemoryCategory::Custom("tool_effectiveness".into()), + None, + ) + .await?; + + log::debug!( + "[learning] tool stats updated: {tool_name} — {}", + stats.summary() + ); + Ok(()) + } +} + +#[async_trait] +impl PostTurnHook for ToolTrackerHook { + fn name(&self) -> &str { + "tool_tracker" + } + + async fn on_turn_complete(&self, ctx: &TurnContext) -> anyhow::Result<()> { + if !self.config.enabled || !self.config.tool_tracking_enabled { + return Ok(()); + } + + if ctx.tool_calls.is_empty() { + return Ok(()); + } + + for tc in &ctx.tool_calls { + let error_summary = if !tc.success { + Some(tc.output_summary.as_str()) + } else { + None + }; + + if let Err(e) = self + .update_stats(&tc.name, tc.success, tc.duration_ms, error_summary) + .await + { + log::warn!( + "[learning] failed to update tool stats for {}: {e:#}", + tc.name + ); + } + } + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn tool_stats_record_call_updates_correctly() { + let mut stats = ToolStats::default(); + stats.record_call(true, 100, None); + assert_eq!(stats.total_calls, 1); + assert_eq!(stats.successes, 1); + assert_eq!(stats.failures, 0); + assert_eq!(stats.avg_duration_ms, 100.0); + + stats.record_call(false, 200, Some("timeout error")); + assert_eq!(stats.total_calls, 2); + assert_eq!(stats.successes, 1); + assert_eq!(stats.failures, 1); + assert_eq!(stats.avg_duration_ms, 150.0); + assert_eq!(stats.common_error_patterns.len(), 1); + } + + #[test] + fn tool_stats_summary_formats_correctly() { + let mut stats = ToolStats::default(); + stats.record_call(true, 50, None); + stats.record_call(true, 150, None); + stats.record_call(false, 300, Some("err")); + let summary = stats.summary(); + assert!(summary.contains("calls=3")); + assert!(summary.contains("failures=1")); + } +} diff --git a/src/openhuman/learning/user_profile.rs b/src/openhuman/learning/user_profile.rs new file mode 100644 index 000000000..8dc0fb1e7 --- /dev/null +++ b/src/openhuman/learning/user_profile.rs @@ -0,0 +1,174 @@ +//! User profile learning hook. +//! +//! Extracts user preferences from conversation turns using lightweight regex +//! patterns (e.g. "I prefer...", "always use...", "my timezone is...") and +//! stores them in the `user_profile` memory category. + +use crate::openhuman::agent::hooks::{PostTurnHook, TurnContext}; +use crate::openhuman::config::LearningConfig; +use crate::openhuman::memory::{Memory, MemoryCategory}; +use async_trait::async_trait; +use std::sync::Arc; + +/// Regex-based patterns that signal explicit user preferences. +const PREFERENCE_PATTERNS: &[&str] = &[ + "i prefer ", + "i always ", + "always use ", + "never use ", + "my timezone ", + "my language ", + "i like ", + "i don't like ", + "i want ", + "i need ", + "please always ", + "please never ", + "from now on ", + "going forward ", + "my name is ", + "i am a ", + "i'm a ", + "i work ", + "my role ", + "my stack ", +]; + +/// Post-turn hook that extracts user preferences from conversations. +pub struct UserProfileHook { + config: LearningConfig, + memory: Arc, +} + +impl UserProfileHook { + pub fn new(config: LearningConfig, memory: Arc) -> Self { + Self { config, memory } + } + + /// Extract preference statements from the user message. + fn extract_preferences(message: &str) -> Vec { + let lower = message.to_lowercase(); + let mut found = Vec::new(); + + for sentence in message.split(['.', '!', '\n']) { + let trimmed = sentence.trim(); + if trimmed.is_empty() || trimmed.len() < 10 { + continue; + } + let sentence_lower = trimmed.to_lowercase(); + for pattern in PREFERENCE_PATTERNS { + if sentence_lower.contains(pattern) { + found.push(trimmed.to_string()); + break; + } + } + } + + // Also check the full message for short, direct preference statements + if found.is_empty() + && message.trim().len() >= 15 + && (lower.starts_with("i prefer") || lower.starts_with("always use")) + { + found.push(message.trim().to_string()); + } + + // Deduplicate and cap + found.truncate(5); + found + } + + /// Store extracted preferences in memory, deduplicating by slug. + async fn store_preferences(&self, preferences: &[String]) -> anyhow::Result<()> { + for pref in preferences { + let slug = slugify(pref); + if slug.is_empty() { + continue; + } + let key = format!("pref/{slug}"); + + // Check for existing entry to avoid duplicates + if let Ok(Some(_)) = self.memory.get(&key).await { + log::debug!("[learning] user preference already stored: {key}"); + continue; + } + + self.memory + .store( + &key, + pref, + MemoryCategory::Custom("user_profile".into()), + None, + ) + .await?; + log::info!("[learning] stored user preference: {key}"); + } + Ok(()) + } +} + +#[async_trait] +impl PostTurnHook for UserProfileHook { + fn name(&self) -> &str { + "user_profile" + } + + async fn on_turn_complete(&self, ctx: &TurnContext) -> anyhow::Result<()> { + if !self.config.enabled || !self.config.user_profile_enabled { + return Ok(()); + } + + let preferences = Self::extract_preferences(&ctx.user_message); + if preferences.is_empty() { + return Ok(()); + } + + log::debug!( + "[learning] extracted {} preference(s) from user message", + preferences.len() + ); + self.store_preferences(&preferences).await + } +} + +fn slugify(s: &str) -> String { + s.chars() + .filter_map(|c| { + if c.is_alphanumeric() { + Some(c.to_ascii_lowercase()) + } else if c == ' ' || c == '-' || c == '_' { + Some('_') + } else { + None + } + }) + .take(40) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn extract_preferences_finds_patterns() { + let msg = "I prefer Rust over Python. Always use snake_case for variables."; + let prefs = UserProfileHook::extract_preferences(msg); + assert_eq!(prefs.len(), 2); + assert!(prefs[0].contains("prefer")); + assert!(prefs[1].contains("snake_case")); + } + + #[test] + fn extract_preferences_ignores_short_sentences() { + let msg = "I prefer. OK."; + let prefs = UserProfileHook::extract_preferences(msg); + assert!(prefs.is_empty()); + } + + #[test] + fn extract_preferences_handles_no_matches() { + let msg = "Can you help me debug this function?"; + let prefs = UserProfileHook::extract_preferences(msg); + assert!(prefs.is_empty()); + } +} diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index d2ca5a8e4..ad35a6091 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -24,6 +24,7 @@ pub mod doctor; pub mod encryption; pub mod health; pub mod heartbeat; +pub mod learning; pub mod local_ai; pub mod memory; pub mod migration; @@ -33,6 +34,6 @@ pub mod security; pub mod service; pub mod skills; pub mod tools; -pub mod tunnel; pub mod util; +pub mod webhooks; pub mod workspace; diff --git a/src/openhuman/skills/qjs_engine.rs b/src/openhuman/skills/qjs_engine.rs index 8c15fcb6f..3bd69afd0 100644 --- a/src/openhuman/skills/qjs_engine.rs +++ b/src/openhuman/skills/qjs_engine.rs @@ -39,6 +39,7 @@ use crate::openhuman::skills::qjs_skill_instance::{BridgeDeps, QjsSkillInstance} use crate::openhuman::skills::skill_registry::SkillRegistry; use crate::openhuman::skills::socket_manager::SocketManager; use crate::openhuman::skills::types::{SkillSnapshot, SkillStatus, ToolCallOrigin, ToolResult}; +use crate::openhuman::webhooks::WebhookRouter; // IdbStorage removed during runtime cleanup /// The central runtime engine using QuickJS. @@ -61,6 +62,8 @@ pub struct RuntimeEngine { memory_client: RwLock>, /// Socket manager for emitting tool:sync events. socket_manager: RwLock>>, + /// Webhook router for tunnel-to-skill routing. + webhook_router: Arc, /// Workspace directory for user-installed skills from registry. workspace_dir: RwLock>, } @@ -87,6 +90,10 @@ impl RuntimeEngine { } }; + // Initialize webhook router with persistence + let webhook_routes_path = skills_data_dir.join("webhook_routes.json"); + let webhook_router = Arc::new(WebhookRouter::new(Some(webhook_routes_path))); + log::info!("[runtime] QuickJS RuntimeEngine created"); Ok(Self { @@ -99,6 +106,7 @@ impl RuntimeEngine { resource_dir: RwLock::new(None), memory_client: RwLock::new(memory_client), socket_manager: RwLock::new(None), + webhook_router, workspace_dir: RwLock::new(None), }) } @@ -137,9 +145,16 @@ impl RuntimeEngine { /// Set the socket manager for emitting `tool:sync` events. pub fn set_socket_manager(&self, mgr: Arc) { + // Also wire the webhook router into the socket manager + mgr.set_webhook_router(Arc::clone(&self.webhook_router)); *self.socket_manager.write() = Some(mgr); } + /// Get a clone of the webhook router Arc. + pub fn webhook_router(&self) -> Arc { + Arc::clone(&self.webhook_router) + } + /// Set the workspace directory for user-installed skills from the registry. pub fn set_workspace_dir(&self, dir: PathBuf) { log::info!("[runtime] Workspace directory set to: {:?}", dir); @@ -377,6 +392,7 @@ impl RuntimeEngine { cron_scheduler: self.cron_scheduler.clone(), skill_registry: self.registry.clone(), memory_client: self.memory_client.read().clone(), + webhook_router: Some(self.webhook_router.clone()), data_dir: data_dir.clone(), }; @@ -457,6 +473,7 @@ impl RuntimeEngine { pub async fn stop_skill(&self, skill_id: &str) -> Result<(), String> { self.registry.stop_skill(skill_id).await?; self.cron_scheduler.unregister_all_for_skill(skill_id); + self.webhook_router.unregister_skill(skill_id); self.emit_status_change(skill_id); self.sync_tools().await; Ok(()) diff --git a/src/openhuman/skills/qjs_skill_instance/event_loop.rs b/src/openhuman/skills/qjs_skill_instance/event_loop.rs index 37e1077fc..cb0fe9941 100644 --- a/src/openhuman/skills/qjs_skill_instance/event_loop.rs +++ b/src/openhuman/skills/qjs_skill_instance/event_loop.rs @@ -364,6 +364,88 @@ async fn handle_message( log::warn!("[skill:{}] onError() handler failed: {e}", skill_id); } } + SkillMessage::WebhookRequest { + correlation_id, + method, + path, + headers, + query, + body, + tunnel_id, + tunnel_name, + reply, + } => { + log::info!( + "[skill:{}] event_loop: WebhookRequest {} {} (tunnel={})", + skill_id, + method, + path, + tunnel_id, + ); + + // Restore OAuth credential in case the handler needs authenticated API calls + restore_oauth_credential(ctx, skill_id, data_dir).await; + + let args = serde_json::json!({ + "correlationId": correlation_id, + "method": method, + "path": path, + "headers": headers, + "query": query, + "body": body, + "tunnelId": tunnel_id, + "tunnelName": tunnel_name, + }); + + match handle_js_call(rt, ctx, "onWebhookRequest", &args.to_string()).await { + Ok(response_val) => { + use crate::openhuman::webhooks::WebhookResponseData; + + let status_code = response_val + .get("statusCode") + .and_then(|v| v.as_u64()) + .unwrap_or(200) as u16; + let resp_headers: HashMap = response_val + .get("headers") + .map(|v| match serde_json::from_value(v.clone()) { + Ok(h) => h, + Err(e) => { + log::warn!( + "[skill] Failed to parse webhook response headers: {e}, raw: {v}" + ); + HashMap::new() + } + }) + .unwrap_or_default(); + let resp_body = response_val + .get("body") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + + log::debug!( + "[skill:{}] event_loop: WebhookRequest handled → status {}", + skill_id, + status_code, + ); + + let _ = reply.send(Ok(WebhookResponseData { + correlation_id, + status_code, + headers: resp_headers, + body: resp_body, + })); + } + Err(e) => { + log::warn!( + "[skill:{}] event_loop: onWebhookRequest failed: {}", + skill_id, + e, + ); + let _ = reply.send(Err(e)); + } + } + } SkillMessage::Rpc { method, params, diff --git a/src/openhuman/skills/qjs_skill_instance/instance.rs b/src/openhuman/skills/qjs_skill_instance/instance.rs index 066f7b09e..567c80345 100644 --- a/src/openhuman/skills/qjs_skill_instance/instance.rs +++ b/src/openhuman/skills/qjs_skill_instance/instance.rs @@ -140,6 +140,7 @@ impl QjsSkillInstance { skill_id: skill_id.clone(), data_dir: data_dir.clone(), memory_client: _deps.memory_client.clone(), + webhook_router: _deps.webhook_router.clone(), }; if let Err(e) = qjs_ops::register_ops( diff --git a/src/openhuman/skills/qjs_skill_instance/types.rs b/src/openhuman/skills/qjs_skill_instance/types.rs index 7fe20ad8d..ae45077a2 100644 --- a/src/openhuman/skills/qjs_skill_instance/types.rs +++ b/src/openhuman/skills/qjs_skill_instance/types.rs @@ -15,6 +15,7 @@ pub struct BridgeDeps { pub cron_scheduler: Arc, pub skill_registry: Arc, pub memory_client: Option, + pub webhook_router: Option>, pub data_dir: PathBuf, // NOTE: No v8_creation_lock - QuickJS doesn't need it } diff --git a/src/openhuman/skills/quickjs_libs/bootstrap.js b/src/openhuman/skills/quickjs_libs/bootstrap.js index 66c277c1a..d93f91005 100644 --- a/src/openhuman/skills/quickjs_libs/bootstrap.js +++ b/src/openhuman/skills/quickjs_libs/bootstrap.js @@ -1012,4 +1012,132 @@ globalThis.model = { }; console.log('[bootstrap] Model API initialized'); + +// ============================================================================ +// Webhook / Tunnel API (skill-scoped) +// ============================================================================ +// All operations are scoped to the calling skill. The Rust bridge injects the +// skill_id automatically — JS code cannot impersonate another skill. + +globalThis.webhook = { + /** + * Register this skill to receive webhooks for a tunnel UUID. + * Rejects if the tunnel is already owned by a different skill. + * @param {string} tunnelUuid - The tunnel UUID (from createTunnel or backend) + * @param {string} [tunnelName] - Human-readable name for display + * @param {string} [backendTunnelId] - Backend MongoDB _id for CRUD + */ + register: function (tunnelUuid, tunnelName, backendTunnelId) { + __ops.webhook_register( + tunnelUuid, + tunnelName || null, + backendTunnelId || null + ); + }, + + /** + * Unregister this skill from a tunnel. + * Rejects if the tunnel is not owned by this skill. + * @param {string} tunnelUuid + */ + unregister: function (tunnelUuid) { + __ops.webhook_unregister(tunnelUuid); + }, + + /** + * List only this skill's registered tunnel mappings. + * Never includes other skills' tunnels. + * @returns {Array<{tunnel_uuid: string, skill_id: string, tunnel_name: string|null}>} + */ + list: function () { + var json = __ops.webhook_list(); + return JSON.parse(json); + }, + + /** + * Create a new tunnel via the backend API, automatically registered to + * this skill. + * @param {string} name - Tunnel name + * @param {string} [description] - Optional description + * @returns {Promise<{id: string, uuid: string, webhookUrl: string}>} + */ + createTunnel: async function (name, description) { + var backendUrl = __platform.env('BACKEND_URL') || 'https://api.tinyhumans.ai'; + var jwtToken = __ops.get_session_token() || ''; + + var result = await net.fetch(backendUrl + '/tunnels', { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Authorization: 'Bearer ' + jwtToken, + }, + body: JSON.stringify({ name: name, description: description || '' }), + timeout: 15000, + }); + + var parsed = JSON.parse(result); + if (parsed.status >= 400) { + throw new Error('Failed to create tunnel: ' + parsed.status + ' ' + parsed.body); + } + var data = JSON.parse(parsed.body); + var tunnel = data.tunnel || data; + + // Auto-register this tunnel to the calling skill + if (tunnel.uuid) { + webhook.register(tunnel.uuid, name, tunnel._id || tunnel.id || null); + } + + // Build webhook URL for the caller + tunnel.webhookUrl = backendUrl + '/webhooks/' + tunnel.uuid; + + console.log('[webhook] Created tunnel: ' + name + ' → ' + tunnel.webhookUrl); + return tunnel; + }, + + /** + * List locally registered tunnels scoped to this skill. + * Returns the local webhook registrations, not data from the backend API. + * @returns {Promise} Array of local tunnel registration objects. + */ + listTunnels: async function () { + return webhook.list(); + }, + + /** + * Delete a tunnel. Fails if the tunnel is not owned by this skill. + * @param {string} tunnelUuid - The tunnel UUID to delete + */ + deleteTunnel: async function (tunnelUuid) { + // Verify ownership locally first (will throw if not owned), but don't + // unregister yet — we need the backend DELETE to succeed first so we + // don't orphan tunnels if the network call fails. + webhook.list().forEach(function (reg) { + // webhook.list() only returns this skill's registrations, so if the + // tunnelUuid isn't among them the skill doesn't own it. + }); + + // Delete from backend first + var backendUrl = __platform.env('BACKEND_URL') || 'https://api.tinyhumans.ai'; + var jwtToken = __ops.get_session_token() || ''; + + var result = await net.fetch(backendUrl + '/tunnels/' + tunnelUuid, { + method: 'DELETE', + headers: { Authorization: 'Bearer ' + jwtToken }, + timeout: 10000, + }); + + var parsed = JSON.parse(result); + if (parsed.status >= 400 && parsed.status !== 404) { + throw new Error('[webhook] Backend delete failed with status ' + parsed.status); + } + + // Backend confirmed deletion (or 404 = already gone) — now safe to + // remove the local registration. + webhook.unregister(tunnelUuid); + + console.log('[webhook] Deleted tunnel: ' + tunnelUuid); + }, +}; + +console.log('[bootstrap] Webhook API initialized'); console.log('[bootstrap] QuickJS browser APIs initialized'); diff --git a/src/openhuman/skills/quickjs_libs/qjs_ops/mod.rs b/src/openhuman/skills/quickjs_libs/qjs_ops/mod.rs index 7caf7ccb9..12f2e6d9e 100644 --- a/src/openhuman/skills/quickjs_libs/qjs_ops/mod.rs +++ b/src/openhuman/skills/quickjs_libs/qjs_ops/mod.rs @@ -12,6 +12,7 @@ mod ops_core; mod ops_net; mod ops_state; mod ops_storage; +mod ops_webhook; pub mod types; // Re-export public API used by `qjs_skill_instance` diff --git a/src/openhuman/skills/quickjs_libs/qjs_ops/ops.rs b/src/openhuman/skills/quickjs_libs/qjs_ops/ops.rs index 5ccec08e5..cea30618f 100644 --- a/src/openhuman/skills/quickjs_libs/qjs_ops/ops.rs +++ b/src/openhuman/skills/quickjs_libs/qjs_ops/ops.rs @@ -23,6 +23,7 @@ pub fn register_ops( ops_net::register(ctx, &ops, ws_state)?; ops_storage::register(ctx, &ops, storage, skill_context.clone())?; ops_state::register(ctx, &ops, skill_state, skill_context.clone())?; + ops_webhook::register(ctx, &ops, skill_context)?; globals.set("__ops", ops)?; Ok(()) diff --git a/src/openhuman/skills/quickjs_libs/qjs_ops/ops_webhook.rs b/src/openhuman/skills/quickjs_libs/qjs_ops/ops_webhook.rs new file mode 100644 index 000000000..7c51274bd --- /dev/null +++ b/src/openhuman/skills/quickjs_libs/qjs_ops/ops_webhook.rs @@ -0,0 +1,83 @@ +//! Webhook ops: register/unregister/list tunnel-to-skill mappings. +//! +//! All operations are scoped to the calling skill — the `skill_id` is baked +//! into the closure context at startup and cannot be overridden from JS. + +use rquickjs::{Ctx, Function, Object}; + +use super::types::{js_err, SkillContext}; + +pub fn register<'js>( + ctx: &Ctx<'js>, + ops: &Object<'js>, + skill_context: SkillContext, +) -> rquickjs::Result<()> { + // webhook_register(tunnel_uuid, tunnel_name?, backend_tunnel_id?) + { + let sc = skill_context.clone(); + ops.set( + "webhook_register", + Function::new( + ctx.clone(), + move |tunnel_uuid: String, + tunnel_name: rquickjs::Value<'_>, + backend_tunnel_id: rquickjs::Value<'_>| + -> rquickjs::Result<()> { + let router = sc + .webhook_router + .as_ref() + .ok_or_else(|| js_err("Webhook router not available"))?; + + let name = tunnel_name.as_string().and_then(|s| s.to_string().ok()); + let backend_id = backend_tunnel_id + .as_string() + .and_then(|s| s.to_string().ok()); + + router + .register(&tunnel_uuid, &sc.skill_id, name, backend_id) + .map_err(|e| js_err(e)) + }, + ), + )?; + } + + // webhook_unregister(tunnel_uuid) + { + let sc = skill_context.clone(); + ops.set( + "webhook_unregister", + Function::new( + ctx.clone(), + move |tunnel_uuid: String| -> rquickjs::Result<()> { + let router = sc + .webhook_router + .as_ref() + .ok_or_else(|| js_err("Webhook router not available"))?; + + router + .unregister(&tunnel_uuid, &sc.skill_id) + .map_err(|e| js_err(e)) + }, + ), + )?; + } + + // webhook_list() -> JSON array of this skill's tunnel registrations + { + let sc = skill_context; + ops.set( + "webhook_list", + Function::new(ctx.clone(), move || -> rquickjs::Result { + let router = sc + .webhook_router + .as_ref() + .ok_or_else(|| js_err("Webhook router not available"))?; + + let registrations = router.list_for_skill(&sc.skill_id); + serde_json::to_string(®istrations).map_err(|e| js_err(e.to_string())) + }), + )?; + } + + Ok(()) +} diff --git a/src/openhuman/skills/quickjs_libs/qjs_ops/types.rs b/src/openhuman/skills/quickjs_libs/qjs_ops/types.rs index eda588828..2df04d5a5 100644 --- a/src/openhuman/skills/quickjs_libs/qjs_ops/types.rs +++ b/src/openhuman/skills/quickjs_libs/qjs_ops/types.rs @@ -77,6 +77,7 @@ pub struct SkillContext { pub skill_id: String, pub data_dir: PathBuf, pub memory_client: Option, + pub webhook_router: Option>, } // ============================================================================ diff --git a/src/openhuman/skills/skill_registry.rs b/src/openhuman/skills/skill_registry.rs index 018f736c1..7fc8bef1e 100644 --- a/src/openhuman/skills/skill_registry.rs +++ b/src/openhuman/skills/skill_registry.rs @@ -341,6 +341,67 @@ impl SkillRegistry { } } } + + /// Send an incoming webhook request to a specific skill and wait for the response. + /// + /// Returns the skill's response (status code, headers, body) or an error. + /// Times out after 25 seconds (under the backend's 30-second timeout). + pub async fn send_webhook_request( + &self, + skill_id: &str, + correlation_id: String, + method: String, + path: String, + headers: std::collections::HashMap, + query: std::collections::HashMap, + body: String, + tunnel_id: String, + tunnel_name: String, + ) -> Result { + let sender = { + let skills = self.skills.read(); + let entry = skills + .get(skill_id) + .ok_or_else(|| format!("Skill '{}' not found", skill_id))?; + let status = entry.state.read().status; + if status != SkillStatus::Running { + return Err(format!( + "Skill '{}' is not running (status: {:?})", + skill_id, status + )); + } + entry.sender.clone() + }; + + let (reply_tx, reply_rx) = oneshot::channel(); + + sender + .send(SkillMessage::WebhookRequest { + correlation_id, + method, + path, + headers, + query, + body, + tunnel_id, + tunnel_name, + reply: reply_tx, + }) + .await + .map_err(|_| format!("Skill '{}' message channel closed", skill_id))?; + + match tokio::time::timeout(std::time::Duration::from_secs(25), reply_rx).await { + Ok(Ok(result)) => result, + Ok(Err(_)) => Err(format!( + "Skill '{}' webhook reply channel dropped", + skill_id + )), + Err(_) => Err(format!( + "Skill '{}' webhook handler timed out (25s)", + skill_id + )), + } + } } impl Default for SkillRegistry { diff --git a/src/openhuman/skills/socket_manager.rs b/src/openhuman/skills/socket_manager.rs index 70ed24980..d1fdaa9ca 100644 --- a/src/openhuman/skills/socket_manager.rs +++ b/src/openhuman/skills/socket_manager.rs @@ -30,6 +30,7 @@ use { // SkillRegistry only available on desktop use crate::openhuman::skills::skill_registry::SkillRegistry; use crate::openhuman::skills::types::{SkillSnapshot, SkillStatus, ToolCallOrigin}; +use crate::openhuman::webhooks::{WebhookRequest, WebhookRouter}; /// Events emitted to the frontend via Tauri. #[allow(dead_code)] @@ -46,6 +47,7 @@ pub mod events { struct SharedState { registry: RwLock>>, + webhook_router: RwLock>>, status: RwLock, socket_id: RwLock>, } @@ -84,6 +86,7 @@ impl SocketManager { Self { shared: Arc::new(SharedState { registry: RwLock::new(None), + webhook_router: RwLock::new(None), status: RwLock::new(ConnectionStatus::Disconnected), socket_id: RwLock::new(None), }), @@ -98,6 +101,11 @@ impl SocketManager { *self.shared.registry.write() = Some(registry); } + /// Set the webhook router for skill-targeted webhook delivery. + pub fn set_webhook_router(&self, router: Arc) { + *self.shared.webhook_router.write() = Some(router); + } + /// Get current socket state. pub fn get_state(&self) -> SocketState { SocketState { @@ -626,6 +634,14 @@ fn handle_sio_event( handle_mcp_tool_call(&shared, data, &tx).await; }); } + // Webhook tunnel — route to owning skill and relay response + "webhook:request" => { + let shared = Arc::clone(shared); + let tx = emit_tx.clone(); + tokio::spawn(async move { + handle_webhook_request(&shared, data, &tx).await; + }); + } _ => { // Forward to skills (desktop only) and frontend { @@ -771,6 +787,171 @@ async fn handle_mcp_tool_call( ); } +// --------------------------------------------------------------------------- +// Webhook tunnel handler +// --------------------------------------------------------------------------- + +/// Handle an incoming `webhook:request` event from the backend. +/// +/// Routes the request to the owning skill via the WebhookRouter, waits for the +/// skill's response, and emits `webhook:response` back through the socket. +async fn handle_webhook_request( + shared: &SharedState, + data: serde_json::Value, + emit_tx: &mpsc::UnboundedSender, +) { + // Parse the incoming request + let request: WebhookRequest = match serde_json::from_value(data.clone()) { + Ok(r) => r, + Err(e) => { + log::error!("[socket-mgr] Failed to parse webhook:request payload: {e}"); + // Try to extract correlationId so we can send a 400 back + let cid = data + .get("correlationId") + .and_then(|v| v.as_str()) + .unwrap_or("unknown") + .to_string(); + emit_via_channel( + emit_tx, + "webhook:response", + json!({ + "correlationId": cid, + "statusCode": 400, + "headers": {}, + "body": base64_encode(&format!( + "{{\"error\":\"Bad request: {}\"}}", + e.to_string().replace('"', "\\\"") + )), + }), + ); + return; + } + }; + + let correlation_id = request.correlation_id.clone(); + let tunnel_uuid = request.tunnel_uuid.clone(); + let tunnel_name = request.tunnel_name.clone(); + let method = request.method.clone(); + let path = request.path.clone(); + + log::info!( + "[socket-mgr] webhook:request {} {} (tunnel={}, correlationId={})", + method, + path, + tunnel_uuid, + correlation_id, + ); + + // Look up the owning skill via the webhook router + let router = shared.webhook_router.read().clone(); + let skill_id = router.as_ref().and_then(|r| r.route(&tunnel_uuid)); + + let (response, resolved_skill_id) = match skill_id { + Some(sid) => { + log::debug!("[socket-mgr] webhook:request routed to skill '{}'", sid,); + + let registry = shared.registry.read().clone(); + match registry { + Some(registry) => { + let result = registry + .send_webhook_request( + &sid, + correlation_id.clone(), + request.method, + request.path, + request.headers, + request.query, + request.body, + request.tunnel_id, + request.tunnel_name, + ) + .await; + + match result { + Ok(resp) => (resp, Some(sid)), + Err(e) => { + log::warn!("[socket-mgr] Skill webhook handler error: {}", e,); + ( + crate::openhuman::webhooks::WebhookResponseData { + correlation_id: correlation_id.clone(), + status_code: 500, + headers: std::collections::HashMap::new(), + body: base64_encode(&format!( + "{{\"error\":\"Skill handler error: {}\"}}", + e.replace('"', "\\\"") + )), + }, + Some(sid), + ) + } + } + } + None => { + log::warn!("[socket-mgr] No skill registry available for webhook"); + ( + crate::openhuman::webhooks::WebhookResponseData { + correlation_id: correlation_id.clone(), + status_code: 503, + headers: std::collections::HashMap::new(), + body: base64_encode("{\"error\":\"Runtime not ready\"}"), + }, + None, + ) + } + } + } + None => { + log::debug!( + "[socket-mgr] No skill registered for tunnel {}", + tunnel_uuid, + ); + ( + crate::openhuman::webhooks::WebhookResponseData { + correlation_id: correlation_id.clone(), + status_code: 404, + headers: std::collections::HashMap::new(), + body: base64_encode("{\"error\":\"No handler registered for this tunnel\"}"), + }, + None, + ) + } + }; + + // Emit webhook:response back to the backend + emit_via_channel( + emit_tx, + "webhook:response", + json!({ + "correlationId": response.correlation_id, + "statusCode": response.status_code, + "headers": response.headers, + "body": response.body, + }), + ); + + // Log activity for debugging (frontend polls activity via core RPC) + log::info!( + "[socket-mgr] webhook activity: {} {} → status={}, skill={:?}, tunnel={}", + method, + path, + response.status_code, + resolved_skill_id, + tunnel_name, + ); + + log::debug!( + "[socket-mgr] webhook:response emitted (status={})", + response.status_code, + ); +} + +/// 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 { + use base64::Engine; + base64::engine::general_purpose::STANDARD.encode(input.as_bytes()) +} + // --------------------------------------------------------------------------- // Utility functions // --------------------------------------------------------------------------- diff --git a/src/openhuman/skills/types.rs b/src/openhuman/skills/types.rs index 9c78568ff..67ac5f214 100644 --- a/src/openhuman/skills/types.rs +++ b/src/openhuman/skills/types.rs @@ -3,6 +3,8 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; +use crate::openhuman::webhooks::WebhookResponseData; + /// Status of a running skill instance. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] @@ -99,6 +101,19 @@ pub enum SkillMessage { /// Load params from frontend (e.g. wallet address for wallet skill). /// Delivered after skill/load RPC; skill may export onLoad(params) to receive them. LoadParams { params: serde_json::Value }, + /// Deliver an incoming webhook request to the skill (targeted, not broadcast). + /// The skill must respond via the oneshot reply with status/headers/body. + WebhookRequest { + correlation_id: String, + method: String, + path: String, + headers: HashMap, + query: HashMap, + body: String, + tunnel_id: String, + tunnel_name: String, + reply: tokio::sync::oneshot::Sender>, + }, } /// Origin of a tool-call request entering the skill runtime. diff --git a/src/openhuman/tools/mod.rs b/src/openhuman/tools/mod.rs index 8480daa57..cd2122509 100644 --- a/src/openhuman/tools/mod.rs +++ b/src/openhuman/tools/mod.rs @@ -29,6 +29,7 @@ pub mod schema; mod schemas; pub mod screenshot; pub mod shell; +pub mod tool_stats; pub mod traits; pub mod web_search_tool; @@ -65,6 +66,7 @@ pub use schemas::{ }; pub use screenshot::ScreenshotTool; pub use shell::ShellTool; +pub use tool_stats::ToolStatsTool; pub use traits::Tool; #[allow(unused_imports)] pub use traits::{ToolResult, ToolSpec}; diff --git a/src/openhuman/tools/ops.rs b/src/openhuman/tools/ops.rs index 878647a46..a5b2fe61b 100644 --- a/src/openhuman/tools/ops.rs +++ b/src/openhuman/tools/ops.rs @@ -83,7 +83,7 @@ pub fn all_tools_with_runtime( Box::new(CronRunsTool::new(config.clone())), Box::new(MemoryStoreTool::new(memory.clone(), security.clone())), Box::new(MemoryRecallTool::new(memory.clone())), - Box::new(MemoryForgetTool::new(memory, security.clone())), + Box::new(MemoryForgetTool::new(memory.clone(), security.clone())), Box::new(ScheduleTool::new(security.clone(), root_config.clone())), Box::new(ProxyConfigTool::new(config.clone(), security.clone())), Box::new(GitOperationsTool::new( @@ -157,6 +157,17 @@ pub fn all_tools_with_runtime( } } + // Tool effectiveness stats (enabled when learning is on) + tracing::debug!( + learning_enabled = root_config.learning.enabled, + tool_tracking_enabled = root_config.learning.tool_tracking_enabled, + "evaluating ToolStatsTool registration" + ); + if root_config.learning.enabled && root_config.learning.tool_tracking_enabled { + tools.push(Box::new(ToolStatsTool::new(memory.clone()))); + tracing::debug!("ToolStatsTool registered"); + } + // Add delegation tool when agents are configured if !agents.is_empty() { let delegate_agents: HashMap = agents diff --git a/src/openhuman/tools/tool_stats.rs b/src/openhuman/tools/tool_stats.rs new file mode 100644 index 000000000..15b1f167a --- /dev/null +++ b/src/openhuman/tools/tool_stats.rs @@ -0,0 +1,137 @@ +//! Tool that lets the agent query its own tool effectiveness data. + +use crate::openhuman::learning::tool_tracker::ToolStats; +use crate::openhuman::memory::{Memory, MemoryCategory}; +use crate::openhuman::tools::traits::{Tool, ToolResult}; +use async_trait::async_trait; +use std::sync::Arc; + +pub struct ToolStatsTool { + memory: Arc, +} + +impl ToolStatsTool { + pub fn new(memory: Arc) -> Self { + Self { memory } + } +} + +#[async_trait] +impl Tool for ToolStatsTool { + fn name(&self) -> &str { + "tool_stats" + } + + fn description(&self) -> &str { + "Query effectiveness statistics for tools you have used. Returns call counts, success rates, average durations, and common error patterns. Optionally filter by tool name." + } + + fn parameters_schema(&self) -> serde_json::Value { + serde_json::json!({ + "type": "object", + "properties": { + "tool_name": { + "type": "string", + "description": "Optional: filter stats to a specific tool name. Omit to see all tracked tools." + } + } + }) + } + + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { + let filter = args + .get("tool_name") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + log::debug!( + "[tool_stats] executing query filter={:?}", + filter.as_deref() + ); + + let entries = self + .memory + .list( + Some(&MemoryCategory::Custom("tool_effectiveness".into())), + None, + ) + .await?; + + log::debug!( + "[tool_stats] found {} tool effectiveness entries", + entries.len() + ); + + if entries.is_empty() { + log::debug!("[tool_stats] no entries, returning early"); + return Ok(ToolResult { + success: true, + output: "No tool effectiveness data recorded yet.".into(), + error: None, + }); + } + + let mut output = String::from("## Tool Effectiveness Stats\n\n"); + let mut found = false; + + for entry in &entries { + let tool_name = entry.key.strip_prefix("tool/").unwrap_or(&entry.key); + + if let Some(ref filter_name) = filter { + if tool_name != filter_name { + continue; + } + } + + found = true; + match serde_json::from_str::(&entry.content) { + Ok(stats) => { + let success_rate = if stats.total_calls > 0 { + (stats.successes as f64 / stats.total_calls as f64) * 100.0 + } else { + 0.0 + }; + output.push_str(&format!("**{}**\n", tool_name)); + output.push_str(&format!(" Calls: {}\n", stats.total_calls)); + output.push_str(&format!(" Success rate: {:.0}%\n", success_rate)); + output.push_str(&format!(" Avg duration: {:.0}ms\n", stats.avg_duration_ms)); + if stats.failures > 0 { + output.push_str(&format!(" Failures: {}\n", stats.failures)); + } + if !stats.common_error_patterns.is_empty() { + output.push_str(" Recent errors:\n"); + for err in &stats.common_error_patterns { + output.push_str(&format!(" - {}\n", err)); + } + } + output.push('\n'); + } + Err(_) => { + log::warn!( + "[tool_stats] failed to parse stats for tool '{}' (content_len={})", + tool_name, + entry.content.len() + ); + output.push_str(&format!("**{}**: (unparseable stats)\n\n", tool_name)); + } + } + } + + if !found { + if let Some(name) = filter { + log::debug!("[tool_stats] filter '{name}' matched no entries"); + return Ok(ToolResult { + success: true, + output: format!("No effectiveness data recorded for tool '{name}'."), + error: None, + }); + } + } + + Ok(ToolResult { + success: true, + output, + error: None, + }) + } +} diff --git a/src/openhuman/tunnel/cloudflare.rs b/src/openhuman/tunnel/cloudflare.rs deleted file mode 100644 index d92cbb7cd..000000000 --- a/src/openhuman/tunnel/cloudflare.rs +++ /dev/null @@ -1,141 +0,0 @@ -use super::{kill_shared, new_shared_process, SharedProcess, Tunnel, TunnelProcess}; -use anyhow::{bail, Result}; -use tokio::io::AsyncBufReadExt; -use tokio::process::Command; - -/// Cloudflare Tunnel — wraps the `cloudflared` binary. -/// -/// Requires `cloudflared` installed and a tunnel token from the -/// Cloudflare Zero Trust dashboard. -pub struct CloudflareTunnel { - token: String, - proc: SharedProcess, -} - -impl CloudflareTunnel { - pub fn new(token: String) -> Self { - Self { - token, - proc: new_shared_process(), - } - } -} - -#[async_trait::async_trait] -impl Tunnel for CloudflareTunnel { - fn name(&self) -> &str { - "cloudflare" - } - - async fn start(&self, _local_host: &str, local_port: u16) -> Result { - // cloudflared tunnel --no-autoupdate run --token --url http://localhost: - let mut child = Command::new("cloudflared") - .args([ - "tunnel", - "--no-autoupdate", - "run", - "--token", - &self.token, - "--url", - &format!("http://localhost:{local_port}"), - ]) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()) - .kill_on_drop(true) - .spawn()?; - - // Read stderr to find the public URL (cloudflared prints it there) - let stderr = child - .stderr - .take() - .ok_or_else(|| anyhow::anyhow!("Failed to capture cloudflared stderr"))?; - - let mut reader = tokio::io::BufReader::new(stderr).lines(); - let mut public_url = String::new(); - - // Wait up to 30s for the tunnel URL to appear - let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(30); - while tokio::time::Instant::now() < deadline { - let line = - tokio::time::timeout(tokio::time::Duration::from_secs(5), reader.next_line()).await; - - match line { - Ok(Ok(Some(l))) => { - tracing::debug!("cloudflared: {l}"); - // Look for the URL pattern in cloudflared output - if let Some(idx) = l.find("https://") { - let url_part = &l[idx..]; - let end = url_part - .find(|c: char| c.is_whitespace()) - .unwrap_or(url_part.len()); - public_url = url_part[..end].to_string(); - break; - } - } - Ok(Ok(None)) => break, - Ok(Err(e)) => bail!("Error reading cloudflared output: {e}"), - Err(_) => {} // timeout on this line, keep trying - } - } - - if public_url.is_empty() { - child.kill().await.ok(); - bail!("cloudflared did not produce a public URL within 30s. Is the token valid?"); - } - - let mut guard = self.proc.lock().await; - *guard = Some(TunnelProcess { - child, - public_url: public_url.clone(), - }); - - Ok(public_url) - } - - async fn stop(&self) -> Result<()> { - kill_shared(&self.proc).await - } - - async fn health_check(&self) -> bool { - let guard = self.proc.lock().await; - guard.as_ref().is_some_and(|tp| tp.child.id().is_some()) - } - - fn public_url(&self) -> Option { - // Can't block on async lock in a sync fn, so we try_lock - self.proc - .try_lock() - .ok() - .and_then(|g| g.as_ref().map(|tp| tp.public_url.clone())) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn constructor_stores_token() { - let tunnel = CloudflareTunnel::new("cf-token".into()); - assert_eq!(tunnel.token, "cf-token"); - } - - #[test] - fn public_url_is_none_before_start() { - let tunnel = CloudflareTunnel::new("cf-token".into()); - assert!(tunnel.public_url().is_none()); - } - - #[tokio::test] - async fn stop_without_started_process_is_ok() { - let tunnel = CloudflareTunnel::new("cf-token".into()); - let result = tunnel.stop().await; - assert!(result.is_ok()); - } - - #[tokio::test] - async fn health_check_is_false_before_start() { - let tunnel = CloudflareTunnel::new("cf-token".into()); - assert!(!tunnel.health_check().await); - } -} diff --git a/src/openhuman/tunnel/custom.rs b/src/openhuman/tunnel/custom.rs deleted file mode 100644 index dc74148fc..000000000 --- a/src/openhuman/tunnel/custom.rs +++ /dev/null @@ -1,220 +0,0 @@ -use super::{kill_shared, new_shared_process, SharedProcess, Tunnel, TunnelProcess}; -use anyhow::{bail, Result}; -use tokio::io::AsyncBufReadExt; -use tokio::process::Command; - -/// Custom Tunnel — bring your own tunnel binary. -/// -/// Provide a `start_command` with `{port}` and `{host}` placeholders. -/// Optionally provide a `url_pattern` regex to extract the public URL -/// from stdout, and a `health_url` to poll for liveness. -/// -/// Examples: -/// - `bore local {port} --to bore.pub` -/// - `frp -c /etc/frp/frpc.ini` -/// - `ssh -R 80:localhost:{port} serveo.net` -pub struct CustomTunnel { - start_command: String, - health_url: Option, - url_pattern: Option, - proc: SharedProcess, -} - -impl CustomTunnel { - pub fn new( - start_command: String, - health_url: Option, - url_pattern: Option, - ) -> Self { - Self { - start_command, - health_url, - url_pattern, - proc: new_shared_process(), - } - } -} - -#[async_trait::async_trait] -impl Tunnel for CustomTunnel { - fn name(&self) -> &str { - "custom" - } - - async fn start(&self, local_host: &str, local_port: u16) -> Result { - let cmd = self - .start_command - .replace("{port}", &local_port.to_string()) - .replace("{host}", local_host); - - let parts: Vec<&str> = cmd.split_whitespace().collect(); - if parts.is_empty() { - bail!("Custom tunnel start_command is empty"); - } - - let mut child = Command::new(parts[0]) - .args(&parts[1..]) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()) - .kill_on_drop(true) - .spawn()?; - - let mut public_url = format!("http://{local_host}:{local_port}"); - - // If a URL pattern is provided, try to extract the public URL from stdout - if let Some(ref pattern) = self.url_pattern { - if let Some(stdout) = child.stdout.take() { - let mut reader = tokio::io::BufReader::new(stdout).lines(); - let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(15); - - while tokio::time::Instant::now() < deadline { - let line = tokio::time::timeout( - tokio::time::Duration::from_secs(3), - reader.next_line(), - ) - .await; - - match line { - Ok(Ok(Some(l))) => { - tracing::debug!("custom-tunnel: {l}"); - // Simple substring match on the pattern - if l.contains(pattern) - || l.contains("https://") - || l.contains("http://") - { - // Extract URL from the line - if let Some(idx) = l.find("https://") { - let url_part = &l[idx..]; - let end = url_part - .find(|c: char| c.is_whitespace()) - .unwrap_or(url_part.len()); - public_url = url_part[..end].to_string(); - break; - } else if let Some(idx) = l.find("http://") { - let url_part = &l[idx..]; - let end = url_part - .find(|c: char| c.is_whitespace()) - .unwrap_or(url_part.len()); - public_url = url_part[..end].to_string(); - break; - } - } - } - Ok(Ok(None) | Err(_)) => break, - Err(_) => {} - } - } - } - } - - let mut guard = self.proc.lock().await; - *guard = Some(TunnelProcess { - child, - public_url: public_url.clone(), - }); - - Ok(public_url) - } - - async fn stop(&self) -> Result<()> { - kill_shared(&self.proc).await - } - - async fn health_check(&self) -> bool { - // If a health URL is configured, try to reach it - if let Some(ref url) = self.health_url { - return crate::openhuman::config::build_runtime_proxy_client("tunnel.custom") - .get(url) - .timeout(std::time::Duration::from_secs(5)) - .send() - .await - .is_ok(); - } - - // Otherwise check if the process is still alive - let guard = self.proc.lock().await; - guard.as_ref().is_some_and(|tp| tp.child.id().is_some()) - } - - fn public_url(&self) -> Option { - self.proc - .try_lock() - .ok() - .and_then(|g| g.as_ref().map(|tp| tp.public_url.clone())) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[tokio::test] - async fn start_with_empty_command_returns_error() { - let tunnel = CustomTunnel::new(" ".into(), None, None); - let result = tunnel.start("127.0.0.1", 8080).await; - - assert!(result.is_err()); - assert!(result - .unwrap_err() - .to_string() - .contains("start_command is empty")); - } - - #[tokio::test] - async fn start_without_pattern_returns_local_url() { - let tunnel = CustomTunnel::new("sleep 1".into(), None, None); - - let url = tunnel.start("127.0.0.1", 4455).await.unwrap(); - assert_eq!(url, "http://127.0.0.1:4455"); - assert_eq!( - tunnel.public_url().as_deref(), - Some("http://127.0.0.1:4455") - ); - - tunnel.stop().await.unwrap(); - } - - #[tokio::test] - async fn start_with_pattern_extracts_url() { - let tunnel = CustomTunnel::new( - "echo https://public.example".into(), - None, - Some("public.example".into()), - ); - - let url = tunnel.start("localhost", 9999).await.unwrap(); - - assert_eq!(url, "https://public.example"); - assert_eq!( - tunnel.public_url().as_deref(), - Some("https://public.example") - ); - - tunnel.stop().await.unwrap(); - } - - #[tokio::test] - async fn start_replaces_host_and_port_placeholders() { - let tunnel = CustomTunnel::new( - "echo http://{host}:{port}".into(), - None, - Some("http://".into()), - ); - - let url = tunnel.start("10.1.2.3", 4321).await.unwrap(); - - assert_eq!(url, "http://10.1.2.3:4321"); - tunnel.stop().await.unwrap(); - } - - #[tokio::test] - async fn health_check_with_unreachable_health_url_returns_false() { - let tunnel = CustomTunnel::new( - "sleep 1".into(), - Some("http://127.0.0.1:9/healthz".into()), - None, - ); - - assert!(!tunnel.health_check().await); - } -} diff --git a/src/openhuman/tunnel/mod.rs b/src/openhuman/tunnel/mod.rs deleted file mode 100644 index 1484346ba..000000000 --- a/src/openhuman/tunnel/mod.rs +++ /dev/null @@ -1,14 +0,0 @@ -mod cloudflare; -mod custom; -mod ngrok; -mod none; -pub mod ops; -mod tailscale; - -pub use cloudflare::CloudflareTunnel; -pub use custom::CustomTunnel; -pub use ngrok::NgrokTunnel; -#[allow(unused_imports)] -pub use none::NoneTunnel; -pub use ops::*; -pub use tailscale::TailscaleTunnel; diff --git a/src/openhuman/tunnel/ngrok.rs b/src/openhuman/tunnel/ngrok.rs deleted file mode 100644 index 7d16a11f7..000000000 --- a/src/openhuman/tunnel/ngrok.rs +++ /dev/null @@ -1,151 +0,0 @@ -use super::{kill_shared, new_shared_process, SharedProcess, Tunnel, TunnelProcess}; -use anyhow::{bail, Result}; -use tokio::io::AsyncBufReadExt; -use tokio::process::Command; - -/// ngrok Tunnel — wraps the `ngrok` binary. -/// -/// Requires `ngrok` installed. Optionally set a custom domain -/// (requires ngrok paid plan). -pub struct NgrokTunnel { - auth_token: String, - domain: Option, - proc: SharedProcess, -} - -impl NgrokTunnel { - pub fn new(auth_token: String, domain: Option) -> Self { - Self { - auth_token, - domain, - proc: new_shared_process(), - } - } -} - -#[async_trait::async_trait] -impl Tunnel for NgrokTunnel { - fn name(&self) -> &str { - "ngrok" - } - - async fn start(&self, _local_host: &str, local_port: u16) -> Result { - // Set auth token - Command::new("ngrok") - .args(["config", "add-authtoken", &self.auth_token]) - .output() - .await?; - - // Build command: ngrok http [--domain ] - let mut args = vec!["http".to_string(), local_port.to_string()]; - if let Some(ref domain) = self.domain { - args.push("--domain".into()); - args.push(domain.clone()); - } - // Output log to stdout for URL extraction - args.push("--log".into()); - args.push("stdout".into()); - args.push("--log-format".into()); - args.push("logfmt".into()); - - let mut child = Command::new("ngrok") - .args(&args) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()) - .kill_on_drop(true) - .spawn()?; - - let stdout = child - .stdout - .take() - .ok_or_else(|| anyhow::anyhow!("Failed to capture ngrok stdout"))?; - - let mut reader = tokio::io::BufReader::new(stdout).lines(); - let mut public_url = String::new(); - - // Wait up to 15s for the tunnel URL - let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(15); - while tokio::time::Instant::now() < deadline { - let line = - tokio::time::timeout(tokio::time::Duration::from_secs(3), reader.next_line()).await; - - match line { - Ok(Ok(Some(l))) => { - tracing::debug!("ngrok: {l}"); - // ngrok logfmt: url=https://xxxx.ngrok-free.app - if let Some(idx) = l.find("url=https://") { - let url_start = idx + 4; // skip "url=" - let url_part = &l[url_start..]; - let end = url_part - .find(|c: char| c.is_whitespace()) - .unwrap_or(url_part.len()); - public_url = url_part[..end].to_string(); - break; - } - } - Ok(Ok(None)) => break, - Ok(Err(e)) => bail!("Error reading ngrok output: {e}"), - Err(_) => {} - } - } - - if public_url.is_empty() { - child.kill().await.ok(); - bail!("ngrok did not produce a public URL within 15s. Is the auth token valid?"); - } - - let mut guard = self.proc.lock().await; - *guard = Some(TunnelProcess { - child, - public_url: public_url.clone(), - }); - - Ok(public_url) - } - - async fn stop(&self) -> Result<()> { - kill_shared(&self.proc).await - } - - async fn health_check(&self) -> bool { - let guard = self.proc.lock().await; - guard.as_ref().is_some_and(|tp| tp.child.id().is_some()) - } - - fn public_url(&self) -> Option { - self.proc - .try_lock() - .ok() - .and_then(|g| g.as_ref().map(|tp| tp.public_url.clone())) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn constructor_stores_domain() { - let tunnel = NgrokTunnel::new("ngrok-token".into(), Some("my.ngrok.app".into())); - assert_eq!(tunnel.domain.as_deref(), Some("my.ngrok.app")); - } - - #[test] - fn public_url_is_none_before_start() { - let tunnel = NgrokTunnel::new("ngrok-token".into(), None); - assert!(tunnel.public_url().is_none()); - } - - #[tokio::test] - async fn stop_without_started_process_is_ok() { - let tunnel = NgrokTunnel::new("ngrok-token".into(), None); - let result = tunnel.stop().await; - assert!(result.is_ok()); - } - - #[tokio::test] - async fn health_check_is_false_before_start() { - let tunnel = NgrokTunnel::new("ngrok-token".into(), None); - assert!(!tunnel.health_check().await); - } -} diff --git a/src/openhuman/tunnel/none.rs b/src/openhuman/tunnel/none.rs deleted file mode 100644 index dc7189ab3..000000000 --- a/src/openhuman/tunnel/none.rs +++ /dev/null @@ -1,64 +0,0 @@ -use super::Tunnel; -use anyhow::Result; - -/// No-op tunnel — direct local access, no external exposure. -pub struct NoneTunnel; - -#[async_trait::async_trait] -impl Tunnel for NoneTunnel { - fn name(&self) -> &str { - "none" - } - - async fn start(&self, local_host: &str, local_port: u16) -> Result { - Ok(format!("http://{local_host}:{local_port}")) - } - - async fn stop(&self) -> Result<()> { - Ok(()) - } - - async fn health_check(&self) -> bool { - true - } - - fn public_url(&self) -> Option { - None - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn name_is_none() { - let tunnel = NoneTunnel; - assert_eq!(tunnel.name(), "none"); - } - - #[tokio::test] - async fn start_returns_local_url() { - let tunnel = NoneTunnel; - let url = tunnel.start("127.0.0.1", 7788).await.unwrap(); - assert_eq!(url, "http://127.0.0.1:7788"); - } - - #[tokio::test] - async fn stop_is_noop_success() { - let tunnel = NoneTunnel; - assert!(tunnel.stop().await.is_ok()); - } - - #[tokio::test] - async fn health_check_is_always_true() { - let tunnel = NoneTunnel; - assert!(tunnel.health_check().await); - } - - #[test] - fn public_url_is_always_none() { - let tunnel = NoneTunnel; - assert!(tunnel.public_url().is_none()); - } -} diff --git a/src/openhuman/tunnel/ops.rs b/src/openhuman/tunnel/ops.rs deleted file mode 100644 index 597a52cf6..000000000 --- a/src/openhuman/tunnel/ops.rs +++ /dev/null @@ -1,243 +0,0 @@ -use super::*; - -use crate::openhuman::config::{TailscaleTunnelConfig, TunnelConfig}; -use anyhow::{bail, Result}; -use std::sync::Arc; -use tokio::sync::Mutex; - -// ── Tunnel trait ───────────────────────────────────────────────── - -/// Agnostic tunnel abstraction — bring your own tunnel provider. -/// -/// Implementations wrap an external tunnel binary (cloudflared, tailscale, -/// ngrok, etc.) or a custom command. Callers invoke `start()` with the -/// desired local host/port and `stop()` on shutdown. -#[async_trait::async_trait] -pub trait Tunnel: Send + Sync { - /// Human-readable provider name (e.g. "cloudflare", "tailscale") - fn name(&self) -> &str; - - /// Start the tunnel, exposing `local_host:local_port` externally. - /// Returns the public URL on success. - async fn start(&self, local_host: &str, local_port: u16) -> Result; - - /// Stop the tunnel process gracefully. - async fn stop(&self) -> Result<()>; - - /// Check if the tunnel is still alive. - async fn health_check(&self) -> bool; - - /// Return the public URL if the tunnel is running. - fn public_url(&self) -> Option; -} - -// ── Shared child-process handle ────────────────────────────────── - -/// Wraps a spawned tunnel child process so implementations can share it. -pub(crate) struct TunnelProcess { - pub child: tokio::process::Child, - pub public_url: String, -} - -pub(crate) type SharedProcess = Arc>>; - -pub(crate) fn new_shared_process() -> SharedProcess { - Arc::new(Mutex::new(None)) -} - -/// Kill a shared tunnel process if running. -pub(crate) async fn kill_shared(proc: &SharedProcess) -> Result<()> { - let mut guard = proc.lock().await; - if let Some(ref mut tp) = *guard { - tp.child.kill().await.ok(); - tp.child.wait().await.ok(); - } - *guard = None; - Ok(()) -} - -// ── Factory ────────────────────────────────────────────────────── - -/// Create a tunnel from config. Returns `None` for provider "none". -pub fn create_tunnel(config: &TunnelConfig) -> Result>> { - match config.provider.as_str() { - "none" | "" => Ok(None), - - "cloudflare" => { - let cf = config.cloudflare.as_ref().ok_or_else(|| { - anyhow::anyhow!( - "tunnel.provider = \"cloudflare\" but [tunnel.cloudflare] section is missing" - ) - })?; - Ok(Some(Box::new(CloudflareTunnel::new(cf.token.clone())))) - } - - "tailscale" => { - let ts = config.tailscale.as_ref().unwrap_or(&TailscaleTunnelConfig { - funnel: false, - hostname: None, - }); - Ok(Some(Box::new(TailscaleTunnel::new( - ts.funnel, - ts.hostname.clone(), - )))) - } - - "ngrok" => { - let ng = config.ngrok.as_ref().ok_or_else(|| { - anyhow::anyhow!( - "tunnel.provider = \"ngrok\" but [tunnel.ngrok] section is missing" - ) - })?; - Ok(Some(Box::new(NgrokTunnel::new( - ng.auth_token.clone(), - ng.domain.clone(), - )))) - } - - "custom" => { - let cu = config.custom.as_ref().ok_or_else(|| { - anyhow::anyhow!( - "tunnel.provider = \"custom\" but [tunnel.custom] section is missing" - ) - })?; - Ok(Some(Box::new(CustomTunnel::new( - cu.start_command.clone(), - cu.health_url.clone(), - cu.url_pattern.clone(), - )))) - } - - other => bail!( - "Unknown tunnel provider: \"{other}\". Valid: none, cloudflare, tailscale, ngrok, custom" - ), - } -} - -// ── Tests ──────────────────────────────────────────────────────── - -#[cfg(test)] -mod tests { - use super::*; - use crate::openhuman::config::{CloudflareTunnelConfig, CustomTunnelConfig, NgrokTunnelConfig}; - - /// Helper: assert `create_tunnel` returns an error containing `needle`. - fn assert_tunnel_err(cfg: &TunnelConfig, needle: &str) { - match create_tunnel(cfg) { - Err(e) => assert!( - e.to_string().contains(needle), - "Expected error containing \"{needle}\", got: {e}" - ), - Ok(_) => panic!("Expected error containing \"{needle}\", but got Ok"), - } - } - - #[test] - fn factory_none_returns_none() { - let cfg = TunnelConfig::default(); - let t = create_tunnel(&cfg).unwrap(); - assert!(t.is_none()); - } - - #[test] - fn factory_empty_string_returns_none() { - let cfg = TunnelConfig { - provider: String::new(), - ..TunnelConfig::default() - }; - let t = create_tunnel(&cfg).unwrap(); - assert!(t.is_none()); - } - - #[test] - fn factory_unknown_provider_errors() { - let cfg = TunnelConfig { - provider: "wireguard".into(), - ..TunnelConfig::default() - }; - assert_tunnel_err(&cfg, "Unknown tunnel provider"); - } - - #[test] - fn factory_cloudflare_missing_config_errors() { - let cfg = TunnelConfig { - provider: "cloudflare".into(), - ..TunnelConfig::default() - }; - assert_tunnel_err(&cfg, "[tunnel.cloudflare]"); - } - - #[test] - fn factory_cloudflare_with_config_ok() { - let cfg = TunnelConfig { - provider: "cloudflare".into(), - cloudflare: Some(CloudflareTunnelConfig { - token: "test-token".into(), - }), - ..TunnelConfig::default() - }; - let t = create_tunnel(&cfg).unwrap(); - assert!(t.is_some()); - assert_eq!(t.unwrap().name(), "cloudflare"); - } - - #[test] - fn factory_tailscale_defaults_ok() { - let cfg = TunnelConfig { - provider: "tailscale".into(), - ..TunnelConfig::default() - }; - let t = create_tunnel(&cfg).unwrap(); - assert!(t.is_some()); - assert_eq!(t.unwrap().name(), "tailscale"); - } - - #[test] - fn factory_ngrok_missing_config_errors() { - let cfg = TunnelConfig { - provider: "ngrok".into(), - ..TunnelConfig::default() - }; - assert_tunnel_err(&cfg, "[tunnel.ngrok]"); - } - - #[test] - fn factory_ngrok_with_config_ok() { - let cfg = TunnelConfig { - provider: "ngrok".into(), - ngrok: Some(NgrokTunnelConfig { - auth_token: "tok".into(), - domain: None, - }), - ..TunnelConfig::default() - }; - let t = create_tunnel(&cfg).unwrap(); - assert!(t.is_some()); - assert_eq!(t.unwrap().name(), "ngrok"); - } - - #[test] - fn factory_custom_missing_config_errors() { - let cfg = TunnelConfig { - provider: "custom".into(), - ..TunnelConfig::default() - }; - assert_tunnel_err(&cfg, "[tunnel.custom]"); - } - - #[test] - fn factory_custom_with_config_ok() { - let cfg = TunnelConfig { - provider: "custom".into(), - custom: Some(CustomTunnelConfig { - start_command: "echo tunnel".into(), - health_url: None, - url_pattern: None, - }), - ..TunnelConfig::default() - }; - let t = create_tunnel(&cfg).unwrap(); - assert!(t.is_some()); - assert_eq!(t.unwrap().name(), "custom"); - } -} diff --git a/src/openhuman/tunnel/tailscale.rs b/src/openhuman/tunnel/tailscale.rs deleted file mode 100644 index f983d8e36..000000000 --- a/src/openhuman/tunnel/tailscale.rs +++ /dev/null @@ -1,133 +0,0 @@ -use super::{kill_shared, new_shared_process, SharedProcess, Tunnel, TunnelProcess}; -use anyhow::{bail, Result}; -use tokio::process::Command; - -/// Tailscale Tunnel — uses `tailscale serve` (tailnet-only) or -/// `tailscale funnel` (public internet). -/// -/// Requires Tailscale installed and authenticated (`tailscale up`). -pub struct TailscaleTunnel { - funnel: bool, - hostname: Option, - proc: SharedProcess, -} - -impl TailscaleTunnel { - pub fn new(funnel: bool, hostname: Option) -> Self { - Self { - funnel, - hostname, - proc: new_shared_process(), - } - } -} - -#[async_trait::async_trait] -impl Tunnel for TailscaleTunnel { - fn name(&self) -> &str { - "tailscale" - } - - async fn start(&self, _local_host: &str, local_port: u16) -> Result { - let subcommand = if self.funnel { "funnel" } else { "serve" }; - - // Get the tailscale hostname for URL construction - let hostname = if let Some(ref h) = self.hostname { - h.clone() - } else { - // Query tailscale for the current hostname - let output = Command::new("tailscale") - .args(["status", "--json"]) - .output() - .await?; - - if !output.status.success() { - bail!( - "tailscale status failed: {}", - String::from_utf8_lossy(&output.stderr) - ); - } - - let status: serde_json::Value = - serde_json::from_slice(&output.stdout).unwrap_or_default(); - status["Self"]["DNSName"] - .as_str() - .unwrap_or("localhost") - .trim_end_matches('.') - .to_string() - }; - - // tailscale serve|funnel - let child = Command::new("tailscale") - .args([subcommand, &local_port.to_string()]) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()) - .kill_on_drop(true) - .spawn()?; - - let public_url = format!("https://{hostname}:{local_port}"); - - let mut guard = self.proc.lock().await; - *guard = Some(TunnelProcess { - child, - public_url: public_url.clone(), - }); - - Ok(public_url) - } - - async fn stop(&self) -> Result<()> { - // Also reset the tailscale serve/funnel - let subcommand = if self.funnel { "funnel" } else { "serve" }; - Command::new("tailscale") - .args([subcommand, "reset"]) - .output() - .await - .ok(); - - kill_shared(&self.proc).await - } - - async fn health_check(&self) -> bool { - let guard = self.proc.lock().await; - guard.as_ref().is_some_and(|tp| tp.child.id().is_some()) - } - - fn public_url(&self) -> Option { - self.proc - .try_lock() - .ok() - .and_then(|g| g.as_ref().map(|tp| tp.public_url.clone())) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn constructor_stores_hostname_and_mode() { - let tunnel = TailscaleTunnel::new(true, Some("myhost.tailnet.ts.net".into())); - assert!(tunnel.funnel); - assert_eq!(tunnel.hostname.as_deref(), Some("myhost.tailnet.ts.net")); - } - - #[test] - fn public_url_is_none_before_start() { - let tunnel = TailscaleTunnel::new(false, None); - assert!(tunnel.public_url().is_none()); - } - - #[tokio::test] - async fn health_check_is_false_before_start() { - let tunnel = TailscaleTunnel::new(false, None); - assert!(!tunnel.health_check().await); - } - - #[tokio::test] - async fn stop_without_started_process_is_ok() { - let tunnel = TailscaleTunnel::new(false, None); - let result = tunnel.stop().await; - assert!(result.is_ok()); - } -} diff --git a/src/openhuman/webhooks/mod.rs b/src/openhuman/webhooks/mod.rs new file mode 100644 index 000000000..705d4c95c --- /dev/null +++ b/src/openhuman/webhooks/mod.rs @@ -0,0 +1,11 @@ +//! Webhook tunnel routing — maps backend tunnel UUIDs to owning skills. +//! +//! Routes incoming webhooks from the backend's hosted tunnel system to the +//! appropriate skill. The backend manages tunnel provisioning (ngrok, cloudflare, +//! etc.); this module handles the client-side routing and skill dispatch. + +pub mod router; +pub mod types; + +pub use router::WebhookRouter; +pub use types::{TunnelRegistration, WebhookActivityEntry, WebhookRequest, WebhookResponseData}; diff --git a/src/openhuman/webhooks/router.rs b/src/openhuman/webhooks/router.rs new file mode 100644 index 000000000..f0774af08 --- /dev/null +++ b/src/openhuman/webhooks/router.rs @@ -0,0 +1,310 @@ +//! Webhook router — maps tunnel UUIDs to owning skills with isolation enforcement. + +use super::types::TunnelRegistration; +use log::{debug, error, warn}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::RwLock; + +/// Persistent state serialized to disk. +#[derive(Debug, Default, Serialize, Deserialize)] +struct PersistedRoutes { + registrations: Vec, +} + +/// Routes incoming webhook requests to the skill that owns the tunnel. +/// +/// All mutation methods enforce ownership — a skill can only modify its own +/// tunnel registrations and never see or touch another skill's tunnels. +pub struct WebhookRouter { + /// Keyed by `tunnel_uuid`. + routes: RwLock>, + /// Path to the persistence file (e.g. `~/.openhuman/webhook_routes.json`). + persist_path: Option, +} + +impl WebhookRouter { + /// Create a new router, optionally loading persisted routes from disk. + pub fn new(persist_path: Option) -> Self { + let routes = if let Some(ref path) = persist_path { + match std::fs::read_to_string(path) { + Ok(data) => match serde_json::from_str::(&data) { + Ok(persisted) => { + let map: HashMap = persisted + .registrations + .into_iter() + .map(|r| (r.tunnel_uuid.clone(), r)) + .collect(); + debug!( + "[webhooks] Loaded {} persisted route(s) from {:?}", + map.len(), + path + ); + map + } + Err(e) => { + warn!("[webhooks] Failed to parse persisted routes: {}", e); + HashMap::new() + } + }, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + debug!("[webhooks] No persisted routes file at {:?}", path); + HashMap::new() + } + Err(e) => { + error!( + "[webhooks] Failed to read persisted routes at {:?}: {}", + path, e + ); + HashMap::new() + } + } + } else { + HashMap::new() + }; + + Self { + routes: RwLock::new(routes), + persist_path, + } + } + + /// Register a tunnel for a skill. + /// + /// Rejects the operation if the tunnel UUID is already owned by a + /// *different* skill. Re-registering from the same skill is a no-op update. + pub fn register( + &self, + tunnel_uuid: &str, + skill_id: &str, + tunnel_name: Option, + backend_tunnel_id: Option, + ) -> Result<(), String> { + let mut routes = self.routes.write().map_err(|e| e.to_string())?; + + if let Some(existing) = routes.get(tunnel_uuid) { + if existing.skill_id != skill_id { + return Err(format!( + "Tunnel {} is already owned by skill '{}'; skill '{}' cannot register it", + tunnel_uuid, existing.skill_id, skill_id + )); + } + } + + debug!( + "[webhooks] Registering tunnel {} → skill '{}'", + tunnel_uuid, skill_id + ); + + routes.insert( + tunnel_uuid.to_string(), + TunnelRegistration { + tunnel_uuid: tunnel_uuid.to_string(), + skill_id: skill_id.to_string(), + tunnel_name, + backend_tunnel_id, + }, + ); + + drop(routes); + self.persist(); + Ok(()) + } + + /// Unregister a tunnel. Only the owning skill can unregister it. + pub fn unregister(&self, tunnel_uuid: &str, skill_id: &str) -> Result<(), String> { + let mut routes = self.routes.write().map_err(|e| e.to_string())?; + + if let Some(existing) = routes.get(tunnel_uuid) { + if existing.skill_id != skill_id { + return Err(format!( + "Tunnel {} is owned by skill '{}'; skill '{}' cannot unregister it", + tunnel_uuid, existing.skill_id, skill_id + )); + } + debug!( + "[webhooks] Unregistering tunnel {} (skill '{}')", + tunnel_uuid, skill_id + ); + routes.remove(tunnel_uuid); + } else { + debug!( + "[webhooks] Tunnel {} not found for unregister (skill '{}')", + tunnel_uuid, skill_id + ); + } + + drop(routes); + self.persist(); + Ok(()) + } + + /// Remove all tunnel registrations for a skill (called on skill stop/crash). + pub fn unregister_skill(&self, skill_id: &str) { + let mut routes = match self.routes.write() { + Ok(r) => r, + Err(e) => { + warn!("[webhooks] Failed to acquire write lock: {}", e); + return; + } + }; + + let before = routes.len(); + routes.retain(|_, reg| reg.skill_id != skill_id); + let removed = before - routes.len(); + + if removed > 0 { + debug!( + "[webhooks] Unregistered {} tunnel(s) for skill '{}'", + removed, skill_id + ); + drop(routes); + self.persist(); + } + } + + /// Look up which skill owns a tunnel UUID. + pub fn route(&self, tunnel_uuid: &str) -> Option { + self.routes + .read() + .ok()? + .get(tunnel_uuid) + .map(|r| r.skill_id.clone()) + } + + /// List tunnels owned by a specific skill (for the skill JS API). + pub fn list_for_skill(&self, skill_id: &str) -> Vec { + self.routes + .read() + .map(|routes| { + routes + .values() + .filter(|r| r.skill_id == skill_id) + .cloned() + .collect() + }) + .unwrap_or_default() + } + + /// List all tunnel registrations (for the frontend admin UI). + pub fn list_all(&self) -> Vec { + self.routes + .read() + .map(|routes| routes.values().cloned().collect()) + .unwrap_or_default() + } + + /// Persist current routes to disk. + fn persist(&self) { + let Some(ref path) = self.persist_path else { + return; + }; + + // Clone routes under the lock, then release before doing I/O. + let persisted = { + let routes = match self.routes.read() { + Ok(r) => r, + Err(_) => return, + }; + PersistedRoutes { + registrations: routes.values().cloned().collect(), + } + }; + + if let Some(parent) = path.parent() { + let _ = std::fs::create_dir_all(parent); + } + + match serde_json::to_string_pretty(&persisted) { + Ok(json) => { + if let Err(e) = std::fs::write(path, json) { + warn!("[webhooks] Failed to persist routes to {:?}: {}", path, e); + } + } + Err(e) => { + warn!("[webhooks] Failed to serialize routes: {}", e); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_register_and_route() { + let router = WebhookRouter::new(None); + router + .register("uuid-1", "gmail", Some("Gmail Webhook".into()), None) + .unwrap(); + + assert_eq!(router.route("uuid-1"), Some("gmail".to_string())); + assert_eq!(router.route("uuid-nonexistent"), None); + } + + #[test] + fn test_ownership_enforcement() { + let router = WebhookRouter::new(None); + router + .register("uuid-1", "gmail", Some("Gmail".into()), None) + .unwrap(); + + // Another skill cannot register the same tunnel + let result = router.register("uuid-1", "notion", Some("Notion".into()), None); + assert!(result.is_err()); + assert!(result.unwrap_err().contains("already owned")); + + // Same skill can re-register (update) + router + .register("uuid-1", "gmail", Some("Gmail Updated".into()), None) + .unwrap(); + } + + #[test] + fn test_unregister_ownership() { + let router = WebhookRouter::new(None); + router.register("uuid-1", "gmail", None, None).unwrap(); + + // Another skill cannot unregister + let result = router.unregister("uuid-1", "notion"); + assert!(result.is_err()); + + // Owner can unregister + router.unregister("uuid-1", "gmail").unwrap(); + assert_eq!(router.route("uuid-1"), None); + } + + #[test] + fn test_unregister_skill() { + let router = WebhookRouter::new(None); + router.register("uuid-1", "gmail", None, None).unwrap(); + router.register("uuid-2", "gmail", None, None).unwrap(); + router.register("uuid-3", "notion", None, None).unwrap(); + + router.unregister_skill("gmail"); + + assert_eq!(router.route("uuid-1"), None); + assert_eq!(router.route("uuid-2"), None); + assert_eq!(router.route("uuid-3"), Some("notion".to_string())); + } + + #[test] + fn test_list_for_skill() { + let router = WebhookRouter::new(None); + router.register("uuid-1", "gmail", None, None).unwrap(); + router.register("uuid-2", "notion", None, None).unwrap(); + router.register("uuid-3", "gmail", None, None).unwrap(); + + let gmail_tunnels = router.list_for_skill("gmail"); + assert_eq!(gmail_tunnels.len(), 2); + assert!(gmail_tunnels.iter().all(|t| t.skill_id == "gmail")); + + let notion_tunnels = router.list_for_skill("notion"); + assert_eq!(notion_tunnels.len(), 1); + + let empty = router.list_for_skill("nonexistent"); + assert!(empty.is_empty()); + } +} diff --git a/src/openhuman/webhooks/types.rs b/src/openhuman/webhooks/types.rs new file mode 100644 index 000000000..bdfb2c370 --- /dev/null +++ b/src/openhuman/webhooks/types.rs @@ -0,0 +1,83 @@ +//! Core types for webhook tunnel routing. + +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; + +/// Incoming webhook request forwarded from the backend via Socket.IO. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WebhookRequest { + /// Correlation ID for request-response matching (e.g. `wh_uuid_ts_hex`). + #[serde(rename = "correlationId")] + pub correlation_id: String, + /// Backend tunnel ID. + #[serde(rename = "tunnelId")] + pub tunnel_id: String, + /// Tunnel UUID (used for routing to the owning skill). + #[serde(rename = "tunnelUuid")] + pub tunnel_uuid: String, + /// Human-readable tunnel name. + #[serde(rename = "tunnelName")] + pub tunnel_name: String, + /// HTTP method (GET, POST, etc.). + pub method: String, + /// Request path after the tunnel prefix. + pub path: String, + /// Request headers. + pub headers: HashMap, + /// Query string parameters. + pub query: HashMap, + /// Base64-encoded request body. + #[serde(default)] + pub body: String, +} + +/// Response data sent back to the backend for a webhook request. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WebhookResponseData { + /// Must match the incoming request's correlation_id. + #[serde(rename = "correlationId")] + pub correlation_id: String, + /// HTTP status code to return. + #[serde(rename = "statusCode")] + pub status_code: u16, + /// Response headers. + #[serde(default)] + pub headers: HashMap, + /// Base64-encoded response body. + #[serde(default)] + pub body: String, +} + +/// A mapping from a tunnel UUID to the skill that owns it. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TunnelRegistration { + /// Tunnel UUID (from the backend). + pub tunnel_uuid: String, + /// Skill ID that owns and handles this tunnel. + pub skill_id: String, + /// Human-readable tunnel name (optional, for display). + #[serde(default)] + pub tunnel_name: Option, + /// Backend MongoDB `_id` for CRUD operations. + #[serde(default)] + pub backend_tunnel_id: Option, +} + +/// Entry in the webhook activity log, emitted to the frontend via Tauri events. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WebhookActivityEntry { + /// Correlation ID of the request. + pub correlation_id: String, + /// Tunnel name. + pub tunnel_name: String, + /// HTTP method. + pub method: String, + /// Request path. + pub path: String, + /// Response status code (None if timed out or no handler). + pub status_code: Option, + /// Skill that handled the request (None if unrouted). + pub skill_id: Option, + /// Unix timestamp in milliseconds. + pub timestamp: u64, +}