mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
Merge remote-tracking branch 'upstream/develop' into develop
This commit is contained in:
@@ -373,6 +373,7 @@ Key updates from recent commits (cd9ebcd to current):
|
||||
- **Unit Tests**: All unit tests live in `__tests__/` folders co-located with the code they test. Use Jest with TypeScript support.
|
||||
- **Runtime Platform Differences**: V8 runtime is desktop-only. Mobile platforms use feature detection and graceful degradation. Skills with platform restrictions are filtered during discovery.
|
||||
- **Socket Management**: Rust-native Socket.io client provides persistent connections with automatic reconnection. Use `connect_to_socket` command instead of frontend-only socket connections for reliability.
|
||||
- **Dual Socket Codebase**: Socket event handling exists in **both** the TypeScript frontend (`src/services/socketService.ts`, `src/utils/tauriSocket.ts`) and the Rust backend (`src-tauri/src/runtime/socket_manager.rs`). **Any new socket event or protocol change must be implemented in both codebases.** The web frontend handles events directly via Socket.io; the Rust backend handles them over raw WebSocket with Engine.IO/Socket.IO framing. Example: `tool:sync` is emitted from both `src/lib/skills/sync.ts` (web mode) and `socket_manager.rs` (Rust mode, on connect + skill lifecycle changes).
|
||||
|
||||
## Platform Gotchas
|
||||
|
||||
|
||||
@@ -307,6 +307,8 @@ pub fn run() {
|
||||
|
||||
// Wire the SkillRegistry into the SocketManager for MCP
|
||||
socket_mgr.set_registry(engine.registry());
|
||||
// Wire the SocketManager into the engine for tool:sync
|
||||
engine.set_socket_manager(socket_mgr.clone());
|
||||
|
||||
app.manage(engine.clone());
|
||||
|
||||
|
||||
@@ -13,6 +13,7 @@ use crate::runtime::cron_scheduler::CronScheduler;
|
||||
use crate::runtime::manifest::SkillManifest;
|
||||
use crate::runtime::preferences::PreferencesStore;
|
||||
use crate::runtime::skill_registry::SkillRegistry;
|
||||
use crate::runtime::socket_manager::SocketManager;
|
||||
use crate::runtime::types::{events, SkillSnapshot, SkillStatus, ToolResult};
|
||||
use crate::runtime::qjs_skill_instance::{BridgeDeps, QjsSkillInstance};
|
||||
use crate::services::quickjs_libs::storage::IdbStorage;
|
||||
@@ -33,6 +34,8 @@ pub struct RuntimeEngine {
|
||||
resource_dir: RwLock<Option<PathBuf>>,
|
||||
/// Tauri app handle for emitting events.
|
||||
app_handle: RwLock<Option<AppHandle>>,
|
||||
/// Socket manager for emitting tool:sync events.
|
||||
socket_manager: RwLock<Option<Arc<SocketManager>>>,
|
||||
}
|
||||
|
||||
impl RuntimeEngine {
|
||||
@@ -53,6 +56,7 @@ impl RuntimeEngine {
|
||||
skills_source_dir: RwLock::new(None),
|
||||
resource_dir: RwLock::new(None),
|
||||
app_handle: RwLock::new(None),
|
||||
socket_manager: RwLock::new(None),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -83,6 +87,19 @@ impl RuntimeEngine {
|
||||
*self.resource_dir.write() = Some(dir);
|
||||
}
|
||||
|
||||
/// Set the socket manager for emitting `tool:sync` events.
|
||||
pub fn set_socket_manager(&self, mgr: Arc<SocketManager>) {
|
||||
*self.socket_manager.write() = Some(mgr);
|
||||
}
|
||||
|
||||
/// Notify the backend of the current tool state via `tool:sync`.
|
||||
async fn sync_tools(&self) {
|
||||
let mgr = { self.socket_manager.read().clone() };
|
||||
if let Some(mgr) = mgr {
|
||||
mgr.sync_tools().await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Get the skills source directory.
|
||||
fn get_skills_source_dir(&self) -> Result<PathBuf, String> {
|
||||
// 1. Explicitly set source dir (highest priority)
|
||||
@@ -274,6 +291,7 @@ impl RuntimeEngine {
|
||||
match current_status {
|
||||
SkillStatus::Running => {
|
||||
self.emit_status_change(&skill_id_owned);
|
||||
self.sync_tools().await;
|
||||
return Ok(instance.snapshot());
|
||||
}
|
||||
SkillStatus::Error => {
|
||||
@@ -321,6 +339,7 @@ impl RuntimeEngine {
|
||||
self.registry.stop_skill(skill_id).await?;
|
||||
self.cron_scheduler.unregister_all_for_skill(skill_id);
|
||||
self.emit_status_change(skill_id);
|
||||
self.sync_tools().await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -391,6 +410,8 @@ impl RuntimeEngine {
|
||||
log::warn!("[runtime] Failed to discover skills for auto-start: {e}");
|
||||
}
|
||||
}
|
||||
// Sync all tool state to backend after auto-start completes
|
||||
self.sync_tools().await;
|
||||
}
|
||||
|
||||
/// Enable a skill.
|
||||
|
||||
@@ -32,6 +32,8 @@ use {
|
||||
// SkillRegistry only available on desktop
|
||||
#[cfg(not(any(target_os = "android", target_os = "ios")))]
|
||||
use crate::runtime::skill_registry::SkillRegistry;
|
||||
#[cfg(not(any(target_os = "android", target_os = "ios")))]
|
||||
use crate::runtime::types::{SkillSnapshot, SkillStatus};
|
||||
|
||||
/// Events emitted to the frontend via Tauri.
|
||||
#[allow(dead_code)]
|
||||
@@ -224,6 +226,27 @@ impl SocketManager {
|
||||
Err("Rust Socket.io not available on Android".to_string())
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Tool sync — notify backend of current skill/tool state
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
/// Emit `tool:sync` with the current skill/tool state.
|
||||
/// Called on socket reconnect and after skill lifecycle changes.
|
||||
#[cfg(not(any(target_os = "android", target_os = "ios")))]
|
||||
pub async fn sync_tools(&self) {
|
||||
let payload = build_tool_sync_payload(&self.shared);
|
||||
if let Some(payload) = payload {
|
||||
if let Err(e) = self.emit("tool:sync", payload).await {
|
||||
log::debug!("[socket-mgr] tool:sync emit failed: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(any(target_os = "android", target_os = "ios"))]
|
||||
pub async fn sync_tools(&self) {
|
||||
// No-op on mobile — skill runtime not available
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Tauri event helpers
|
||||
// -----------------------------------------------------------------------
|
||||
@@ -663,6 +686,10 @@ fn handle_sio_event(
|
||||
log::info!("[socket-mgr] Server ready — auth successful");
|
||||
*shared.status.write() = ConnectionStatus::Connected;
|
||||
SocketManager::emit_state_change(shared);
|
||||
|
||||
// Sync current tool state to backend on connect/reconnect
|
||||
#[cfg(not(any(target_os = "android", target_os = "ios")))]
|
||||
sync_tools_via_channel(emit_tx, shared);
|
||||
}
|
||||
"error" => {
|
||||
log::error!("[socket-mgr] Server error event: {}", data);
|
||||
@@ -869,6 +896,70 @@ fn emit_via_channel(
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Tool sync helpers (desktop only — requires SkillRegistry)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Derive a unified connection status string from a Rust-side SkillSnapshot.
|
||||
/// Mirrors the frontend's `deriveConnectionStatus()` logic in `src/lib/skills/hooks.ts`.
|
||||
#[cfg(not(any(target_os = "android", target_os = "ios")))]
|
||||
fn derive_connection_status(snap: &SkillSnapshot) -> &'static str {
|
||||
match snap.status {
|
||||
SkillStatus::Error => "error",
|
||||
SkillStatus::Pending | SkillStatus::Stopped => "offline",
|
||||
SkillStatus::Initializing => "connecting",
|
||||
SkillStatus::Stopping => "disconnected",
|
||||
SkillStatus::Running => {
|
||||
// Check the skill's self-reported connection/auth state
|
||||
let conn = snap.state.get("connection_status").and_then(|v| v.as_str());
|
||||
let auth = snap.state.get("auth_status").and_then(|v| v.as_str());
|
||||
|
||||
match (conn, auth) {
|
||||
(Some("error"), _) | (_, Some("error")) => "error",
|
||||
(Some("connected"), Some("authenticated")) => "connected",
|
||||
(Some("connecting"), _) | (_, Some("authenticating")) => "connecting",
|
||||
(Some("connected"), Some("not_authenticated")) => "not_authenticated",
|
||||
(Some("disconnected"), _) => "disconnected",
|
||||
// Running with no explicit connection state = connected
|
||||
_ => "connected",
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Build the `tool:sync` payload from the current registry state.
|
||||
#[cfg(not(any(target_os = "android", target_os = "ios")))]
|
||||
fn build_tool_sync_payload(shared: &SharedState) -> Option<serde_json::Value> {
|
||||
let registry = shared.registry.read().clone()?;
|
||||
let skills = registry.list_skills();
|
||||
let tools: Vec<serde_json::Value> = skills
|
||||
.iter()
|
||||
.map(|snap| {
|
||||
let status = derive_connection_status(snap);
|
||||
let tool_names: Vec<String> = snap.tools.iter().map(|t| t.name.clone()).collect();
|
||||
json!({
|
||||
"skillId": snap.skill_id,
|
||||
"name": snap.name,
|
||||
"status": status,
|
||||
"tools": tool_names,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
Some(json!({ "tools": tools }))
|
||||
}
|
||||
|
||||
/// Emit `tool:sync` synchronously via the emit channel (for use from event handlers).
|
||||
#[cfg(not(any(target_os = "android", target_os = "ios")))]
|
||||
fn sync_tools_via_channel(emit_tx: &mpsc::UnboundedSender<String>, shared: &SharedState) {
|
||||
if let Some(payload) = build_tool_sync_payload(shared) {
|
||||
emit_via_channel(emit_tx, "tool:sync", payload);
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// SIO event parsing
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Parse a Socket.IO EVENT payload: `["eventName", data]` or `<ackId>["eventName", data]`.
|
||||
#[cfg(not(target_os = "android"))]
|
||||
fn parse_sio_event(text: &str) -> Option<(String, serde_json::Value)> {
|
||||
|
||||
@@ -13,7 +13,7 @@ import type {
|
||||
* Derive a unified connection status from the skill's lifecycle status
|
||||
* and its self-reported connection/auth state.
|
||||
*/
|
||||
function deriveConnectionStatus(
|
||||
export function deriveConnectionStatus(
|
||||
lifecycleStatus: string | undefined,
|
||||
setupComplete: boolean | undefined,
|
||||
skillState: Record<string, unknown> | undefined,
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
export { SkillTransport } from "./transport";
|
||||
export { SkillRuntime } from "./runtime";
|
||||
export { skillManager } from "./manager";
|
||||
export { syncToolsToBackend } from "./sync";
|
||||
export * from "./types";
|
||||
export * from "./paths";
|
||||
export * from "./hooks";
|
||||
|
||||
@@ -8,6 +8,7 @@
|
||||
import { invoke } from "@tauri-apps/api/core";
|
||||
|
||||
import { SkillRuntime } from "./runtime";
|
||||
import { syncToolsToBackend } from "./sync";
|
||||
import type {
|
||||
SkillManifest,
|
||||
SkillStatus,
|
||||
@@ -130,6 +131,7 @@ class SkillManager {
|
||||
const tools = await runtime.listTools();
|
||||
store.dispatch(setSkillTools({ skillId, tools }));
|
||||
store.dispatch(setSkillStatus({ skillId, status: "ready" }));
|
||||
syncToolsToBackend();
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
store.dispatch(setSkillError({ skillId, error: msg }));
|
||||
@@ -314,6 +316,7 @@ class SkillManager {
|
||||
store.dispatch(setSkillSetupComplete({ skillId, complete: false }));
|
||||
store.dispatch(setSkillOAuthCredential({ skillId, credential: undefined }));
|
||||
store.dispatch(setSkillState({ skillId, state: {} }));
|
||||
syncToolsToBackend();
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -331,6 +334,7 @@ class SkillManager {
|
||||
}
|
||||
this.runtimes.delete(skillId);
|
||||
store.dispatch(setSkillStatus({ skillId, status: "installed" }));
|
||||
syncToolsToBackend();
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -407,6 +411,7 @@ class SkillManager {
|
||||
type: "skills/setSkillState",
|
||||
payload: { skillId, state: newState },
|
||||
});
|
||||
syncToolsToBackend();
|
||||
return { ok: true };
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
/**
|
||||
* Sync tool/skill state to the backend via `tool:sync` socket event.
|
||||
*
|
||||
* Called whenever skill connection state changes or the socket reconnects,
|
||||
* so the backend always has an up-to-date picture of connected tools.
|
||||
*/
|
||||
|
||||
import { isTauri as coreIsTauri } from '@tauri-apps/api/core';
|
||||
|
||||
import { socketService } from '../../services/socketService';
|
||||
import { store } from '../../store';
|
||||
import { emitViaRustSocket } from '../../utils/tauriSocket';
|
||||
import { deriveConnectionStatus } from './hooks';
|
||||
import type { SkillConnectionStatus } from './types';
|
||||
|
||||
interface ToolSyncEntry {
|
||||
skillId: string;
|
||||
name: string;
|
||||
status: SkillConnectionStatus;
|
||||
tools: string[];
|
||||
}
|
||||
|
||||
function isTauri(): boolean {
|
||||
try {
|
||||
return coreIsTauri();
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Read all skills from Redux, derive their connection status,
|
||||
* and emit a `tool:sync` event with the full list.
|
||||
*/
|
||||
export function syncToolsToBackend(): void {
|
||||
const state = store.getState();
|
||||
const skills = state.skills.skills;
|
||||
const skillStates = state.skills.skillStates;
|
||||
|
||||
const tools: ToolSyncEntry[] = [];
|
||||
|
||||
for (const [skillId, skill] of Object.entries(skills)) {
|
||||
const connectionStatus = deriveConnectionStatus(
|
||||
skill.status,
|
||||
skill.setupComplete,
|
||||
skillStates[skillId],
|
||||
);
|
||||
|
||||
tools.push({
|
||||
skillId,
|
||||
name: skill.manifest.name,
|
||||
status: connectionStatus,
|
||||
tools: (skill.tools ?? []).map(t => t.name),
|
||||
});
|
||||
}
|
||||
|
||||
const payload = { tools };
|
||||
|
||||
if (isTauri()) {
|
||||
emitViaRustSocket('tool:sync', payload);
|
||||
} else {
|
||||
socketService.emit('tool:sync', payload);
|
||||
}
|
||||
}
|
||||
@@ -3,7 +3,7 @@ import debug from 'debug';
|
||||
import { io, Socket } from 'socket.io-client';
|
||||
|
||||
import { MCPTool, MCPToolCall, SocketIOMCPTransportImpl } from '../lib/mcp';
|
||||
import { skillManager } from '../lib/skills';
|
||||
import { skillManager, syncToolsToBackend } from '../lib/skills';
|
||||
import { store } from '../store';
|
||||
import { resetForUser, setSocketIdForUser, setStatusForUser } from '../store/socketSlice';
|
||||
import { BACKEND_URL } from '../utils/config';
|
||||
@@ -134,6 +134,7 @@ class SocketService {
|
||||
socketLog('Connected', { socketId, userId: uid });
|
||||
store.dispatch(setStatusForUser({ userId: uid, status: 'connected' }));
|
||||
store.dispatch(setSocketIdForUser({ userId: uid, socketId }));
|
||||
syncToolsToBackend();
|
||||
});
|
||||
|
||||
this.socket.on('ready', () => {
|
||||
|
||||
@@ -16,6 +16,7 @@
|
||||
import { isTauri as coreIsTauri, invoke } from '@tauri-apps/api/core';
|
||||
import { listen, UnlistenFn } from '@tauri-apps/api/event';
|
||||
|
||||
import { syncToolsToBackend } from '../lib/skills/sync';
|
||||
import { store } from '../store';
|
||||
import { setSocketIdForUser, setStatusForUser } from '../store/socketSlice';
|
||||
import { BACKEND_URL } from './config';
|
||||
@@ -149,6 +150,10 @@ export async function setupTauriSocketListeners(): Promise<void> {
|
||||
const mappedStatus: ReduxStatus = statusMap[status] || 'disconnected';
|
||||
store.dispatch(setStatusForUser({ userId: uid, status: mappedStatus }));
|
||||
store.dispatch(setSocketIdForUser({ userId: uid, socketId: socket_id ?? null }));
|
||||
|
||||
if (mappedStatus === 'connected') {
|
||||
syncToolsToBackend();
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user