diff --git a/.env.example b/.env.example index 4ea08a9be..252a06352 100644 --- a/.env.example +++ b/.env.example @@ -140,6 +140,9 @@ SKILLS_REGISTRY_URL= # 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= +# [optional] Enable sync-derived user working memory extraction (default: true). +# Set to false to disable persisting `working.user.*` docs from skill sync payloads. +OPENHUMAN_SKILLS_WORKING_MEMORY_ENABLED=true # --------------------------------------------------------------------------- # Error Reporting (Sentry) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index ce772474b..56f7a219b 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -249,6 +249,8 @@ AI Response to User Memory encryption keys derive from user credentials via Argon2id, ensuring memory files are unreadable without authentication. The hybrid search combines semantic understanding (vector similarity) with keyword precision (SQLite FTS5) for reliable recall. +Skill sync now also feeds a bounded **user working memory** layer (preferences, goals, constraints, recurring entities) via the core runtime sync worker. See [`docs/SKILL-WORKING-MEMORY.md`](./SKILL-WORKING-MEMORY.md) for hook locations, privacy rules, and extension guidance. + --- ## Security Architecture diff --git a/docs/SKILL-WORKING-MEMORY.md b/docs/SKILL-WORKING-MEMORY.md new file mode 100644 index 000000000..918cb174b --- /dev/null +++ b/docs/SKILL-WORKING-MEMORY.md @@ -0,0 +1,60 @@ +# Skill Sync Working Memory + +This document describes how OpenHuman turns successful skill sync payloads into durable user working memory for agent personalization. + +## Definition + +- **User working memory**: persisted, user-scoped facts that remain useful across turns (preferences, goals, constraints, recurring entities). +- **Ephemeral chat context**: transient per-turn conversation state and prompt history; not persisted by this flow. +- **TTL policy**: no TTL by default (`ttl = "none"`), but growth is bounded with deterministic upsert keys. + +## Hook location + +- Sync entrypoint: `src/openhuman/skills/qjs_skill_instance/event_loop/rpc_handlers.rs` (`handle_sync`). +- Sync persistence worker: `src/openhuman/skills/qjs_skill_instance/event_loop/mod.rs` (`spawn_memory_write_worker`). +- Working-memory extraction: `src/openhuman/skills/working_memory.rs`. +- Agent recall/injection: `src/openhuman/agent/loop_/memory_context.rs` and `src/openhuman/agent/memory_loader.rs`. + +Flow: +1. `skills.sync` triggers `skill/sync`. +2. On success, the event loop enqueues a memory write job. +3. The memory worker stores raw sync history and runs working-memory extraction. +4. Extracted working-memory documents are upserted into `global` with fixed keys: + - `working.user..preferences` + - `working.user..goals` + - `working.user..constraints` + - `working.user..entities` + - `working.user..summary` + +Control switch: +- `OPENHUMAN_SKILLS_WORKING_MEMORY_ENABLED=false` disables this extraction/persistence path. + +## Privacy and safety + +- Sensitive keys (`token`, `secret`, `password`, `credential`, OAuth/auth fields, API keys, JWT/cookies) are skipped. +- Sensitive value heuristics are applied to avoid persisting secret-like blobs. +- Common PII patterns (email, phone) are redacted before persistence. + +## Logging and observability + +Per sync batch, the worker logs: +- scalar fields scanned +- sensitive fields skipped +- extracted preferences/goals/constraints/entities +- generated/persisted/failed working-memory docs + +Log prefix: `[skills-working-memory]`. + +## Agent usage (controlled) + +- Agent context assembly now appends a bounded `[User working memory]` section. +- Only `working.user.*` keys are included, with relevance threshold + small caps. +- This keeps personalization available while preventing unbounded prompt growth. + +## Extending for new skills + +When a new integration needs better extraction quality: +1. Add or tune classification heuristics in `classify_into_buckets` and `looks_like_*` helpers in `src/openhuman/skills/working_memory.rs`. +2. Keep persistence bounded by reusing deterministic keys (do not introduce unbounded per-item keys by default). +3. Add/update tests with mocked sync payloads in `src/openhuman/skills/working_memory.rs`. +4. Verify degraded behavior remains non-fatal (sync success should not fail due to memory extraction issues). diff --git a/src/openhuman/agent/loop_/memory_context.rs b/src/openhuman/agent/loop_/memory_context.rs index 5deb20e85..2ad747545 100644 --- a/src/openhuman/agent/loop_/memory_context.rs +++ b/src/openhuman/agent/loop_/memory_context.rs @@ -1,6 +1,10 @@ use crate::openhuman::memory::Memory; +use std::collections::HashSet; use std::fmt::Write; +pub(crate) const WORKING_MEMORY_KEY_PREFIX: &str = "working.user."; +pub(crate) const WORKING_MEMORY_LIMIT: usize = 3; + /// Build context preamble by searching memory for relevant entries. /// Entries with a hybrid score below `min_relevance_score` are dropped to /// prevent unrelated memories from bleeding into the conversation. @@ -10,6 +14,7 @@ pub(crate) async fn build_context( min_relevance_score: f64, ) -> String { let mut context = String::new(); + let mut seen_keys = HashSet::new(); // Pull relevant memories for this message if let Ok(entries) = mem.recall(user_msg, 5, None).await { @@ -24,6 +29,34 @@ pub(crate) async fn build_context( if !relevant.is_empty() { context.push_str("[Memory context]\n"); for entry in &relevant { + seen_keys.insert(entry.key.clone()); + let _ = writeln!(context, "- {}: {}", entry.key, entry.content); + } + context.push('\n'); + } + } + + // Explicitly load bounded user working memory entries so sync-derived profile + // facts can influence the turn in a controlled way. + let working_query = format!("working.user {user_msg}"); + if let Ok(entries) = mem + .recall(&working_query, WORKING_MEMORY_LIMIT + 2, None) + .await + { + let working: Vec<_> = entries + .iter() + .filter(|entry| entry.key.starts_with(WORKING_MEMORY_KEY_PREFIX)) + .filter(|entry| !seen_keys.contains(&entry.key)) + .filter(|entry| match entry.score { + Some(score) => score >= min_relevance_score, + None => true, + }) + .take(WORKING_MEMORY_LIMIT) + .collect(); + + if !working.is_empty() { + context.push_str("[User working memory]\n"); + for entry in &working { let _ = writeln!(context, "- {}: {}", entry.key, entry.content); } context.push('\n'); diff --git a/src/openhuman/agent/loop_/mod.rs b/src/openhuman/agent/loop_/mod.rs index 828b2ee7e..581c3e060 100644 --- a/src/openhuman/agent/loop_/mod.rs +++ b/src/openhuman/agent/loop_/mod.rs @@ -4,7 +4,7 @@ pub(crate) mod context_guard; mod credentials; mod history; mod instructions; -mod memory_context; +pub(crate) mod memory_context; mod parse; mod session; mod tool_loop; diff --git a/src/openhuman/agent/memory_loader.rs b/src/openhuman/agent/memory_loader.rs index e78c8c1e2..cfe0601d3 100644 --- a/src/openhuman/agent/memory_loader.rs +++ b/src/openhuman/agent/memory_loader.rs @@ -1,5 +1,8 @@ use crate::openhuman::memory::Memory; use async_trait::async_trait; +use std::collections::HashSet; + +use super::loop_::memory_context::{WORKING_MEMORY_KEY_PREFIX, WORKING_MEMORY_LIMIT}; #[async_trait] pub trait MemoryLoader: Send + Sync { @@ -47,18 +50,15 @@ impl MemoryLoader for DefaultMemoryLoader { user_message: &str, ) -> anyhow::Result { let entries = memory.recall(user_message, self.limit, None).await?; - if entries.is_empty() { - return Ok(String::new()); - } - - let header = "[Memory context]\n"; - let mut context = String::from(header); + let mut context = String::new(); let budget = if self.max_context_chars > 0 { self.max_context_chars } else { usize::MAX }; + let mut seen_keys = HashSet::new(); + let header = "[Memory context]\n"; for entry in entries { if let Some(score) = entry.score { if score < self.min_relevance_score { @@ -66,6 +66,12 @@ impl MemoryLoader for DefaultMemoryLoader { } } let line = format!("- {}: {}\n", entry.key, entry.content); + if context.is_empty() { + if header.len() >= budget { + return Ok(String::new()); + } + context.push_str(header); + } if context.len() + line.len() > budget { tracing::debug!( budget, @@ -75,14 +81,51 @@ impl MemoryLoader for DefaultMemoryLoader { ); break; } + seen_keys.insert(entry.key); context.push_str(&line); } - // If all entries were below threshold, return empty - if context == header { - return Ok(String::new()); + // Explicit bounded recall for sync-derived user working memory. + let working_query = format!("working.user {user_message}"); + let working_entries = memory + .recall(&working_query, WORKING_MEMORY_LIMIT + 2, None) + .await + .unwrap_or_default(); + let mut appended_working_header = false; + for entry in working_entries + .into_iter() + .filter(|entry| entry.key.starts_with(WORKING_MEMORY_KEY_PREFIX)) + .filter(|entry| !seen_keys.contains(&entry.key)) + .filter(|entry| match entry.score { + Some(score) => score >= self.min_relevance_score, + None => true, + }) + .take(WORKING_MEMORY_LIMIT) + { + if !appended_working_header { + let section = "[User working memory]\n"; + if context.len() + section.len() > budget { + break; + } + context.push_str(section); + appended_working_header = true; + } + let line = format!("- {}: {}\n", entry.key, entry.content); + if context.len() + line.len() > budget { + tracing::debug!( + budget, + current_len = context.len(), + skipped_line_len = line.len(), + "[memory_loader] context budget reached while appending working memory" + ); + break; + } + context.push_str(&line); } + if context.is_empty() { + return Ok(String::new()); + } context.push('\n'); Ok(context) } @@ -109,13 +152,25 @@ mod tests { async fn recall( &self, - _query: &str, + query: &str, limit: usize, _session_id: Option<&str>, ) -> anyhow::Result> { if limit == 0 { return Ok(vec![]); } + if query.contains("working.user") { + return Ok(vec![MemoryEntry { + id: "2".into(), + key: "working.user.gmail.summary".into(), + content: "User prefers concise updates.".into(), + namespace: Some("global".into()), + category: MemoryCategory::Core, + timestamp: "now".into(), + session_id: None, + score: Some(0.95), + }]); + } Ok(vec![MemoryEntry { id: "1".into(), key: "k".into(), @@ -163,5 +218,7 @@ mod tests { let context = loader.load_context(&MockMemory, "hello").await.unwrap(); assert!(context.contains("[Memory context]")); assert!(context.contains("- k: v")); + assert!(context.contains("[User working memory]")); + assert!(context.contains("working.user.gmail.summary")); } } diff --git a/src/openhuman/skills/mod.rs b/src/openhuman/skills/mod.rs index a1acd31af..44c40ff56 100644 --- a/src/openhuman/skills/mod.rs +++ b/src/openhuman/skills/mod.rs @@ -22,6 +22,7 @@ mod schemas; pub mod skill_registry; pub mod types; pub mod utils; +pub mod working_memory; pub use ops::*; pub use qjs_engine::{global_engine, require_engine, set_global_engine}; diff --git a/src/openhuman/skills/qjs_skill_instance/event_loop/mod.rs b/src/openhuman/skills/qjs_skill_instance/event_loop/mod.rs index 54c6e6ad7..fe9bebec7 100644 --- a/src/openhuman/skills/qjs_skill_instance/event_loop/mod.rs +++ b/src/openhuman/skills/qjs_skill_instance/event_loop/mod.rs @@ -22,6 +22,7 @@ use crate::openhuman::{ skills::{ quickjs_libs::qjs_ops, types::{SkillMessage, SkillStatus, ToolResult}, + working_memory::{skills_working_memory_enabled, working_memory_documents_from_sync}, }, tool_timeout::{tool_execution_timeout_duration, tool_execution_timeout_secs}, }; @@ -45,6 +46,11 @@ pub(crate) struct MemoryWriteJob { title: String, /// Stringified JSON content of the skill's published state. content: String, + /// Whether to derive and persist user working-memory documents from this + /// payload. Set `true` only for explicit `skill/sync` enqueues; leave + /// `false` for Tick and CronTrigger enqueues so that `working.user.*` + /// documents are only produced from intentional sync payloads. + extract_working_memory: bool, } /// Maximum number of memory-write jobs that can be buffered before back-pressure @@ -92,6 +98,60 @@ fn spawn_memory_write_worker() -> mpsc::Sender { } log::debug!("[memory] store_skill_sync succeeded for '{}'", job.title); + if job.extract_working_memory { + if skills_working_memory_enabled() { + let working_memory = + working_memory_documents_from_sync(&job.skill, &job.content); + let mut working_persisted = 0usize; + let mut working_failed = 0usize; + for doc in working_memory.documents { + let key = doc.key.clone(); + match job.client.put_doc(doc).await { + Ok(_) => { + working_persisted += 1; + } + Err(e) => { + working_failed += 1; + log::warn!( + "[skills-working-memory] put_doc failed for skill='{}' key='{}': {}", + job.skill, + key, + e + ); + } + } + } + log::info!( + "[skills-working-memory] sync_batch skill='{}' title='{}' scalar_fields={} \ + skipped_sensitive={} prefs={} goals={} constraints={} entities={} \ + generated_docs={} persisted_docs={} failed_docs={}", + job.skill, + job.title, + working_memory.stats.scalar_fields_seen, + working_memory.stats.sensitive_fields_skipped, + working_memory.stats.preferences, + working_memory.stats.goals, + working_memory.stats.constraints, + working_memory.stats.entities, + working_memory.stats.documents_generated, + working_persisted, + working_failed, + ); + } else { + log::debug!( + "[skills-working-memory] disabled by OPENHUMAN_SKILLS_WORKING_MEMORY_ENABLED; skill='{}' title='{}'", + job.skill, + job.title + ); + } + } else { + log::debug!( + "[skills-working-memory] skipping working-memory extraction for non-sync job; skill='{}' title='{}'", + job.skill, + job.title + ); + } + let namespace = format!("skill-{}", job.skill.trim()); let skill = job.skill.trim().to_lowercase(); @@ -266,12 +326,17 @@ async fn ingest_single_doc( } /// Snapshot the skill's published state and queue it for background memory persistence. +/// +/// `extract_working_memory` should be `true` only for explicit `skill/sync` enqueues so +/// that user working-memory documents (`working.user.*`) are derived only from intentional +/// sync payloads, not from every Tick or CronTrigger write. pub(crate) fn persist_state_to_memory( skill_id: &str, title_suffix: &str, ops_state: &Arc>, memory_client: &Option, memory_write_tx: &mpsc::Sender, + extract_working_memory: bool, ) { let state_snapshot = ops_state.read().data.clone(); log::debug!( @@ -299,6 +364,7 @@ pub(crate) fn persist_state_to_memory( skill, title: title.clone(), content, + extract_working_memory, }) { log::warn!( "[memory] persist_state_to_memory: channel full, dropping write for '{title}': {e}" @@ -526,6 +592,7 @@ async fn handle_message( ops_state, memory_client, memory_write_tx, + false, ); } Err(e) => { @@ -604,6 +671,7 @@ async fn handle_message( ops_state, memory_client, memory_write_tx, + false, ); } let _ = reply.send(result); diff --git a/src/openhuman/skills/qjs_skill_instance/event_loop/rpc_handlers.rs b/src/openhuman/skills/qjs_skill_instance/event_loop/rpc_handlers.rs index 2c2deb3ce..9b96a0fb4 100644 --- a/src/openhuman/skills/qjs_skill_instance/event_loop/rpc_handlers.rs +++ b/src/openhuman/skills/qjs_skill_instance/event_loop/rpc_handlers.rs @@ -385,13 +385,16 @@ pub(crate) async fn handle_sync( ) -> Result { let result = handle_js_call(rt, ctx, "onSync", "{}").await; if result.is_ok() { - // Only persist to memory if the JS sync was successful + // Only persist to memory if the JS sync was successful; set + // extract_working_memory=true so this explicit sync payload derives + // working.user.* documents. persist_state_to_memory( skill_id, "periodic sync", ops_state, memory_client, memory_write_tx, + true, ); } result diff --git a/src/openhuman/skills/working_memory.rs b/src/openhuman/skills/working_memory.rs new file mode 100644 index 000000000..13baa5e73 --- /dev/null +++ b/src/openhuman/skills/working_memory.rs @@ -0,0 +1,649 @@ +//! Working-memory extraction for skill sync payloads. +//! +//! This module defines what "user working memory" means for skill sync data: +//! durable, user-scoped facts (preferences, goals, constraints, recurring entities) +//! extracted from successful integration sync payloads. +//! +//! Scope model: +//! - **Working memory**: persisted in `global` namespace with deterministic keys. +//! - **Ephemeral chat context**: transient prompt/history data, not persisted here. +//! - **TTL**: none by default (durable), with bounded growth via fixed keys + caps. + +use std::collections::BTreeSet; +use std::sync::OnceLock; + +use regex::Regex; +use serde_json::Value; + +use crate::openhuman::memory::NamespaceDocumentInput; + +const GLOBAL_NAMESPACE: &str = "global"; +const MAX_SCALAR_FIELDS: usize = 1200; +const MAX_ARRAY_ITEMS_PER_NODE: usize = 40; +const MAX_DOC_ITEMS_PER_BUCKET: usize = 12; +const MAX_ITEM_CHARS: usize = 180; + +const KIND_PREFERENCES: &str = "preferences"; +const KIND_GOALS: &str = "goals"; +const KIND_CONSTRAINTS: &str = "constraints"; +const KIND_ENTITIES: &str = "entities"; +const KIND_SUMMARY: &str = "summary"; + +/// Metrics describing extraction quality for a sync batch. +#[derive(Debug, Clone, Default)] +pub(crate) struct WorkingMemoryExtractionStats { + pub scalar_fields_seen: usize, + pub sensitive_fields_skipped: usize, + pub preferences: usize, + pub goals: usize, + pub constraints: usize, + pub entities: usize, + pub documents_generated: usize, +} + +/// Output for a sync batch: deterministic memory documents + extraction stats. +#[derive(Debug, Clone, Default)] +pub(crate) struct WorkingMemoryExtractionOutcome { + pub documents: Vec, + pub stats: WorkingMemoryExtractionStats, +} + +/// Whether skill-sync working-memory extraction is enabled. +/// +/// Product-level allow switch (consent/rollout safety): +/// - `OPENHUMAN_SKILLS_WORKING_MEMORY_ENABLED=0|false|off|no` disables extraction. +/// - Any other value (or unset) enables extraction. +pub(crate) fn skills_working_memory_enabled() -> bool { + static ENABLED: OnceLock = OnceLock::new(); + *ENABLED.get_or_init(|| { + std::env::var("OPENHUMAN_SKILLS_WORKING_MEMORY_ENABLED") + .map(|raw| { + !matches!( + raw.trim().to_ascii_lowercase().as_str(), + "0" | "false" | "off" | "no" + ) + }) + .unwrap_or(true) + }) +} + +#[derive(Debug, Default)] +struct WorkingMemoryBuckets { + preferences: BTreeSet, + goals: BTreeSet, + constraints: BTreeSet, + entities: BTreeSet, +} + +/// Build bounded working-memory documents from a skill sync payload. +/// +/// The generated documents are user-scoped, deterministic, and upsert-friendly. +/// Keys are fixed per `(skill, kind)` to prevent unbounded growth. +pub(crate) fn working_memory_documents_from_sync( + skill_id: &str, + sync_content: &str, +) -> WorkingMemoryExtractionOutcome { + let skill_slug = sanitize_skill_id(skill_id); + let source_namespace = format!("skill-{skill_slug}"); + let mut stats = WorkingMemoryExtractionStats::default(); + + let Ok(payload) = serde_json::from_str::(sync_content) else { + log::debug!( + "[skills-working-memory] non-JSON sync payload for skill='{}'; skipping extraction", + skill_slug + ); + return WorkingMemoryExtractionOutcome { + documents: Vec::new(), + stats, + }; + }; + + let mut scalars = Vec::new(); + collect_scalar_fields( + &payload, + "$", + &mut scalars, + &mut stats.sensitive_fields_skipped, + ); + stats.scalar_fields_seen = scalars.len(); + + let mut buckets = WorkingMemoryBuckets::default(); + for (path, raw_value) in scalars { + let normalized = normalize_text(&raw_value); + if normalized.len() < 3 { + continue; + } + if is_sensitive_value(&normalized) { + stats.sensitive_fields_skipped += 1; + continue; + } + + // Redact PII on the full normalized text first, then clip, so that a + // PII pattern spanning the clip boundary is never partially preserved. + let sanitized = redact_common_pii(&normalized); + let clipped = clip(&sanitized, MAX_ITEM_CHARS); + classify_into_buckets(&path, &clipped, &mut buckets); + } + + cap_set(&mut buckets.preferences, MAX_DOC_ITEMS_PER_BUCKET); + cap_set(&mut buckets.goals, MAX_DOC_ITEMS_PER_BUCKET); + cap_set(&mut buckets.constraints, MAX_DOC_ITEMS_PER_BUCKET); + cap_set(&mut buckets.entities, MAX_DOC_ITEMS_PER_BUCKET); + + stats.preferences = buckets.preferences.len(); + stats.goals = buckets.goals.len(); + stats.constraints = buckets.constraints.len(); + stats.entities = buckets.entities.len(); + + let mut documents = Vec::new(); + if !buckets.preferences.is_empty() { + documents.push(build_doc( + &skill_slug, + &source_namespace, + KIND_PREFERENCES, + "User preferences extracted from skill sync", + &render_bucket(&buckets.preferences), + )); + } + if !buckets.goals.is_empty() { + documents.push(build_doc( + &skill_slug, + &source_namespace, + KIND_GOALS, + "User goals extracted from skill sync", + &render_bucket(&buckets.goals), + )); + } + if !buckets.constraints.is_empty() { + documents.push(build_doc( + &skill_slug, + &source_namespace, + KIND_CONSTRAINTS, + "User constraints extracted from skill sync", + &render_bucket(&buckets.constraints), + )); + } + if !buckets.entities.is_empty() { + documents.push(build_doc( + &skill_slug, + &source_namespace, + KIND_ENTITIES, + "Recurring entities extracted from skill sync", + &render_bucket(&buckets.entities), + )); + } + + if !documents.is_empty() { + documents.push(build_doc( + &skill_slug, + &source_namespace, + KIND_SUMMARY, + "Working memory summary from skill sync", + &render_summary(&buckets), + )); + } + + stats.documents_generated = documents.len(); + + WorkingMemoryExtractionOutcome { documents, stats } +} + +fn collect_scalar_fields( + value: &Value, + path: &str, + out: &mut Vec<(String, String)>, + sensitive_skipped: &mut usize, +) { + if out.len() >= MAX_SCALAR_FIELDS { + return; + } + + match value { + Value::Object(map) => { + for (key, v) in map { + if out.len() >= MAX_SCALAR_FIELDS { + break; + } + if is_sensitive_key(key) { + *sensitive_skipped += 1; + continue; + } + let child_path = if path == "$" { + format!("$.{key}") + } else { + format!("{path}.{key}") + }; + collect_scalar_fields(v, &child_path, out, sensitive_skipped); + } + } + Value::Array(items) => { + for item in items.iter().take(MAX_ARRAY_ITEMS_PER_NODE) { + if out.len() >= MAX_SCALAR_FIELDS { + break; + } + collect_scalar_fields(item, &format!("{path}[]"), out, sensitive_skipped); + } + } + Value::String(s) => out.push((path.to_string(), s.clone())), + Value::Bool(b) => out.push((path.to_string(), b.to_string())), + Value::Number(n) => out.push((path.to_string(), n.to_string())), + Value::Null => {} + } +} + +fn classify_into_buckets(path: &str, value: &str, buckets: &mut WorkingMemoryBuckets) { + let path_l = path.to_ascii_lowercase(); + let value_l = value.to_ascii_lowercase(); + + if looks_like_preference(&path_l, &value_l) { + buckets + .preferences + .insert(format!("{} ({})", value, summarize_path(&path_l))); + } + if looks_like_goal(&path_l, &value_l) { + buckets + .goals + .insert(format!("{} ({})", value, summarize_path(&path_l))); + } + if looks_like_constraint(&path_l, &value_l) { + buckets + .constraints + .insert(format!("{} ({})", value, summarize_path(&path_l))); + } + if looks_like_entity(&path_l, value) { + buckets.entities.insert(value.to_string()); + } +} + +fn render_bucket(items: &BTreeSet) -> String { + items + .iter() + .map(|item| format!("- {item}")) + .collect::>() + .join("\n") +} + +fn render_summary(buckets: &WorkingMemoryBuckets) -> String { + let mut sections = Vec::new(); + if !buckets.preferences.is_empty() { + sections.push(format!( + "Preferences:\n{}", + render_bucket(&buckets.preferences) + )); + } + if !buckets.goals.is_empty() { + sections.push(format!("Goals:\n{}", render_bucket(&buckets.goals))); + } + if !buckets.constraints.is_empty() { + sections.push(format!( + "Constraints:\n{}", + render_bucket(&buckets.constraints) + )); + } + if !buckets.entities.is_empty() { + sections.push(format!( + "Recurring entities:\n{}", + render_bucket(&buckets.entities) + )); + } + sections.join("\n\n") +} + +fn build_doc( + skill_slug: &str, + source_namespace: &str, + kind: &str, + title: &str, + content: &str, +) -> NamespaceDocumentInput { + NamespaceDocumentInput { + namespace: GLOBAL_NAMESPACE.to_string(), + key: format!("working.user.{skill_slug}.{kind}"), + title: format!("{skill_slug} • {kind}"), + content: format!("{title}\n\n{content}"), + source_type: "skill_sync_working_memory".to_string(), + priority: "high".to_string(), + tags: vec![ + "working-memory".to_string(), + "skill-sync".to_string(), + skill_slug.to_string(), + kind.to_string(), + ], + metadata: serde_json::json!({ + "memory_scope": "user_working_memory", + "source": "skills.sync", + "source_namespace": source_namespace, + "skill_id": skill_slug, + "kind": kind, + "ttl": "none", + "version": 1, + }), + category: "core".to_string(), + session_id: None, + document_id: None, + } +} + +fn sanitize_skill_id(skill_id: &str) -> String { + let normalized = skill_id + .trim() + .to_ascii_lowercase() + .chars() + .map(|ch| { + if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' { + ch + } else { + '-' + } + }) + .collect::(); + + if normalized.is_empty() { + "unknown-skill".to_string() + } else { + normalized + } +} + +fn is_sensitive_key(key: &str) -> bool { + let key_l = key.to_ascii_lowercase(); + [ + "token", + "secret", + "password", + "credential", + "oauth", + "auth", + "apikey", + "api_key", + "jwt", + "cookie", + "bearer", + "refresh", + "access", + "clientkey", + "client_key", + ] + .iter() + .any(|needle| key_l.contains(needle)) +} + +fn is_sensitive_value(value: &str) -> bool { + let value_l = value.to_ascii_lowercase(); + if value_l.contains("bearer ") || value_l.contains("api_key") { + return true; + } + // Heuristic for opaque secrets/tokens: a whitespace-separated token of + // 32+ chars where every character is alphanumeric or a common separator + // used in API keys, UUIDs, and similar credentials ('-', '_', '.'). + for token in value.split_whitespace() { + if token.len() < 32 { + continue; + } + if token + .chars() + .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.')) + { + return true; + } + // JWT-like: three dot-delimited segments of base64url-safe characters. + let parts: Vec<&str> = token.split('.').collect(); + if parts.len() == 3 + && parts.iter().all(|seg| { + !seg.is_empty() + && seg + .chars() + .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '=')) + }) + { + return true; + } + } + false +} + +fn looks_like_preference(path: &str, value: &str) -> bool { + [ + "preference", + "pref", + "setting", + "timezone", + "language", + "style", + ] + .iter() + .any(|needle| path.contains(needle)) + || [ + "prefer", "likes", "favorite", "timezone", "language", "style", + ] + .iter() + .any(|needle| value.contains(needle)) +} + +fn looks_like_goal(path: &str, value: &str) -> bool { + [ + "goal", + "objective", + "target", + "milestone", + "roadmap", + "plan", + ] + .iter() + .any(|needle| path.contains(needle)) + || [ + "goal", + "objective", + "target", + "milestone", + "plan", + "ship", + "deliver", + ] + .iter() + .any(|needle| value.contains(needle)) +} + +fn looks_like_constraint(path: &str, value: &str) -> bool { + [ + "constraint", + "limit", + "restriction", + "policy", + "deadline", + "availability", + ] + .iter() + .any(|needle| path.contains(needle)) + || [ + "must", + "cannot", + "can't", + "do not", + "deadline", + "limited", + "restriction", + ] + .iter() + .any(|needle| value.contains(needle)) +} + +fn looks_like_entity(path: &str, value: &str) -> bool { + if value.len() > 80 { + return false; + } + [ + "name", + "title", + "project", + "workspace", + "team", + "customer", + "account", + "label", + ] + .iter() + .any(|needle| path.contains(needle)) +} + +fn summarize_path(path: &str) -> String { + path.trim_start_matches("$.") + .replace("[]", "") + .replace('.', " > ") +} + +fn cap_set(set: &mut BTreeSet, max_len: usize) { + while set.len() > max_len { + let last = set.iter().next_back().cloned(); + if let Some(value) = last { + set.remove(&value); + } else { + break; + } + } +} + +fn normalize_text(input: &str) -> String { + input.split_whitespace().collect::>().join(" ") +} + +fn redact_common_pii(input: &str) -> String { + static EMAIL_RE: OnceLock = OnceLock::new(); + static PHONE_RE: OnceLock = OnceLock::new(); + + let email_re = EMAIL_RE.get_or_init(|| { + Regex::new(r"(?i)\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b") + .expect("email regex should compile") + }); + let phone_re = PHONE_RE + .get_or_init(|| Regex::new(r"\+?\d[\d\-\s]{8,}\d").expect("phone regex should compile")); + let redacted = email_re.replace_all(input, "[redacted-email]"); + let redacted = phone_re.replace_all(&redacted, "[redacted-phone]"); + redacted.to_string() +} + +fn clip(input: &str, max_chars: usize) -> String { + if input.chars().count() <= max_chars { + return input.to_string(); + } + let mut clipped = String::new(); + for ch in input.chars().take(max_chars) { + clipped.push(ch); + } + clipped +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + + use crate::openhuman::memory::{embeddings::NoopEmbedding, UnifiedMemory}; + + #[test] + fn extracts_preferences_goals_constraints_entities_and_redacts_sensitive_fields() { + let payload = serde_json::json!({ + "user": { + "timezone": "America/Los_Angeles", + "email": "alice@example.com", + "api_key": "super-secret-value" + }, + "preferences": { + "writing_style": "prefers concise updates", + "language": "English" + }, + "goals": [ + "Ship onboarding redesign by Friday" + ], + "constraints": [ + "No meetings after 3pm" + ], + "projects": [ + {"name": "Atlas"}, + {"name": "Hermes"} + ] + }) + .to_string(); + + let outcome = working_memory_documents_from_sync("gmail", &payload); + assert!(!outcome.documents.is_empty()); + + let all_content = outcome + .documents + .iter() + .map(|doc| doc.content.clone()) + .collect::>() + .join("\n"); + + assert!(all_content.contains("prefers concise updates")); + assert!(all_content.contains("Ship onboarding redesign by Friday")); + assert!(all_content.contains("No meetings after 3pm")); + assert!(all_content.contains("Atlas")); + assert!(!all_content.contains("super-secret-value")); + assert!(!all_content.contains("alice@example.com")); + assert!(outcome.stats.sensitive_fields_skipped > 0); + assert!(outcome.stats.preferences > 0); + assert!(outcome.stats.goals > 0); + assert!(outcome.stats.constraints > 0); + assert!(outcome.stats.entities > 0); + } + + #[tokio::test] + async fn generated_working_memory_docs_upsert_without_growth() { + let tmp = tempfile::tempdir().unwrap(); + let memory = UnifiedMemory::new(tmp.path(), Arc::new(NoopEmbedding), None).unwrap(); + + let payload_v1 = serde_json::json!({ + "preferences": {"style": "prefers concise responses"}, + "goals": ["Ship v1 this week"], + "projects": [{"name": "Atlas"}] + }) + .to_string(); + let payload_v2 = serde_json::json!({ + "preferences": {"style": "prefers concise responses"}, + "goals": ["Ship v2 next week"], + "projects": [{"name": "Atlas"}, {"name": "Hermes"}] + }) + .to_string(); + + let outcome_v1 = working_memory_documents_from_sync("notion", &payload_v1); + let expected_max_docs = outcome_v1.documents.len(); + for doc in outcome_v1.documents { + memory.upsert_document(doc).await.unwrap(); + } + + let outcome_v2 = working_memory_documents_from_sync("notion", &payload_v2); + for doc in outcome_v2.documents { + memory.upsert_document(doc).await.unwrap(); + } + + let docs = memory.list_documents(Some("global")).await.unwrap(); + let working_docs = docs + .get("documents") + .and_then(Value::as_array) + .unwrap() + .iter() + .filter(|doc| { + doc.get("key") + .and_then(Value::as_str) + .is_some_and(|key| key.starts_with("working.user.notion.")) + }) + .count(); + + assert!( + working_docs <= expected_max_docs, + "working memory docs should not grow unbounded for repeated syncs" + ); + + let ranked = memory + .query_namespace_ranked("global", "Ship v2 next week", 5) + .await + .unwrap(); + assert!( + ranked + .iter() + .any(|hit| hit.content.contains("Ship v2 next week")), + "updated goal should be discoverable in global working memory" + ); + } +} diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index f75d7e422..a3a348954 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -1303,7 +1303,8 @@ fn write_test_skill(workspace: &Path, skill_id: &str) { ) .expect("write manifest"); - // Minimal JS skill that exports one tool: "echo" + // Minimal JS skill that exports one tool: "echo" and a deterministic onSync + // payload so we can assert sync → working-memory extraction end to end. let js = r#" globalThis.__skill = { name: "E2E Runtime Skill", @@ -1331,6 +1332,18 @@ fn write_test_skill(workspace: &Path, skill_id: &str) { } } + async function onSync() { + if (globalThis.state && typeof globalThis.state.set === "function") { + globalThis.state.set("sync_payload", { + preferences: { writing_style: "prefers concise updates", language: "English" }, + goals: ["Ship e2e integration"], + constraints: ["No meetings after 3pm"], + projects: [{ name: "Atlas" }] + }); + } + return { status: "ok", synced: true }; + } + init(); "#; std::fs::write(skill_dir.join("index.js"), js).expect("write index.js"); @@ -1347,6 +1360,9 @@ async fn json_rpc_skills_runtime_start_tools_call_stop() { let _home_guard = EnvVarGuard::set_to_path("HOME", home); let _workspace_guard = EnvVarGuard::set_to_path("OPENHUMAN_WORKSPACE", &workspace); + // Ensure working-memory extraction is not disabled by an ambient env var so + // the assertions below are deterministic regardless of the host environment. + let _wm_guard = EnvVarGuard::unset("OPENHUMAN_SKILLS_WORKING_MEMORY_ENABLED"); // Write a minimal skill to the workspace write_test_skill(&workspace, "e2e-runtime"); @@ -1486,10 +1502,59 @@ async fn json_rpc_skills_runtime_start_tools_call_stop() { json!({"skill_id": "e2e-runtime"}), ) .await; - // 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"); + // 5a. Poll until the async memory worker has written working-memory docs into + // the global namespace, instead of relying on a fixed sleep. + let poll_deadline = tokio::time::Instant::now() + Duration::from_secs(10); + let (docs_result, docs_arr) = loop { + let docs = post_json_rpc( + &rpc_base, + 241, + "openhuman.memory_list_documents", + json!({"namespace":"global"}), + ) + .await; + let arr = { + let result = assert_no_jsonrpc_error(&docs, "memory_list_documents"); + result + .get("documents") + .or_else(|| result.get("data").and_then(|d| d.get("documents"))) + .and_then(Value::as_array) + .cloned() + .unwrap_or_default() + }; + let has_summary = arr.iter().any(|doc| { + doc.get("key").and_then(Value::as_str) == Some("working.user.e2e-runtime.summary") + }); + if has_summary { + break (docs, arr); + } + assert!( + tokio::time::Instant::now() < poll_deadline, + "Timeout waiting for working.user.e2e-runtime.summary to appear. docs={docs}" + ); + tokio::time::sleep(Duration::from_millis(200)).await; + }; + + let wm_keys: Vec = docs_arr + .iter() + .filter_map(|doc| doc.get("key").and_then(Value::as_str)) + .filter(|key| key.starts_with("working.user.e2e-runtime.")) + .map(ToString::to_string) + .collect(); + + assert!( + !wm_keys.is_empty(), + "Expected working memory docs after skills_sync, found none. docs={docs_result}" + ); + assert!( + wm_keys + .iter() + .any(|key| key == "working.user.e2e-runtime.summary"), + "Expected summary working-memory key. keys={wm_keys:?}" + ); + // 6. Stop the skill let stop = post_json_rpc( &rpc_base,