From c134d96056056a1d00749570f1df7977df7b959b Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Wed, 1 Apr 2026 22:00:57 -0700 Subject: [PATCH] fix(skills): route sync RPC to onSync handler and persist state to memory (#183) * feat(debug): add script for Notion sync memory verification and enhance memory persistence - Introduced `debug-notion-sync-memory.sh` to facilitate live testing of the Notion skill with memory verification. - The script validates the full flow from skill start to memory persistence, ensuring required environment variables are set. - Updated `handle_skills_sync` to change the RPC method from `skill/tick` to `skill/sync` for better clarity in the sync process. - Implemented `persist_state_to_memory` function to ensure published ops state is saved to memory after sync, cron, and tick events. - Enhanced end-to-end tests to validate memory persistence during skill sync and tick operations, ensuring robust functionality. This update improves the debugging process for the Notion skill and enhances the overall reliability of memory operations. * style: apply cargo fmt formatting Co-Authored-By: Claude Opus 4.6 (1M context) * feat(memory): implement background memory-write worker for skill state persistence - Introduced a `MemoryWriteJob` struct to encapsulate the details of memory write operations. - Added a bounded background worker using `tokio::spawn` to handle memory writes asynchronously, improving performance and responsiveness. - Updated `persist_state_to_memory` to queue memory write jobs instead of executing them directly, allowing for better flow control and error handling. - Enhanced logging for memory persistence operations to provide clearer insights into success and failure cases. - Modified event handling in `handle_message` to ensure state persistence occurs only after successful operations, reducing the risk of data loss. This update significantly enhances the memory management capabilities of the skill instance, ensuring more reliable state persistence. * feat(skills): enhance skills synchronization and memory persistence - Added detailed debug logging in `handle_skills_sync` to track skill synchronization events, improving observability during skill operations. - Updated `persist_state_to_memory` to include a memory write transaction, ensuring state persistence is handled more robustly and efficiently. - Enhanced event handling in `handle_message` to ensure memory writes are queued correctly, improving the reliability of state persistence during skill operations. These changes improve the overall functionality and reliability of skills synchronization and memory management. * feat(skills): enhance skills synchronization and memory persistence - Added detailed debug logging in `handle_skills_sync` to track skill synchronization events, improving observability during skill operations. - Updated `persist_state_to_memory` to include a `memory_write_tx` parameter, allowing for more efficient state persistence in the event loop. - Enhanced tests for skills synchronization to ensure robust memory handling and state persistence during skill operations. This update improves the reliability and traceability of skills synchronization processes, ensuring better memory management and debugging capabilities. * refactor(tests): update JSON-RPC sync handling in end-to-end tests - Enhanced comments in `json_rpc_skills_runtime_start_tools_call_stop` to clarify the sync process routing through the `skill/sync` RPC. - Updated assertions in `skills_sync_rpc_calls_on_sync_not_on_tick` to ensure proper handling of the new sync flow, verifying that the `skills_sync` method routes correctly to `onSync`. - Improved logging for better observability during skill synchronization tests, ensuring that the expected behavior aligns with the updated RPC structure. These changes improve the clarity and reliability of the end-to-end tests related to skills synchronization. * feat(env): enhance environment configuration for skills development - Updated `.env.example` to include `SKILLS_LOCAL_DIR` for local skills source directory, allowing developers to specify a path for skill discovery and installation. - Improved comments in the environment file to clarify the usage of `SKILLS_REGISTRY_URL` for both remote and local paths, enhancing the development experience. - Refactored `qjs_engine.rs` to prioritize the `SKILLS_LOCAL_DIR` for skill source directory resolution, improving local development workflows. - Added utility functions in `registry_ops.rs` to support local file path handling for skill registries, enhancing flexibility in skill management. These changes streamline the development process for skills by providing clearer configuration options and improving local development capabilities. * feat(tests): enhance skills directory discovery in test files - Updated `try_find_skills_dir` function across multiple test files to include support for the `SKILLS_LOCAL_DIR` environment variable, improving the flexibility of skills directory resolution. - Enhanced comments to clarify the order of directory search priorities, ensuring better understanding for developers. - Improved error handling and logging for cases where the specified directory does not exist, aiding in debugging and test reliability. These changes streamline the skills directory discovery process in tests, enhancing the overall development experience. * feat(tests): enhance skills directory discovery in test files - Updated `try_find_skills_dir` function to include support for the `SKILLS_LOCAL_DIR` environment variable, allowing for more flexible skills directory resolution. - Improved documentation to clarify the priority order for skills directory discovery. - Refactored multiple test files to utilize the updated skills directory discovery logic, enhancing consistency and maintainability across tests. These changes streamline the skills directory discovery process in tests, improving the overall testing framework. * refactor(tests): streamline memory client verification in Notion live tests - Updated the memory client verification process in `notion_live_with_real_data` to utilize `MemoryClient::new_local()` for improved clarity and consistency. - Enhanced comments to clarify the memory store check location and removed redundant error handling for workspace directory creation. - Simplified the error logging to focus on the memory client creation failure, improving readability and maintainability of the test code. These changes enhance the reliability of memory verification in the Notion live tests, ensuring a clearer understanding of the memory client initialization process. --------- Co-authored-by: Claude Opus 4.6 (1M context) --- .env.example | 13 +- scripts/debug-notion-sync-memory.sh | 73 ++++ src/openhuman/skills/qjs_engine.rs | 15 +- .../skills/qjs_skill_instance/event_loop.rs | 171 ++++++-- src/openhuman/skills/registry_ops.rs | 187 +++++++-- src/openhuman/skills/schemas.rs | 7 +- tests/json_rpc_e2e.rs | 11 +- tests/skills_debug_e2e.rs | 41 +- tests/skills_notion_live.rs | 103 ++++- tests/skills_rpc_e2e.rs | 6 + tests/skills_sync_memory_test.rs | 388 ++++++++++++++++++ 11 files changed, 919 insertions(+), 96 deletions(-) create mode 100755 scripts/debug-notion-sync-memory.sh create mode 100644 tests/skills_sync_memory_test.rs 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"); +}