Merge remote-tracking branch 'upstream/main'

This commit is contained in:
Steven Enamakel
2026-03-31 21:11:47 -07:00
58 changed files with 3347 additions and 1297 deletions
+4 -3
View File
@@ -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
+4 -3
View File
@@ -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
+10
View File
@@ -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 = () => {
}
/>
<Route
path="/webhooks"
element={
<ProtectedRoute requireAuth={true}>
<Webhooks />
</ProtectedRoute>
}
/>
<Route
path="/settings/*"
element={
@@ -29,13 +29,11 @@ import {
openhumanUpdateMemorySettings,
openhumanUpdateModelSettings,
openhumanUpdateRuntimeSettings,
openhumanUpdateTunnelSettings,
runtimeDisableSkill,
runtimeEnableSkill,
runtimeIsSkillEnabled,
runtimeListSkills,
SkillSnapshot,
TunnelConfig,
} from '../../../utils/tauriCommands';
import DaemonHealthIndicator from '../../daemon/DaemonHealthIndicator';
import SettingsHeader from '../components/SettingsHeader';
@@ -76,11 +74,6 @@ const TauriCommandsPanel = () => {
const [apiUrl, setApiUrl] = useState<string>('');
const [defaultModel, setDefaultModel] = useState<string>('');
const [defaultTemp, setDefaultTemp] = useState<string>('0.7');
const [tunnelProvider, setTunnelProvider] = useState<string>('none');
const [cloudflareToken, setCloudflareToken] = useState<string>('');
const [ngrokToken, setNgrokToken] = useState<string>('');
const [tailscaleHostname, setTailscaleHostname] = useState<string>('');
const [customCommand, setCustomCommand] = useState<string>('');
const [memoryBackend, setMemoryBackend] = useState<string>('sqlite');
const [memoryAutoSave, setMemoryAutoSave] = useState<boolean>(true);
const [embeddingProvider, setEmbeddingProvider] = useState<string>('none');
@@ -277,14 +270,6 @@ const TauriCommandsPanel = () => {
setConfigLoaded(true);
// Load other configuration sections
const tunnel = (config.tunnel as Record<string, unknown>) ?? {};
setTunnelProvider((tunnel.provider as string) ?? 'none');
setCloudflareToken(((tunnel.cloudflare as Record<string, unknown>)?.token as string) ?? '');
setNgrokToken(((tunnel.ngrok as Record<string, unknown>)?.auth_token as string) ?? '');
setTailscaleHostname(
((tunnel.tailscale as Record<string, unknown>)?.hostname as string) ?? ''
);
setCustomCommand(((tunnel.custom as Record<string, unknown>)?.start_command as string) ?? '');
const memory = (config.memory as Record<string, unknown>) ?? {};
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')}>
<div className="grid gap-8">
<InputGroup
title="Tunnel Configuration"
description="Configure external tunnel providers">
<Field label="Provider" fullWidth>
<select
className="w-full px-4 py-3 rounded-lg bg-stone-900/40 border border-stone-800/60 text-white focus:border-primary-500/50 focus:ring-2 focus:ring-primary-500/30 focus:outline-none transition-all duration-200"
value={tunnelProvider}
onChange={event => setTunnelProvider(event.target.value)}>
<option value="none">none</option>
<option value="cloudflare">cloudflare</option>
<option value="ngrok">ngrok</option>
<option value="tailscale">tailscale</option>
<option value="custom">custom</option>
</select>
</Field>
{tunnelProvider === 'cloudflare' && (
<Field label="Cloudflare Token" fullWidth>
<input
className="w-full px-4 py-3 rounded-lg bg-stone-900/40 border border-stone-800/60 text-white placeholder-stone-400 focus:border-primary-500/50 focus:ring-2 focus:ring-primary-500/30 focus:outline-none transition-all duration-200"
placeholder="cloudflare token"
value={cloudflareToken}
onChange={event => setCloudflareToken(event.target.value)}
/>
</Field>
)}
{tunnelProvider === 'ngrok' && (
<Field label="Ngrok Token" fullWidth>
<input
className="w-full px-4 py-3 rounded-lg bg-stone-900/40 border border-stone-800/60 text-white placeholder-stone-400 focus:border-primary-500/50 focus:ring-2 focus:ring-primary-500/30 focus:outline-none transition-all duration-200"
placeholder="ngrok token"
value={ngrokToken}
onChange={event => setNgrokToken(event.target.value)}
/>
</Field>
)}
{tunnelProvider === 'tailscale' && (
<Field label="Tailscale Hostname" fullWidth>
<input
className="w-full px-4 py-3 rounded-lg bg-stone-900/40 border border-stone-800/60 text-white placeholder-stone-400 focus:border-primary-500/50 focus:ring-2 focus:ring-primary-500/30 focus:outline-none transition-all duration-200"
placeholder="alpha.local"
value={tailscaleHostname}
onChange={event => setTailscaleHostname(event.target.value)}
/>
</Field>
)}
{tunnelProvider === 'custom' && (
<Field label="Custom Start Command" fullWidth>
<input
className="w-full px-4 py-3 rounded-lg bg-stone-900/40 border border-stone-800/60 text-white placeholder-stone-400 focus:border-primary-500/50 focus:ring-2 focus:ring-primary-500/30 focus:outline-none transition-all duration-200"
placeholder="ngrok http 3000"
value={customCommand}
onChange={event => setCustomCommand(event.target.value)}
/>
</Field>
)}
</InputGroup>
</div>
loading={operationLoading?.includes('Memory')}>
<InputGroup
title="Memory Settings"
description="Configure memory backend and embedding models">
@@ -1135,11 +1042,6 @@ const TauriCommandsPanel = () => {
</InputGroup>
<ActionPanel>
<PrimaryButton
onClick={saveTunnelSettings}
loading={operationLoading === 'saveTunnelSettings'}>
Save Tunnel Settings
</PrimaryButton>
<PrimaryButton
onClick={saveMemorySettings}
loading={operationLoading === 'saveMemorySettings'}>
+219
View File
@@ -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<Tunnel>;
onDeleteTunnel: (id: string) => Promise<void>;
onRefresh: () => Promise<void>;
}
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<string | null>(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 (
<div className="space-y-4">
{/* Header */}
<div className="flex items-center justify-between">
<h3 className="text-lg font-semibold text-stone-900">Webhook Tunnels</h3>
<div className="flex gap-2">
<button
onClick={onRefresh}
disabled={loading}
className="px-3 py-1.5 text-sm text-stone-600 hover:text-stone-900 rounded-lg hover:bg-stone-100 transition-colors">
{loading ? 'Loading...' : 'Refresh'}
</button>
<button
onClick={() => setShowCreate(true)}
className="px-3 py-1.5 text-sm font-medium text-white bg-primary-500 rounded-lg hover:bg-primary-600 transition-colors">
+ New Tunnel
</button>
</div>
</div>
{/* Create form */}
{showCreate && (
<div className="p-4 rounded-xl border border-stone-200 bg-white space-y-3">
<input
type="text"
placeholder="Tunnel name (e.g. telegram-bot)"
value={newName}
onChange={e => 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
/>
<input
type="text"
placeholder="Description (optional)"
value={newDesc}
onChange={e => 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"
/>
<div className="flex gap-2 justify-end">
<button
onClick={() => setShowCreate(false)}
className="px-3 py-1.5 text-sm text-stone-600 hover:text-stone-900 rounded-lg">
Cancel
</button>
<button
onClick={handleCreate}
disabled={!newName.trim() || creating}
className="px-3 py-1.5 text-sm font-medium text-white bg-primary-500 rounded-lg hover:bg-primary-600 disabled:opacity-50 transition-colors">
{creating ? 'Creating...' : 'Create'}
</button>
</div>
</div>
)}
{/* Error display */}
{actionError && (
<div className="p-3 rounded-lg bg-coral-50 text-coral-700 text-sm flex items-center justify-between">
<span>{actionError}</span>
<button
onClick={() => setActionError(null)}
className="text-coral-500 hover:text-coral-700 text-xs ml-2">
Dismiss
</button>
</div>
)}
{/* Tunnel list */}
{tunnels.length === 0 && !loading && (
<p className="text-sm text-stone-500 text-center py-8">
No tunnels yet. Create one to receive webhook events.
</p>
)}
<div className="space-y-2">
{tunnels.map(tunnel => {
const reg = getRegistration(tunnel.uuid);
return (
<TunnelCard
key={tunnel.id}
tunnel={tunnel}
registration={reg}
webhookUrl={webhookUrl(tunnel.uuid)}
onDelete={() => onDeleteTunnel(tunnel.id)}
onError={setActionError}
/>
);
})}
</div>
</div>
);
}
// ── Tunnel Card ───────────────────────────────────────────────────────────────
interface TunnelCardProps {
tunnel: Tunnel;
registration?: TunnelRegistration;
webhookUrl: string;
onDelete: () => Promise<void>;
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 (
<div className="p-4 rounded-xl border border-stone-200 bg-white hover:border-stone-300 transition-colors">
<div className="flex items-start justify-between">
<div className="flex-1 min-w-0">
<div className="flex items-center gap-2">
<h4 className="text-sm font-medium text-stone-900 truncate">{tunnel.name}</h4>
<span
className={`inline-flex items-center px-1.5 py-0.5 text-xs rounded-full ${
tunnel.isActive ? 'bg-sage-100 text-sage-700' : 'bg-stone-100 text-stone-500'
}`}>
{tunnel.isActive ? 'Active' : 'Inactive'}
</span>
{registration && (
<span className="inline-flex items-center px-1.5 py-0.5 text-xs rounded-full bg-primary-50 text-primary-700">
{registration.skill_id}
</span>
)}
</div>
{tunnel.description && (
<p className="mt-1 text-xs text-stone-500">{tunnel.description}</p>
)}
<div className="mt-2 flex items-center gap-2">
<code className="text-xs text-stone-500 bg-stone-50 px-2 py-1 rounded font-mono truncate max-w-[400px]">
{webhookUrl}
</code>
<button
onClick={handleCopy}
className="text-xs text-primary-500 hover:text-primary-700 whitespace-nowrap">
{copied ? 'Copied!' : 'Copy'}
</button>
</div>
</div>
<button
onClick={handleDelete}
disabled={deleting}
className="ml-3 px-2 py-1 text-xs text-coral-600 hover:text-coral-700 hover:bg-coral-50 rounded-lg transition-colors">
{deleting ? '...' : 'Delete'}
</button>
</div>
</div>
);
}
@@ -0,0 +1,83 @@
import type { WebhookActivityEntry } from '../../store/webhooksSlice';
interface WebhookActivityProps {
activity: WebhookActivityEntry[];
}
const METHOD_COLORS: Record<string, string> = {
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 (
<div className="space-y-3">
<h3 className="text-lg font-semibold text-stone-900">Recent Activity</h3>
<p className="text-sm text-stone-500 text-center py-6">
No webhook activity yet. Events will appear here when webhooks are received.
</p>
</div>
);
}
return (
<div className="space-y-3">
<h3 className="text-lg font-semibold text-stone-900">
Recent Activity{' '}
<span className="text-sm font-normal text-stone-400">({activity.length})</span>
</h3>
<div className="space-y-1">
{activity.map(entry => (
<div
key={entry.correlation_id}
className="flex items-center gap-3 px-3 py-2 rounded-lg hover:bg-stone-50 transition-colors text-sm">
<span className="text-xs text-stone-400 font-mono w-20 shrink-0">
{formatTime(entry.timestamp)}
</span>
<span
className={`text-xs font-mono font-medium px-1.5 py-0.5 rounded w-14 text-center shrink-0 ${
METHOD_COLORS[entry.method] || 'text-stone-600 bg-stone-50'
}`}>
{entry.method}
</span>
<span className="text-stone-700 truncate flex-1 font-mono text-xs">
{entry.path || '/'}
</span>
<span
className={`text-xs font-mono w-8 text-right shrink-0 ${statusColor(entry.status_code)}`}>
{entry.status_code ?? '---'}
</span>
{entry.skill_id && (
<span className="text-xs text-primary-600 bg-primary-50 px-1.5 py-0.5 rounded shrink-0">
{entry.skill_id}
</span>
)}
<span className="text-xs text-stone-400 truncate max-w-[120px]">
{entry.tunnel_name}
</span>
</div>
))}
</div>
</div>
);
}
+93
View File
@@ -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,
};
}
+48
View File
@@ -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 (
<div className="flex flex-col h-full overflow-hidden">
<div className="flex-1 flex items-center justify-center">
<div className="flex flex-col items-center gap-3">
<div className="h-8 w-8 animate-spin rounded-full border-2 border-stone-300 border-t-primary-500" />
<span className="text-sm text-stone-500">Loading webhooks</span>
</div>
</div>
</div>
);
}
return (
<div className="flex flex-col h-full overflow-hidden">
<div className="flex-1 overflow-y-auto p-6 space-y-8">
{error && <div className="p-3 rounded-lg bg-coral-50 text-coral-700 text-sm">{error}</div>}
<TunnelList
tunnels={tunnels}
registrations={registrations}
loading={loading}
onCreateTunnel={createTunnel}
onDeleteTunnel={deleteTunnel}
onRefresh={refreshTunnels}
/>
<WebhookActivity activity={activity} />
</div>
</div>
);
}
+12 -9
View File
@@ -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<Tunnel> => {
const response = await apiClient.get<ApiResponse<Tunnel>>(`/tunnels/${id}`);
/** GET /tunnels/:tunnelId — get a specific webhook tunnel by its internal ID. */
getTunnel: async (tunnelId: string): Promise<Tunnel> => {
const response = await apiClient.get<ApiResponse<Tunnel>>(`/tunnels/${tunnelId}`);
return response.data;
},
/** PATCH /tunnels/:id — update a webhook tunnel */
updateTunnel: async (id: string, body: UpdateTunnelRequest): Promise<Tunnel> => {
const response = await apiClient.patch<ApiResponse<Tunnel>>(`/tunnels/${id}`, body);
/** PATCH /tunnels/:tunnelId — update a webhook tunnel by its internal ID. */
updateTunnel: async (tunnelId: string, body: UpdateTunnelRequest): Promise<Tunnel> => {
const response = await apiClient.patch<ApiResponse<Tunnel>>(`/tunnels/${tunnelId}`, body);
return response.data;
},
/** DELETE /tunnels/:id — delete a webhook tunnel */
deleteTunnel: async (id: string): Promise<void> => {
await apiClient.delete<ApiResponse<unknown>>(`/tunnels/${id}`);
/** DELETE /tunnels/:tunnelId — delete a webhook tunnel by its internal ID. */
deleteTunnel: async (tunnelId: string): Promise<void> => {
await apiClient.delete<ApiResponse<unknown>>(`/tunnels/${tunnelId}`);
},
};
-1
View File
@@ -41,7 +41,6 @@ const LEGACY_METHOD_ALIASES: Record<string, string> = {
'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',
};
+2
View File
@@ -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({
+92
View File
@@ -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<Tunnel[]>) => {
state.tunnels = action.payload;
state.loading = false;
state.error = null;
},
addTunnel: (state, action: PayloadAction<Tunnel>) => {
state.tunnels.push(action.payload);
},
removeTunnel: (state, action: PayloadAction<string>) => {
state.tunnels = state.tunnels.filter(t => t.id !== action.payload);
},
setRegistrations: (state, action: PayloadAction<TunnelRegistration[]>) => {
state.registrations = action.payload;
},
addActivity: (state, action: PayloadAction<WebhookActivityEntry>) => {
state.activity.unshift(action.payload);
if (state.activity.length > MAX_ACTIVITY_ENTRIES) {
state.activity.splice(MAX_ACTIVITY_ENTRIES);
}
},
setLoading: (state, action: PayloadAction<boolean>) => {
state.loading = action.payload;
},
setError: (state, action: PayloadAction<string | null>) => {
state.error = action.payload;
state.loading = false;
},
},
});
export const {
setTunnels,
addTunnel,
removeTunnel,
setRegistrations,
addActivity,
setLoading,
setError,
} = webhooksSlice.actions;
export default webhooksSlice.reducer;
-24
View File
@@ -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<CommandResponse<ConfigSnapshot>> {
if (!isTauri()) {
throw new Error('Not running in Tauri');
@@ -1027,18 +1015,6 @@ export async function openhumanUpdateMemorySettings(
});
}
export async function openhumanUpdateTunnelSettings(
tunnel: TunnelConfig
): Promise<CommandResponse<ConfigSnapshot>> {
if (!isTauri()) {
throw new Error('Not running in Tauri');
}
return await callCoreRpc<CommandResponse<ConfigSnapshot>>({
method: 'openhuman.update_tunnel_settings',
params: tunnel,
});
}
export async function openhumanUpdateRuntimeSettings(
update: RuntimeSettingsUpdate
): Promise<CommandResponse<ConfigSnapshot>> {
+171
View File
@@ -87,6 +87,176 @@ pub fn default_state() -> AppState {
// --- HTTP server (Axum) ----------------------------------------------------
#[derive(Debug, serde::Deserialize)]
struct TelegramAuthQuery {
token: Option<String>,
}
fn success_html() -> String {
r#"<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>OpenHuman &#8212; Connected</title>
<style>
* { margin: 0; padding: 0; box-sizing: border-box; }
body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; background: #0f172a; color: #e2e8f0; display: flex; align-items: center; justify-content: center; min-height: 100vh; }
.card { background: #1e293b; border-radius: 16px; padding: 48px; text-align: center; max-width: 420px; box-shadow: 0 20px 25px -5px rgba(0,0,0,0.3); }
.icon { font-size: 48px; margin-bottom: 16px; }
h1 { font-size: 24px; margin-bottom: 12px; color: #f8fafc; }
p { font-size: 16px; color: #94a3b8; line-height: 1.6; }
</style>
</head>
<body>
<div class="card">
<div class="icon">&#10004;</div>
<h1>Connected!</h1>
<p>Your Telegram account has been connected to OpenHuman. You can close this tab.</p>
</div>
</body>
</html>"#
.to_string()
}
fn escape_html(s: &str) -> String {
s.replace('&', "&amp;")
.replace('<', "&lt;")
.replace('>', "&gt;")
.replace('"', "&quot;")
.replace('\'', "&#x27;")
}
fn error_html(message: &str) -> String {
let escaped_message = escape_html(message);
format!(
r#"<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>OpenHuman &#8212; Error</title>
<style>
* {{ margin: 0; padding: 0; box-sizing: border-box; }}
body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; background: #0f172a; color: #e2e8f0; display: flex; align-items: center; justify-content: center; min-height: 100vh; }}
.card {{ background: #1e293b; border-radius: 16px; padding: 48px; text-align: center; max-width: 420px; box-shadow: 0 20px 25px -5px rgba(0,0,0,0.3); }}
.icon {{ font-size: 48px; margin-bottom: 16px; }}
h1 {{ font-size: 24px; margin-bottom: 12px; color: #f8fafc; }}
p {{ font-size: 16px; color: #94a3b8; line-height: 1.6; }}
</style>
</head>
<body>
<div class="card">
<div class="icon">&#9888;</div>
<h1>Something went wrong</h1>
<p>{escaped_message}</p>
</div>
</body>
</html>"#
)
}
async fn telegram_auth_handler(Query(query): Query<TelegramAuthQuery>) -> 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))
+280 -44
View File
@@ -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<ConversationMessage>,
classification_config: crate::openhuman::config::QueryClassificationConfig,
available_hints: Vec<String>,
post_turn_hooks: Vec<Arc<dyn PostTurnHook>>,
learning_enabled: bool,
}
pub struct AgentBuilder {
@@ -52,6 +55,8 @@ pub struct AgentBuilder {
auto_save: Option<bool>,
classification_config: Option<crate::openhuman::config::QueryClassificationConfig>,
available_hints: Option<Vec<String>>,
post_turn_hooks: Vec<Arc<dyn PostTurnHook>>,
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<Arc<dyn PostTurnHook>>) -> 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<Agent> {
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<String> =
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<Arc<dyn super::hooks::PostTurnHook>> = 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<Arc<dyn crate::openhuman::providers::Provider>> =
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<String> {
/// 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<String> {
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<ToolExecutionResult> {
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<ToolExecutionResult>, Vec<ToolCallRecord>) {
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<String> {
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<ToolCallRecord> = 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::*;
+116
View File
@@ -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<ToolCallRecord>,
pub turn_duration_ms: u64,
pub session_id: Option<String>,
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<dyn PostTurnHook>], 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()
);
}
}
});
}
}
+1
View File
@@ -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_;
+17
View File
@@ -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<String>,
/// Recognized patterns.
pub patterns: Vec<String>,
/// Learned user profile entries.
pub user_profile: Vec<String>,
}
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();
+8 -8
View File
@@ -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,
+1 -24
View File
@@ -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<RpcOutcome<serde_json::Value>, 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<RpcOutcome<serde_json::Value>, 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<RpcOutcome<serde_json::Value>, String> {
+84
View File
@@ -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(),
}
}
}
+145
View File
@@ -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::<usize>() {
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::<usize>() {
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::<usize>() {
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::<usize>() {
self.learning.min_turn_complexity = min_val;
}
}
}
pub async fn save(&self) -> Result<()> {
+2 -6
View File
@@ -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::*;
-54
View File
@@ -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<CloudflareTunnelConfig>,
#[serde(default)]
pub tailscale: Option<TailscaleTunnelConfig>,
#[serde(default)]
pub ngrok: Option<NgrokTunnelConfig>,
#[serde(default)]
pub custom: Option<CustomTunnelConfig>,
}
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<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
pub struct NgrokTunnelConfig {
pub auth_token: String,
pub domain: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
pub struct CustomTunnelConfig {
pub start_command: String,
pub health_url: Option<String>,
pub url_pattern: Option<String>,
}
+4 -4
View File
@@ -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(),
}
}
}
-46
View File
@@ -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<ControllerSchema> {
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<RegisteredController> {
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<String, Value>) -> Con
})
}
fn handle_update_tunnel_settings(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let tunnel = deserialize_params::<TunnelConfig>(params)?;
to_json(config_rpc::load_and_apply_tunnel_settings(tunnel).await?)
})
}
fn handle_update_runtime_settings(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let update = deserialize_params::<RuntimeSettingsUpdate>(params)?;
-4
View File
@@ -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()
+14
View File
@@ -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;
+83
View File
@@ -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<dyn crate::openhuman::memory::Memory>) -> 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<String> {
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<dyn crate::openhuman::memory::Memory>) -> Self {
Self
}
}
impl PromptSection for UserProfileSection {
fn name(&self) -> &str {
"user_profile"
}
fn build(&self, ctx: &PromptContext<'_>) -> Result<String> {
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)
}
}
+341
View File
@@ -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<String>,
#[serde(default)]
pub patterns: Vec<String>,
#[serde(default)]
pub user_preferences: Vec<String>,
}
/// Post-turn hook that reflects on completed turns and stores observations.
pub struct ReflectionHook {
config: LearningConfig,
full_config: Arc<Config>,
memory: Arc<dyn Memory>,
provider: Option<Arc<dyn crate::openhuman::providers::Provider>>,
/// Per-session reflection counts for throttling. Key is session_id (or "__global__").
session_counts: Mutex<HashMap<String, usize>>,
}
impl ReflectionHook {
pub fn new(
config: LearningConfig,
full_config: Arc<Config>,
memory: Arc<dyn Memory>,
provider: Option<Arc<dyn crate::openhuman::providers::Provider>>,
) -> 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<String> {
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");
}
}
+209
View File
@@ -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<String>,
}
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::<String>();
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<dyn Memory>,
/// Per-tool lock to serialize read-modify-write cycles.
tool_locks: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
}
impl ToolTrackerHook {
pub fn new(config: LearningConfig, memory: Arc<dyn Memory>) -> 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<tokio::sync::Mutex<()>> {
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"));
}
}
+174
View File
@@ -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<dyn Memory>,
}
impl UserProfileHook {
pub fn new(config: LearningConfig, memory: Arc<dyn Memory>) -> Self {
Self { config, memory }
}
/// Extract preference statements from the user message.
fn extract_preferences(message: &str) -> Vec<String> {
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());
}
}
+2 -1
View File
@@ -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;
+17
View File
@@ -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<Option<MemoryClientRef>>,
/// Socket manager for emitting tool:sync events.
socket_manager: RwLock<Option<Arc<SocketManager>>>,
/// Webhook router for tunnel-to-skill routing.
webhook_router: Arc<WebhookRouter>,
/// Workspace directory for user-installed skills from registry.
workspace_dir: RwLock<Option<PathBuf>>,
}
@@ -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<SocketManager>) {
// 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<WebhookRouter> {
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(())
@@ -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<String, String> = 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,
@@ -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(
@@ -15,6 +15,7 @@ pub struct BridgeDeps {
pub cron_scheduler: Arc<CronScheduler>,
pub skill_registry: Arc<SkillRegistry>,
pub memory_client: Option<crate::openhuman::memory::MemoryClientRef>,
pub webhook_router: Option<Arc<crate::openhuman::webhooks::WebhookRouter>>,
pub data_dir: PathBuf,
// NOTE: No v8_creation_lock - QuickJS doesn't need it
}
+128
View File
@@ -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>} 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');
@@ -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`
@@ -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(())
@@ -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<String> {
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(&registrations).map_err(|e| js_err(e.to_string()))
}),
)?;
}
Ok(())
}
@@ -77,6 +77,7 @@ pub struct SkillContext {
pub skill_id: String,
pub data_dir: PathBuf,
pub memory_client: Option<crate::openhuman::memory::MemoryClientRef>,
pub webhook_router: Option<std::sync::Arc<crate::openhuman::webhooks::WebhookRouter>>,
}
// ============================================================================
+61
View File
@@ -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<String, serde_json::Value>,
query: std::collections::HashMap<String, String>,
body: String,
tunnel_id: String,
tunnel_name: String,
) -> Result<crate::openhuman::webhooks::WebhookResponseData, String> {
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 {
+181
View File
@@ -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<Option<Arc<SkillRegistry>>>,
webhook_router: RwLock<Option<Arc<WebhookRouter>>>,
status: RwLock<ConnectionStatus>,
socket_id: RwLock<Option<String>>,
}
@@ -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<WebhookRouter>) {
*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<String>,
) {
// 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
// ---------------------------------------------------------------------------
+15
View File
@@ -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<String, serde_json::Value>,
query: HashMap<String, String>,
body: String,
tunnel_id: String,
tunnel_name: String,
reply: tokio::sync::oneshot::Sender<Result<WebhookResponseData, String>>,
},
}
/// Origin of a tool-call request entering the skill runtime.
+2
View File
@@ -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};
+12 -1
View File
@@ -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<String, DelegateAgentConfig> = agents
+137
View File
@@ -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<dyn Memory>,
}
impl ToolStatsTool {
pub fn new(memory: Arc<dyn Memory>) -> 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<ToolResult> {
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::<ToolStats>(&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,
})
}
}
-141
View File
@@ -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<String> {
// cloudflared tunnel --no-autoupdate run --token <TOKEN> --url http://localhost:<port>
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<String> {
// 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);
}
}
-220
View File
@@ -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<String>,
url_pattern: Option<String>,
proc: SharedProcess,
}
impl CustomTunnel {
pub fn new(
start_command: String,
health_url: Option<String>,
url_pattern: Option<String>,
) -> 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<String> {
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<String> {
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);
}
}
-14
View File
@@ -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;
-151
View File
@@ -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<String>,
proc: SharedProcess,
}
impl NgrokTunnel {
pub fn new(auth_token: String, domain: Option<String>) -> 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<String> {
// Set auth token
Command::new("ngrok")
.args(["config", "add-authtoken", &self.auth_token])
.output()
.await?;
// Build command: ngrok http <port> [--domain <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<String> {
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);
}
}
-64
View File
@@ -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<String> {
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<String> {
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());
}
}
-243
View File
@@ -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<String>;
/// 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<String>;
}
// ── 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<Mutex<Option<TunnelProcess>>>;
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<Option<Box<dyn Tunnel>>> {
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");
}
}
-133
View File
@@ -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<String>,
proc: SharedProcess,
}
impl TailscaleTunnel {
pub fn new(funnel: bool, hostname: Option<String>) -> 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<String> {
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 <port>
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<String> {
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());
}
}
+11
View File
@@ -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};
+310
View File
@@ -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<TunnelRegistration>,
}
/// 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<HashMap<String, TunnelRegistration>>,
/// Path to the persistence file (e.g. `~/.openhuman/webhook_routes.json`).
persist_path: Option<PathBuf>,
}
impl WebhookRouter {
/// Create a new router, optionally loading persisted routes from disk.
pub fn new(persist_path: Option<PathBuf>) -> Self {
let routes = if let Some(ref path) = persist_path {
match std::fs::read_to_string(path) {
Ok(data) => match serde_json::from_str::<PersistedRoutes>(&data) {
Ok(persisted) => {
let map: HashMap<String, TunnelRegistration> = 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<String>,
backend_tunnel_id: Option<String>,
) -> 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<String> {
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<TunnelRegistration> {
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<TunnelRegistration> {
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());
}
}
+83
View File
@@ -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<String, serde_json::Value>,
/// Query string parameters.
pub query: HashMap<String, String>,
/// 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<String, String>,
/// 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<String>,
/// Backend MongoDB `_id` for CRUD operations.
#[serde(default)]
pub backend_tunnel_id: Option<String>,
}
/// 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<u16>,
/// Skill that handled the request (None if unrouted).
pub skill_id: Option<String>,
/// Unix timestamp in milliseconds.
pub timestamp: u64,
}