diff --git a/.env.example b/.env.example index 9d4c2cd7d..7dd775bf1 100644 --- a/.env.example +++ b/.env.example @@ -119,8 +119,19 @@ OLLAMA_BIN= # --------------------------------------------------------------------------- # Skills # --------------------------------------------------------------------------- -# [optional] Override skills registry URL +# [optional] Override skills registry URL. +# Supports remote HTTP URLs and local file paths for development: +# SKILLS_REGISTRY_URL=https://example.com/registry.json (remote) +# SKILLS_REGISTRY_URL=/path/to/openhuman-skills/skills/registry.json (local) +# When set to a local path, the registry is read directly from disk on every +# call (no caching), so changes are picked up immediately. SKILLS_REGISTRY_URL= +# [optional] Local skills source directory for development. +# Points to the built skills directory (the folder containing per-skill subdirs +# with manifest.json + index.js). When set, this takes highest priority for +# skill discovery and install will copy from this directory instead of downloading. +# Example: SKILLS_LOCAL_DIR=/Users/you/work/openhuman-skills/skills +SKILLS_LOCAL_DIR= # --------------------------------------------------------------------------- # Error Reporting (Sentry) diff --git a/scripts/debug-notion-sync-memory.sh b/scripts/debug-notion-sync-memory.sh new file mode 100755 index 000000000..8b8c152d5 --- /dev/null +++ b/scripts/debug-notion-sync-memory.sh @@ -0,0 +1,73 @@ +#!/usr/bin/env bash +# +# debug-notion-sync-memory.sh — Run the Notion live test with memory verification. +# +# Tests the full flow: skill start → sync → memory persistence → verify documents +# +# Prerequisites: +# - .env with BACKEND_URL, JWT_TOKEN, CREDENTIAL_ID, SKILLS_DATA_DIR +# - OAuth credential at $SKILLS_DATA_DIR/notion/oauth_credential.json +# - openhuman-skills repo available (auto-detected or via SKILL_DEBUG_DIR) +# +# Usage: +# bash scripts/debug-notion-sync-memory.sh +# +# Environment variables (set in .env or export before running): +# BACKEND_URL — backend API URL (e.g. https://staging-api.alphahuman.xyz) +# JWT_TOKEN — session JWT for OAuth proxy +# CREDENTIAL_ID — OAuth credential ID for the proxy +# SKILLS_DATA_DIR — path to skills data dir (contains notion/ subdir) +# RUST_LOG — Rust log filter (default: info) +# +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)" + +# Load .env +if [ -f "$REPO_ROOT/.env" ]; then + source "$SCRIPT_DIR/load-dotenv.sh" "$REPO_ROOT/.env" +fi + +export RUN_LIVE_NOTION=1 +export RUST_LOG="${RUST_LOG:-info}" + +echo "========================================" +echo " Notion Sync + Memory Verification" +echo "========================================" +echo " BACKEND_URL: ${BACKEND_URL:-}" +echo " JWT_TOKEN: ${JWT_TOKEN:+}" +echo " CREDENTIAL_ID: ${CREDENTIAL_ID:-}" +echo " SKILLS_DATA_DIR: ${SKILLS_DATA_DIR:-}" +echo " RUST_LOG: ${RUST_LOG}" +echo "" + +# Verify required vars +for var in BACKEND_URL JWT_TOKEN CREDENTIAL_ID SKILLS_DATA_DIR; do + if [ -z "${!var:-}" ]; then + echo "ERROR: $var is not set. Add it to .env or export it." + exit 1 + fi +done + +# Verify OAuth credential exists +CRED_FILE="$SKILLS_DATA_DIR/notion/oauth_credential.json" +if [ -f "$CRED_FILE" ]; then + echo " OAuth credential: present ($(wc -c < "$CRED_FILE" | tr -d ' ') bytes)" +else + echo " WARNING: $CRED_FILE not found" + echo " The skill will start without OAuth — API calls will fail" +fi +echo "" + +cd "$REPO_ROOT" + +echo "--- Running Notion live test with memory verification ---" +echo "" + +cargo test --test skills_notion_live -- --nocapture notion_live_with_real_data 2>&1 + +echo "" +echo "========================================" +echo " DONE" +echo "========================================" diff --git a/src/openhuman/skills/qjs_engine.rs b/src/openhuman/skills/qjs_engine.rs index 3bd69afd0..f34e127db 100644 --- a/src/openhuman/skills/qjs_engine.rs +++ b/src/openhuman/skills/qjs_engine.rs @@ -171,7 +171,20 @@ impl RuntimeEngine { /// Get the skills source directory. fn get_skills_source_dir(&self) -> Result { - // 1. Explicitly set source dir (highest priority) + // 0. SKILLS_LOCAL_DIR env var (highest priority — explicit local dev override) + if let Ok(local_dir) = std::env::var("SKILLS_LOCAL_DIR") { + let local_path = PathBuf::from(&local_dir); + if local_path.exists() { + log::info!("[runtime] Using SKILLS_LOCAL_DIR: {:?}", local_path); + return Ok(local_path); + } + log::warn!( + "[runtime] SKILLS_LOCAL_DIR set to {:?} but directory does not exist", + local_path + ); + } + + // 1. Explicitly set source dir (programmatic override) if let Some(dir) = self.skills_source_dir.read().as_ref() { log::info!("[runtime] Using explicit skills source dir: {:?}", dir); return Ok(dir.clone()); diff --git a/src/openhuman/skills/qjs_skill_instance/event_loop.rs b/src/openhuman/skills/qjs_skill_instance/event_loop.rs index cb0fe9941..bf0e95eb1 100644 --- a/src/openhuman/skills/qjs_skill_instance/event_loop.rs +++ b/src/openhuman/skills/qjs_skill_instance/event_loop.rs @@ -16,6 +16,101 @@ use super::js_handlers::{ use super::js_helpers::{drive_jobs, restore_oauth_credential}; use super::types::SkillState; +/// Payload queued for the background memory-write worker. +struct MemoryWriteJob { + client: MemoryClientRef, + skill: String, + title: String, + content: String, +} + +/// Maximum number of memory-write jobs that can be buffered before back-pressure +/// causes `persist_state_to_memory` to drop new writes. +const MEMORY_WRITE_CHANNEL_CAPACITY: usize = 16; + +/// Spawn a bounded background worker that consumes `MemoryWriteJob` items and +/// calls `store_skill_sync` sequentially. Returns the sender half; dropping it +/// shuts down the worker. +fn spawn_memory_write_worker() -> mpsc::Sender { + let (tx, mut rx) = mpsc::channel::(MEMORY_WRITE_CHANNEL_CAPACITY); + tokio::spawn(async move { + while let Some(job) = rx.recv().await { + log::debug!("[memory] store_skill_sync: title={}", job.title); + if let Err(e) = job + .client + .store_skill_sync( + &job.skill, + "default", + &job.title, + &job.content, + None, + None, + None, + None, + None, + None, + ) + .await + { + log::warn!("[memory] store_skill_sync failed for '{}': {e}", job.title); + } else { + log::info!("[memory] store_skill_sync succeeded for '{}'", job.title); + } + } + log::debug!("[memory] memory-write worker shutting down"); + }); + tx +} + +/// Snapshot the skill's published ops state and queue it for memory persistence. +/// +/// 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( + skill_id: &str, + title_suffix: &str, + ops_state: &Arc>, + memory_client: &Option, + memory_write_tx: &mpsc::Sender, +) { + let state_snapshot = ops_state.read().data.clone(); + log::debug!( + "[skill:{}] persist_state_to_memory({}): {} keys in snapshot", + skill_id, + title_suffix, + state_snapshot.len(), + ); + if state_snapshot.is_empty() { + return; + } + let Some(client) = memory_client.clone() else { + log::debug!( + "[skill:{}] persist_state_to_memory: no memory client available, skipping", + skill_id, + ); + return; + }; + let skill = skill_id.to_string(); + let content = serde_json::to_string_pretty(&serde_json::Value::Object(state_snapshot)) + .unwrap_or_else(|_| "{}".to_string()); + let title = format!("{} {}", skill, title_suffix); + if let Err(e) = memory_write_tx.try_send(MemoryWriteJob { + client, + skill, + title: title.clone(), + content, + }) { + log::warn!( + "[memory] persist_state_to_memory: channel full, dropping write for '{title}': {e}" + ); + } +} + /// Pending async tool call that is being driven by the event loop. struct PendingToolCall { reply: tokio::sync::oneshot::Sender>, @@ -48,6 +143,10 @@ pub(crate) async fn run_event_loop( // 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 = None; @@ -78,6 +177,7 @@ pub(crate) async fn run_event_loop( &memory_client, ops_state, data_dir, + &memory_write_tx, ) .await; if should_stop { @@ -215,6 +315,7 @@ async fn handle_message( memory_client: &Option, ops_state: &Arc>, data_dir: &std::path::Path, + memory_write_tx: &mpsc::Sender, ) -> bool { match msg { SkillMessage::CallTool { @@ -274,7 +375,25 @@ async fn handle_message( let _ = handle_server_event(rt, ctx, &event, data).await; } SkillMessage::CronTrigger { schedule_id } => { - let _ = handle_cron_trigger(rt, ctx, &schedule_id).await; + 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), + ops_state, + memory_client, + memory_write_tx, + ); + } + Err(e) => { + log::warn!( + "[skill:{}] cron trigger '{}' failed, skipping memory persistence: {e}", + skill_id, + schedule_id, + ); + } + } } SkillMessage::Stop { reply } => { let _ = call_lifecycle(rt, ctx, "stop").await; @@ -336,6 +455,16 @@ 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", + ops_state, + memory_client, + memory_write_tx, + ); + } let _ = reply.send(result); } SkillMessage::LoadParams { params } => { @@ -495,36 +624,18 @@ async fn handle_message( "skill/ping" => handle_js_call(rt, ctx, "onPing", "{}").await, "skill/sync" => { let result = handle_js_call(rt, ctx, "onSync", "{}").await; - // Fire-and-forget: persist published ops state to TinyHumans memory. - // Skills publish data via state.set()/setPartial() into ops_state.data, - // not as the return value of onSync() (which is typically undefined). - let state_snapshot = ops_state.read().data.clone(); - log::info!( - "[memory] store_skill_sync: payload → state_snapshot={:?}", - state_snapshot - ); - if !state_snapshot.is_empty() { - if let Some(client) = memory_client_opt.clone() { - let skill = skill_id.to_string(); - let content = serde_json::to_string_pretty(&serde_json::Value::Object( - state_snapshot, - )) - .unwrap_or_else(|_| "{}".to_string()); - let title = format!("{} periodic sync", skill); - tokio::spawn(async move { - if let Err(e) = client - .store_skill_sync( - &skill, "default", &title, &content, None, None, None, - None, None, None, - ) - .await - { - log::warn!("[memory] store_skill_sync failed: {e}"); - } - }); - } + 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 } "oauth/revoked" => { diff --git a/src/openhuman/skills/registry_ops.rs b/src/openhuman/skills/registry_ops.rs index dc0fdd0e0..2018912fe 100644 --- a/src/openhuman/skills/registry_ops.rs +++ b/src/openhuman/skills/registry_ops.rs @@ -1,4 +1,4 @@ -use std::path::Path; +use std::path::{Path, PathBuf}; use sha2::{Digest, Sha256}; @@ -17,6 +17,28 @@ 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 { + 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, 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") } @@ -59,11 +81,37 @@ fn tag_categories(registry: &mut RemoteSkillRegistry) { } } -/// Fetch the remote skill registry. Uses cache unless `force` is true or cache is stale. +/// Fetch the skill registry. Supports both remote HTTP URLs and local file paths. +/// +/// When `SKILLS_REGISTRY_URL` points to a local file (absolute path or `file://` URI), +/// the registry is read directly from disk — no caching is applied so changes are +/// picked up immediately (ideal for local development). +/// +/// For remote URLs, uses a 1-hour disk cache unless `force` is true. pub async fn registry_fetch( workspace_dir: &Path, force: bool, ) -> Result { + let url = registry_url(); + + // --- Local file path: read directly, skip cache for instant dev feedback --- + if is_local_path(&url) { + log::info!("[registry] reading local registry from {url}"); + let bytes = read_local_file(&url)?; + let body = String::from_utf8(bytes) + .map_err(|e| format!("registry file is not valid UTF-8: {e}"))?; + let mut registry: RemoteSkillRegistry = serde_json::from_str(&body) + .map_err(|e| format!("failed to parse local registry JSON: {e}"))?; + tag_categories(&mut registry); + log::info!( + "[registry] loaded {} core + {} third-party skills from local file", + registry.skills.core.len(), + registry.skills.third_party.len() + ); + return Ok(registry); + } + + // --- Remote URL: use disk cache --- if !force { if let Some(cached) = read_cache(workspace_dir) { if is_cache_fresh(&cached) { @@ -73,7 +121,6 @@ pub async fn registry_fetch( } } - let url = registry_url(); log::info!("[registry] fetching registry from {url}"); let client = reqwest::Client::builder() @@ -156,8 +203,59 @@ pub async fn registry_search( Ok(results) } -/// Install a skill by downloading its JS bundle and manifest from the registry. +/// Fetch bytes from a URL — supports both local file paths and HTTP URLs. +async fn fetch_url_bytes(url: &str) -> Result, String> { + if is_local_path(url) { + return read_local_file(url); + } + + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build() + .map_err(|e| format!("failed to create HTTP client: {e}"))?; + + let resp = client + .get(url) + .send() + .await + .map_err(|e| format!("failed to fetch {url}: {e}"))?; + + if !resp.status().is_success() { + return Err(format!("fetch {url} failed with status {}", resp.status())); + } + + resp.bytes() + .await + .map(|b| b.to_vec()) + .map_err(|e| format!("failed to read response from {url}: {e}")) +} + +/// Install a skill from the registry or local directory. +/// +/// When `SKILLS_LOCAL_DIR` is set, copies files directly from the local skills +/// directory (e.g. `$SKILLS_LOCAL_DIR//`) instead of downloading. +/// This also works when registry entry URLs are local file paths. pub async fn skill_install(workspace_dir: &Path, skill_id: &str) -> Result<(), String> { + // --- Fast path: SKILLS_LOCAL_DIR copies directly from local dev directory --- + if let Some(local_dir) = local_skills_dir() { + let local_skill = local_dir.join(skill_id); + if local_skill.exists() { + log::info!( + "[registry] installing '{skill_id}' from local dir: {}", + local_skill.display() + ); + let skill_dir = workspace_dir.join("skills").join(skill_id); + copy_dir_recursive(&local_skill, &skill_dir)?; + log::info!("[registry] skill '{skill_id}' installed from local dir"); + return Ok(()); + } + log::warn!( + "[registry] SKILLS_LOCAL_DIR set but '{skill_id}' not found at {}; falling back to registry", + local_skill.display() + ); + } + + // --- Standard path: fetch from registry (remote or local URLs) --- let registry = registry_fetch(workspace_dir, false).await?; let entry = registry @@ -172,46 +270,19 @@ pub async fn skill_install(workspace_dir: &Path, skill_id: &str) -> Result<(), S let skill_dir = workspace_dir.join("skills").join(skill_id); std::fs::create_dir_all(&skill_dir).map_err(|e| format!("failed to create skill dir: {e}"))?; - let client = reqwest::Client::builder() - .timeout(std::time::Duration::from_secs(30)) - .build() - .map_err(|e| format!("failed to create HTTP client: {e}"))?; + // Fetch manifest (local or remote) + log::info!( + "[registry] fetching manifest for '{skill_id}' from {}", + entry.manifest_url + ); + let manifest_bytes = fetch_url_bytes(&entry.manifest_url).await?; - // Download manifest - log::info!("[registry] downloading manifest for '{skill_id}'"); - let manifest_resp = client - .get(&entry.manifest_url) - .send() - .await - .map_err(|e| format!("failed to download manifest: {e}"))?; - if !manifest_resp.status().is_success() { - return Err(format!( - "manifest download failed with status {}", - manifest_resp.status() - )); - } - let manifest_bytes = manifest_resp - .bytes() - .await - .map_err(|e| format!("failed to read manifest: {e}"))?; - - // Download JS bundle - log::info!("[registry] downloading JS bundle for '{skill_id}'"); - let js_resp = client - .get(&entry.download_url) - .send() - .await - .map_err(|e| format!("failed to download JS bundle: {e}"))?; - if !js_resp.status().is_success() { - return Err(format!( - "JS download failed with status {}", - js_resp.status() - )); - } - let js_bytes = js_resp - .bytes() - .await - .map_err(|e| format!("failed to read JS bundle: {e}"))?; + // Fetch JS bundle (local or remote) + log::info!( + "[registry] fetching JS bundle for '{skill_id}' from {}", + entry.download_url + ); + let js_bytes = fetch_url_bytes(&entry.download_url).await?; // Verify checksum if present if let Some(expected) = &entry.checksum_sha256 { @@ -219,7 +290,6 @@ pub async fn skill_install(workspace_dir: &Path, skill_id: &str) -> Result<(), S hasher.update(&js_bytes); let actual = format!("{:x}", hasher.finalize()); if actual != *expected { - // Clean up the directory we created let _ = std::fs::remove_dir_all(&skill_dir); return Err(format!( "checksum mismatch for '{skill_id}': expected {expected}, got {actual}" @@ -238,6 +308,35 @@ pub async fn skill_install(workspace_dir: &Path, skill_id: &str) -> Result<(), S Ok(()) } +/// Recursively copy a directory tree from `src` to `dst`. +fn copy_dir_recursive(src: &Path, dst: &Path) -> Result<(), String> { + std::fs::create_dir_all(dst) + .map_err(|e| format!("failed to create dir {}: {e}", dst.display()))?; + + let entries = + std::fs::read_dir(src).map_err(|e| format!("failed to read dir {}: {e}", src.display()))?; + + for entry in entries { + let entry = entry.map_err(|e| format!("failed to read dir entry: {e}"))?; + let src_path = entry.path(); + let dst_path = dst.join(entry.file_name()); + + if src_path.is_dir() { + copy_dir_recursive(&src_path, &dst_path)?; + } else { + std::fs::copy(&src_path, &dst_path).map_err(|e| { + format!( + "failed to copy {} -> {}: {e}", + src_path.display(), + dst_path.display() + ) + })?; + } + } + + Ok(()) +} + /// Uninstall a skill by removing its directory from the workspace. pub async fn skill_uninstall(workspace_dir: &Path, skill_id: &str) -> Result<(), String> { let skill_dir = workspace_dir.join("skills").join(skill_id); diff --git a/src/openhuman/skills/schemas.rs b/src/openhuman/skills/schemas.rs index 6fdeb70d1..94e7f1021 100644 --- a/src/openhuman/skills/schemas.rs +++ b/src/openhuman/skills/schemas.rs @@ -773,9 +773,14 @@ fn handle_skills_sync(params: Map) -> ControllerFuture { Box::pin(async move { let p: SkillIdParams = serde_json::from_value(Value::Object(params)).map_err(|e| e.to_string())?; + log::debug!("[skills] handle_skills_sync: skill_id={}", p.skill_id); let engine = require_engine()?; + log::debug!( + "[skills] handle_skills_sync: dispatching skill/sync to engine for '{}'", + p.skill_id + ); engine - .rpc(&p.skill_id, "skill/tick", serde_json::json!({})) + .rpc(&p.skill_id, "skill/sync", serde_json::json!({})) .await }) } diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index ca32388d0..5e5d48bfc 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -1125,7 +1125,7 @@ async fn json_rpc_skills_runtime_start_tools_call_stop() { "echo tool should return the message: {call_result}" ); - // 5. Trigger sync (tick) + // 5. Trigger sync (routes to onSync via skill/sync RPC) let sync = post_json_rpc( &rpc_base, 24, @@ -1133,12 +1133,9 @@ async fn json_rpc_skills_runtime_start_tools_call_stop() { json!({"skill_id": "e2e-runtime"}), ) .await; - let sync_result = assert_no_jsonrpc_error(&sync, "skills_sync"); - assert_eq!( - sync_result.get("ok"), - Some(&json!(true)), - "sync should acknowledge: {sync_result}" - ); + // skills_sync now routes through "skill/sync" → onSync(). The e2e skill + // does not export onSync, so handle_js_call returns null (no error). + let _sync_result = assert_no_jsonrpc_error(&sync, "skills_sync"); // 6. Stop the skill let stop = post_json_rpc( diff --git a/tests/skills_debug_e2e.rs b/tests/skills_debug_e2e.rs index e2df002b8..31cd8f5e0 100644 --- a/tests/skills_debug_e2e.rs +++ b/tests/skills_debug_e2e.rs @@ -64,9 +64,10 @@ fn info(msg: &str) { /// /// Priority: /// 1. SKILL_DEBUG_DIR env var -/// 2. ../openhuman-skills/skills (sibling repo) -/// 3. openhuman-skills/skills (subdir) -/// 4. Broader search in parent workspace +/// 2. SKILLS_LOCAL_DIR env var (shared with runtime) +/// 3. ../openhuman-skills/skills (sibling repo) +/// 4. openhuman-skills/skills (subdir) +/// 5. Broader search in parent workspace fn try_find_skills_dir() -> Option { if let Ok(dir) = std::env::var("SKILL_DEBUG_DIR") { let p = PathBuf::from(&dir); @@ -77,6 +78,14 @@ fn try_find_skills_dir() -> Option { return None; } + if let Ok(dir) = std::env::var("SKILLS_LOCAL_DIR") { + let p = PathBuf::from(&dir); + if p.exists() { + return Some(p); + } + eprintln!("SKILLS_LOCAL_DIR={dir} does not exist"); + } + let cwd = std::env::current_dir().expect("cwd"); let candidates = [ @@ -363,8 +372,8 @@ async fn skill_full_lifecycle() { } } - // ── 6. Tick / Sync ── - step("TICK (sync)"); + // ── 6a. Tick ── + step("TICK"); let tick_result = tokio::time::timeout( Duration::from_secs(15), engine.rpc(&skill_id, "skill/tick", json!({})), @@ -383,6 +392,28 @@ async fn skill_full_lifecycle() { } } + // ── 6b. Sync (calls onSync, not onTick) ── + step("SYNC (skill/sync → onSync)"); + let sync_result = tokio::time::timeout( + Duration::from_secs(15), + engine.rpc(&skill_id, "skill/sync", json!({})), + ) + .await; + + match sync_result { + Ok(Ok(val)) => { + ok(&format!("skill/sync returned: {val}")); + } + Ok(Err(e)) => { + info(&format!( + "skill/sync error: {e} (expected if skill has no onSync handler)" + )); + } + Err(_) => { + fail("skill/sync TIMED OUT (15s)"); + } + } + // ── 7. Session lifecycle ── step("SESSION LIFECYCLE"); let session_id = "debug-session-1"; diff --git a/tests/skills_notion_live.rs b/tests/skills_notion_live.rs index 10fd9b4b3..1f4158bb6 100644 --- a/tests/skills_notion_live.rs +++ b/tests/skills_notion_live.rs @@ -18,11 +18,31 @@ use serde_json::{json, Value}; use openhuman_core::openhuman::skills::qjs_engine::{set_global_engine, RuntimeEngine}; +/// Truncate a string to at most `max_bytes` without panicking on multi-byte +/// char boundaries. Falls back to the nearest char boundary at or before +/// `max_bytes`. +fn truncate_str(s: &str, max_bytes: usize) -> &str { + if s.len() <= max_bytes { + return s; + } + let mut end = max_bytes; + while end > 0 && !s.is_char_boundary(end) { + end -= 1; + } + &s[..end] +} + fn try_find_skills_dir() -> Option { if let Ok(dir) = std::env::var("SKILL_DEBUG_DIR") { let p = PathBuf::from(&dir); return if p.exists() { Some(p) } else { None }; } + if let Ok(dir) = std::env::var("SKILLS_LOCAL_DIR") { + let p = PathBuf::from(&dir); + if p.exists() { + return Some(p); + } + } let cwd = std::env::current_dir().expect("cwd"); for candidate in &[ "../openhuman-skills/skills", @@ -121,7 +141,7 @@ async fn notion_live_with_real_data() { eprintln!(" GET /settings → HTTP {status}"); if status.is_success() { eprintln!(" ✓ Backend reachable, JWT valid"); - eprintln!(" Body: {}...", &body[..body.len().min(200)]); + eprintln!(" Body: {}...", truncate_str(&body, 200)); } else if status.as_u16() == 502 { eprintln!(" ✗ Backend DOWN (502 Bad Gateway)"); eprintln!(" The staging server is unreachable. OAuth proxy will fail."); @@ -154,11 +174,11 @@ async fn notion_live_with_real_data() { eprintln!(" HTTP {status}"); if status.is_success() { eprintln!(" ✓ Notion API accessible via proxy"); - eprintln!(" Body: {}...", &body[..body.len().min(300)]); + eprintln!(" Body: {}...", truncate_str(&body, 300)); } else { eprintln!( " ✗ Proxy returned {status}: {}...", - &body[..body.len().min(200)] + truncate_str(&body, 200) ); } } @@ -258,7 +278,7 @@ async fn notion_live_with_real_data() { for content in &result.content { match content { openhuman_core::openhuman::skills::types::ToolContent::Text { text } => { - eprintln!(" Result: {}...", &text[..text.len().min(500)]); + eprintln!(" Result: {}...", truncate_str(text, 500)); } _ => {} } @@ -282,7 +302,7 @@ async fn notion_live_with_real_data() { for content in &result.content { match content { openhuman_core::openhuman::skills::types::ToolContent::Text { text } => { - eprintln!(" Result: {}...", &text[..text.len().min(500)]); + eprintln!(" Result: {}...", truncate_str(text, 500)); } _ => {} } @@ -292,8 +312,77 @@ async fn notion_live_with_real_data() { Err(_) => eprintln!(" ✗ TIMED OUT"), } - // ── Step 7: Final state ── - eprintln!("\n--- Step 7: Final Skill State ---"); + // ── Step 7: Sync + Memory Persistence ── + eprintln!("\n--- Step 7: Sync + Memory Verification ---"); + eprintln!(" Calling skill/sync to trigger onSync() + memory persistence..."); + + let sync_result = tokio::time::timeout( + Duration::from_secs(60), + engine.rpc("notion", "skill/sync", json!({})), + ) + .await; + + match &sync_result { + Ok(Ok(val)) => eprintln!(" skill/sync returned: {val}"), + Ok(Err(e)) => eprintln!(" skill/sync error: {e}"), + Err(_) => eprintln!(" skill/sync TIMED OUT (60s)"), + } + + // Wait for fire-and-forget memory persistence to complete + eprintln!(" Waiting 3s for async memory persistence..."); + tokio::time::sleep(Duration::from_secs(3)).await; + + // Verify memory documents were created. + // The RuntimeEngine initializes its MemoryClient via new_local() which + // uses ~/.openhuman/workspace, so we check there. + eprintln!(" Checking local memory store (~/.openhuman/workspace)..."); + match openhuman_core::openhuman::memory::MemoryClient::new_local() { + Ok(memory_client) => { + let namespace = "skill-notion"; + match memory_client.list_documents(Some(namespace)).await { + Ok(docs) => { + let doc_array = docs + .get("documents") + .and_then(|d| d.as_array()) + .cloned() + .unwrap_or_default(); + eprintln!(" Documents in '{namespace}': {}", doc_array.len()); + for doc in &doc_array { + let title = doc.get("title").and_then(|t| t.as_str()).unwrap_or("?"); + let content_len = doc + .get("content") + .and_then(|c| c.as_str()) + .map(|c| c.len()) + .unwrap_or(0); + eprintln!(" - {title} ({content_len} bytes)"); + } + if doc_array.is_empty() { + eprintln!(" WARNING: No memory documents found after sync"); + eprintln!(" This could mean:"); + eprintln!(" 1. onSync() didn't publish any state via state.set()"); + eprintln!(" 2. The memory client wasn't wired into the engine"); + eprintln!(" 3. store_skill_sync failed silently"); + } else { + eprintln!(" PASS: Memory documents created after sync"); + } + } + Err(e) => eprintln!(" Failed to list documents: {e}"), + } + + match memory_client.list_namespaces().await { + Ok(namespaces) => { + eprintln!(" All namespaces: {:?}", namespaces); + } + Err(e) => eprintln!(" Failed to list namespaces: {e}"), + } + } + Err(e) => { + eprintln!(" Could not create MemoryClient: {e}"); + } + } + + // ── Step 8: Final state ── + eprintln!("\n--- Step 8: Final Skill State ---"); if let Some(snap) = engine.get_skill_state("notion") { eprintln!(" Status: {:?}", snap.status); for (k, v) in &snap.state { diff --git a/tests/skills_rpc_e2e.rs b/tests/skills_rpc_e2e.rs index c2857a736..8bc69be90 100644 --- a/tests/skills_rpc_e2e.rs +++ b/tests/skills_rpc_e2e.rs @@ -34,6 +34,12 @@ fn try_find_skills_dir() -> Option { let p = PathBuf::from(&dir); return if p.exists() { Some(p) } else { None }; } + if let Ok(dir) = std::env::var("SKILLS_LOCAL_DIR") { + let p = PathBuf::from(&dir); + if p.exists() { + return Some(p); + } + } let cwd = std::env::current_dir().expect("cwd"); for candidate in &[ "../openhuman-skills/skills", diff --git a/tests/skills_sync_memory_test.rs b/tests/skills_sync_memory_test.rs new file mode 100644 index 000000000..021790dea --- /dev/null +++ b/tests/skills_sync_memory_test.rs @@ -0,0 +1,388 @@ +//! Integration test: skill sync → memory persistence. +//! +//! Verifies that calling `skill/sync` via RPC: +//! 1. Invokes the skill's `onSync()` JS handler +//! 2. Persists published state to the local memory store +//! +//! Also tests that cron-triggered syncs and tick persist to memory. +//! +//! Run: +//! cargo test --test skills_sync_memory_test -- --nocapture + +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex, OnceLock}; +use std::time::Duration; + +use serde_json::json; +use tempfile::tempdir; + +use openhuman_core::core::all::try_invoke_registered_rpc; +use openhuman_core::openhuman::memory::MemoryClient; +use openhuman_core::openhuman::skills::qjs_engine::{set_global_engine, RuntimeEngine}; + +/// Serializes tests in this binary: `set_global_engine` mutates the process-wide +/// GLOBAL_ENGINE, so parallel tests would cross-wire engine instances. +static ENV_LOCK: OnceLock> = OnceLock::new(); + +fn env_lock() -> std::sync::MutexGuard<'static, ()> { + ENV_LOCK + .get_or_init(|| Mutex::new(())) + .lock() + .expect("skills_sync_memory_test env lock poisoned") +} + +// ── Helpers ────────────────────────────────────────────────────────────────── + +fn try_find_skills_dir() -> Option { + if let Ok(dir) = std::env::var("SKILL_DEBUG_DIR") { + let p = PathBuf::from(&dir); + return if p.exists() { Some(p) } else { None }; + } + if let Ok(dir) = std::env::var("SKILLS_LOCAL_DIR") { + let p = PathBuf::from(&dir); + if p.exists() { + return Some(p); + } + } + let cwd = std::env::current_dir().expect("cwd"); + for candidate in &[ + "../openhuman-skills/skills", + "openhuman-skills/skills", + "../alphahuman/skills/skills", + ] { + let p = cwd.join(candidate); + if p.exists() { + return Some(p.canonicalize().unwrap()); + } + } + if let Some(parent) = cwd.parent() { + for entry in std::fs::read_dir(parent).into_iter().flatten().flatten() { + let c = entry.path().join("skills/skills"); + if c.join("example-skill/manifest.json").exists() { + return Some(c.canonicalize().unwrap()); + } + } + } + None +} + +macro_rules! require_skills_dir { + () => { + match try_find_skills_dir() { + Some(dir) => dir, + None => { + eprintln!("SKIPPED: no skills directory available (set SKILL_DEBUG_DIR for CI)"); + return; + } + } + }; +} + +async fn create_engine_with_memory( + skills_dir: &Path, + data_dir: &Path, + workspace_dir: &Path, +) -> (Arc, Arc) { + let engine = + RuntimeEngine::new(data_dir.to_path_buf()).expect("RuntimeEngine::new should succeed"); + let engine = Arc::new(engine); + engine.set_skills_source_dir(skills_dir.to_path_buf()); + + // Create a MemoryClient pointing at the temp workspace + let memory_client = + MemoryClient::from_workspace_dir(workspace_dir.to_path_buf()).expect("MemoryClient"); + let memory_client = Arc::new(memory_client); + + // Wire the memory client into the engine so event_loop can use it + engine.set_memory_client(memory_client.clone()); + + // Set as global so RPC handlers can find it + set_global_engine(engine.clone()); + + (engine, memory_client) +} + +// ── Tests ──────────────────────────────────────────────────────────────────── + +/// Test that `skill/sync` RPC triggers `onSync()` and persists state to memory. +#[tokio::test] +async fn sync_rpc_persists_to_memory() { + let _lock = env_lock(); + let _ = env_logger::builder() + .filter_level(log::LevelFilter::Debug) + .is_test(true) + .try_init(); + + let skills_dir = require_skills_dir!(); + let tmp = tempdir().expect("tempdir"); + let data_dir = tmp.path().join("skills_data"); + let workspace_dir = tmp.path().join("workspace"); + std::fs::create_dir_all(&data_dir).unwrap(); + std::fs::create_dir_all(&workspace_dir).unwrap(); + + let skill_id = "example-skill"; + + eprintln!("\n=== sync_rpc_persists_to_memory ==="); + eprintln!(" Skills dir: {}", skills_dir.display()); + eprintln!(" Data dir: {}", data_dir.display()); + eprintln!(" Workspace dir: {}", workspace_dir.display()); + + let (engine, memory_client) = + create_engine_with_memory(&skills_dir, &data_dir, &workspace_dir).await; + + // ── Start skill ── + eprintln!("\n--- Start skill '{skill_id}' ---"); + let snap = engine.start_skill(skill_id).await.expect("start"); + eprintln!(" Status: {:?}, tools: {}", snap.status, snap.tools.len()); + eprintln!( + " Published state keys: {:?}", + snap.state.keys().collect::>() + ); + + // ── Verify no memory documents exist yet ── + eprintln!("\n--- Check memory before sync ---"); + let namespace = format!("skill-{}", skill_id); + let before = memory_client + .list_documents(Some(&namespace)) + .await + .expect("list_documents"); + let before_count = before + .get("documents") + .and_then(|d| d.as_array()) + .map(|a| a.len()) + .unwrap_or(0); + eprintln!(" Documents in '{namespace}' before sync: {before_count}"); + + // ── Call skill/sync via RPC ── + eprintln!("\n--- Call skill/sync RPC ---"); + let sync_result = tokio::time::timeout( + Duration::from_secs(30), + engine.rpc(skill_id, "skill/sync", json!({})), + ) + .await; + + match &sync_result { + Ok(Ok(val)) => eprintln!(" skill/sync returned: {val}"), + Ok(Err(e)) => eprintln!(" skill/sync error: {e} (may be expected for example-skill)"), + Err(_) => panic!("skill/sync TIMED OUT"), + } + + // ── Wait for fire-and-forget memory persistence ── + eprintln!("\n--- Waiting for memory persistence (2s) ---"); + tokio::time::sleep(Duration::from_secs(2)).await; + + // ── Verify documents were created ── + eprintln!("\n--- Check memory after sync ---"); + let after = memory_client + .list_documents(Some(&namespace)) + .await + .expect("list_documents"); + let after_docs = after + .get("documents") + .and_then(|d| d.as_array()) + .cloned() + .unwrap_or_default(); + eprintln!( + " Documents in '{namespace}' after sync: {}", + after_docs.len() + ); + + for doc in &after_docs { + let title = doc.get("title").and_then(|t| t.as_str()).unwrap_or("?"); + let doc_id = doc + .get("documentId") + .and_then(|d| d.as_str()) + .unwrap_or("?"); + eprintln!(" - title: {title}, doc_id: {doc_id}"); + } + + // The example-skill may or may not publish state during onSync. + // If it does, we should see at least one document. + // If it doesn't (no state.set calls in onSync), the snapshot will be empty + // and no document is created — that's correct behavior. + // + // We check published_state to know what to expect. + let final_state = engine.get_skill_state(skill_id); + let has_published_state = final_state + .as_ref() + .map(|s| !s.state.is_empty()) + .unwrap_or(false); + eprintln!( + " Skill has published state: {} ({} keys)", + has_published_state, + final_state.as_ref().map(|s| s.state.len()).unwrap_or(0) + ); + + if has_published_state { + assert!( + after_docs.len() > before_count, + "Expected at least one new document in namespace '{namespace}' after sync, \ + but found {} (before: {before_count}). Published state is non-empty, \ + so store_skill_sync should have been called.", + after_docs.len() + ); + eprintln!(" PASS: Memory document created after sync"); + } else { + eprintln!(" NOTE: Skill has no published state — no memory write expected"); + eprintln!(" (This is correct behavior; persist_state_to_memory skips empty state)"); + } + + // ── Cleanup ── + let _ = engine.stop_skill(skill_id).await; + eprintln!("\n=== sync_rpc_persists_to_memory COMPLETE ===\n"); +} + +/// Test that `skill/tick` also persists state to memory. +#[tokio::test] +async fn tick_persists_to_memory() { + let _lock = env_lock(); + let _ = env_logger::builder() + .filter_level(log::LevelFilter::Debug) + .is_test(true) + .try_init(); + + let skills_dir = require_skills_dir!(); + let tmp = tempdir().expect("tempdir"); + let data_dir = tmp.path().join("skills_data"); + let workspace_dir = tmp.path().join("workspace"); + std::fs::create_dir_all(&data_dir).unwrap(); + std::fs::create_dir_all(&workspace_dir).unwrap(); + + let skill_id = "example-skill"; + + eprintln!("\n=== tick_persists_to_memory ==="); + + let (engine, memory_client) = + create_engine_with_memory(&skills_dir, &data_dir, &workspace_dir).await; + + // Start skill + let snap = engine.start_skill(skill_id).await.expect("start"); + eprintln!(" Started: {:?}", snap.status); + + // Call skill/tick + let tick_result = tokio::time::timeout( + Duration::from_secs(15), + engine.rpc(skill_id, "skill/tick", json!({})), + ) + .await; + + match &tick_result { + Ok(Ok(val)) => eprintln!(" skill/tick returned: {val}"), + Ok(Err(e)) => eprintln!(" skill/tick error: {e}"), + Err(_) => panic!("skill/tick TIMED OUT"), + } + + // Wait for persistence + tokio::time::sleep(Duration::from_secs(2)).await; + + // Check memory + let namespace = format!("skill-{}", skill_id); + let docs = memory_client + .list_documents(Some(&namespace)) + .await + .expect("list_documents"); + let doc_count = docs + .get("documents") + .and_then(|d| d.as_array()) + .map(|a| a.len()) + .unwrap_or(0); + eprintln!(" Documents in '{namespace}' after tick: {doc_count}"); + + let has_published_state = engine + .get_skill_state(skill_id) + .map(|s| !s.state.is_empty()) + .unwrap_or(false); + + if has_published_state { + assert!( + doc_count > 0, + "Expected memory documents after tick with published state" + ); + eprintln!(" PASS: Memory persisted after tick"); + } else { + eprintln!(" NOTE: No published state — no memory write expected"); + } + + let _ = engine.stop_skill(skill_id).await; + eprintln!("=== tick_persists_to_memory COMPLETE ===\n"); +} + +/// Verify that the `skills_sync` RPC schema routes to `skill/sync` (not `skill/tick`). +/// This is a regression test for the routing bug where `handle_skills_sync` sent +/// `"skill/tick"` instead of `"skill/sync"`. +#[tokio::test] +async fn skills_sync_rpc_calls_on_sync_not_on_tick() { + let _lock = env_lock(); + let _ = env_logger::builder() + .filter_level(log::LevelFilter::Debug) + .is_test(true) + .try_init(); + + let skills_dir = require_skills_dir!(); + let tmp = tempdir().expect("tempdir"); + let data_dir = tmp.path().join("skills_data"); + let workspace_dir = tmp.path().join("workspace"); + std::fs::create_dir_all(&data_dir).unwrap(); + std::fs::create_dir_all(&workspace_dir).unwrap(); + + let skill_id = "example-skill"; + + eprintln!("\n=== skills_sync_rpc_calls_on_sync_not_on_tick ==="); + + let (engine, memory_client) = + create_engine_with_memory(&skills_dir, &data_dir, &workspace_dir).await; + + let snap = engine.start_skill(skill_id).await.expect("start"); + eprintln!(" Started: {:?}", snap.status); + + // Exercise the full controller path via try_invoke_registered_rpc so we + // verify handle_skills_sync routes to "skill/sync" (not "skill/tick"). + // This catches regressions in the controller layer that engine.rpc() would bypass. + let rpc_params: serde_json::Map = + serde_json::from_value(json!({"skill_id": skill_id})).unwrap(); + let sync_result = tokio::time::timeout( + Duration::from_secs(15), + try_invoke_registered_rpc("openhuman.skills_sync", rpc_params), + ) + .await; + + match &sync_result { + Ok(Some(Ok(val))) => { + eprintln!(" openhuman.skills_sync returned: {val}"); + // The result comes from handle_js_call("onSync") which returns the + // JS value (typically null/undefined). The old buggy path returned + // {"ok": true} from handle_js_void_call("onTick"). + } + Ok(Some(Err(e))) => { + eprintln!(" openhuman.skills_sync error: {e}"); + // An error from onSync not being defined is still acceptable — + // the important thing is it tried onSync, not onTick. + } + Ok(None) => { + panic!("openhuman.skills_sync not found in registered controllers"); + } + Err(_) => panic!("openhuman.skills_sync TIMED OUT"), + } + + // Also verify the namespace gets a title with "periodic sync" (not "tick sync") + tokio::time::sleep(Duration::from_secs(2)).await; + + let namespace = format!("skill-{}", skill_id); + let docs = memory_client + .list_documents(Some(&namespace)) + .await + .expect("list_documents"); + if let Some(doc_array) = docs.get("documents").and_then(|d| d.as_array()) { + for doc in doc_array { + let title = doc.get("title").and_then(|t| t.as_str()).unwrap_or("?"); + eprintln!(" Memory doc title: {title}"); + assert!( + title.contains("periodic sync"), + "Expected title to contain 'periodic sync' (from skill/sync handler), got: {title}" + ); + } + } + + let _ = engine.stop_skill(skill_id).await; + eprintln!("=== skills_sync_rpc_calls_on_sync_not_on_tick COMPLETE ===\n"); +}