feat(skills): advanced authentication modes for skills (#315)

* feat(auth): implement advanced authentication modes and UI selector

- Introduced a new `AuthModeSelector` component for selecting authentication methods (managed, self-hosted, text).
- Enhanced the `SkillManager` to handle revocation of both OAuth and auth credentials.
- Updated the API to support advanced auth configurations with multiple modes.
- Added functionality to persist and restore auth credentials in the skill's data directory.
- Updated relevant types and interfaces to accommodate the new auth structure.

* feat(auth): enhance SkillSetupWizard with multi-phase authentication flow

- Updated `SkillSetupWizard` to support a two-phase authentication process: an optional auth mode selection followed by the setup phase.
- Integrated `AuthModeSelector` for users to choose between managed, self-hosted, and text authentication methods.
- Improved state management to handle various auth phases, including error handling and transitions to the setup phase upon successful authentication.
- Refactored relevant types and interfaces to accommodate the new authentication structure.

* refactor(skills): remove unused loader module and clean up bridge imports

- Deleted the unused `loader.rs` module, which was reserved for future ES module import support.
- Removed references to the `loader` module in `mod.rs`.
- Cleaned up the `bridge` module by removing several unused bridge files, including `cron_bridge.rs`, `db.rs`, `log_bridge.rs`, `skills_bridge.rs`, `store.rs`, and `tauri_bridge.rs`.
- Added comprehensive tests for the `manifest.rs`, `ops.rs`, and `preferences.rs` modules to ensure functionality remains intact after the refactor.

* feat(registry): introduce registry cache management for remote skill registry

- Added a new `registry_cache.rs` module to handle disk-based caching for the remote skill registry.
- Implemented functions for managing cache, including reading, writing, and checking cache freshness.
- Updated `registry_ops.rs` to utilize the new caching functions, improving performance and reducing redundant network calls.
- Introduced a new `event_loop` module to manage the QuickJS runtime, including handling incoming messages and persisting state to memory.
- Added webhook request handling and RPC message handlers to facilitate communication with external services.
- Removed the obsolete `event_loop.rs` file, consolidating functionality into the new structure.

* fix(tests): update imports in registry_ops tests to include cache management functions

- Modified test module imports in `registry_ops.rs` to include new cache management functions from `registry_cache`.
- This change ensures that tests can utilize the updated caching functionality introduced in the recent registry cache management feature.

* chore(deps): update OpenHuman version to 0.51.6 and clean up code formatting

- Updated the OpenHuman dependency version in `Cargo.lock` to 0.51.6.
- Refactored code in several files for improved readability by adjusting formatting and removing unnecessary line breaks.
- Ensured consistent logging format in `event_loop` and `rpc_handlers` modules for better clarity in log messages.

* refactor(conversations): improve team usage display logic and formatting

- Refactored the Conversations component to enhance the display of team usage information, including clearer conditional rendering for budget exhaustion messages.
- Improved formatting of budget display for both 5-hour and weekly limits, ensuring consistent presentation.
- Streamlined the import of the core HTTP base URL in the WebhooksDebugPanel for better code organization.

* refactor(auth): enhance credential management and validation process

- Updated the SkillManager to attempt revoking auth credentials even if the auth mode is unknown, improving robustness.
- Introduced a new deserialization function to ensure that at least one authentication mode is provided in the SkillAuthConfig.
- Enhanced the handling of temporary credential injection during the auth completion process, ensuring credentials are only persisted upon successful validation.
- Improved the handling of OAuth credentials in JavaScript, ensuring stale credentials are cleared for non-managed modes.
- Added checks for existing Authorization headers to prevent redundant injections in self-hosted modes.

* test(manifest): add unit tests for authentication mode deserialization

- Introduced tests to validate the deserialization of known authentication modes in the SkillManifest.
- Added checks to ensure invalid authentication types and empty mode lists are correctly rejected.
- These tests enhance the robustness of the authentication configuration handling in the manifest module.

* refactor(messaging): simplify MessagingPanel by removing unused constants and improving input handling

- Removed unused constants related to channel statuses, authentication modes, and fallback definitions to streamline the MessagingPanel component.
- Updated input handling to use a more concise onChange function, enhancing code readability.
- Commented out placeholder and className properties for input fields to improve clarity and focus on essential functionality.

* refactor(webhooks): streamline TunnelList imports for improved clarity

- Combined the import statements for Tunnel type and tunnelsApi from the same module into a single line, enhancing code readability and organization.
This commit is contained in:
Steven Enamakel
2026-04-04 02:20:39 -07:00
committed by GitHub
parent 4a228ee318
commit acfa5497ef
32 changed files with 1996 additions and 1016 deletions
@@ -19,120 +19,11 @@ import type {
ChannelType,
} from '../../../types/channels';
import { openUrl } from '../../../utils/openUrl';
import ChannelCapabilities from '../../channels/ChannelCapabilities';
import ChannelFieldInput from '../../channels/ChannelFieldInput';
import ChannelStatusBadge from '../../channels/ChannelStatusBadge';
import SettingsHeader from '../components/SettingsHeader';
import { useSettingsNavigation } from '../hooks/useSettingsNavigation';
const STATUS_STYLES: Record<ChannelConnectionStatus, { label: string; className: string }> = {
connected: { label: 'Connected', className: 'bg-sage-50 text-sage-700 border-sage-200' },
connecting: { label: 'Connecting', className: 'bg-amber-50 text-amber-700 border-amber-200' },
disconnected: {
label: 'Disconnected',
className: 'bg-stone-100 text-stone-600 border-stone-200',
},
error: { label: 'Error', className: 'bg-coral-50 text-coral-600 border-coral-200' },
};
const AUTH_MODE_LABELS: Record<string, string> = {
managed_dm: 'Managed DM',
oauth: 'OAuth Sign-in',
bot_token: 'Bot Token',
api_key: 'API Key',
};
/** Fallback definitions used when the core sidecar is unreachable. */
const FALLBACK_DEFINITIONS: ChannelDefinition[] = [
{
id: 'telegram',
display_name: 'Telegram',
description: 'Send and receive messages via Telegram.',
icon: 'telegram',
auth_modes: [
{
mode: 'bot_token',
description: 'Provide your own Telegram Bot token from @BotFather.',
fields: [
{
key: 'bot_token',
label: 'Bot Token',
field_type: 'secret',
required: true,
placeholder: '123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11',
},
{
key: 'allowed_users',
label: 'Allowed Users',
field_type: 'string',
required: false,
placeholder: 'Comma-separated Telegram usernames',
},
],
auth_action: undefined,
},
{
mode: 'managed_dm',
description: 'Message the OpenHuman Telegram bot directly.',
fields: [],
auth_action: 'telegram_managed_dm',
},
],
capabilities: ['send_text', 'receive_text', 'typing', 'draft_updates'],
},
{
id: 'discord',
display_name: 'Discord',
description: 'Send and receive messages via Discord.',
icon: 'discord',
auth_modes: [
{
mode: 'bot_token',
description: 'Provide your own Discord bot token.',
fields: [
{
key: 'bot_token',
label: 'Bot Token',
field_type: 'secret',
required: true,
placeholder: 'Your Discord bot token',
},
{
key: 'guild_id',
label: 'Server (Guild) ID',
field_type: 'string',
required: false,
placeholder: 'Optional: restrict to a specific server',
},
],
auth_action: undefined,
},
{
mode: 'oauth',
description: 'Install the OpenHuman bot to your Discord server via OAuth.',
fields: [],
auth_action: 'discord_oauth',
},
],
capabilities: ['send_text', 'receive_text', 'typing', 'threaded_replies'],
},
{
id: 'web',
display_name: 'Web',
description: 'Chat via the built-in web UI.',
icon: 'web',
auth_modes: [
{
mode: 'managed_dm',
description: 'Use the embedded web chat — no setup required.',
fields: [],
auth_action: undefined,
},
],
capabilities: ['send_text', 'send_rich_text', 'receive_text'],
},
];
const MessagingPanel = () => {
const { navigateBack } = useSettingsNavigation();
const dispatch = useAppDispatch();
@@ -364,9 +255,9 @@ const MessagingPanel = () => {
key={field.key}
field={field}
value={fieldValues[compositeKey]?.[field.key] ?? ''}
onChange={e => updateField(compositeKey, field.key, e.target.value)}
placeholder={field.placeholder || field.label}
className="w-full rounded-lg border border-stone-200 bg-white px-3 py-2 text-sm text-stone-900 placeholder:text-stone-400 focus:outline-none focus:border-primary-500/60"
onChange={value => updateField(compositeKey, field.key, value)}
// placeholder={field.placeholder || field.label}
// className="w-full rounded-lg border border-stone-200 bg-white px-3 py-2 text-sm text-stone-900 placeholder:text-stone-400 focus:outline-none focus:border-primary-500/60"
/>
))}
</div>
@@ -1,7 +1,7 @@
import { useCallback, useEffect, useMemo, useState } from 'react';
import { getCoreHttpBaseUrl } from '../../../services/coreRpcClient';
import { tunnelsApi } from '../../../services/api/tunnelsApi';
import { getCoreHttpBaseUrl } from '../../../services/coreRpcClient';
import { BACKEND_URL } from '../../../utils/config';
import {
openhumanWebhooksClearLogs,
@@ -0,0 +1,139 @@
/**
* Card-based selector for skill authentication modes.
* Displays available auth modes (managed, self_hosted, text) as clickable cards.
* For managed mode, clicking immediately triggers the OAuth flow.
*/
import type { ReactElement } from "react";
import type { AuthMode } from "../../lib/skills/types.ts";
interface AuthModeSelectorProps {
modes: AuthMode[];
onSelect: (mode: AuthMode) => void;
disabled?: boolean;
}
const MODE_ICONS: Record<string, (props: { className: string }) => ReactElement> = {
managed: ({ className }) => (
<svg className={className} fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={1.5}
d="M9 12.75L11.25 15 15 9.75m-3-7.036A11.959 11.959 0 013.598 6 11.99 11.99 0 003 9.749c0 5.592 3.824 10.29 9 11.623 5.176-1.332 9-6.03 9-11.622 0-1.31-.21-2.571-.598-3.751h-.152c-3.196 0-6.1-1.248-8.25-3.285z"
/>
</svg>
),
self_hosted: ({ className }) => (
<svg className={className} fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={1.5}
d="M5.25 14.25h13.5m-13.5 0a3 3 0 01-3-3m3 3a3 3 0 100 6h13.5a3 3 0 100-6m-16.5-3a3 3 0 013-3h13.5a3 3 0 013 3m-19.5 0a4.5 4.5 0 01.9-2.7L5.737 5.1a3.375 3.375 0 012.7-1.35h7.126c1.062 0 2.062.5 2.7 1.35l2.587 3.45a4.5 4.5 0 01.9 2.7m0 0a3 3 0 01-3 3m0 3h.008v.008h-.008v-.008zm0-6h.008v.008h-.008v-.008zm-3 6h.008v.008h-.008v-.008zm0-6h.008v.008h-.008v-.008z"
/>
</svg>
),
text: ({ className }) => (
<svg className={className} fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={1.5}
d="M19.5 14.25v-2.625a3.375 3.375 0 00-3.375-3.375h-1.5A1.125 1.125 0 0113.5 7.125v-1.5a3.375 3.375 0 00-3.375-3.375H8.25m0 12.75h7.5m-7.5 3H12M10.5 2.25H5.625c-.621 0-1.125.504-1.125 1.125v17.25c0 .621.504 1.125 1.125 1.125h12.75c.621 0 1.125-.504 1.125-1.125V11.25a9 9 0 00-9-9z"
/>
</svg>
),
};
const DEFAULT_LABELS: Record<string, string> = {
managed: "OpenHuman Managed",
self_hosted: "Self-hosted",
text: "Credential Text",
};
const DEFAULT_DESCRIPTIONS: Record<string, string> = {
managed: "One-click setup through OpenHuman",
self_hosted: "Enter your own API credentials",
text: "Paste credential content directly",
};
function formatProviderName(provider: string): string {
const names: Record<string, string> = {
notion: "Notion",
google: "Google",
github: "GitHub",
slack: "Slack",
discord: "Discord",
twitter: "Twitter",
linear: "Linear",
gitlab: "GitLab",
};
return names[provider] ?? provider.charAt(0).toUpperCase() + provider.slice(1);
}
export default function AuthModeSelector({
modes,
onSelect,
disabled,
}: AuthModeSelectorProps) {
return (
<div className="space-y-3">
<div className="text-center mb-4">
<h3 className="text-lg font-semibold text-stone-900">
Choose how to connect
</h3>
<p className="text-sm text-stone-500 mt-1">
Select an authentication method
</p>
</div>
{modes.map((mode) => {
const IconComponent = MODE_ICONS[mode.type] ?? MODE_ICONS.self_hosted;
const label =
mode.type === "managed" && mode.provider
? mode.label ?? `Connect with ${formatProviderName(mode.provider)}`
: mode.label ?? DEFAULT_LABELS[mode.type] ?? mode.type;
const description =
mode.description ?? DEFAULT_DESCRIPTIONS[mode.type] ?? "";
return (
<button
key={mode.type}
type="button"
onClick={() => onSelect(mode)}
disabled={disabled}
className="w-full flex items-center gap-4 p-4 bg-stone-50 border border-stone-200 rounded-xl hover:bg-stone-100 hover:border-stone-300 transition-colors text-left disabled:opacity-50 disabled:cursor-not-allowed group"
>
<div className="flex-shrink-0 w-10 h-10 rounded-lg bg-white border border-stone-200 flex items-center justify-center group-hover:border-primary-300 transition-colors">
<IconComponent className="w-5 h-5 text-stone-500 group-hover:text-primary-500 transition-colors" />
</div>
<div className="flex-1 min-w-0">
<div className="text-sm font-medium text-stone-900">
{label}
</div>
{description && (
<div className="text-xs text-stone-500 mt-0.5">
{description}
</div>
)}
</div>
<svg
className="w-4 h-4 text-stone-400 group-hover:text-stone-600 transition-colors flex-shrink-0"
fill="none"
stroke="currentColor"
viewBox="0 0 24 24"
>
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={2}
d="M9 5l7 7-7 7"
/>
</svg>
</button>
);
})}
</div>
);
}
@@ -174,6 +174,22 @@ export default function SetupFormRenderer({
</select>
)}
{/* Textarea */}
{field.type === "textarea" && (
<textarea
value={String(value ?? "")}
onChange={(e) => updateValue(field.name, e.target.value)}
placeholder={field.placeholder ?? undefined}
disabled={loading}
rows={8}
className={`w-full px-3 py-2.5 bg-stone-50 border rounded-xl text-sm text-stone-900 placeholder-stone-400 font-mono focus:outline-none focus:ring-1 transition-colors resize-y ${
fieldError
? "border-coral-500/50 focus:ring-coral-500/30"
: "border-stone-200 focus:ring-primary-500/30 focus:border-primary-500/50"
}`}
/>
)}
{/* Boolean toggle */}
{field.type === "boolean" && (
<label className="flex items-center space-x-3 cursor-pointer">
+338 -71
View File
@@ -1,19 +1,28 @@
/**
* Multi-step setup wizard for a single skill.
* Manages the state machine: start -> render form -> submit -> next/error/complete.
* For OAuth skills, shows a login button instead of form steps.
* Ensures the skill is running (starts it if needed) before starting the setup flow.
*
* Supports two sequential phases:
* 1. **Auth phase** (optional): If the skill declares `setup.auth`, the user
* picks an auth mode (managed / self_hosted / text), enters credentials,
* and the skill validates them via `onAuthComplete`.
* 2. **Setup phase**: The normal setup flow runs (`onSetupStart` → form steps
* → `onSetupSubmit`).
*
* For legacy skills with only `setup.oauth`, the OAuth config is auto-synthesized
* as a single "managed" auth mode.
*/
import { useState, useEffect, useCallback } from "react";
import { useSkillSnapshot } from "../../lib/skills/hooks.ts";
import { skillManager } from "../../lib/skills/manager.ts";
import { listAvailable, setSetupComplete, startSkill } from "../../lib/skills/skillsApi.ts";
import { callCoreRpc } from "../../services/coreRpcClient.ts";
import { apiClient } from "../../services/apiClient.ts";
import { openUrl } from "../../utils/openUrl.ts";
import type { SetupStep, SetupFieldError } from "../../lib/skills/types.ts";
import type { SetupStep, SetupFieldError, AuthMode, SkillAuthConfig } from "../../lib/skills/types.ts";
import SetupFormRenderer from "./SetupFormRenderer.tsx";
import {IS_DEV} from "../../utils/config.ts";
import AuthModeSelector from "./AuthModeSelector.tsx";
import { IS_DEV } from "../../utils/config.ts";
interface SkillSetupWizardProps {
skillId: string;
@@ -27,9 +36,16 @@ interface OAuthConfig {
}
type WizardState =
// Auth phases
| { phase: "loading" }
| { phase: "auth_mode_select"; auth: SkillAuthConfig }
| { phase: "auth_form"; mode: AuthMode; errors?: SetupFieldError[] | null }
| { phase: "auth_submitting"; mode: AuthMode }
| { phase: "auth_managed_waiting"; mode: AuthMode }
// Legacy OAuth (when no setup.auth, only setup.oauth)
| { phase: "oauth"; oauth: OAuthConfig }
| { phase: "oauth_waiting"; oauth: OAuthConfig }
// Setup phases (run after auth succeeds, or directly for non-auth skills)
| { phase: "step"; step: SetupStep; errors?: SetupFieldError[] | null }
| { phase: "submitting"; step: SetupStep }
| { phase: "complete"; message?: string }
@@ -42,23 +58,148 @@ export default function SkillSetupWizard({
}: SkillSetupWizardProps) {
const [state, setState] = useState<WizardState>({ phase: "loading" });
// Watch skill snapshot for OAuth completion via RPC-backed hook
// Watch skill snapshot for OAuth/managed completion
const snap = useSkillSnapshot(skillId);
const isConnected = snap?.connection_status === "connected" || snap?.setup_complete === true;
// When skill state changes to connected during OAuth waiting, mark complete
// --- Transition from auth to setup ---
const transitionToSetup = useCallback(async () => {
try {
// Ensure skill runtime is running
try {
await startSkill(skillId);
} catch {
// May already be running
}
const firstStep = await skillManager.startSetup(skillId);
if (!firstStep) {
// No setup steps needed — auth alone is sufficient
await setSetupComplete(skillId, true);
setState({ phase: "complete", message: "Successfully connected!" });
} else {
setState({ phase: "step", step: firstStep });
}
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
setState({ phase: "error", message: msg });
}
}, [skillId]);
// --- Managed OAuth ---
const handleManagedAuth = useCallback(
async (mode: AuthMode) => {
if (!mode.provider) {
setState({ phase: "error", message: "Managed mode requires a provider." });
return;
}
try {
const shouldShowJson = IS_DEV ? 'responseType=json&' : '';
const data = await apiClient.get<{ oauthUrl?: string }>(
`/auth/${mode.provider}/connect?${shouldShowJson}skillId=${skillId}`,
);
if (!data.oauthUrl) {
setState({ phase: "error", message: "Failed to get OAuth URL from backend." });
return;
}
await openUrl(data.oauthUrl);
setState({ phase: "auth_managed_waiting", mode });
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
setState({ phase: "error", message: `OAuth connection failed: ${msg}` });
}
},
[skillId],
);
// --- Auth mode selection ---
const handleAuthModeSelected = useCallback(
(mode: AuthMode) => {
if (mode.type === "managed") {
handleManagedAuth(mode);
} else {
// self_hosted or text: show credential form
setState({ phase: "auth_form", mode });
}
},
[handleManagedAuth],
);
// --- Auth form submission ---
const handleAuthFormSubmit = useCallback(
async (values: Record<string, unknown>) => {
if (state.phase !== "auth_form") return;
const { mode } = state;
setState({ phase: "auth_submitting", mode });
try {
// Ensure skill runtime is running
try {
await startSkill(skillId);
} catch {
// May already be running
}
// Build credential payload
const credentials =
mode.type === "text"
? { content: values.content }
: values;
// Send auth/complete RPC to skill
const result = await callCoreRpc({
method: "openhuman.skills_rpc",
params: {
skill_id: skillId,
method: "auth/complete",
params: { mode: mode.type, credentials },
},
}) as { status?: string; errors?: SetupFieldError[]; message?: string } | null;
if (result?.status === "error" && result.errors) {
setState({ phase: "auth_form", mode, errors: result.errors });
return;
}
if (result?.status === "error") {
setState({
phase: "error",
message: result.message ?? "Auth validation failed.",
});
return;
}
// Auth succeeded — transition to setup phase
await transitionToSetup();
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
setState({ phase: "error", message: msg });
}
},
[state, skillId, transitionToSetup],
);
// Detect managed OAuth completion
useEffect(() => {
if (state.phase === "auth_managed_waiting" && isConnected) {
setTimeout(() => { transitionToSetup(); }, 0);
}
// Legacy OAuth completion
if (
(state.phase === "oauth" || state.phase === "oauth_waiting") &&
isConnected
) {
setSetupComplete(skillId, true).catch(() => {});
// Schedule state update to avoid synchronous setState inside an effect
setTimeout(() => {
setState({ phase: "complete", message: "Successfully connected!" });
}, 0);
}
}, [isConnected, state.phase, skillId]);
}, [isConnected, state.phase, skillId, transitionToSetup]);
// Start the setup flow on mount
useEffect(() => {
@@ -68,7 +209,6 @@ export default function SkillSetupWizard({
try {
console.log("[SkillSetupWizard] initSetup", skillId);
// Find the available skill entry from the registry for OAuth config
const available = await listAvailable();
const entry = available.find(e => e.id === skillId);
@@ -82,9 +222,26 @@ export default function SkillSetupWizard({
return;
}
const setup = entry.setup as { required?: boolean; oauth?: OAuthConfig } | null | undefined;
const setup = entry.setup as {
required?: boolean;
oauth?: OAuthConfig;
auth?: SkillAuthConfig;
} | null | undefined;
// If the skill has OAuth config, show OAuth login directly
// Priority 1: Advanced auth config
if (setup?.auth && setup.auth.modes.length > 0) {
if (!cancelled) {
if (setup.auth.modes.length === 1) {
const mode = setup.auth.modes[0];
handleAuthModeSelected(mode);
} else {
setState({ phase: "auth_mode_select", auth: setup.auth });
}
}
return;
}
// Priority 2: Legacy OAuth config
if (setup?.oauth) {
if (!cancelled) {
setState({
@@ -98,7 +255,7 @@ export default function SkillSetupWizard({
return;
}
// Non-OAuth skills need the runtime running for setup steps
// Priority 3: Non-auth — start form-based setup directly
try {
await startSkill(skillId);
console.log("[SkillSetupWizard] skill started via RPC", skillId);
@@ -120,7 +277,7 @@ export default function SkillSetupWizard({
if (!firstStep) {
setState({
phase: "error",
message: "This skill requires OAuth setup but no setup steps were returned. Try restarting the app.",
message: "This skill requires setup but no setup steps were returned. Try restarting the app.",
});
} else {
setState({ phase: "step", step: firstStep });
@@ -139,36 +296,11 @@ export default function SkillSetupWizard({
return () => {
cancelled = true;
};
}, [skillId]);
}, [skillId, handleAuthModeSelected]);
const handleOAuthLogin = useCallback(async () => {
if (state.phase !== "oauth") return;
// --- Setup form submission (existing flow) ---
const { oauth } = state;
try {
const shouldShowJson = IS_DEV ? 'responseType=json&' : ''
// Call backend to get the real OAuth authorization URL
const data = await apiClient.get<{ oauthUrl?: string }>(
`/auth/${oauth.provider}/connect?${shouldShowJson}skillId=${skillId}`,
);
if (!data.oauthUrl) {
console.error("[SkillSetupWizard] Backend did not return oauthUrl:", data);
setState({ phase: "error", message: "Failed to get OAuth URL from backend." });
return;
}
await openUrl(data.oauthUrl);
setState({ phase: "oauth_waiting", oauth });
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
console.error("[SkillSetupWizard] OAuth connect error:", err);
setState({ phase: "error", message: `OAuth connection failed: ${msg}` });
}
}, [state, skillId]);
const handleSubmit = useCallback(
const handleSetupSubmit = useCallback(
async (values: Record<string, unknown>) => {
if (state.phase !== "step") return;
@@ -210,8 +342,44 @@ export default function SkillSetupWizard({
[state, skillId],
);
// --- Legacy OAuth handler ---
const handleOAuthLogin = useCallback(async () => {
if (state.phase !== "oauth") return;
const { oauth } = state;
try {
const shouldShowJson = IS_DEV ? 'responseType=json&' : '';
const data = await apiClient.get<{ oauthUrl?: string }>(
`/auth/${oauth.provider}/connect?${shouldShowJson}skillId=${skillId}`,
);
if (!data.oauthUrl) {
console.error("[SkillSetupWizard] Backend did not return oauthUrl:", data);
setState({ phase: "error", message: "Failed to get OAuth URL from backend." });
return;
}
await openUrl(data.oauthUrl);
setState({ phase: "oauth_waiting", oauth });
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
console.error("[SkillSetupWizard] OAuth connect error:", err);
setState({ phase: "error", message: `OAuth connection failed: ${msg}` });
}
}, [state, skillId]);
// --- Cancel ---
const handleCancel = useCallback(async () => {
if (state.phase !== "oauth" && state.phase !== "oauth_waiting") {
if (
state.phase !== "oauth" &&
state.phase !== "oauth_waiting" &&
state.phase !== "auth_mode_select" &&
state.phase !== "auth_form" &&
state.phase !== "auth_managed_waiting"
) {
try {
await skillManager.cancelSetup(skillId);
} catch {
@@ -221,37 +389,61 @@ export default function SkillSetupWizard({
onCancel();
}, [skillId, onCancel, state.phase]);
// Render based on current wizard state
// --- Render ---
switch (state.phase) {
case "loading":
return <LoadingView />;
case "auth_mode_select":
return (
<div className="flex items-center justify-center py-12">
<svg
className="animate-spin h-6 w-6 text-primary-500"
xmlns="http://www.w3.org/2000/svg"
fill="none"
viewBox="0 0 24 24"
>
<circle
className="opacity-25"
cx="12"
cy="12"
r="10"
stroke="currentColor"
strokeWidth="4"
/>
<path
className="opacity-75"
fill="currentColor"
d="M4 12a8 8 0 018-8V0C5.373 0 0 5.373 0 12h4z"
/>
</svg>
<span className="ml-3 text-sm text-stone-400">
Starting setup...
</span>
<div className="py-4">
<AuthModeSelector
modes={state.auth.modes}
onSelect={handleAuthModeSelected}
/>
<div className="mt-4">
<button
onClick={handleCancel}
className="w-full py-2.5 text-sm font-medium text-stone-600 bg-stone-100 border border-stone-200 rounded-xl hover:bg-stone-200 transition-colors"
>
Cancel
</button>
</div>
</div>
);
case "auth_form":
return (
<SetupFormRenderer
step={buildAuthFormStep(state.mode)}
errors={state.errors}
loading={false}
onSubmit={handleAuthFormSubmit}
onCancel={handleCancel}
/>
);
case "auth_submitting":
return (
<SetupFormRenderer
step={buildAuthFormStep(state.mode)}
loading={true}
onSubmit={() => {}}
onCancel={() => {}}
/>
);
case "auth_managed_waiting":
return (
<OAuthLoginView
provider={state.mode.provider ?? "service"}
onLogin={() => handleManagedAuth(state.mode)}
onCancel={handleCancel}
waiting={true}
/>
);
case "oauth":
return (
<OAuthLoginView
@@ -278,7 +470,7 @@ export default function SkillSetupWizard({
step={state.step}
errors={state.errors}
loading={false}
onSubmit={handleSubmit}
onSubmit={handleSetupSubmit}
onCancel={handleCancel}
/>
);
@@ -288,8 +480,8 @@ export default function SkillSetupWizard({
<SetupFormRenderer
step={state.step}
loading={true}
onSubmit={() => { }}
onCancel={() => { }}
onSubmit={() => {}}
onCancel={() => {}}
/>
);
@@ -361,6 +553,81 @@ export default function SkillSetupWizard({
}
}
// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------
/** Build a synthetic SetupStep from an AuthMode for rendering via SetupFormRenderer. */
function buildAuthFormStep(mode: AuthMode): SetupStep {
if (mode.type === "text") {
return {
id: "auth_text",
title: mode.label ?? "Credential Text",
description: mode.textDescription ?? "Paste your credential content below.",
fields: [
{
name: "content",
type: "textarea",
label: "Credential Content",
required: true,
placeholder: mode.textPlaceholder ?? undefined,
},
],
};
}
// self_hosted: use the mode's declared fields
return {
id: `auth_${mode.type}`,
title: mode.label ?? "Enter Credentials",
description: mode.description ?? "Provide your credentials to connect.",
fields: (mode.fields ?? []).map(f => ({
name: f.name ?? "",
type: f.type ?? "text",
label: f.label ?? f.name ?? "",
required: f.required ?? false,
default: f.default,
placeholder: f.placeholder,
description: f.description,
options: f.options,
})),
};
}
// ---------------------------------------------------------------------------
// Loading View
// ---------------------------------------------------------------------------
function LoadingView() {
return (
<div className="flex items-center justify-center py-12">
<svg
className="animate-spin h-6 w-6 text-primary-500"
xmlns="http://www.w3.org/2000/svg"
fill="none"
viewBox="0 0 24 24"
>
<circle
className="opacity-25"
cx="12"
cy="12"
r="10"
stroke="currentColor"
strokeWidth="4"
/>
<path
className="opacity-75"
fill="currentColor"
d="M4 12a8 8 0 018-8V0C5.373 0 0 5.373 0 12h4z"
/>
</svg>
<span className="ml-3 text-sm text-stone-400">
Starting setup...
</span>
</div>
);
}
// ---------------------------------------------------------------------------
// OAuth Login View
// ---------------------------------------------------------------------------
+1 -2
View File
@@ -1,7 +1,6 @@
import { useState } from 'react';
import type { Tunnel } from '../../services/api/tunnelsApi';
import { tunnelsApi } from '../../services/api/tunnelsApi';
import { type Tunnel, tunnelsApi } from '../../services/api/tunnelsApi';
import type { TunnelRegistration } from '../../store/webhooksSlice';
import { BACKEND_URL } from '../../utils/config';
+43 -11
View File
@@ -13,6 +13,8 @@ import {
setSetupComplete as rpcSetSetupComplete,
revokeOAuth as rpcRevokeOAuth,
removePersistedOAuthCredential,
revokeAuth as rpcRevokeAuth,
removePersistedAuthCredential,
} from "./skillsApi";
import { syncToolsToBackend } from "./sync";
import type {
@@ -376,30 +378,40 @@ class SkillManager {
}
/**
* Disconnect a skill — revoke OAuth credentials, stop it, and reset setup state.
* Disconnect a skill — revoke OAuth and/or auth credentials, stop it, and reset setup state.
*/
async disconnectSkill(skillId: string): Promise<void> {
// Read the stored credential ID so oauth/revoked clears the right memory bucket.
// Read the stored credential IDs so revoke handlers clear the right data.
let credentialId: string | undefined;
let authMode: string | undefined;
try {
const snap = await getSkillSnapshot(skillId);
const cred = snap?.state?.__oauth_credential as
const oauthCred = snap?.state?.__oauth_credential as
| { credentialId?: string }
| string
| undefined;
if (cred && typeof cred === "object") {
credentialId = cred.credentialId;
if (oauthCred && typeof oauthCred === "object") {
credentialId = oauthCred.credentialId;
}
const authCred = snap?.state?.__auth_credential as
| { mode?: string }
| string
| undefined;
if (authCred && typeof authCred === "object") {
authMode = authCred.mode;
}
} catch {
// Snapshot may fail if skill isn't registered yet
}
// Revoke OAuth credential before stopping so the running skill can clean up
// its in-memory state and the event loop deletes oauth_credential.json.
let revokeSucceeded = false;
// Revoke credentials before stopping so the running skill can clean up.
let oauthRevokeSucceeded = false;
let authRevokeSucceeded = false;
// Try revoking OAuth credential
try {
await rpcRevokeOAuth(skillId, credentialId ?? "default");
revokeSucceeded = true;
oauthRevokeSucceeded = true;
} catch (err) {
console.debug(
"[SkillManager] oauth/revoked failed (runtime may be stopped):",
@@ -407,13 +419,33 @@ class SkillManager {
);
}
// Try revoking auth credential (attempt even if authMode is unknown)
try {
await rpcRevokeAuth(skillId, authMode ?? "unknown");
authRevokeSucceeded = true;
} catch (err) {
console.debug(
"[SkillManager] auth/revoked failed (runtime may be stopped):",
err,
);
}
try {
await this.stopSkill(skillId);
} finally {
if (!revokeSucceeded) {
// Host-side fallback cleanup if RPC revoke failed
if (!oauthRevokeSucceeded) {
await removePersistedOAuthCredential(skillId).catch((err) => {
console.debug(
"[SkillManager] host-side credential cleanup failed:",
"[SkillManager] host-side OAuth credential cleanup failed:",
err,
);
});
}
if (!authRevokeSucceeded) {
await removePersistedAuthCredential(skillId).catch((err) => {
console.debug(
"[SkillManager] host-side auth credential cleanup failed:",
err,
);
});
+39 -1
View File
@@ -28,7 +28,22 @@ export interface AvailableSkillEntryRpc {
entry: string;
auto_start: boolean;
platforms?: string[] | null;
setup?: { required?: boolean; label?: string; oauth?: { provider: string; scopes: string[]; apiBaseUrl: string } } | null;
setup?: {
required?: boolean;
label?: string;
oauth?: { provider: string; scopes: string[]; apiBaseUrl: string };
auth?: { modes: Array<{
type: string;
label?: string;
description?: string;
provider?: string;
scopes?: string[];
apiBaseUrl?: string;
fields?: Array<Record<string, unknown>>;
textDescription?: string;
textPlaceholder?: string;
}> };
} | null;
ignore_in_production: boolean;
download_url: string;
manifest_url: string;
@@ -140,6 +155,29 @@ export async function removePersistedOAuthCredential(skillId: string): Promise<v
});
}
/** Revoke advanced auth credential via skill RPC. */
export async function revokeAuth(skillId: string, mode?: string): Promise<void> {
await callCoreRpc({
method: 'openhuman.skills_rpc',
params: {
skill_id: skillId,
method: 'auth/revoked',
params: { mode: mode ?? 'unknown' },
},
});
}
/**
* Host-side fallback: delete auth_credential.json from the skill's data dir.
* Used when the runtime is already stopped so auth/revoked RPC can't reach it.
*/
export async function removePersistedAuthCredential(skillId: string): Promise<void> {
await callCoreRpc({
method: 'openhuman.skills_data_write',
params: { skill_id: skillId, filename: 'auth_credential.json', content: '' },
});
}
export async function disableSkill(skillId: string): Promise<void> {
await callCoreRpc({
method: 'openhuman.skills_disable',
+32 -1
View File
@@ -25,11 +25,14 @@ export interface SkillManifest {
setup?: {
required: boolean;
label?: string;
/** Legacy OAuth configuration. Prefer `auth` for new skills. */
oauth?: {
provider: string;
scopes: string[];
apiBaseUrl: string;
};
/** Advanced auth configuration with multiple auth modes. */
auth?: SkillAuthConfig;
};
/** Platform filter. When present, only listed platforms load this skill.
* When absent or empty, the skill is available on all platforms. */
@@ -51,7 +54,7 @@ export interface SetupFieldOption {
export interface SetupField {
name: string;
type: "text" | "number" | "password" | "select" | "multiselect" | "boolean";
type: "text" | "number" | "password" | "select" | "multiselect" | "boolean" | "textarea";
label: string;
description?: string | null;
required: boolean;
@@ -79,6 +82,34 @@ export interface SetupResult {
message?: string | null;
}
// ---------------------------------------------------------------------------
// Advanced Auth Configuration
// ---------------------------------------------------------------------------
/** A single authentication mode that a skill supports. */
export interface AuthMode {
type: "managed" | "self_hosted" | "text";
label?: string;
description?: string;
/** Managed mode: OAuth provider name. */
provider?: string;
/** Managed mode: OAuth scopes. */
scopes?: string[];
/** Managed mode: base URL for API proxying. */
apiBaseUrl?: string;
/** Self-hosted mode: form fields for credential input. */
fields?: SetupField[];
/** Text mode: hint above the textarea. */
textDescription?: string;
/** Text mode: placeholder in the textarea. */
textPlaceholder?: string;
}
/** Advanced auth configuration with multiple auth modes. */
export interface SkillAuthConfig {
modes: AuthMode[];
}
// ---------------------------------------------------------------------------
// JSON-RPC 2.0
// ---------------------------------------------------------------------------
+50 -37
View File
@@ -1185,36 +1185,39 @@ const Conversations = () => {
)}
<div className="flex-shrink-0 border-t border-stone-200 px-4 py-3">
{teamUsage && (teamUsage.remainingUsd <= 0 || (teamUsage.fiveHourCapUsd > 0 && teamUsage.fiveHourSpendUsd >= teamUsage.fiveHourCapUsd)) && (
<div className="mb-3 p-3 rounded-xl bg-coral-50 border border-coral-200 flex items-center justify-between gap-3">
<div className="flex items-center gap-2 min-w-0">
<svg
className="w-4 h-4 text-coral-400 flex-shrink-0"
fill="none"
stroke="currentColor"
viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={2}
d="M12 9v2m0 4h.01m-6.938 4h13.856c1.54 0 2.502-1.667 1.732-3L13.732 4c-.77-1.333-2.694-1.333-3.464 0L3.34 16c-.77 1.333.192 3 1.732 3z"
/>
</svg>
<p className="text-xs text-coral-600 truncate">
{teamUsage.remainingUsd <= 0
? 'Weekly inference budget exhausted. Top up to continue.'
: `5-hour rate limit reached.${teamUsage.fiveHourResetsAt ? ` Resets ${formatResetTime(teamUsage.fiveHourResetsAt)}.` : ''}`}
</p>
{teamUsage &&
(teamUsage.remainingUsd <= 0 ||
(teamUsage.fiveHourCapUsd > 0 &&
teamUsage.fiveHourSpendUsd >= teamUsage.fiveHourCapUsd)) && (
<div className="mb-3 p-3 rounded-xl bg-coral-50 border border-coral-200 flex items-center justify-between gap-3">
<div className="flex items-center gap-2 min-w-0">
<svg
className="w-4 h-4 text-coral-400 flex-shrink-0"
fill="none"
stroke="currentColor"
viewBox="0 0 24 24">
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth={2}
d="M12 9v2m0 4h.01m-6.938 4h13.856c1.54 0 2.502-1.667 1.732-3L13.732 4c-.77-1.333-2.694-1.333-3.464 0L3.34 16c-.77 1.333.192 3 1.732 3z"
/>
</svg>
<p className="text-xs text-coral-600 truncate">
{teamUsage.remainingUsd <= 0
? 'Weekly inference budget exhausted. Top up to continue.'
: `5-hour rate limit reached.${teamUsage.fiveHourResetsAt ? ` Resets ${formatResetTime(teamUsage.fiveHourResetsAt)}.` : ''}`}
</p>
</div>
{teamUsage.remainingUsd <= 0 && (
<button
onClick={() => navigate('/settings/billing')}
className="flex-shrink-0 px-3 py-1.5 rounded-lg bg-coral-500 hover:bg-coral-400 text-white text-xs font-medium transition-colors">
Top Up
</button>
)}
</div>
{teamUsage.remainingUsd <= 0 && (
<button
onClick={() => navigate('/settings/billing')}
className="flex-shrink-0 px-3 py-1.5 rounded-lg bg-coral-500 hover:bg-coral-400 text-white text-xs font-medium transition-colors">
Top Up
</button>
)}
</div>
)}
)}
<div className="flex items-center gap-2 mb-2">
{isLoadingModels ? (
@@ -1297,15 +1300,23 @@ const Conversations = () => {
<div className="flex items-center gap-2">
<LimitPill
label="5h"
usedPct={teamUsage.fiveHourCapUsd > 0
? Math.min(1, teamUsage.fiveHourSpendUsd / teamUsage.fiveHourCapUsd)
: 0}
usedPct={
teamUsage.fiveHourCapUsd > 0
? Math.min(1, teamUsage.fiveHourSpendUsd / teamUsage.fiveHourCapUsd)
: 0
}
/>
<LimitPill
label="7d"
usedPct={teamUsage.cycleBudgetUsd > 0
? Math.min(1, (teamUsage.cycleBudgetUsd - teamUsage.remainingUsd) / teamUsage.cycleBudgetUsd)
: 0}
usedPct={
teamUsage.cycleBudgetUsd > 0
? Math.min(
1,
(teamUsage.cycleBudgetUsd - teamUsage.remainingUsd) /
teamUsage.cycleBudgetUsd
)
: 0
}
/>
</div>
) : (
@@ -1317,7 +1328,8 @@ const Conversations = () => {
<div className="flex items-center justify-between gap-4">
<span className="text-stone-400">5-hour limit</span>
<span>
${teamUsage.fiveHourSpendUsd.toFixed(2)} / ${teamUsage.fiveHourCapUsd.toFixed(2)}
${teamUsage.fiveHourSpendUsd.toFixed(2)} / $
{teamUsage.fiveHourCapUsd.toFixed(2)}
{teamUsage.fiveHourResetsAt && (
<span className="text-stone-400 ml-1">
resets {formatResetTime(teamUsage.fiveHourResetsAt)}
@@ -1328,7 +1340,8 @@ const Conversations = () => {
<div className="flex items-center justify-between gap-4">
<span className="text-stone-400">Weekly limit</span>
<span>
${teamUsage.remainingUsd.toFixed(2)} / ${teamUsage.cycleBudgetUsd.toFixed(2)} left
${teamUsage.remainingUsd.toFixed(2)} / $
{teamUsage.cycleBudgetUsd.toFixed(2)} left
{teamUsage.cycleEndsAt && (
<span className="text-stone-400 ml-1">
resets {formatResetTime(teamUsage.cycleEndsAt)}
@@ -1,35 +0,0 @@
//! Cron scheduling bridge for skills.
//!
//! Thin wrapper around `CronScheduler` methods exposed to skill JS contexts.
//! Skills register cron schedules here and receive callbacks via
//! `globalThis.onCronTrigger(scheduleId)`.
//!
//! Cron expressions use 6 fields (sec min hour dom month dow):
//! "0 */5 * * * *" = every 5 minutes
//! "0 0 * * * *" = every hour
//! "0 0 9 * * Mon" = 9 AM every Monday
use std::sync::Arc;
use crate::openhuman::skills::cron_scheduler::CronScheduler;
/// Register a cron schedule for a skill.
pub fn register(
scheduler: &Arc<CronScheduler>,
skill_id: &str,
schedule_id: &str,
cron_expr: &str,
) -> Result<(), String> {
scheduler.register(skill_id, schedule_id, cron_expr)
}
/// Unregister a cron schedule for a skill.
pub fn unregister(scheduler: &Arc<CronScheduler>, skill_id: &str, schedule_id: &str) {
scheduler.unregister(skill_id, schedule_id)
}
/// List all cron schedules for a skill.
/// Returns pairs of (schedule_id, next_fire_time_rfc3339).
pub fn list(scheduler: &Arc<CronScheduler>, skill_id: &str) -> Vec<(String, String)> {
scheduler.list_schedules(skill_id)
}
-231
View File
@@ -1,231 +0,0 @@
//! @openhuman/db bridge — scoped SQLite access for each skill.
//!
//! Each skill gets its own SQLite database at `{app_data_dir}/skills/{skill_id}/skill.db`.
//! The bridge exposes async functions callable from JS:
//! - db.exec(sql, params?) → run a statement (INSERT/UPDATE/DELETE/CREATE)
//! - db.get(sql, params?) → fetch one row as object
//! - db.all(sql, params?) → fetch all rows as array of objects
//! - db.kvGet(key) → get a value from the built-in __kv table
//! - db.kvSet(key, value) → set a value in the built-in __kv table
use rusqlite::{params_from_iter, Connection, OpenFlags};
use serde_json::Value as JsonValue;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
/// Handle to a skill's scoped SQLite database.
#[derive(Clone)]
pub struct SkillDb {
conn: Arc<Mutex<Connection>>,
#[allow(dead_code)]
path: PathBuf,
}
impl SkillDb {
/// Open (or create) a skill database at the given directory.
/// Initializes WAL mode and creates the __kv table.
pub fn open(skill_data_dir: &Path) -> Result<Self, String> {
std::fs::create_dir_all(skill_data_dir)
.map_err(|e| format!("Failed to create skill data dir: {e}"))?;
let db_path = skill_data_dir.join("skill.db");
let conn = Connection::open_with_flags(
&db_path,
OpenFlags::SQLITE_OPEN_READ_WRITE
| OpenFlags::SQLITE_OPEN_CREATE
| OpenFlags::SQLITE_OPEN_FULL_MUTEX,
)
.map_err(|e| format!("Failed to open skill DB at {}: {e}", db_path.display()))?;
// Enable WAL mode for better concurrent read performance
conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000;")
.map_err(|e| format!("Failed to set WAL mode: {e}"))?;
// Create the built-in key-value table
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS __kv (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);",
)
.map_err(|e| format!("Failed to create __kv table: {e}"))?;
Ok(Self {
conn: Arc::new(Mutex::new(conn)),
path: db_path,
})
}
/// Execute a SQL statement (no result rows expected).
pub fn exec(&self, sql: &str, params: &[JsonValue]) -> Result<usize, String> {
let conn = self
.conn
.lock()
.map_err(|e| format!("DB lock error: {e}"))?;
let sqlite_params = json_values_to_sqlite(params);
conn.execute(sql, params_from_iter(sqlite_params.iter()))
.map_err(|e| format!("SQL exec error: {e}"))
}
/// Fetch a single row as a JSON object. Returns `null` if no row matches.
pub fn get(&self, sql: &str, params: &[JsonValue]) -> Result<JsonValue, String> {
let conn = self
.conn
.lock()
.map_err(|e| format!("DB lock error: {e}"))?;
let sqlite_params = json_values_to_sqlite(params);
let mut stmt = conn
.prepare(sql)
.map_err(|e| format!("SQL prepare error: {e}"))?;
let column_names: Vec<String> = stmt
.column_names()
.into_iter()
.map(|s| s.to_string())
.collect();
let mut rows = stmt
.query(params_from_iter(sqlite_params.iter()))
.map_err(|e| format!("SQL query error: {e}"))?;
match rows.next().map_err(|e| format!("SQL row error: {e}"))? {
Some(row) => row_to_json(row, &column_names),
None => Ok(JsonValue::Null),
}
}
/// Fetch all matching rows as a JSON array of objects.
pub fn all(&self, sql: &str, params: &[JsonValue]) -> Result<JsonValue, String> {
let conn = self
.conn
.lock()
.map_err(|e| format!("DB lock error: {e}"))?;
let sqlite_params = json_values_to_sqlite(params);
let mut stmt = conn
.prepare(sql)
.map_err(|e| format!("SQL prepare error: {e}"))?;
let column_names: Vec<String> = stmt
.column_names()
.into_iter()
.map(|s| s.to_string())
.collect();
let mut rows_out = Vec::new();
let mut rows = stmt
.query(params_from_iter(sqlite_params.iter()))
.map_err(|e| format!("SQL query error: {e}"))?;
while let Some(row) = rows.next().map_err(|e| format!("SQL row error: {e}"))? {
rows_out.push(row_to_json(row, &column_names)?);
}
Ok(JsonValue::Array(rows_out))
}
/// Get a value from the __kv table.
pub fn kv_get(&self, key: &str) -> Result<JsonValue, String> {
let result = self.get(
"SELECT value FROM __kv WHERE key = ?",
&[JsonValue::String(key.to_string())],
)?;
match result {
JsonValue::Null => Ok(JsonValue::Null),
obj => {
let val_str = obj.get("value").and_then(|v| v.as_str()).unwrap_or("null");
serde_json::from_str(val_str)
.map_err(|e| format!("Failed to parse KV value for '{key}': {e}"))
}
}
}
/// Set a value in the __kv table. Value is stored as JSON string.
pub fn kv_set(&self, key: &str, value: &JsonValue) -> Result<(), String> {
let serialized =
serde_json::to_string(value).map_err(|e| format!("Failed to serialize value: {e}"))?;
self.exec(
"INSERT INTO __kv (key, value, updated_at) VALUES (?, ?, datetime('now'))
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at",
&[
JsonValue::String(key.to_string()),
JsonValue::String(serialized),
],
)?;
Ok(())
}
}
/// Wrapper to hold a SQLite value that implements `rusqlite::ToSql`.
enum SqliteValue {
Null,
Integer(i64),
Real(f64),
Text(String),
}
impl rusqlite::types::ToSql for SqliteValue {
fn to_sql(&self) -> rusqlite::Result<rusqlite::types::ToSqlOutput<'_>> {
match self {
SqliteValue::Null => Ok(rusqlite::types::ToSqlOutput::Owned(
rusqlite::types::Value::Null,
)),
SqliteValue::Integer(i) => Ok(rusqlite::types::ToSqlOutput::Owned(
rusqlite::types::Value::Integer(*i),
)),
SqliteValue::Real(f) => Ok(rusqlite::types::ToSqlOutput::Owned(
rusqlite::types::Value::Real(*f),
)),
SqliteValue::Text(s) => Ok(rusqlite::types::ToSqlOutput::Owned(
rusqlite::types::Value::Text(s.clone()),
)),
}
}
}
/// Convert JSON values to SQLite-compatible parameter values.
fn json_values_to_sqlite(params: &[JsonValue]) -> Vec<SqliteValue> {
params
.iter()
.map(|v| match v {
JsonValue::Null => SqliteValue::Null,
JsonValue::Bool(b) => SqliteValue::Integer(if *b { 1 } else { 0 }),
JsonValue::Number(n) => {
if let Some(i) = n.as_i64() {
SqliteValue::Integer(i)
} else if let Some(f) = n.as_f64() {
SqliteValue::Real(f)
} else {
SqliteValue::Text(n.to_string())
}
}
JsonValue::String(s) => SqliteValue::Text(s.clone()),
other => SqliteValue::Text(other.to_string()),
})
.collect()
}
/// Convert a SQLite row to a JSON object using column names.
fn row_to_json(row: &rusqlite::Row<'_>, column_names: &[String]) -> Result<JsonValue, String> {
let mut obj = serde_json::Map::new();
for (i, name) in column_names.iter().enumerate() {
let value: JsonValue = match row.get_ref(i) {
Ok(rusqlite::types::ValueRef::Null) => JsonValue::Null,
Ok(rusqlite::types::ValueRef::Integer(n)) => JsonValue::Number(n.into()),
Ok(rusqlite::types::ValueRef::Real(f)) => serde_json::Number::from_f64(f)
.map(JsonValue::Number)
.unwrap_or(JsonValue::Null),
Ok(rusqlite::types::ValueRef::Text(s)) => {
let text = String::from_utf8_lossy(s).to_string();
JsonValue::String(text)
}
Ok(rusqlite::types::ValueRef::Blob(b)) => JsonValue::String(base64::Engine::encode(
&base64::engine::general_purpose::STANDARD,
b,
)),
Err(e) => return Err(format!("Failed to read column '{}': {}", name, e)),
};
obj.insert(name.clone(), value);
}
Ok(JsonValue::Object(obj))
}
-49
View File
@@ -1,49 +0,0 @@
//! @openhuman/log bridge — structured logging from JS skills.
//!
//! Exposes log levels: debug, info, warn, error.
//! Logs are forwarded to Rust's `log` crate AND emitted as Tauri events
//! so the frontend can display them.
use serde::{Deserialize, Serialize};
/// A log entry produced by a skill.
#[allow(dead_code)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SkillLogEntry {
pub skill_id: String,
pub level: LogLevel,
pub message: String,
pub timestamp: String,
}
/// Log severity levels.
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum LogLevel {
Debug,
Info,
Warn,
Error,
}
/// Log a message from a skill. Forwards to Rust `log` crate with skill context.
pub fn skill_log(skill_id: &str, level: LogLevel, message: &str) {
let prefixed = format!("[skill:{}] {}", skill_id, message);
match level {
LogLevel::Debug => log::debug!("{}", prefixed),
LogLevel::Info => log::info!("{}", prefixed),
LogLevel::Warn => log::warn!("{}", prefixed),
LogLevel::Error => log::error!("{}", prefixed),
}
}
/// Create a `SkillLogEntry` with the current timestamp.
#[allow(dead_code)]
pub fn make_log_entry(skill_id: &str, level: LogLevel, message: &str) -> SkillLogEntry {
SkillLogEntry {
skill_id: skill_id.to_string(),
level,
message: message.to_string(),
timestamp: chrono::Utc::now().to_rfc3339(),
}
}
-15
View File
@@ -1,18 +1,3 @@
//! JS-to-Rust bridge modules for the skill runtime.
//!
//! Currently only `net` is actively used (by V8 ops).
//! Other modules are preserved for potential future use.
#[allow(dead_code)]
pub mod cron_bridge;
#[allow(dead_code)]
pub mod db;
#[allow(dead_code)]
pub mod log_bridge;
pub mod net;
#[allow(dead_code)]
pub mod skills_bridge;
#[allow(dead_code)]
pub mod store;
#[allow(dead_code)]
pub mod tauri_bridge;
@@ -1,46 +0,0 @@
//! Inter-skill communication bridge.
//!
//! Deliberately restricted: running skills cannot invoke tools owned by
//! other skills. Keep this boundary explicit even if bridge APIs are
//! reintroduced in the future.
use std::sync::Arc;
use crate::openhuman::skills::skill_registry::SkillRegistry;
/// List all running skills.
/// Returns a JSON string: `[{"skillId":"...","name":"...","status":"..."}]`
pub fn list_skills(registry: &Arc<SkillRegistry>) -> String {
let skills = registry.list_skills();
let simplified: Vec<serde_json::Value> = skills
.iter()
.map(|s| {
serde_json::json!({
"skillId": s.skill_id,
"name": s.name,
"status": s.status,
})
})
.collect();
serde_json::to_string(&simplified).unwrap_or_else(|_| "[]".to_string())
}
/// Call a tool on another skill.
///
/// Spawns an OS thread with a mini Tokio runtime to make the async
/// registry call without conflicting with the V8 runtime or the
/// outer Tokio runtime.
///
/// Returns the ToolResult as a JSON string.
pub fn call_tool(
_registry: &Arc<SkillRegistry>,
caller_skill_id: &str,
target_skill_id: &str,
_tool_name: &str,
_arguments_json: &str,
) -> Result<String, String> {
Err(format!(
"Cross-skill tool invocation is disabled: '{}' cannot call '{}'",
caller_skill_id, target_skill_id
))
}
-52
View File
@@ -1,52 +0,0 @@
//! @openhuman/store bridge — persisted key-value state for skills.
//!
//! Thin wrapper around the __kv table in each skill's SQLite database.
//! Provides a simpler API than raw SQL for common state persistence patterns.
//!
//! JS API:
//! store.get(key) → Promise<any>
//! store.set(key, value) → Promise<void>
//! store.delete(key) → Promise<void>
//! store.keys() → Promise<string[]>
use super::db::SkillDb;
use serde_json::Value as JsonValue;
/// Store operations delegate to the skill's SkillDb __kv table.
#[derive(Clone)]
pub struct SkillStore {
db: SkillDb,
}
impl SkillStore {
pub fn new(db: SkillDb) -> Self {
Self { db }
}
pub fn get(&self, key: &str) -> Result<JsonValue, String> {
self.db.kv_get(key)
}
pub fn set(&self, key: &str, value: &JsonValue) -> Result<(), String> {
self.db.kv_set(key, value)
}
pub fn delete(&self, key: &str) -> Result<(), String> {
self.db.exec(
"DELETE FROM __kv WHERE key = ?",
&[JsonValue::String(key.to_string())],
)?;
Ok(())
}
pub fn keys(&self) -> Result<Vec<String>, String> {
let rows = self.db.all("SELECT key FROM __kv ORDER BY key", &[])?;
match rows {
JsonValue::Array(arr) => Ok(arr
.into_iter()
.filter_map(|v| v.get("key").and_then(|k| k.as_str()).map(|s| s.to_string()))
.collect()),
_ => Ok(Vec::new()),
}
}
}
@@ -1,34 +0,0 @@
//! Platform bridge for skills.
//!
//! Provides platform detection and native OS features like notifications.
/// Get the current platform name.
pub fn get_platform() -> &'static str {
match std::env::consts::OS {
"macos" => "macos",
"windows" => "windows",
"linux" => "linux",
_ => "unknown",
}
}
/// Send a native OS notification.
pub fn send_notification(title: &str, body: &str) -> Result<(), String> {
log::info!("[runtime] notification requested: {title} - {body}");
Ok(())
}
/// Whitelisted environment values exposed to skills via `platform.env(key)`.
/// Skills should never hardcode host-specific URLs or secrets; use this instead.
pub fn get_skill_env(key: &str) -> Option<String> {
match key {
"BACKEND_URL" => Some(
std::env::var("VITE_BACKEND_URL")
.ok()
.filter(|url| !url.trim().is_empty())
.unwrap_or_else(|| "http://localhost:5005".to_string()),
),
"PLATFORM" => Some(get_platform().to_string()),
_ => None,
}
}
-9
View File
@@ -1,9 +0,0 @@
//! Custom module resolver and loader for `@openhuman/*` imports.
//!
//! NOTE: Currently unused. Skills access bridge APIs via globals (db, store, console)
//! injected by qjs_skill_instance.rs. This module is reserved for future ES module
//! import support (e.g., `import { db } from '@openhuman/db'`).
//!
//! The globals-based approach was chosen because:
//! - Globals are simpler and sufficient for the initial implementation
//! - Per-skill module loaders can be added later if needed
+186
View File
@@ -14,7 +14,73 @@ pub struct SkillSetup {
pub required: bool,
pub label: Option<String>,
/// OAuth configuration (provider, scopes, apiBaseUrl).
/// Legacy — prefer `auth` for new skills.
pub oauth: Option<serde_json::Value>,
/// Advanced auth configuration with multiple auth modes.
/// When present, the UI shows a mode selector (managed / self_hosted / text).
#[serde(default)]
pub auth: Option<SkillAuthConfig>,
}
/// Auth mode type: managed, self_hosted, or text.
#[derive(Debug, Deserialize, Serialize, Clone, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum SkillAuthType {
Managed,
SelfHosted,
Text,
}
/// Advanced auth configuration declaring available authentication modes.
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct SkillAuthConfig {
/// Available auth modes. At least one required.
#[serde(deserialize_with = "deserialize_non_empty_modes")]
pub modes: Vec<SkillAuthMode>,
}
fn deserialize_non_empty_modes<'de, D>(deserializer: D) -> Result<Vec<SkillAuthMode>, D::Error>
where
D: serde::Deserializer<'de>,
{
let modes = Vec::<SkillAuthMode>::deserialize(deserializer)?;
if modes.is_empty() {
return Err(serde::de::Error::custom(
"auth.modes must contain at least one mode",
));
}
Ok(modes)
}
/// A single authentication mode that a skill supports.
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct SkillAuthMode {
/// Mode type: managed, self_hosted, or text.
#[serde(rename = "type")]
pub mode_type: SkillAuthType,
/// Display label for this mode in the UI.
pub label: Option<String>,
/// Short description shown below the label.
pub description: Option<String>,
// --- Managed mode fields ---
/// OAuth provider name (e.g. "google", "github", "notion").
pub provider: Option<String>,
/// OAuth scopes to request.
pub scopes: Option<Vec<String>>,
/// Base URL for API requests proxied through backend.
#[serde(rename = "apiBaseUrl")]
pub api_base_url: Option<String>,
// --- Self-hosted mode fields ---
/// Form fields for credential input (SetupField-compatible JSON objects).
#[serde(default)]
pub fields: Vec<serde_json::Value>,
// --- Text mode fields ---
/// Hint text shown above the textarea (e.g. "Paste your service account JSON").
#[serde(rename = "textDescription")]
pub text_description: Option<String>,
/// Placeholder text for the textarea.
#[serde(rename = "textPlaceholder")]
pub text_placeholder: Option<String>,
}
/// Raw manifest as it appears on disk.
@@ -124,3 +190,123 @@ impl SkillManifest {
}
}
}
#[cfg(test)]
mod tests {
use super::*;
const MINIMAL_MANIFEST: &str = r#"{"id":"test","name":"Test Skill"}"#;
const FULL_MANIFEST: &str = r#"{
"id": "price-tracker",
"name": "Price Tracker",
"runtime": "quickjs",
"entry": "main.js",
"memory_limit_mb": 128,
"auto_start": true,
"version": "1.0.0",
"description": "Tracks prices",
"platforms": ["macos", "linux"]
}"#;
#[test]
fn parse_minimal_manifest() {
let m = SkillManifest::from_json(MINIMAL_MANIFEST).unwrap();
assert_eq!(m.id, "test");
assert_eq!(m.name, "Test Skill");
assert_eq!(m.runtime, "v8"); // default
assert_eq!(m.entry, "index.js"); // default
assert!(!m.auto_start);
assert!(m.memory_limit_mb.is_none());
assert!(m.platforms.is_none());
}
#[test]
fn parse_full_manifest() {
let m = SkillManifest::from_json(FULL_MANIFEST).unwrap();
assert_eq!(m.id, "price-tracker");
assert_eq!(m.runtime, "quickjs");
assert_eq!(m.entry, "main.js");
assert_eq!(m.memory_limit_mb, Some(128));
assert!(m.auto_start);
assert_eq!(m.platforms.as_ref().unwrap().len(), 2);
}
#[test]
fn parse_invalid_json_returns_error() {
assert!(SkillManifest::from_json("not json").is_err());
}
#[test]
fn is_javascript_known_runtimes() {
for rt in &["v8", "javascript", "quickjs"] {
let json = format!(r#"{{"id":"t","name":"T","runtime":"{}"}}"#, rt);
let m = SkillManifest::from_json(&json).unwrap();
assert!(m.is_javascript(), "expected is_javascript() for '{rt}'");
}
}
#[test]
fn is_javascript_unknown_runtime() {
let m = SkillManifest::from_json(r#"{"id":"t","name":"T","runtime":"python"}"#).unwrap();
assert!(!m.is_javascript());
}
#[test]
fn supports_current_platform_no_restriction() {
let m = SkillManifest::from_json(MINIMAL_MANIFEST).unwrap();
assert!(m.supports_current_platform());
}
#[test]
fn supports_current_platform_empty_vec() {
let m = SkillManifest::from_json(r#"{"id":"t","name":"T","platforms":[]}"#).unwrap();
assert!(m.supports_current_platform());
}
#[test]
fn to_config_memory_limit_conversion() {
let m = SkillManifest::from_json(FULL_MANIFEST).unwrap();
let cfg = m.to_config();
assert_eq!(cfg.memory_limit, 128 * 1024 * 1024);
assert_eq!(cfg.skill_id, "price-tracker");
assert_eq!(cfg.entry_point, "main.js");
assert!(cfg.auto_start);
}
#[test]
fn to_config_default_memory_limit() {
let m = SkillManifest::from_json(MINIMAL_MANIFEST).unwrap();
let cfg = m.to_config();
assert_eq!(cfg.memory_limit, 64 * 1024 * 1024);
}
#[test]
fn auth_mode_type_deserializes_known_types() {
let json = r#"{"id":"t","name":"T","setup":{"required":true,"auth":{"modes":[
{"type":"managed","label":"Cloud","provider":"google"},
{"type":"self_hosted","label":"Own"},
{"type":"text","label":"Token"}
]}}}"#;
let m = SkillManifest::from_json(json).unwrap();
let modes = &m.setup.unwrap().auth.unwrap().modes;
assert_eq!(modes.len(), 3);
assert_eq!(modes[0].mode_type, SkillAuthType::Managed);
assert_eq!(modes[1].mode_type, SkillAuthType::SelfHosted);
assert_eq!(modes[2].mode_type, SkillAuthType::Text);
}
#[test]
fn auth_mode_type_rejects_invalid_type() {
let json = r#"{"id":"t","name":"T","setup":{"required":true,"auth":{"modes":[
{"type":"banana","label":"Bad"}
]}}}"#;
assert!(SkillManifest::from_json(json).is_err());
}
#[test]
fn auth_modes_rejects_empty() {
let json = r#"{"id":"t","name":"T","setup":{"required":true,"auth":{"modes":[]}}}"#;
assert!(SkillManifest::from_json(json).is_err());
}
}
+1 -1
View File
@@ -1,6 +1,5 @@
pub mod bridge;
pub mod cron_scheduler;
pub mod loader;
pub mod manifest;
pub mod ops;
pub mod ping_scheduler;
@@ -8,6 +7,7 @@ pub mod preferences;
pub mod qjs_engine;
pub mod qjs_skill_instance;
pub mod quickjs_libs;
mod registry_cache;
pub mod registry_ops;
pub mod registry_types;
mod schemas;
+89
View File
@@ -138,3 +138,92 @@ fn read_skill_md_description(path: &Path) -> Option<String> {
}
None
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn init_skills_dir_creates_dir_and_readme() {
let dir = tempfile::tempdir().unwrap();
init_skills_dir(dir.path()).unwrap();
let skills_dir = dir.path().join("skills");
assert!(skills_dir.is_dir());
let readme = skills_dir.join("README.md");
assert!(readme.exists());
let content = std::fs::read_to_string(&readme).unwrap();
assert!(content.contains("Skills"));
}
#[test]
fn init_skills_dir_idempotent() {
let dir = tempfile::tempdir().unwrap();
init_skills_dir(dir.path()).unwrap();
init_skills_dir(dir.path()).unwrap(); // should not fail
}
#[test]
fn load_skills_empty_dir() {
let dir = tempfile::tempdir().unwrap();
init_skills_dir(dir.path()).unwrap();
let skills = load_skills(dir.path());
assert!(skills.is_empty());
}
#[test]
fn load_skills_with_manifest() {
let dir = tempfile::tempdir().unwrap();
init_skills_dir(dir.path()).unwrap();
let skill_dir = dir.path().join("skills").join("my-skill");
std::fs::create_dir_all(&skill_dir).unwrap();
std::fs::write(
skill_dir.join("skill.json"),
r#"{"name":"My Skill","description":"A test","version":"1.0"}"#,
)
.unwrap();
let skills = load_skills(dir.path());
assert_eq!(skills.len(), 1);
assert_eq!(skills[0].name, "My Skill");
assert_eq!(skills[0].description, "A test");
assert_eq!(skills[0].version, "1.0");
}
#[test]
fn load_skills_fallback_name_from_dir() {
let dir = tempfile::tempdir().unwrap();
init_skills_dir(dir.path()).unwrap();
let skill_dir = dir.path().join("skills").join("fallback-name");
std::fs::create_dir_all(&skill_dir).unwrap();
// No skill.json, but has a SKILL.md
std::fs::write(
skill_dir.join("SKILL.md"),
"# Title\n\nThis is the description.",
)
.unwrap();
let skills = load_skills(dir.path());
assert_eq!(skills.len(), 1);
assert_eq!(skills[0].name, "fallback-name");
assert_eq!(skills[0].description, "This is the description.");
}
#[test]
fn load_skills_ignores_hidden_dirs() {
let dir = tempfile::tempdir().unwrap();
init_skills_dir(dir.path()).unwrap();
let hidden = dir.path().join("skills").join(".hidden");
std::fs::create_dir_all(&hidden).unwrap();
let skills = load_skills(dir.path());
assert!(skills.is_empty());
}
#[test]
fn parse_skill_manifest_missing_fields_uses_defaults() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("skill.json");
std::fs::write(&path, "{}").unwrap();
let skill = parse_skill_manifest(&path, "fallback");
assert_eq!(skill.name, "fallback");
assert!(skill.description.is_empty());
assert!(skill.tags.is_empty());
}
}
+96
View File
@@ -136,3 +136,99 @@ impl PreferencesStore {
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn temp_store() -> (PreferencesStore, tempfile::TempDir) {
let dir = tempfile::tempdir().unwrap();
let store = PreferencesStore::new(&dir.path().to_path_buf());
(store, dir)
}
#[test]
fn enable_disable_roundtrip() {
let (store, _dir) = temp_store();
assert_eq!(store.is_enabled("my-skill"), None);
store.set_enabled("my-skill", true);
assert_eq!(store.is_enabled("my-skill"), Some(true));
store.set_enabled("my-skill", false);
assert_eq!(store.is_enabled("my-skill"), Some(false));
}
#[test]
fn setup_complete_also_enables() {
let (store, _dir) = temp_store();
store.set_setup_complete("s1", true);
assert!(store.is_setup_complete("s1"));
assert_eq!(store.is_enabled("s1"), Some(true));
}
#[test]
fn persistence_across_reload() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().to_path_buf();
{
let store = PreferencesStore::new(&path);
store.set_enabled("x", true);
store.set_setup_complete("x", true);
}
// Reload from disk
let store2 = PreferencesStore::new(&path);
assert_eq!(store2.is_enabled("x"), Some(true));
assert!(store2.is_setup_complete("x"));
}
#[test]
fn missing_file_returns_defaults() {
let dir = tempfile::tempdir().unwrap();
let store = PreferencesStore::new(&dir.path().to_path_buf());
assert_eq!(store.is_enabled("nonexistent"), None);
assert!(!store.is_setup_complete("nonexistent"));
}
#[test]
fn corrupt_file_returns_defaults() {
let dir = tempfile::tempdir().unwrap();
let file = dir.path().join("skill-preferences.json");
std::fs::write(&file, "not valid json {{{}").unwrap();
let store = PreferencesStore::new(&dir.path().to_path_buf());
assert_eq!(store.is_enabled("any"), None);
}
#[test]
fn resolve_should_start_setup_complete_overrides() {
let (store, _dir) = temp_store();
store.set_setup_complete("s1", true);
// Even if manifest says false, setup_complete wins
assert!(store.resolve_should_start("s1", false));
}
#[test]
fn resolve_should_start_falls_back_to_enabled() {
let (store, _dir) = temp_store();
store.set_enabled("s1", true);
assert!(store.resolve_should_start("s1", false));
store.set_enabled("s1", false);
assert!(!store.resolve_should_start("s1", true));
}
#[test]
fn resolve_should_start_falls_back_to_manifest() {
let (store, _dir) = temp_store();
assert!(store.resolve_should_start("unknown", true));
assert!(!store.resolve_should_start("unknown", false));
}
#[test]
fn get_all_returns_complete_map() {
let (store, _dir) = temp_store();
store.set_enabled("a", true);
store.set_enabled("b", false);
let all = store.get_all();
assert_eq!(all.len(), 2);
assert!(all["a"].enabled);
assert!(!all["b"].enabled);
}
}
@@ -1,3 +1,9 @@
//! Skill event loop — drives the QuickJS runtime, routes incoming messages,
//! and persists state to memory.
mod rpc_handlers;
mod webhook_handler;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
@@ -17,14 +23,14 @@ use crate::openhuman::{
};
use super::js_handlers::{
call_lifecycle, handle_cron_trigger, handle_js_call, handle_js_void_call, handle_server_event,
call_lifecycle, handle_cron_trigger, handle_js_call, handle_js_void_call,
read_pending_tool_result, start_async_tool_call,
};
use super::js_helpers::{drive_jobs, restore_oauth_credential};
use super::js_helpers::{drive_jobs, restore_auth_credential, restore_oauth_credential};
use super::types::SkillState;
/// Payload queued for the background memory-write worker.
struct MemoryWriteJob {
pub(crate) struct MemoryWriteJob {
client: MemoryClientRef,
skill: String,
title: String,
@@ -257,11 +263,7 @@ async fn ingest_single_doc(
/// Called after sync, cron, and tick handlers so that data published via
/// `state.set()` / `state.setPartial()` during the JS handler is written to the
/// local memory store (SQLite + vector embeddings).
///
/// Writes are dispatched to a bounded background worker (see
/// [`spawn_memory_write_worker`]). If the worker is busy the write is dropped
/// rather than blocking the event loop.
fn persist_state_to_memory(
pub(crate) fn persist_state_to_memory(
skill_id: &str,
title_suffix: &str,
ops_state: &Arc<RwLock<qjs_ops::SkillState>>,
@@ -308,13 +310,6 @@ struct PendingToolCall {
}
/// The main event loop that drives the QuickJS runtime.
/// This continuously:
/// 1. Polls for ready timers and fires their callbacks
/// 2. Checks for incoming messages (non-blocking)
/// 3. Runs the QuickJS job queue for promises/async ops
/// 4. Checks if a pending async tool call has completed
/// 5. Syncs published state from ops → instance
/// 6. Sleeps efficiently when idle
pub(crate) async fn run_event_loop(
rt: &rquickjs::AsyncRuntime,
ctx: &rquickjs::AsyncContext,
@@ -326,35 +321,21 @@ pub(crate) async fn run_event_loop(
memory_client: Option<MemoryClientRef>,
data_dir: &std::path::Path,
) {
// Maximum sleep duration when no timers are pending
const MAX_IDLE_SLEEP: Duration = Duration::from_millis(100);
// Minimum sleep to prevent busy-spinning
const MIN_SLEEP: Duration = Duration::from_millis(1);
// Faster polling when waiting for an async tool call
const TOOL_POLL_SLEEP: Duration = Duration::from_millis(5);
// Bounded background worker for memory writes — limits concurrent in-flight
// store_skill_sync calls and applies backpressure when the channel is full.
let memory_write_tx = spawn_memory_write_worker();
// Tracks an in-flight async tool call whose Promise hasn't resolved yet.
let mut pending_tool: Option<PendingToolCall> = None;
loop {
// 1. Poll and fire ready timers
let ready_timers = {
let (ready, _next) = qjs_ops::poll_timers(timer_state);
ready
};
// Fire timer callbacks in JavaScript
let (ready_timers, _) = qjs_ops::poll_timers(timer_state);
for timer_id in ready_timers {
fire_timer_callback(ctx, timer_id).await;
}
// 2. Check for incoming messages (non-blocking).
// While an async tool call is in flight, still process other
// messages (events, pings, etc.) but queue any new CallTool.
// 2. Check for incoming messages (non-blocking)
match rx.try_recv() {
Ok(msg) => {
let should_stop = handle_message(
@@ -374,11 +355,8 @@ pub(crate) async fn run_event_loop(
break;
}
}
Err(mpsc::error::TryRecvError::Empty) => {
// No message - that's fine
}
Err(mpsc::error::TryRecvError::Empty) => {}
Err(mpsc::error::TryRecvError::Disconnected) => {
// Channel closed, exit
log::info!(
"[skill:{}] Message channel disconnected, stopping",
skill_id
@@ -428,7 +406,6 @@ pub(crate) async fn run_event_loop(
skill_id,
tool_execution_timeout_secs()
);
// Dump JS error state for debugging
let error_info = ctx
.with(|js_ctx| {
js_ctx
@@ -452,12 +429,11 @@ pub(crate) async fn run_event_loop(
}
}
// 5. Sync ops-level published state instance published_state + emit event
// 5. Sync ops-level published state -> instance published_state
{
let mut ops = ops_state.write();
if ops.dirty {
ops.dirty = false;
// Convert serde_json::Map → HashMap for the instance snapshot
let new_map: HashMap<String, serde_json::Value> = ops
.data
.iter()
@@ -467,9 +443,8 @@ pub(crate) async fn run_event_loop(
}
}
// 6. Calculate sleep duration based on next timer and pending tool call
// 6. Calculate sleep duration
let sleep_duration = if pending_tool.is_some() {
// Poll faster while waiting for an async tool result
TOOL_POLL_SLEEP
} else {
let (_, next_timer) = qjs_ops::poll_timers(timer_state);
@@ -481,7 +456,6 @@ pub(crate) async fn run_event_loop(
}
};
// Sleep efficiently - this yields the thread when no work is needed
tokio::time::sleep(sleep_duration).await;
}
}
@@ -497,8 +471,7 @@ async fn fire_timer_callback(ctx: &rquickjs::AsyncContext, timer_id: u32) {
.await;
}
/// Handle a single message from the channel.
/// Returns true if the skill should stop.
/// Handle a single message from the channel. Returns true if the skill should stop.
async fn handle_message(
rt: &rquickjs::AsyncRuntime,
ctx: &rquickjs::AsyncContext,
@@ -523,16 +496,14 @@ async fn handle_message(
tool_name
);
// Lazy-load persisted OAuth credential before calling the tool
restore_oauth_credential(ctx, skill_id, data_dir).await;
restore_auth_credential(ctx, skill_id, data_dir).await;
log::debug!(
"[skill:{}] event_loop: OAuth credential restored for tool '{}'",
"[skill:{}] event_loop: credentials restored for tool '{}'",
skill_id,
tool_name
);
// Start the async tool execution. The JS code stores the result
// in globals when done. The main event loop checks for completion.
match start_async_tool_call(ctx, &tool_name, arguments).await {
Ok(Some(sync_result)) => {
log::info!(
@@ -571,7 +542,6 @@ async fn handle_message(
SkillMessage::CronTrigger { schedule_id } => {
match handle_cron_trigger(rt, ctx, &schedule_id).await {
Ok(_) => {
// Persist state to memory after successful cron-triggered sync
persist_state_to_memory(
skill_id,
&format!("cron sync ({})", schedule_id),
@@ -592,7 +562,6 @@ async fn handle_message(
SkillMessage::Stop { reply } => {
let _ = call_lifecycle(rt, ctx, "stop").await;
// Clear OAuth credential from memory and mark as disconnected in store
let clear_code = r#"(function() {
if (typeof globalThis.oauth !== 'undefined' && globalThis.oauth.__setCredential) {
globalThis.oauth.__setCredential(null);
@@ -609,7 +578,7 @@ async fn handle_message(
log::info!("[skill:{}] Stopped (OAuth credential cleared)", skill_id);
let _ = reply.send(());
return true; // Signal to stop
return true;
}
SkillMessage::SetupStart { reply } => {
let result = handle_js_call(rt, ctx, "onSetupStart", "{}").await;
@@ -650,7 +619,6 @@ async fn handle_message(
SkillMessage::Tick { reply } => {
let result = handle_js_void_call(rt, ctx, "onTick", "{}").await;
if result.is_ok() {
// Persist any state published during tick to memory
persist_state_to_memory(
skill_id,
"tick sync",
@@ -698,190 +666,68 @@ async fn handle_message(
tunnel_name,
reply,
} => {
log::info!(
"[skill:{}] event_loop: WebhookRequest {} {} (tunnel={})",
let result = webhook_handler::handle_webhook_request(
rt,
ctx,
skill_id,
correlation_id,
method,
path,
headers,
query,
body,
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));
}
}
tunnel_name,
data_dir,
)
.await;
let _ = reply.send(result);
}
SkillMessage::Rpc {
method,
params,
reply,
} => {
let memory_client_opt = memory_client.clone();
let result = match method.as_str() {
"oauth/complete" => {
// Set credential on the oauth bridge + in-memory state
let cred_json =
serde_json::to_string(&params).unwrap_or_else(|_| "null".to_string());
let code = format!(
r#"(function() {{
if (typeof globalThis.oauth !== 'undefined' && globalThis.oauth.__setCredential) {{
globalThis.oauth.__setCredential({cred});
}}
if (typeof globalThis.state !== 'undefined' && globalThis.state.set) {{
globalThis.state.set('__oauth_credential', {cred});
}}
}})()"#,
cred = cred_json
);
ctx.with(|js_ctx| {
let _ = js_ctx.eval::<rquickjs::Value, _>(code.as_bytes());
})
.await;
// Persist credential to disk so it survives restarts
let cred_path = data_dir.join("oauth_credential.json");
if let Err(e) = std::fs::write(&cred_path, &cred_json) {
log::error!(
"[skill:{}] Failed to persist OAuth credential: {e}",
skill_id
);
} else {
log::info!(
"[skill:{}] OAuth credential persisted to {}",
skill_id,
cred_path.display()
);
}
let params_str =
serde_json::to_string(&params).unwrap_or_else(|_| "{}".to_string());
handle_js_call(rt, ctx, "onOAuthComplete", &params_str).await
rpc_handlers::handle_oauth_complete(rt, ctx, skill_id, params, data_dir).await
}
"skill/ping" => handle_js_call(rt, ctx, "onPing", "{}").await,
"skill/sync" => {
let result = handle_js_call(rt, ctx, "onSync", "{}").await;
if result.is_ok() {
// Persist published ops state to memory after onSync() succeeds.
// Skills publish data via state.set()/setPartial() into ops_state.data,
// not as the return value of onSync() (which is typically undefined).
persist_state_to_memory(
skill_id,
"periodic sync",
ops_state,
&memory_client_opt,
memory_write_tx,
);
}
result
rpc_handlers::handle_sync(
rt,
ctx,
skill_id,
ops_state,
memory_client,
memory_write_tx,
)
.await
}
"oauth/revoked" => {
// Clear credential: set to empty string so it's clearly "disconnected"
let clear_code = r#"(function() {
if (typeof globalThis.oauth !== 'undefined' && globalThis.oauth.__setCredential) {
globalThis.oauth.__setCredential(null);
}
if (typeof globalThis.state !== 'undefined' && globalThis.state.set) {
globalThis.state.set('__oauth_credential', '');
}
})()"#;
ctx.with(|js_ctx| {
let _ = js_ctx.eval::<rquickjs::Value, _>(clear_code.as_bytes());
})
.await;
// Remove persisted credential file
let cred_path = data_dir.join("oauth_credential.json");
let _ = std::fs::remove_file(&cred_path);
log::info!(
"[skill:{}] OAuth credential cleared from store and disk",
skill_id
);
// Fire-and-forget: delete memory for this integration
if let Some(client) = memory_client_opt {
let skill = skill_id.to_string();
let integration_id = params
.get("integrationId")
.and_then(|v| v.as_str())
.unwrap_or("unknown")
.to_string();
tokio::spawn(async move {
if let Err(e) = client.clear_skill_memory(&skill, &integration_id).await
{
log::warn!("[memory] clear_skill_memory failed: {e}");
} else {
log::info!(
"[memory] Cleared memory for {}:{}",
skill,
integration_id
);
}
});
}
let params_str =
serde_json::to_string(&params).unwrap_or_else(|_| "{}".to_string());
handle_js_void_call(rt, ctx, "onOAuthRevoked", &params_str)
.await
.map(|()| serde_json::json!({ "ok": true }))
rpc_handlers::handle_oauth_revoked(
rt,
ctx,
skill_id,
params,
data_dir,
memory_client,
)
.await
}
"auth/complete" => {
rpc_handlers::handle_auth_complete(rt, ctx, skill_id, params, data_dir).await
}
"auth/revoked" => {
rpc_handlers::handle_auth_revoked(
rt,
ctx,
skill_id,
params,
data_dir,
memory_client,
)
.await
}
_ => {
let args = serde_json::json!({ "method": method, "params": params });
@@ -891,9 +737,11 @@ async fn handle_message(
let _ = reply.send(result);
}
}
false // Don't stop
false
}
use super::js_handlers::handle_server_event;
#[cfg(test)]
mod tests {
use super::*;
@@ -0,0 +1,344 @@
//! RPC message handlers for the skill event loop.
//!
//! Each function handles one RPC method (oauth/complete, auth/complete, etc.)
//! and returns the result as a `Result<serde_json::Value, String>`.
use std::sync::Arc;
use parking_lot::RwLock;
use tokio::sync::mpsc;
use crate::openhuman::{memory::MemoryClientRef, skills::quickjs_libs::qjs_ops};
use super::{persist_state_to_memory, MemoryWriteJob};
use crate::openhuman::skills::qjs_skill_instance::js_handlers::{
handle_js_call, handle_js_void_call,
};
/// Handle `oauth/complete` RPC: inject credential into JS, persist to disk, call onOAuthComplete.
pub(crate) async fn handle_oauth_complete(
rt: &rquickjs::AsyncRuntime,
ctx: &rquickjs::AsyncContext,
skill_id: &str,
params: serde_json::Value,
data_dir: &std::path::Path,
) -> Result<serde_json::Value, String> {
let cred_json = serde_json::to_string(&params).unwrap_or_else(|_| "null".to_string());
let code = format!(
r#"(function() {{
if (typeof globalThis.oauth !== 'undefined' && globalThis.oauth.__setCredential) {{
globalThis.oauth.__setCredential({cred});
}}
if (typeof globalThis.state !== 'undefined' && globalThis.state.set) {{
globalThis.state.set('__oauth_credential', {cred});
}}
}})()"#,
cred = cred_json
);
ctx.with(|js_ctx| {
let _ = js_ctx.eval::<rquickjs::Value, _>(code.as_bytes());
})
.await;
let cred_path = data_dir.join("oauth_credential.json");
if let Err(e) = std::fs::write(&cred_path, &cred_json) {
log::error!(
"[skill:{}] Failed to persist OAuth credential: {e}",
skill_id
);
} else {
log::info!(
"[skill:{}] OAuth credential persisted to {}",
skill_id,
cred_path.display()
);
}
let params_str = serde_json::to_string(&params).unwrap_or_else(|_| "{}".to_string());
handle_js_call(rt, ctx, "onOAuthComplete", &params_str).await
}
/// Handle `oauth/revoked` RPC: clear credential from JS/disk/memory, call onOAuthRevoked.
pub(crate) async fn handle_oauth_revoked(
rt: &rquickjs::AsyncRuntime,
ctx: &rquickjs::AsyncContext,
skill_id: &str,
params: serde_json::Value,
data_dir: &std::path::Path,
memory_client: &Option<MemoryClientRef>,
) -> Result<serde_json::Value, String> {
let clear_code = r#"(function() {
if (typeof globalThis.oauth !== 'undefined' && globalThis.oauth.__setCredential) {
globalThis.oauth.__setCredential(null);
}
if (typeof globalThis.state !== 'undefined' && globalThis.state.set) {
globalThis.state.set('__oauth_credential', '');
}
})()"#;
ctx.with(|js_ctx| {
let _ = js_ctx.eval::<rquickjs::Value, _>(clear_code.as_bytes());
})
.await;
let cred_path = data_dir.join("oauth_credential.json");
let _ = std::fs::remove_file(&cred_path);
log::info!(
"[skill:{}] OAuth credential cleared from store and disk",
skill_id
);
// Fire-and-forget: delete memory for this integration
if let Some(client) = memory_client.clone() {
let skill = skill_id.to_string();
let integration_id = params
.get("integrationId")
.and_then(|v| v.as_str())
.unwrap_or("unknown")
.to_string();
tokio::spawn(async move {
if let Err(e) = client.clear_skill_memory(&skill, &integration_id).await {
log::warn!("[memory] clear_skill_memory failed: {e}");
} else {
log::info!("[memory] Cleared memory for {}:{}", skill, integration_id);
}
});
}
let params_str = serde_json::to_string(&params).unwrap_or_else(|_| "{}".to_string());
handle_js_void_call(rt, ctx, "onOAuthRevoked", &params_str)
.await
.map(|()| serde_json::json!({ "ok": true }))
}
/// Handle `auth/complete` RPC: validate via onAuthComplete first, then inject
/// credentials and persist to disk only on success.
pub(crate) async fn handle_auth_complete(
rt: &rquickjs::AsyncRuntime,
ctx: &rquickjs::AsyncContext,
skill_id: &str,
params: serde_json::Value,
data_dir: &std::path::Path,
) -> Result<serde_json::Value, String> {
let cred_json = serde_json::to_string(&params).unwrap_or_else(|_| "null".to_string());
let is_managed = params
.get("mode")
.and_then(|v| v.as_str())
.map(|m| m == "managed")
.unwrap_or(false);
// Temporarily inject credentials so onAuthComplete can use them for validation
// (e.g. making test API calls). We'll clear them if validation fails.
let temp_code = format!(
r#"(function() {{
if (typeof globalThis.auth !== 'undefined' && globalThis.auth.__setCredential) {{
globalThis.auth.__setCredential({cred});
}}
}})()"#,
cred = cred_json
);
ctx.with(|js_ctx| {
let _ = js_ctx.eval::<rquickjs::Value, _>(temp_code.as_bytes());
})
.await;
// Call skill's onAuthComplete lifecycle hook for validation
let params_str = serde_json::to_string(&params).unwrap_or_else(|_| "{}".to_string());
let result = handle_js_call(rt, ctx, "onAuthComplete", &params_str).await;
// Check if validation failed (error return or {status:"error"} response)
let validation_failed = match &result {
Err(_) => true,
Ok(val) => val.get("status").and_then(|s| s.as_str()) == Some("error"),
};
if validation_failed {
// Clear the temporary credential injection
let clear_code = r#"(function() {
if (typeof globalThis.auth !== 'undefined' && globalThis.auth.__setCredential) {
globalThis.auth.__setCredential(null);
}
})()"#;
ctx.with(|js_ctx| {
let _ = js_ctx.eval::<rquickjs::Value, _>(clear_code.as_bytes());
})
.await;
log::info!(
"[skill:{}] auth/complete validation failed, credentials not persisted",
skill_id
);
return result;
}
// Validation succeeded — now persist credentials into state and disk
// Build managed-mode bridge code (inject into oauth globals too)
let managed_bridge = if is_managed {
let creds_json = serde_json::to_string(
params
.get("credentials")
.unwrap_or(&serde_json::Value::Null),
)
.unwrap_or_else(|_| "null".to_string());
format!(
r#"
if (typeof globalThis.oauth !== 'undefined' && globalThis.oauth.__setCredential) {{
globalThis.oauth.__setCredential({creds});
}}
if (typeof globalThis.state !== 'undefined' && globalThis.state.set) {{
globalThis.state.set('__oauth_credential', {creds});
}}"#,
creds = creds_json
)
} else {
String::new()
};
let persist_code = format!(
r#"(function() {{
if (typeof globalThis.state !== 'undefined' && globalThis.state.set) {{
globalThis.state.set('__auth_credential', {cred});
}}
{managed_bridge}
}})()"#,
cred = cred_json,
managed_bridge = managed_bridge
);
ctx.with(|js_ctx| {
let _ = js_ctx.eval::<rquickjs::Value, _>(persist_code.as_bytes());
})
.await;
// Persist auth credential to disk
let cred_path = data_dir.join("auth_credential.json");
if let Err(e) = std::fs::write(&cred_path, &cred_json) {
log::error!(
"[skill:{}] Failed to persist auth credential: {e}",
skill_id
);
} else {
log::info!(
"[skill:{}] Auth credential persisted to {}",
skill_id,
cred_path.display()
);
}
// For managed mode, also persist as oauth_credential.json for backward compat
if is_managed {
let oauth_cred_json = serde_json::to_string(
params
.get("credentials")
.unwrap_or(&serde_json::Value::Null),
)
.unwrap_or_else(|_| "null".to_string());
let oauth_path = data_dir.join("oauth_credential.json");
if let Err(e) = std::fs::write(&oauth_path, &oauth_cred_json) {
log::error!(
"[skill:{}] Failed to persist managed OAuth credential: {e}",
skill_id
);
}
}
result
}
/// Handle `auth/revoked` RPC: clear auth credential from JS/disk/memory, call onAuthRevoked.
pub(crate) async fn handle_auth_revoked(
rt: &rquickjs::AsyncRuntime,
ctx: &rquickjs::AsyncContext,
skill_id: &str,
params: serde_json::Value,
data_dir: &std::path::Path,
memory_client: &Option<MemoryClientRef>,
) -> Result<serde_json::Value, String> {
let is_managed = params
.get("mode")
.and_then(|v| v.as_str())
.map(|m| m == "managed")
.unwrap_or(false);
let managed_clear = if is_managed {
r#"
if (typeof globalThis.oauth !== 'undefined' && globalThis.oauth.__setCredential) {
globalThis.oauth.__setCredential(null);
}
if (typeof globalThis.state !== 'undefined' && globalThis.state.set) {
globalThis.state.set('__oauth_credential', '');
}"#
} else {
""
};
let clear_code = format!(
r#"(function() {{
if (typeof globalThis.auth !== 'undefined' && globalThis.auth.__setCredential) {{
globalThis.auth.__setCredential(null);
}}
if (typeof globalThis.state !== 'undefined' && globalThis.state.set) {{
globalThis.state.set('__auth_credential', '');
}}
{managed_clear}
}})()"#,
managed_clear = managed_clear
);
ctx.with(|js_ctx| {
let _ = js_ctx.eval::<rquickjs::Value, _>(clear_code.as_bytes());
})
.await;
// Remove persisted credential files
let cred_path = data_dir.join("auth_credential.json");
let _ = std::fs::remove_file(&cred_path);
if is_managed {
let oauth_path = data_dir.join("oauth_credential.json");
let _ = std::fs::remove_file(&oauth_path);
}
log::info!(
"[skill:{}] Auth credential cleared from store and disk",
skill_id
);
// Fire-and-forget: delete memory for this integration
if let Some(client) = memory_client.clone() {
let skill = skill_id.to_string();
let integration_id = params
.get("integrationId")
.and_then(|v| v.as_str())
.unwrap_or("unknown")
.to_string();
tokio::spawn(async move {
if let Err(e) = client.clear_skill_memory(&skill, &integration_id).await {
log::warn!("[memory] clear_skill_memory failed: {e}");
} else {
log::info!("[memory] Cleared memory for {}:{}", skill, integration_id);
}
});
}
let params_str = serde_json::to_string(&params).unwrap_or_else(|_| "{}".to_string());
handle_js_void_call(rt, ctx, "onAuthRevoked", &params_str)
.await
.map(|()| serde_json::json!({ "ok": true }))
}
/// Handle `skill/sync` RPC: call onSync, persist state to memory on success.
pub(crate) async fn handle_sync(
rt: &rquickjs::AsyncRuntime,
ctx: &rquickjs::AsyncContext,
skill_id: &str,
ops_state: &Arc<RwLock<qjs_ops::SkillState>>,
memory_client: &Option<MemoryClientRef>,
memory_write_tx: &mpsc::Sender<MemoryWriteJob>,
) -> Result<serde_json::Value, String> {
let result = handle_js_call(rt, ctx, "onSync", "{}").await;
if result.is_ok() {
persist_state_to_memory(
skill_id,
"periodic sync",
ops_state,
memory_client,
memory_write_tx,
);
}
result
}
@@ -0,0 +1,97 @@
//! Webhook request handler for the skill event loop.
use std::collections::HashMap;
use crate::openhuman::webhooks::WebhookResponseData;
use crate::openhuman::skills::qjs_skill_instance::js_handlers::handle_js_call;
use crate::openhuman::skills::qjs_skill_instance::js_helpers::{
restore_auth_credential, restore_oauth_credential,
};
/// Handle an incoming webhook request: restore credentials, call onWebhookRequest,
/// parse the JS response into a `WebhookResponseData`.
pub(crate) async fn handle_webhook_request(
rt: &rquickjs::AsyncRuntime,
ctx: &rquickjs::AsyncContext,
skill_id: &str,
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,
data_dir: &std::path::Path,
) -> Result<WebhookResponseData, String> {
log::info!(
"[skill:{}] event_loop: WebhookRequest {} {} (tunnel={})",
skill_id,
method,
path,
tunnel_id,
);
// Restore credentials in case the handler needs authenticated API calls
restore_oauth_credential(ctx, skill_id, data_dir).await;
restore_auth_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) => {
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,
);
Ok(WebhookResponseData {
correlation_id,
status_code,
headers: resp_headers,
body: resp_body,
})
}
Err(e) => {
log::warn!(
"[skill:{}] event_loop: onWebhookRequest failed: {}",
skill_id,
e,
);
Err(e)
}
}
}
@@ -9,7 +9,10 @@ use crate::openhuman::skills::types::{SkillConfig, SkillMessage, SkillSnapshot,
use super::event_loop::run_event_loop;
use super::js_handlers::{call_lifecycle, handle_js_call};
use super::js_helpers::{drive_jobs, extract_tools, format_js_exception, restore_oauth_credential};
use super::js_helpers::{
drive_jobs, extract_tools, format_js_exception, restore_auth_credential,
restore_oauth_credential,
};
use super::types::{BridgeDeps, QjsSkillInstance, SkillState};
impl QjsSkillInstance {
@@ -52,7 +55,7 @@ impl QjsSkillInstance {
pub fn spawn(
&self,
mut rx: mpsc::Receiver<SkillMessage>,
_deps: BridgeDeps,
deps: BridgeDeps,
) -> tokio::task::JoinHandle<()> {
let config = self.config.clone();
let state = self.state.clone();
@@ -139,8 +142,8 @@ impl QjsSkillInstance {
let skill_context = qjs_ops::SkillContext {
skill_id: skill_id.clone(),
data_dir: data_dir.clone(),
memory_client: _deps.memory_client.clone(),
webhook_router: _deps.webhook_router.clone(),
memory_client: deps.memory_client.clone(),
webhook_router: deps.webhook_router.clone(),
};
if let Err(e) = qjs_ops::register_ops(
@@ -193,6 +196,7 @@ impl QjsSkillInstance {
}
restore_oauth_credential(&ctx, &config.skill_id, &data_dir).await;
restore_auth_credential(&ctx, &config.skill_id, &data_dir).await;
// Call init() lifecycle
if let Err(e) = call_lifecycle(&rt, &ctx, "init").await {
@@ -242,7 +246,7 @@ impl QjsSkillInstance {
&config.skill_id,
&timer_state,
&published_state,
_deps.memory_client.clone(),
deps.memory_client.clone(),
&data_dir,
)
.await;
@@ -120,3 +120,67 @@ pub(crate) async fn restore_oauth_credential(
);
}
}
/// Load a persisted auth credential from the skill's data directory and inject
/// it into the JS context so tools have access to the credential.
///
/// Reads `{data_dir}/auth_credential.json` which is written by the
/// `auth/complete` handler and deleted by `auth/revoked`.
///
/// For managed mode, also bridges to `globalThis.oauth.__setCredential()`.
pub(crate) async fn restore_auth_credential(
ctx: &rquickjs::AsyncContext,
skill_id: &str,
data_dir: &std::path::Path,
) {
let cred_path = data_dir.join("auth_credential.json");
let cred_json = match std::fs::read_to_string(&cred_path) {
Ok(s) if !s.is_empty() => s,
_ => return,
};
// Inject credential into auth bridge and in-memory state.
// For managed mode, also bridge to the oauth credential system.
let code = format!(
r#"(function() {{
var cred = {cred};
if (typeof globalThis.auth !== 'undefined' && globalThis.auth.__setCredential) {{
globalThis.auth.__setCredential(cred);
}}
if (typeof globalThis.state !== 'undefined' && globalThis.state.set) {{
globalThis.state.set('__auth_credential', cred);
}}
// For managed mode, bridge to oauth system for backward compat.
// For non-managed modes, clear any stale oauth credentials.
if (cred && cred.mode === 'managed' && cred.credentials) {{
if (typeof globalThis.oauth !== 'undefined' && globalThis.oauth.__setCredential) {{
globalThis.oauth.__setCredential(cred.credentials);
}}
if (typeof globalThis.state !== 'undefined' && globalThis.state.set) {{
globalThis.state.set('__oauth_credential', cred.credentials);
}}
}} else {{
if (typeof globalThis.oauth !== 'undefined' && globalThis.oauth.__setCredential) {{
globalThis.oauth.__setCredential(null);
}}
if (typeof globalThis.state !== 'undefined' && globalThis.state.set) {{
globalThis.state.set('__oauth_credential', null);
}}
}}
return true;
}})()"#,
cred = cred_json
);
let restored = ctx
.with(|js_ctx| js_ctx.eval::<bool, _>(code.as_bytes()).unwrap_or(false))
.await;
if restored {
log::info!(
"[skill:{}] Restored auth credential from {}",
skill_id,
cred_path.display()
);
}
}
@@ -10,7 +10,6 @@ use crate::openhuman::skills::skill_registry::SkillRegistry;
use crate::openhuman::skills::types::{SkillConfig, SkillMessage, SkillStatus, ToolDefinition};
/// Dependencies passed to a skill instance for bridge installation.
#[allow(dead_code)]
pub struct BridgeDeps {
pub cron_scheduler: Arc<CronScheduler>,
pub skill_registry: Arc<SkillRegistry>,
+123
View File
@@ -914,6 +914,129 @@ globalThis.data = {
};
})();
// ============================================================================
// Advanced Auth Bridge API (self-hosted / text credential management)
// ============================================================================
(function () {
globalThis.__authCredential = null;
globalThis.auth = {
/**
* Get the full auth credential object, or null if not set.
* Shape: { mode: "managed"|"self_hosted"|"text", credentials: {...} }
*/
getCredential: function () {
return globalThis.__authCredential;
},
/**
* Get the auth mode string, or null.
* @returns {"managed"|"self_hosted"|"text"|null}
*/
getMode: function () {
return globalThis.__authCredential ? globalThis.__authCredential.mode : null;
},
/**
* Get just the credentials map (e.g. { client_id, client_secret, url, content }).
* @returns {Object|null}
*/
getCredentials: function () {
return globalThis.__authCredential ? globalThis.__authCredential.credentials : null;
},
/**
* Make an HTTP request with credentials auto-injected.
* For self_hosted: auto-adds Authorization header from client_id/client_secret.
* For managed: delegates to oauth.fetch.
* For text: no auto-injection (skill should handle manually).
*/
fetch: async function (url, options) {
if (!globalThis.__authCredential) {
return {
status: 401,
headers: {},
body: JSON.stringify({ error: 'No auth credential. Complete auth setup first.' }),
};
}
var mode = globalThis.__authCredential.mode;
var creds = globalThis.__authCredential.credentials;
// Managed mode: delegate to oauth.fetch for proxy behavior
if (mode === 'managed' && typeof globalThis.oauth !== 'undefined') {
return globalThis.oauth.fetch(url, options);
}
var headers = {};
if (options && options.headers) {
for (var k in options.headers) {
headers[k] = options.headers[k];
}
}
if (!headers['Content-Type']) {
headers['Content-Type'] = 'application/json';
}
// Check for existing Authorization header (case-insensitive)
var hasAuth = false;
for (var hk in headers) {
if (hk.toLowerCase() === 'authorization') { hasAuth = true; break; }
}
// Self-hosted: auto-inject basic auth if client_id + client_secret present
if (mode === 'self_hosted' && creds.client_id && creds.client_secret && !hasAuth) {
headers['Authorization'] = 'Basic ' + btoa(creds.client_id + ':' + creds.client_secret);
hasAuth = true;
}
// Self-hosted with access_token or refresh_token: inject Bearer token
if (mode === 'self_hosted' && !hasAuth && creds.access_token) {
headers['Authorization'] = 'Bearer ' + creds.access_token;
hasAuth = true;
}
// Do NOT auto-inject session JWT for text mode or when auth is already set.
// Text mode credentials are opaque — the skill handles auth manually.
var fetchOpts = {
method: (options && options.method) || 'GET',
headers: headers,
body: options ? options.body : undefined,
timeout: options ? options.timeout : undefined,
};
console.log('[auth.fetch] ' + fetchOpts.method + ' ' + url + ' (mode=' + mode + ')');
var result = await net.fetch(url, fetchOpts);
console.log('[auth.fetch] response status=' + result.status);
// Auto-clear on 401/403 so user is prompted to re-auth
if (result.status === 401 || result.status === 403) {
console.warn('[auth.fetch] Got ' + result.status + ' — clearing invalid credential');
globalThis.__authCredential = null;
if (typeof globalThis.state !== 'undefined' && globalThis.state.set) {
globalThis.state.set('__auth_credential', '');
globalThis.state.setPartial({
connection_status: 'error',
connection_error: 'Auth credential expired or invalid. Please reconnect.',
auth_status: 'not_authenticated',
});
}
}
return result;
},
/**
* Internal: set credential (called by runtime on auth/complete).
* Skills should not call this directly.
*/
__setCredential: function (cred) {
globalThis.__authCredential = cred;
},
};
})();
// ============================================================================
// Tools array (skills assign to this global)
// ============================================================================
+80
View File
@@ -0,0 +1,80 @@
//! Registry cache management — disk-based caching for the remote skill registry.
use std::path::{Path, PathBuf};
use super::registry_types::{CachedRegistry, RemoteSkillRegistry, SkillCategory};
/// Cache TTL in seconds (1 hour).
pub(crate) const CACHE_TTL_SECS: i64 = 3600;
/// Default registry URL (GitHub raw on the `build` branch).
pub(crate) const DEFAULT_REGISTRY_URL: &str = "https://raw.githubusercontent.com/tinyhumansai/openhuman-skills/refs/heads/build/skills/registry.json";
pub(crate) fn registry_url() -> String {
std::env::var("SKILLS_REGISTRY_URL").unwrap_or_else(|_| DEFAULT_REGISTRY_URL.to_string())
}
/// If `SKILLS_LOCAL_DIR` is set, return the local skills directory path.
pub(crate) fn local_skills_dir() -> Option<PathBuf> {
std::env::var("SKILLS_LOCAL_DIR").ok().map(PathBuf::from)
}
/// Check if a URL is a local file path (absolute path or file:// URI).
pub(crate) fn is_local_path(url: &str) -> bool {
url.starts_with('/') || url.starts_with("file://")
}
/// Read a file from a local path or file:// URI.
pub(crate) fn read_local_file(url: &str) -> Result<Vec<u8>, String> {
let path = if let Some(stripped) = url.strip_prefix("file://") {
PathBuf::from(stripped)
} else {
PathBuf::from(url)
};
std::fs::read(&path).map_err(|e| format!("failed to read local file {}: {e}", path.display()))
}
pub(crate) fn cache_path(workspace_dir: &Path) -> PathBuf {
workspace_dir.join("skills").join(".registry-cache.json")
}
pub(crate) fn is_cache_fresh(cached: &CachedRegistry) -> bool {
let Ok(fetched) = chrono::DateTime::parse_from_rfc3339(&cached.fetched_at) else {
return false;
};
let now = chrono::Utc::now();
(now - fetched.to_utc()).num_seconds() < CACHE_TTL_SECS
}
pub(crate) fn read_cache(workspace_dir: &Path) -> Option<CachedRegistry> {
let path = cache_path(workspace_dir);
let content = std::fs::read_to_string(&path).ok()?;
serde_json::from_str(&content).ok()
}
pub(crate) fn write_cache(
workspace_dir: &Path,
registry: &RemoteSkillRegistry,
) -> Result<(), String> {
let cached = CachedRegistry {
fetched_at: chrono::Utc::now().to_rfc3339(),
registry: registry.clone(),
};
let path = cache_path(workspace_dir);
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).map_err(|e| format!("failed to create cache dir: {e}"))?;
}
let json = serde_json::to_string_pretty(&cached).map_err(|e| e.to_string())?;
std::fs::write(&path, json).map_err(|e| format!("failed to write cache: {e}"))?;
Ok(())
}
/// Tag each entry with its category based on which list it came from.
pub(crate) fn tag_categories(registry: &mut RemoteSkillRegistry) {
for entry in &mut registry.skills.core {
entry.category = SkillCategory::Core;
}
for entry in &mut registry.skills.third_party {
entry.category = SkillCategory::ThirdParty;
}
}
+7 -77
View File
@@ -2,84 +2,13 @@ use std::path::{Path, PathBuf};
use sha2::{Digest, Sha256};
use super::registry_cache::{
cache_path, is_cache_fresh, is_local_path, local_skills_dir, read_cache, read_local_file,
registry_url, tag_categories, write_cache,
};
use super::registry_types::{
AvailableSkillEntry, CachedRegistry, InstalledSkillInfo, RegistrySkillEntry,
RemoteSkillRegistry, SkillCategory,
AvailableSkillEntry, InstalledSkillInfo, RegistrySkillEntry, RemoteSkillRegistry,
};
/// Cache TTL in seconds (1 hour).
const CACHE_TTL_SECS: i64 = 3600;
/// Default registry URL (GitHub raw on the `build` branch).
const DEFAULT_REGISTRY_URL: &str = "https://raw.githubusercontent.com/tinyhumansai/openhuman-skills/refs/heads/build/skills/registry.json";
fn registry_url() -> String {
std::env::var("SKILLS_REGISTRY_URL").unwrap_or_else(|_| DEFAULT_REGISTRY_URL.to_string())
}
/// If `SKILLS_LOCAL_DIR` is set, return the local skills directory path.
/// This enables local development by reading skills directly from disk
/// instead of downloading from the remote registry.
fn local_skills_dir() -> Option<PathBuf> {
std::env::var("SKILLS_LOCAL_DIR").ok().map(PathBuf::from)
}
/// Check if a URL is a local file path (absolute path or file:// URI).
fn is_local_path(url: &str) -> bool {
url.starts_with('/') || url.starts_with("file://")
}
/// Read a file from a local path or file:// URI.
fn read_local_file(url: &str) -> Result<Vec<u8>, String> {
let path = if let Some(stripped) = url.strip_prefix("file://") {
PathBuf::from(stripped)
} else {
PathBuf::from(url)
};
std::fs::read(&path).map_err(|e| format!("failed to read local file {}: {e}", path.display()))
}
fn cache_path(workspace_dir: &Path) -> std::path::PathBuf {
workspace_dir.join("skills").join(".registry-cache.json")
}
fn is_cache_fresh(cached: &CachedRegistry) -> bool {
let Ok(fetched) = chrono::DateTime::parse_from_rfc3339(&cached.fetched_at) else {
return false;
};
let now = chrono::Utc::now();
(now - fetched.to_utc()).num_seconds() < CACHE_TTL_SECS
}
fn read_cache(workspace_dir: &Path) -> Option<CachedRegistry> {
let path = cache_path(workspace_dir);
let content = std::fs::read_to_string(&path).ok()?;
serde_json::from_str(&content).ok()
}
fn write_cache(workspace_dir: &Path, registry: &RemoteSkillRegistry) -> Result<(), String> {
let cached = CachedRegistry {
fetched_at: chrono::Utc::now().to_rfc3339(),
registry: registry.clone(),
};
let path = cache_path(workspace_dir);
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).map_err(|e| format!("failed to create cache dir: {e}"))?;
}
let json = serde_json::to_string_pretty(&cached).map_err(|e| e.to_string())?;
std::fs::write(&path, json).map_err(|e| format!("failed to write cache: {e}"))?;
Ok(())
}
/// Tag each entry with its category based on which list it came from.
fn tag_categories(registry: &mut RemoteSkillRegistry) {
for entry in &mut registry.skills.core {
entry.category = SkillCategory::Core;
}
for entry in &mut registry.skills.third_party {
entry.category = SkillCategory::ThirdParty;
}
}
/// Fetch the skill registry. Supports both remote HTTP URLs and local file paths.
///
@@ -457,7 +386,8 @@ pub async fn skills_list_available(
#[cfg(test)]
mod tests {
use super::super::registry_types::RegistrySkillCategories;
use super::super::registry_cache::{cache_path, is_cache_fresh, write_cache};
use super::super::registry_types::{CachedRegistry, RegistrySkillCategories, SkillCategory};
use super::*;
fn make_workspace() -> tempfile::TempDir {
+171 -6
View File
@@ -40,7 +40,6 @@ pub enum SkillMessage {
data: serde_json::Value,
},
/// Trigger a cron job by name.
#[allow(dead_code)]
CronTrigger { schedule_id: String },
/// Request the skill to stop.
Stop {
@@ -85,7 +84,6 @@ pub enum SkillMessage {
reply: tokio::sync::oneshot::Sender<Result<(), String>>,
},
/// Notify the skill of an error from an async operation.
#[allow(dead_code)]
Error {
error_type: String,
message: String,
@@ -251,10 +249,177 @@ fn default_memory_limit() -> usize {
}
/// Events emitted from the runtime to the frontend via Tauri.
#[allow(dead_code)]
pub mod events {
pub const SKILL_STATUS_CHANGED: &str = "runtime:skill-status-changed";
/// Skill published state has changed.
pub const SKILL_STATE_CHANGED: &str = "runtime:skill-state-changed";
pub const SKILL_TOOLS_CHANGED: &str = "runtime:skill-tools-changed";
pub const SKILL_LOG: &str = "runtime:skill-log";
}
#[cfg(test)]
mod tests {
use super::*;
fn state_with(entries: &[(&str, &str)]) -> HashMap<String, serde_json::Value> {
entries
.iter()
.map(|(k, v)| (k.to_string(), serde_json::Value::String(v.to_string())))
.collect()
}
// --- Status-based early returns ---
#[test]
fn pending_returns_offline() {
assert_eq!(
derive_connection_status(SkillStatus::Pending, true, &HashMap::new()),
"offline"
);
}
#[test]
fn stopped_returns_offline() {
assert_eq!(
derive_connection_status(SkillStatus::Stopped, true, &HashMap::new()),
"offline"
);
}
#[test]
fn stopping_returns_offline() {
assert_eq!(
derive_connection_status(SkillStatus::Stopping, true, &HashMap::new()),
"offline"
);
}
#[test]
fn error_status_returns_error() {
assert_eq!(
derive_connection_status(SkillStatus::Error, true, &HashMap::new()),
"error"
);
}
#[test]
fn initializing_returns_connecting() {
assert_eq!(
derive_connection_status(SkillStatus::Initializing, false, &HashMap::new()),
"connecting"
);
}
// --- No published state ---
#[test]
fn running_setup_complete_no_state_returns_connected() {
assert_eq!(
derive_connection_status(SkillStatus::Running, true, &HashMap::new()),
"connected"
);
}
#[test]
fn running_no_setup_no_state_returns_setup_required() {
assert_eq!(
derive_connection_status(SkillStatus::Running, false, &HashMap::new()),
"setup_required"
);
}
// --- Published connection_status ---
#[test]
fn conn_error_returns_error() {
let st = state_with(&[("connection_status", "error")]);
assert_eq!(
derive_connection_status(SkillStatus::Running, true, &st),
"error"
);
}
#[test]
fn auth_error_returns_error() {
let st = state_with(&[("auth_status", "error")]);
assert_eq!(
derive_connection_status(SkillStatus::Running, true, &st),
"error"
);
}
#[test]
fn conn_connecting_returns_connecting() {
let st = state_with(&[("connection_status", "connecting")]);
assert_eq!(
derive_connection_status(SkillStatus::Running, true, &st),
"connecting"
);
}
#[test]
fn auth_authenticating_returns_connecting() {
let st = state_with(&[("auth_status", "authenticating")]);
assert_eq!(
derive_connection_status(SkillStatus::Running, true, &st),
"connecting"
);
}
#[test]
fn conn_connected_no_auth_returns_connected() {
let st = state_with(&[("connection_status", "connected")]);
assert_eq!(
derive_connection_status(SkillStatus::Running, true, &st),
"connected"
);
}
#[test]
fn conn_connected_auth_authenticated_returns_connected() {
let st = state_with(&[
("connection_status", "connected"),
("auth_status", "authenticated"),
]);
assert_eq!(
derive_connection_status(SkillStatus::Running, true, &st),
"connected"
);
}
#[test]
fn conn_connected_auth_not_authenticated() {
let st = state_with(&[
("connection_status", "connected"),
("auth_status", "not_authenticated"),
]);
assert_eq!(
derive_connection_status(SkillStatus::Running, true, &st),
"not_authenticated"
);
}
#[test]
fn conn_disconnected_setup_complete_returns_disconnected() {
let st = state_with(&[("connection_status", "disconnected")]);
assert_eq!(
derive_connection_status(SkillStatus::Running, true, &st),
"disconnected"
);
}
#[test]
fn conn_disconnected_no_setup_returns_setup_required() {
let st = state_with(&[("connection_status", "disconnected")]);
assert_eq!(
derive_connection_status(SkillStatus::Running, false, &st),
"setup_required"
);
}
#[test]
fn unknown_state_falls_through_to_connecting() {
let st = state_with(&[("connection_status", "unknown_value")]);
assert_eq!(
derive_connection_status(SkillStatus::Running, true, &st),
"connecting"
);
}
}