//! libSQL/Turso-backed control-plane store. //! //! This implementation intentionally mirrors the SQLite store while keeping the //! public ControlPlaneStore trait synchronous for the rest of mnote-web. use std::collections::BTreeMap; use std::future::Future; use std::path::PathBuf; use std::sync::{Arc, Mutex, MutexGuard}; use std::time::Duration; use libsql::{Builder, Connection, Database, Value}; use crate::error::ControlPlaneError; use crate::migrations; use crate::model::{ password_hash_v1, share_token_hash_v1, AiAgentProfileAccessRecord, AiAgentProfileGrantRecord, AiAgentProfileRecord, AiExternalConversationBindingRecord, AiFilePatchRecord, AiPolicyRecord, AiRuntimeEventRecord, AiRuntimeJournalEventRecord, AiRuntimeRunRecord, AiToolEventRecord, AppendAiFilePatchInput, AppendAiRuntimeEventInput, AppendAiToolEventInput, AppendAuditInput, AuditLogRecord, AuthSessionRecord, AuthenticatePasswordInput, CreatePasswordIdentityInput, CreateSessionInput, CreateShareLinkInput, CreatedShareLink, DirectoryGrantInput, DirectoryGrantLookup, DirectoryGrantRecord, NavigationRecentRecord, OutboxEventInput, OutboxEventRecord, ResolvedAccess, ResolvedAuthSession, ShareLinkRecord, SidebarShortcutRecord, SyncStateRecord, UpsertAiExternalConversationBindingInput, UpsertAiPolicyInput, UpsertAiRuntimeRunInput, UpsertNavigationRecentInput, UpsertSidebarShortcutInput, UpsertSyncStateInput, UpsertUserInput, UpsertUserUiPreferenceInput, UpsertWorkspaceInput, UserRecord, UserUiPreferenceRecord, WorkspaceRecord, }; use crate::store::ControlPlaneStore; #[derive(Debug, Clone)] pub enum TursoControlPlaneMode { Local, Remote, RemoteReplica, Synced, } #[derive(Debug, Clone)] pub struct TursoControlPlaneConfig { pub mode: TursoControlPlaneMode, pub path: Option, pub remote_url: Option, pub auth_token: Option, pub sync_interval: Option, } pub struct TursoControlPlaneStore { conn: Mutex, _database: Database, } #[derive(Clone)] struct TursoConnection { conn: Connection, runtime: Arc, } struct TursoStatement { conn: TursoConnection, sql: String, } struct TursoMappedRows { rows: std::vec::IntoIter>, } struct TursoTransaction { conn: TursoConnection, committed: bool, } trait OptionalExtension { fn optional(self) -> libsql::Result>; } impl OptionalExtension for libsql::Result { fn optional(self) -> libsql::Result> { match self { Ok(value) => Ok(Some(value)), Err(libsql::Error::QueryReturnedNoRows) => Ok(None), Err(error) => Err(error), } } } trait ToTursoValue { fn to_turso_value(self) -> Value; } fn to_turso_value(value: T) -> Value { value.to_turso_value() } macro_rules! params { ($($value:expr),* $(,)?) => { vec![$(to_turso_value(&$value)),*] }; } trait IntoTursoParams { fn into_turso_values(self) -> libsql::Result>; } impl IntoTursoParams for Vec { fn into_turso_values(self) -> libsql::Result> { Ok(self) } } impl IntoTursoParams for [T; N] { fn into_turso_values(self) -> libsql::Result> { Ok(self.into_iter().map(to_turso_value).collect()) } } impl IntoTursoParams for () { fn into_turso_values(self) -> libsql::Result> { Ok(Vec::new()) } } impl ToTursoValue for () { fn to_turso_value(self) -> Value { Value::Null } } impl ToTursoValue for Value { fn to_turso_value(self) -> Value { self } } impl ToTursoValue for &Value { fn to_turso_value(self) -> Value { self.clone() } } impl ToTursoValue for &&Value { fn to_turso_value(self) -> Value { (*self).clone() } } impl ToTursoValue for String { fn to_turso_value(self) -> Value { Value::Text(self) } } impl ToTursoValue for &String { fn to_turso_value(self) -> Value { Value::Text(self.clone()) } } impl ToTursoValue for &&String { fn to_turso_value(self) -> Value { Value::Text((*self).clone()) } } impl ToTursoValue for &str { fn to_turso_value(self) -> Value { Value::Text(self.to_string()) } } impl ToTursoValue for &&str { fn to_turso_value(self) -> Value { Value::Text((*self).to_string()) } } impl ToTursoValue for &&&str { fn to_turso_value(self) -> Value { Value::Text((**self).to_string()) } } impl ToTursoValue for Option { fn to_turso_value(self) -> Value { self.map(Value::Text).unwrap_or(Value::Null) } } impl ToTursoValue for &Option { fn to_turso_value(self) -> Value { self.clone().map(Value::Text).unwrap_or(Value::Null) } } impl ToTursoValue for &&Option { fn to_turso_value(self) -> Value { (*self).clone().map(Value::Text).unwrap_or(Value::Null) } } impl ToTursoValue for Option<&str> { fn to_turso_value(self) -> Value { self.map(|value| Value::Text(value.to_string())) .unwrap_or(Value::Null) } } impl ToTursoValue for &Option<&str> { fn to_turso_value(self) -> Value { self.map(|value| Value::Text(value.to_string())) .unwrap_or(Value::Null) } } impl ToTursoValue for Option<&String> { fn to_turso_value(self) -> Value { self.cloned().map(Value::Text).unwrap_or(Value::Null) } } impl ToTursoValue for &Option<&String> { fn to_turso_value(self) -> Value { self.cloned().map(Value::Text).unwrap_or(Value::Null) } } impl ToTursoValue for i64 { fn to_turso_value(self) -> Value { Value::Integer(self) } } impl ToTursoValue for &i64 { fn to_turso_value(self) -> Value { Value::Integer(*self) } } impl ToTursoValue for i32 { fn to_turso_value(self) -> Value { Value::Integer(self as i64) } } impl ToTursoValue for &i32 { fn to_turso_value(self) -> Value { Value::Integer(*self as i64) } } impl ToTursoValue for usize { fn to_turso_value(self) -> Value { Value::Integer(self as i64) } } impl ToTursoValue for &usize { fn to_turso_value(self) -> Value { Value::Integer(*self as i64) } } impl ToTursoValue for u64 { fn to_turso_value(self) -> Value { Value::Integer(self as i64) } } impl ToTursoValue for &u64 { fn to_turso_value(self) -> Value { Value::Integer(*self as i64) } } impl ToTursoValue for bool { fn to_turso_value(self) -> Value { Value::Integer(if self { 1 } else { 0 }) } } impl ToTursoValue for &bool { fn to_turso_value(self) -> Value { Value::Integer(if *self { 1 } else { 0 }) } } impl TursoConnection { fn new(conn: Connection, runtime: Arc) -> Self { Self { conn, runtime } } fn execute

(&self, sql: &str, params: P) -> libsql::Result where P: IntoTursoParams, { let values = params.into_turso_values()?; block_on_runtime(self.runtime.as_ref(), async { self.conn .execute(sql, values) .await .map(|changed| changed as usize) }) } fn execute_batch(&self, sql: &str) -> libsql::Result<()> { block_on_runtime(self.runtime.as_ref(), async { self.conn.execute_batch(sql).await.map(|_| ()) }) } fn prepare(&self, sql: &str) -> libsql::Result { Ok(TursoStatement { conn: self.clone(), sql: sql.to_string(), }) } fn query_row(&self, sql: &str, params: P, mapper: F) -> libsql::Result where P: IntoTursoParams, F: FnOnce(&libsql::Row) -> libsql::Result, { let values = params.into_turso_values()?; block_on_runtime(self.runtime.as_ref(), async { let mut rows = self.conn.query(sql, values).await?; let Some(row) = rows.next().await? else { return Err(libsql::Error::QueryReturnedNoRows); }; mapper(&row) }) } fn transaction(&self) -> libsql::Result { block_on_runtime(self.runtime.as_ref(), async { self.conn .execute("BEGIN IMMEDIATE", ()) .await .map(|_| TursoTransaction { conn: self.clone(), committed: false, }) }) } } impl TursoTransaction { fn execute

(&self, sql: &str, params: P) -> libsql::Result where P: IntoTursoParams, { self.conn.execute(sql, params) } fn query_row(&self, sql: &str, params: P, mapper: F) -> libsql::Result where P: IntoTursoParams, F: FnOnce(&libsql::Row) -> libsql::Result, { self.conn.query_row(sql, params, mapper) } fn commit(mut self) -> libsql::Result<()> { self.conn.execute_batch("COMMIT")?; self.committed = true; Ok(()) } } impl Drop for TursoTransaction { fn drop(&mut self) { if !self.committed { let _ = self.conn.execute_batch("ROLLBACK"); } } } impl Iterator for TursoMappedRows { type Item = libsql::Result; fn next(&mut self) -> Option { self.rows.next() } } impl TursoStatement { fn query_map(&mut self, params: P, mut mapper: F) -> libsql::Result> where P: IntoTursoParams, F: FnMut(&libsql::Row) -> libsql::Result, { let values = params.into_turso_values()?; let rows = block_on_runtime(self.conn.runtime.as_ref(), async { let mut rows = self.conn.conn.query(&self.sql, values).await?; let mut mapped = Vec::new(); while let Some(row) = rows.next().await? { mapped.push(mapper(&row)); } Ok::<_, libsql::Error>(mapped) })?; Ok(TursoMappedRows { rows: rows.into_iter(), }) } } impl TursoControlPlaneStore { pub fn open(path: impl AsRef) -> Result { Self::open_local(path) } pub fn open_local(path: impl AsRef) -> Result { Self::open_with_config(TursoControlPlaneConfig { mode: TursoControlPlaneMode::Local, path: Some(path.as_ref().to_path_buf()), remote_url: None, auth_token: None, sync_interval: None, }) } pub fn in_memory() -> Result { Self::open_local(":memory:") } pub fn open_remote(url: String, auth_token: String) -> Result { Self::open_with_config(TursoControlPlaneConfig { mode: TursoControlPlaneMode::Remote, path: None, remote_url: Some(url), auth_token: Some(auth_token), sync_interval: None, }) } pub fn open_remote_replica( path: impl AsRef, url: String, auth_token: String, sync_interval: Option, ) -> Result { Self::open_with_config(TursoControlPlaneConfig { mode: TursoControlPlaneMode::RemoteReplica, path: Some(path.as_ref().to_path_buf()), remote_url: Some(url), auth_token: Some(auth_token), sync_interval, }) } pub fn open_synced( path: impl AsRef, url: String, auth_token: String, sync_interval: Option, ) -> Result { Self::open_with_config(TursoControlPlaneConfig { mode: TursoControlPlaneMode::Synced, path: Some(path.as_ref().to_path_buf()), remote_url: Some(url), auth_token: Some(auth_token), sync_interval, }) } pub fn open_with_config(config: TursoControlPlaneConfig) -> Result { let runtime = Arc::new( tokio::runtime::Builder::new_multi_thread() .enable_all() .build() .map_err(|error| ControlPlaneError::Storage(format!("libSQL runtime: {error}")))?, ); let mode = config.mode.clone(); let database = block_on_runtime(runtime.as_ref(), async { build_database(config).await })?; let raw_conn = database.connect()?; let conn = TursoConnection::new(raw_conn.clone(), runtime); configure_connection(&conn, &mode)?; block_on_runtime( conn.runtime.as_ref(), migrations::run_libsql_migrations(&raw_conn), )?; Ok(Self { conn: Mutex::new(conn), _database: database, }) } fn lock_conn(&self) -> Result, ControlPlaneError> { self.conn.lock().map_err(|error| { ControlPlaneError::Storage(format!("libSQL control-plane lock poisoned: {error}")) }) } } fn block_on_runtime(runtime: &tokio::runtime::Runtime, future: F) -> F::Output where F: Future, { match tokio::runtime::Handle::try_current() { Ok(handle) if matches!( handle.runtime_flavor(), tokio::runtime::RuntimeFlavor::MultiThread ) => { tokio::task::block_in_place(|| runtime.block_on(future)) } _ => runtime.block_on(future), } } async fn build_database(config: TursoControlPlaneConfig) -> Result { match config.mode { TursoControlPlaneMode::Local => { let path = config .path .ok_or_else(|| ControlPlaneError::InvalidInput("缺少 libSQL local path".into()))?; Builder::new_local(path) .build() .await .map_err(ControlPlaneError::from) } TursoControlPlaneMode::Remote => { let url = config .remote_url .filter(|value| !value.trim().is_empty()) .ok_or_else(|| ControlPlaneError::InvalidInput("缺少 Turso URL".into()))?; let token = config .auth_token .filter(|value| !value.trim().is_empty()) .ok_or_else(|| ControlPlaneError::InvalidInput("缺少 Turso auth token".into()))?; Builder::new_remote(url, token) .build() .await .map_err(ControlPlaneError::from) } TursoControlPlaneMode::RemoteReplica => { let path = config.path.ok_or_else(|| { ControlPlaneError::InvalidInput("缺少 Turso local replica path".into()) })?; let url = config .remote_url .filter(|value| !value.trim().is_empty()) .ok_or_else(|| ControlPlaneError::InvalidInput("缺少 Turso URL".into()))?; let token = config .auth_token .filter(|value| !value.trim().is_empty()) .ok_or_else(|| ControlPlaneError::InvalidInput("缺少 Turso auth token".into()))?; let mut builder = Builder::new_remote_replica(path, url, token); if let Some(interval) = config.sync_interval { builder = builder.sync_interval(interval); } builder.build().await.map_err(ControlPlaneError::from) } TursoControlPlaneMode::Synced => { let path = config .path .ok_or_else(|| ControlPlaneError::InvalidInput("缺少 Turso synced path".into()))?; let url = config .remote_url .filter(|value| !value.trim().is_empty()) .ok_or_else(|| ControlPlaneError::InvalidInput("缺少 Turso URL".into()))?; let token = config .auth_token .filter(|value| !value.trim().is_empty()) .ok_or_else(|| ControlPlaneError::InvalidInput("缺少 Turso auth token".into()))?; let mut builder = Builder::new_synced_database(path, url, token); if let Some(interval) = config.sync_interval { builder = builder.sync_interval(interval); } builder.build().await.map_err(ControlPlaneError::from) } } } fn configure_connection( conn: &TursoConnection, mode: &TursoControlPlaneMode, ) -> Result<(), ControlPlaneError> { match mode { TursoControlPlaneMode::Local | TursoControlPlaneMode::RemoteReplica | TursoControlPlaneMode::Synced => conn.execute_batch( "PRAGMA journal_mode = WAL; PRAGMA synchronous = NORMAL; PRAGMA foreign_keys = ON; PRAGMA busy_timeout = 5000;", )?, TursoControlPlaneMode::Remote => { conn.execute_batch("PRAGMA foreign_keys = ON;")?; } } Ok(()) } fn now_text() -> String { chrono::Utc::now().to_rfc3339() } fn new_id(prefix: &str) -> String { format!("{prefix}_{}", uuid::Uuid::new_v4().simple()) } fn capabilities_json(capabilities: &[String]) -> Result { serde_json::to_string(capabilities).map_err(ControlPlaneError::from) } fn file_uri_to_legacy_path(value: &str) -> String { value .trim() .strip_prefix("file://") .unwrap_or("") .to_string() } fn row_to_user(row: &libsql::Row) -> libsql::Result { Ok(UserRecord { id: row.get(0)?, email: row.get(1)?, username: row.get(2)?, display_name: row.get(3)?, role: row.get(4)?, status: row.get(5)?, created_at: row.get(6)?, updated_at: row.get(7)?, revision: row.get(8)?, }) } fn row_to_auth_session(row: &libsql::Row) -> libsql::Result { Ok(AuthSessionRecord { id: row.get(0)?, user_id: row.get(1)?, token_hash: row.get(2)?, user_agent: row.get(3)?, ip_hash: row.get(4)?, created_at: row.get(5)?, expires_at: row.get(6)?, revoked_at: row.get(7)?, last_seen_at: row.get(8)?, }) } fn row_to_workspace(row: &libsql::Row) -> libsql::Result { Ok(WorkspaceRecord { id: row.get(0)?, owner_user_id: row.get(1)?, name: row.get(2)?, kind: row.get(3)?, root_uri: row.get(4)?, root_path: row.get(5)?, source_kind: row.get(6)?, status: row.get(7)?, created_at: row.get(8)?, updated_at: row.get(9)?, revision: row.get(10)?, }) } fn row_to_grant(row: &libsql::Row) -> libsql::Result { Ok(DirectoryGrantRecord { id: row.get(0)?, user_id: row.get(1)?, workspace_id: row.get(2)?, root_uri: row.get(3)?, root_path: row.get(4)?, permission: row.get(5)?, recursive: row.get::(6)? != 0, capabilities_json: row.get(7)?, source: row.get(8)?, status: row.get(9)?, created_by: row.get(10)?, created_at: row.get(11)?, updated_at: row.get(12)?, revision: row.get(13)?, }) } fn row_to_outbox(row: &libsql::Row) -> libsql::Result { Ok(OutboxEventRecord { id: row.get(0)?, topic: row.get(1)?, event_type: row.get(2)?, payload_json: row.get(3)?, created_at: row.get(4)?, delivered_at: row.get(5)?, attempts: row.get(6)?, }) } fn row_to_share_link(row: &libsql::Row) -> libsql::Result { Ok(ShareLinkRecord { id: row.get(0)?, workspace_id: row.get(1)?, resource_kind: row.get(2)?, resource_id: row.get(3)?, token_hash: row.get(4)?, permission: row.get(5)?, created_by: row.get(6)?, expires_at: row.get(7)?, revoked_at: row.get(8)?, created_at: row.get(9)?, updated_at: row.get(10)?, revision: row.get(11)?, }) } fn row_to_audit_log(row: &libsql::Row) -> libsql::Result { Ok(AuditLogRecord { id: row.get(0)?, actor_user_id: row.get(1)?, action: row.get(2)?, target_kind: row.get(3)?, target_id: row.get(4)?, metadata_json: row.get(5)?, created_at: row.get(6)?, }) } fn row_to_sidebar_shortcut(row: &libsql::Row) -> libsql::Result { Ok(SidebarShortcutRecord { id: row.get(0)?, user_id: row.get(1)?, workspace_id: row.get(2)?, root_uri: row.get(3)?, kind: row.get(4)?, source_kind: row.get(5)?, target_id: row.get(6)?, relative_path: row.get(7)?, document_id: row.get(8)?, title: row.get(9)?, icon: row.get(10)?, sort_order: row.get(11)?, status: row.get(12)?, metadata_json: row.get(13)?, created_at: row.get(14)?, updated_at: row.get(15)?, revision: row.get(16)?, }) } fn empty_string_to_option(value: String) -> Option { if value.trim().is_empty() { None } else { Some(value) } } fn option_to_stored_text(value: Option) -> String { value .map(|item| item.trim().to_string()) .filter(|item| !item.is_empty()) .unwrap_or_default() } fn sanitize_profile_component(value: &str) -> String { let mut sanitized = value .trim() .chars() .map(|ch| { if ch.is_ascii_alphanumeric() { ch.to_ascii_lowercase() } else { '-' } }) .collect::(); while sanitized.contains("--") { sanitized = sanitized.replace("--", "-"); } sanitized = sanitized.trim_matches('-').to_string(); if sanitized.is_empty() { "user".to_string() } else { sanitized } } fn row_to_ai_agent_profile(row: &libsql::Row) -> libsql::Result { let owner_user_id: String = row.get(3)?; Ok(AiAgentProfileRecord { id: row.get(0)?, agent_id: row.get(1)?, profile_kind: row.get(2)?, owner_user_id: empty_string_to_option(owner_user_id), base_profile_name: row.get(4)?, isolated_profile_name: row.get(5)?, display_name: row.get(6)?, status: row.get(7)?, created_at: row.get(8)?, updated_at: row.get(9)?, revision: row.get(10)?, }) } fn row_to_ai_agent_profile_access(row: &libsql::Row) -> libsql::Result { Ok(AiAgentProfileAccessRecord { profile: row_to_ai_agent_profile(row)?, grant: AiAgentProfileGrantRecord { id: row.get(11)?, profile_id: row.get(12)?, user_id: empty_string_to_option(row.get(13)?), role: row.get(14)?, can_run: row.get::(15)? != 0, can_manage_skills: row.get::(16)? != 0, can_manage_config: row.get::(17)? != 0, created_at: row.get(18)?, updated_at: row.get(19)?, revision: row.get(20)?, }, }) } fn row_to_user_ui_preference(row: &libsql::Row) -> libsql::Result { Ok(UserUiPreferenceRecord { id: row.get(0)?, user_id: row.get(1)?, workspace_id: empty_string_to_option(row.get(2)?), source_kind: empty_string_to_option(row.get(3)?), scope_kind: row.get(4)?, scope_id: row.get(5)?, key: row.get(6)?, value_json: row.get(7)?, status: row.get(8)?, created_at: row.get(9)?, updated_at: row.get(10)?, revision: row.get(11)?, }) } fn row_to_navigation_recent(row: &libsql::Row) -> libsql::Result { Ok(NavigationRecentRecord { id: row.get(0)?, user_id: row.get(1)?, workspace_id: empty_string_to_option(row.get(2)?), kind: row.get(3)?, source_kind: row.get(4)?, root_uri: row.get(5)?, target_key: row.get(6)?, relative_path: row.get(7)?, document_id: row.get(8)?, title: row.get(9)?, status: row.get(10)?, metadata_json: row.get(11)?, visited_at: row.get(12)?, created_at: row.get(13)?, updated_at: row.get(14)?, revision: row.get(15)?, }) } fn navigation_recent_target_key( kind: &str, root_uri: &str, relative_path: Option<&str>, document_id: Option<&str>, ) -> Result { match kind { "folder" => Ok(format!( "folder:{}:{}", root_uri.trim(), relative_path.map(str::trim).unwrap_or_default() )), "page" => { let document_id = document_id .map(str::trim) .filter(|value| !value.is_empty()) .ok_or_else(|| { ControlPlaneError::InvalidInput( "navigation recent page 需要 document_id".to_string(), ) })?; Ok(format!("page:{}:{document_id}", root_uri.trim())) } _ => Err(ControlPlaneError::InvalidInput( "navigation recent kind 只能是 folder 或 page".to_string(), )), } } fn validate_ai_external_conversation_status(status: &str) -> Result<(), ControlPlaneError> { match status { "active" | "local_deleted" | "remote_deleted" | "remote_delete_failed" | "remote_missing" => Ok(()), _ => Err(ControlPlaneError::InvalidInput( "external conversation status 不合法".to_string(), )), } } fn row_to_ai_policy(row: &libsql::Row) -> libsql::Result { Ok(AiPolicyRecord { id: row.get(0)?, user_id: row.get(1)?, workspace_id: row.get(2)?, allowed_roots_json: row.get(3)?, model_policy_json: row.get(4)?, quota_json: row.get(5)?, created_at: row.get(6)?, updated_at: row.get(7)?, revision: row.get(8)?, }) } fn row_to_sync_state(row: &libsql::Row) -> libsql::Result { Ok(SyncStateRecord { id: row.get(0)?, workspace_id: row.get(1)?, remote_kind: row.get(2)?, remote_id: row.get(3)?, cursor: row.get(4)?, last_synced_at: row.get(5)?, status: row.get(6)?, error_json: row.get(7)?, updated_at: row.get(8)?, revision: row.get(9)?, }) } fn row_to_ai_runtime_run(row: &libsql::Row) -> libsql::Result { Ok(AiRuntimeRunRecord { id: row.get(0)?, user_id: row.get(1)?, workspace_id: row.get(2)?, document_id: row.get(3)?, session_id: row.get(4)?, run_id: row.get(5)?, title: row.get(6)?, profile: row.get(7)?, acp_runtime: row.get(8)?, trace_id: row.get(9)?, status: row.get(10)?, runtime_json: row.get(11)?, payload_json: row.get(12)?, deleted_at: row.get(13)?, created_at: row.get(14)?, updated_at: row.get(15)?, revision: row.get(16)?, }) } fn row_to_ai_runtime_event(row: &libsql::Row) -> libsql::Result { Ok(AiRuntimeEventRecord { id: row.get(0)?, user_id: row.get(1)?, workspace_id: row.get(2)?, document_id: row.get(3)?, session_id: row.get(4)?, run_id: row.get(5)?, profile: row.get(6)?, acp_runtime: row.get(7)?, event_type: row.get(8)?, payload_json: row.get(9)?, created_at: row.get(10)?, }) } fn row_to_ai_runtime_journal_event( row: &libsql::Row, ) -> libsql::Result { Ok(AiRuntimeJournalEventRecord { seq: row.get(0)?, event: AiRuntimeEventRecord { id: row.get(1)?, user_id: row.get(2)?, workspace_id: row.get(3)?, document_id: row.get(4)?, session_id: row.get(5)?, run_id: row.get(6)?, profile: row.get(7)?, acp_runtime: row.get(8)?, event_type: row.get(9)?, payload_json: row.get(10)?, created_at: row.get(11)?, }, }) } fn row_to_ai_external_conversation_binding( row: &libsql::Row, ) -> libsql::Result { Ok(AiExternalConversationBindingRecord { id: row.get(0)?, user_id: row.get(1)?, workspace_id: row.get(2)?, mnote_session_id: row.get(3)?, acp_session_id: row.get(4)?, agent_id: row.get(5)?, profile: row.get(6)?, provider: row.get(7)?, remote_conversation_id: row.get(8)?, remote_url: row.get(9)?, status: row.get(10)?, metadata_json: row.get(11)?, created_at: row.get(12)?, updated_at: row.get(13)?, deleted_at: row.get(14)?, revision: row.get(15)?, }) } fn row_to_ai_tool_event(row: &libsql::Row) -> libsql::Result { Ok(AiToolEventRecord { id: row.get(0)?, user_id: row.get(1)?, workspace_id: row.get(2)?, session_id: row.get(3)?, run_id: row.get(4)?, provider: row.get(5)?, provider_session_id: row.get(6)?, tool_name: row.get(7)?, allowed: row.get(8)?, deny_reason: row.get(9)?, root_uri: row.get(10)?, page_path: row.get(11)?, normalized_file_path: row.get(12)?, diff_summary: row.get(13)?, citation_count: row.get(14)?, before_file_version: row.get(15)?, after_file_version: row.get(16)?, payload_json: row.get(17)?, created_at: row.get(18)?, deleted_at: row.get(19)?, }) } fn row_to_ai_file_patch(row: &libsql::Row) -> libsql::Result { Ok(AiFilePatchRecord { id: row.get(0)?, user_id: row.get(1)?, workspace_id: row.get(2)?, session_id: row.get(3)?, run_id: row.get(4)?, tool_event_id: row.get(5)?, root_uri: row.get(6)?, relative_path: row.get(7)?, before_file_version: row.get(8)?, after_file_version: row.get(9)?, patch_summary_json: row.get(10)?, created_at: row.get(11)?, deleted_at: row.get(12)?, }) } fn derive_ai_runtime_title(payload_json: &str, fallback: &str) -> String { let title = serde_json::from_str::(payload_json) .ok() .and_then(|payload| { payload .get("title") .or_else(|| payload.get("message")) .and_then(serde_json::Value::as_str) .map(str::trim) .filter(|value| !value.is_empty()) .map(ToOwned::to_owned) }) .unwrap_or_else(|| fallback.to_string()); title.chars().take(80).collect() } fn prune_navigation_recent_for_kind( conn: &TursoConnection, user_id: &str, kind: &str, ) -> Result<(), ControlPlaneError> { let keep = match kind { "folder" => 10_i64, "page" => 20_i64, _ => 20_i64, }; conn.execute( "UPDATE user_navigation_recent SET status = 'removed', updated_at = ?1, revision = revision + 1 WHERE user_id = ?2 AND kind = ?3 AND status = 'active' AND id NOT IN ( SELECT id FROM user_navigation_recent WHERE user_id = ?2 AND kind = ?3 AND status = 'active' ORDER BY visited_at DESC, updated_at DESC LIMIT ?4 )", params![now_text(), user_id, kind, keep], )?; Ok(()) } fn list_ai_agent_profile_access_rows( conn: &TursoConnection, user_id: &str, ) -> Result, ControlPlaneError> { let mut stmt = conn.prepare( "SELECT p.id, p.agent_id, p.profile_kind, p.owner_user_id, p.base_profile_name, p.isolated_profile_name, p.display_name, p.status, p.created_at, p.updated_at, p.revision, g.id, g.profile_id, g.user_id, g.role, g.can_run, g.can_manage_skills, g.can_manage_config, g.created_at, g.updated_at, g.revision FROM ai_agent_profiles p JOIN ai_agent_profile_grants g ON g.profile_id = p.id WHERE p.agent_id = 'hermes' AND p.status = 'active' AND g.can_run = 1 AND ( (p.profile_kind = 'personal' AND p.owner_user_id = ?1 AND g.user_id = ?1) OR (p.profile_kind = 'shared' AND (g.user_id = '' OR g.user_id = ?1)) ) ORDER BY CASE p.profile_kind WHEN 'personal' THEN 0 ELSE 1 END, p.display_name ASC, g.can_manage_skills DESC, g.can_manage_config DESC", )?; let rows = stmt .query_map(params![user_id], row_to_ai_agent_profile_access)? .collect::, _>>() .map_err(ControlPlaneError::from)?; let mut by_profile = BTreeMap::::new(); for row in rows { by_profile .entry(row.profile.id.clone()) .and_modify(|existing| { if row.grant.can_manage_skills && !existing.grant.can_manage_skills { *existing = row.clone(); } }) .or_insert(row); } Ok(by_profile.into_values().collect()) } impl ControlPlaneStore for TursoControlPlaneStore { fn upsert_user(&self, input: UpsertUserInput) -> Result { if input.username.trim().is_empty() { return Err(ControlPlaneError::InvalidInput( "username 不能为空".to_string(), )); } if input.display_name.trim().is_empty() { return Err(ControlPlaneError::InvalidInput( "display_name 不能为空".to_string(), )); } let conn = self.lock_conn()?; let now = now_text(); let id = input.id.unwrap_or_else(|| new_id("usr")); let role = input.role.unwrap_or_else(|| "user".to_string()); let existing = conn .query_row( "SELECT id, email, username, display_name, role, status, created_at, updated_at, revision FROM users WHERE id = ?1", params![id], row_to_user, ) .optional()?; if let Some(existing) = existing { let revision = existing.revision + 1; conn.execute( "UPDATE users SET email = ?1, username = ?2, display_name = ?3, role = ?4, updated_at = ?5, revision = ?6 WHERE id = ?7", params![ input.email, input.username, input.display_name, role, now, revision, existing.id ], )?; return Ok(UserRecord { id: existing.id, email: input.email, username: input.username, display_name: input.display_name, role, status: existing.status, created_at: existing.created_at, updated_at: now, revision, }); } conn.execute( "INSERT INTO users (id, email, username, display_name, role, status, created_at, updated_at, revision) VALUES (?1, ?2, ?3, ?4, ?5, 'active', ?6, ?7, 1)", params![id, input.email, input.username, input.display_name, role, now, now], )?; Ok(UserRecord { id, email: input.email, username: input.username, display_name: input.display_name, role, status: "active".to_string(), created_at: now.clone(), updated_at: now, revision: 1, }) } fn list_users(&self, limit: usize) -> Result, ControlPlaneError> { let conn = self.lock_conn()?; let limit = limit.min(1000).max(1); let mut stmt = conn.prepare( "SELECT id, email, username, display_name, role, status, created_at, updated_at, revision FROM users ORDER BY created_at ASC, id ASC LIMIT ?1", )?; let rows = stmt .query_map(params![limit as i64], row_to_user)? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(rows) } fn create_password_identity( &self, input: CreatePasswordIdentityInput, ) -> Result<(), ControlPlaneError> { let username = input.username.trim(); if username.is_empty() { return Err(ControlPlaneError::InvalidInput( "username 不能为空".to_string(), )); } if input.password.is_empty() { return Err(ControlPlaneError::InvalidInput( "password 不能为空".to_string(), )); } let conn = self.lock_conn()?; let now = now_text(); let user_id = input.user_id; let password_hash = password_hash_v1(&input.password); conn.execute( "INSERT INTO auth_identities (id, user_id, provider, provider_subject, password_hash, password_version, created_at, updated_at) VALUES (?1, ?2, 'password_username', ?3, ?4, 1, ?5, ?6) ON CONFLICT(provider, provider_subject) DO UPDATE SET user_id = excluded.user_id, password_hash = excluded.password_hash, updated_at = excluded.updated_at", params![ new_id("ident"), user_id, username, password_hash, now, now ], )?; if let Some(email) = input .email .map(|value| value.trim().to_string()) .filter(|value| !value.is_empty()) { conn.execute( "INSERT INTO auth_identities (id, user_id, provider, provider_subject, password_hash, password_version, created_at, updated_at) VALUES (?1, ?2, 'password_email', ?3, ?4, 1, ?5, ?6) ON CONFLICT(provider, provider_subject) DO UPDATE SET user_id = excluded.user_id, password_hash = excluded.password_hash, updated_at = excluded.updated_at", params![ new_id("ident"), user_id, email, password_hash_v1(&input.password), now, now ], )?; } Ok(()) } fn authenticate_password( &self, input: AuthenticatePasswordInput, ) -> Result { let account = input.account.trim(); if account.is_empty() { return Err(ControlPlaneError::InvalidInput( "account 不能为空".to_string(), )); } let conn = self.lock_conn()?; let provider = if account.contains('@') { "password_email" } else { "password_username" }; let expected_hash = password_hash_v1(&input.password); let user = conn .query_row( "SELECT u.id, u.email, u.username, u.display_name, u.role, u.status, u.created_at, u.updated_at, u.revision FROM auth_identities i JOIN users u ON u.id = i.user_id WHERE i.provider = ?1 AND i.provider_subject = ?2 AND i.password_hash = ?3 AND u.status = 'active' LIMIT 1", params![provider, account, expected_hash], row_to_user, ) .optional()? .ok_or_else(|| ControlPlaneError::Unauthorized("账号或密码错误".to_string()))?; drop(conn); let session = self.create_session(CreateSessionInput { id: input.session_id, user_id: user.id.clone(), token_hash: input.token_hash, user_agent: input.user_agent, ip_hash: input.ip_hash, expires_at: input.expires_at, })?; Ok(ResolvedAuthSession { session, user }) } fn create_session( &self, input: CreateSessionInput, ) -> Result { let token_hash = input.token_hash.trim(); if token_hash.is_empty() { return Err(ControlPlaneError::InvalidInput( "token_hash 不能为空".to_string(), )); } let conn = self.lock_conn()?; let now = now_text(); let expires_at = input .expires_at .unwrap_or_else(|| (chrono::Utc::now() + chrono::Duration::days(30)).to_rfc3339()); let record = AuthSessionRecord { id: input.id.unwrap_or_else(|| new_id("sess")), user_id: input.user_id, token_hash: token_hash.to_string(), user_agent: input.user_agent, ip_hash: input.ip_hash, created_at: now.clone(), expires_at, revoked_at: None, last_seen_at: now.clone(), }; conn.execute( "INSERT INTO auth_sessions (id, user_id, token_hash, user_agent, ip_hash, created_at, expires_at, revoked_at, last_seen_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL, ?8)", params![ record.id, record.user_id, record.token_hash, record.user_agent, record.ip_hash, record.created_at, record.expires_at, record.last_seen_at ], )?; Ok(record) } fn get_session_by_token_hash( &self, token_hash: &str, ) -> Result, ControlPlaneError> { let token_hash = token_hash.trim(); if token_hash.is_empty() { return Ok(None); } let conn = self.lock_conn()?; let now = now_text(); let resolved = conn .query_row( "SELECT s.id, s.user_id, s.token_hash, s.user_agent, s.ip_hash, s.created_at, s.expires_at, s.revoked_at, s.last_seen_at, u.id, u.email, u.username, u.display_name, u.role, u.status, u.created_at, u.updated_at, u.revision FROM auth_sessions s JOIN users u ON u.id = s.user_id WHERE s.token_hash = ?1 AND s.revoked_at IS NULL AND s.expires_at > ?2 AND u.status = 'active' LIMIT 1", params![token_hash, now], |row| { Ok(ResolvedAuthSession { session: row_to_auth_session(row)?, user: UserRecord { id: row.get(9)?, email: row.get(10)?, username: row.get(11)?, display_name: row.get(12)?, role: row.get(13)?, status: row.get(14)?, created_at: row.get(15)?, updated_at: row.get(16)?, revision: row.get(17)?, }, }) }, ) .optional()?; if let Some(resolved) = &resolved { conn.execute( "UPDATE auth_sessions SET last_seen_at = ?1 WHERE id = ?2", params![now, resolved.session.id], )?; } Ok(resolved) } fn revoke_session(&self, session_id: &str) -> Result<(), ControlPlaneError> { let session_id = session_id.trim(); if session_id.is_empty() { return Err(ControlPlaneError::InvalidInput( "session_id 不能为空".to_string(), )); } let conn = self.lock_conn()?; conn.execute( "UPDATE auth_sessions SET revoked_at = ?1 WHERE id = ?2 AND revoked_at IS NULL", params![now_text(), session_id], )?; Ok(()) } fn ensure_default_workspace( &self, actor_id: &str, ) -> Result { let conn = self.lock_conn()?; let existing = conn .query_row( "SELECT id, owner_user_id, name, kind, root_uri, root_path, source_kind, status, created_at, updated_at, revision FROM workspaces WHERE owner_user_id = ?1 AND kind = 'personal' AND status = 'active' LIMIT 1", params![actor_id], row_to_workspace, ) .optional()?; if let Some(workspace) = existing { return Ok(workspace); } let user = conn .query_row( "SELECT id, email, username, display_name, role, status, created_at, updated_at, revision FROM users WHERE id = ?1", params![actor_id], row_to_user, ) .optional()? .ok_or_else(|| ControlPlaneError::NotFound(format!("user not found: {actor_id}")))?; let now = now_text(); let workspace = WorkspaceRecord { id: new_id("ws"), owner_user_id: user.id.clone(), name: format!("{} 的空间", user.display_name), kind: "personal".to_string(), root_uri: format!("local://users/{}/workspaces/my-space", user.id), root_path: format!( "/mnt/Data1T/Mnote_data/users/{}/workspaces/my-space", user.id ), source_kind: "local_folder".to_string(), status: "active".to_string(), created_at: now.clone(), updated_at: now.clone(), revision: 1, }; conn.execute( "INSERT INTO workspaces (id, owner_user_id, name, kind, root_uri, root_path, source_kind, status, created_at, updated_at, revision) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)", params![ workspace.id, workspace.owner_user_id, workspace.name, workspace.kind, workspace.root_uri, workspace.root_path, workspace.source_kind, workspace.status, workspace.created_at, workspace.updated_at, workspace.revision ], )?; conn.execute( "INSERT INTO workspace_members (id, workspace_id, user_id, role, status, created_at, updated_at, revision) VALUES (?1, ?2, ?3, 'owner', 'active', ?4, ?5, 1)", params![new_id("wsm"), workspace.id, user.id, now, now], )?; conn.execute( "INSERT INTO directory_grants (id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision) VALUES (?1, ?2, ?3, ?4, ?5, 'write', 1, ?6, 'auto', 'active', ?7, ?8, ?9, 1)", params![ new_id("grant"), workspace.owner_user_id, workspace.id, workspace.root_uri, workspace.root_path, capabilities_json(&["ai".to_string(), "share".to_string()])?, workspace.owner_user_id, now, now ], )?; Ok(workspace) } fn upsert_workspace( &self, input: UpsertWorkspaceInput, ) -> Result { let owner_user_id = input.owner_user_id.trim().to_string(); let name = input.name.trim().to_string(); let root_uri = input.root_uri.trim().to_string(); let root_path = input.root_path.trim().to_string(); if owner_user_id.is_empty() || name.is_empty() || root_uri.is_empty() || root_path.is_empty() { return Err(ControlPlaneError::InvalidInput( "workspace owner/name/root 不能为空".to_string(), )); } let kind = input .kind .as_deref() .map(str::trim) .filter(|value| !value.is_empty()) .unwrap_or("personal") .to_string(); let source_kind = input .source_kind .as_deref() .map(str::trim) .filter(|value| !value.is_empty()) .unwrap_or("local_folder") .to_string(); let now = now_text(); let id = input.id.unwrap_or_else(|| new_id("ws")); let conn = self.lock_conn()?; let tx = conn.transaction()?; tx.query_row( "SELECT id FROM users WHERE id = ?1", params![&owner_user_id], |row| row.get::(0), ) .optional()? .ok_or_else(|| ControlPlaneError::NotFound(format!("user not found: {owner_user_id}")))?; tx.execute( "INSERT INTO workspaces (id, owner_user_id, name, kind, root_uri, root_path, source_kind, status, created_at, updated_at, revision) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 'active', ?8, ?9, 1) ON CONFLICT(id) DO UPDATE SET owner_user_id = excluded.owner_user_id, name = excluded.name, kind = excluded.kind, root_uri = excluded.root_uri, root_path = excluded.root_path, source_kind = excluded.source_kind, updated_at = excluded.updated_at, revision = workspaces.revision + 1", params![ &id, &owner_user_id, &name, &kind, &root_uri, &root_path, &source_kind, &now, &now ], )?; tx.execute( "INSERT OR IGNORE INTO workspace_members (id, workspace_id, user_id, role, status, created_at, updated_at, revision) VALUES (?1, ?2, ?3, 'owner', 'active', ?4, ?5, 1)", params![new_id("wsm"), &id, &owner_user_id, &now, &now], )?; let workspace = tx.query_row( "SELECT id, owner_user_id, name, kind, root_uri, root_path, source_kind, status, created_at, updated_at, revision FROM workspaces WHERE id = ?1", params![&id], row_to_workspace, )?; tx.commit()?; Ok(workspace) } fn grant_directory_access( &self, input: DirectoryGrantInput, ) -> Result { let conn = self.lock_conn()?; let now = now_text(); let record = DirectoryGrantRecord { id: new_id("grant"), user_id: input.user_id, workspace_id: input.workspace_id, root_uri: input.root_uri, root_path: input.root_path, permission: input.permission, recursive: input.recursive, capabilities_json: capabilities_json(&input.capabilities)?, source: input.source, status: "active".to_string(), created_by: input.created_by, created_at: now.clone(), updated_at: now, revision: 1, }; conn.execute( "INSERT INTO directory_grants (id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)", params![ record.id, record.user_id, record.workspace_id, record.root_uri, record.root_path, record.permission, if record.recursive { 1 } else { 0 }, record.capabilities_json, record.source, record.status, record.created_by, record.created_at, record.updated_at, record.revision ], )?; Ok(record) } fn list_directory_grants(&self) -> Result, ControlPlaneError> { let conn = self.lock_conn()?; let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision FROM directory_grants WHERE status = 'active' ORDER BY created_at ASC", )?; let grants = stmt .query_map((), row_to_grant)? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(grants) } fn list_directory_grants_for_actor( &self, actor_id: &str, ) -> Result, ControlPlaneError> { let actor_id = actor_id.trim(); if actor_id.is_empty() { return Ok(Vec::new()); } let conn = self.lock_conn()?; let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision FROM directory_grants WHERE user_id = ?1 AND status = 'active' ORDER BY created_at ASC", )?; let grants = stmt .query_map(params![actor_id], row_to_grant)? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(grants) } fn find_directory_grants( &self, lookup: DirectoryGrantLookup, ) -> Result, ControlPlaneError> { let conn = self.lock_conn()?; let grant_id = lookup .grant_id .as_deref() .map(str::trim) .filter(|value| !value.is_empty()); let user_id = lookup .user_id .as_deref() .map(str::trim) .filter(|value| !value.is_empty()); let root_uri = lookup .root_uri .as_deref() .map(str::trim) .filter(|value| !value.is_empty()); let status = if lookup.include_revoked { "%" } else { "active" }; let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision FROM directory_grants WHERE (?1 IS NULL OR id = ?1) AND (?2 IS NULL OR user_id = ?2) AND (?3 IS NULL OR root_uri = ?3) AND status LIKE ?4 ORDER BY created_at ASC", )?; let grants = stmt .query_map(params![grant_id, user_id, root_uri, status], row_to_grant)? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(grants) } fn revoke_directory_grant( &self, grant_id: &str, expected_revision: Option, ) -> Result<(), ControlPlaneError> { let grant_id = grant_id.trim(); if grant_id.is_empty() { return Err(ControlPlaneError::InvalidInput( "grant_id 不能为空".to_string(), )); } let conn = self.lock_conn()?; let existing_revision = conn .query_row( "SELECT revision FROM directory_grants WHERE id = ?1 AND status = 'active'", params![grant_id], |row| row.get::(0), ) .optional()? .ok_or_else(|| ControlPlaneError::NotFound(format!("grant not found: {grant_id}")))?; if let Some(expected_revision) = expected_revision { if expected_revision != existing_revision { return Err(ControlPlaneError::Conflict( "目录授权 revision 已变化".to_string(), )); } } conn.execute( "UPDATE directory_grants SET status = 'revoked', updated_at = ?1, revision = revision + 1 WHERE id = ?2 AND status = 'active'", params![now_text(), grant_id], )?; Ok(()) } fn resolve_access( &self, actor_id: &str, root_uri: &str, ) -> Result { let conn = self.lock_conn()?; let root_path = file_uri_to_legacy_path(root_uri); let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision FROM directory_grants WHERE user_id = ?1 AND status = 'active' AND ( ?2 = root_uri OR (recursive = 1 AND ?2 LIKE root_uri || '/%') OR (recursive = 1 AND root_uri LIKE '%/' AND ?2 LIKE root_uri || '%') OR (?3 != '' AND ?3 = root_path) OR (?3 != '' AND recursive = 1 AND ?3 LIKE root_path || '/%') )", )?; let grants = stmt .query_map(params![actor_id, root_uri, root_path], row_to_grant)? .collect::, _>>()?; let mut permission = "none".to_string(); for grant in &grants { permission = max_permission(&permission, &grant.permission).to_string(); } Ok(ResolvedAccess { user_id: actor_id.to_string(), root_uri: root_uri.to_string(), permission, grant_ids: grants.into_iter().map(|grant| grant.id).collect(), }) } fn create_share_link( &self, input: CreateShareLinkInput, ) -> Result { let workspace_id = input.workspace_id.trim().to_string(); if workspace_id.is_empty() { return Err(ControlPlaneError::InvalidInput( "workspace_id 不能为空".to_string(), )); } let resource_kind = input.resource_kind.trim().to_string(); let resource_id = input.resource_id.trim().to_string(); let permission = input.permission.trim().to_string(); if resource_kind.is_empty() || resource_id.is_empty() || permission.is_empty() { return Err(ControlPlaneError::InvalidInput( "share_link 资源与权限不能为空".to_string(), )); } let token = input .token .filter(|value| !value.trim().is_empty()) .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()); let token_hash = share_token_hash_v1(&token); let conn = self.lock_conn()?; let now = now_text(); let link = ShareLinkRecord { id: input.id.unwrap_or_else(|| new_id("share")), workspace_id, resource_kind, resource_id, token_hash: token_hash.clone(), permission, created_by: input.created_by, expires_at: input.expires_at, revoked_at: None, created_at: now.clone(), updated_at: now.clone(), revision: 1, }; conn.execute( "INSERT INTO share_links (id, workspace_id, resource_kind, resource_id, token_hash, permission, created_by, expires_at, revoked_at, created_at, updated_at, revision) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, NULL, ?9, ?10, 1)", params![ link.id, link.workspace_id, link.resource_kind, link.resource_id, link.token_hash, link.permission, link.created_by, link.expires_at, link.created_at, link.updated_at ], )?; Ok(CreatedShareLink { link, token }) } fn list_share_links( &self, workspace_id: &str, ) -> Result, ControlPlaneError> { let workspace_id = workspace_id.trim(); if workspace_id.is_empty() { return Ok(Vec::new()); } let conn = self.lock_conn()?; let mut stmt = conn.prepare( "SELECT id, workspace_id, resource_kind, resource_id, token_hash, permission, created_by, expires_at, revoked_at, created_at, updated_at, revision FROM share_links WHERE workspace_id = ?1 ORDER BY created_at ASC", )?; let links = stmt .query_map(params![workspace_id], row_to_share_link)? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(links) } fn resolve_share_link( &self, token_hash: &str, ) -> Result, ControlPlaneError> { let token_hash = token_hash.trim(); if token_hash.is_empty() { return Ok(None); } let conn = self.lock_conn()?; let link = conn .query_row( "SELECT id, workspace_id, resource_kind, resource_id, token_hash, permission, created_by, expires_at, revoked_at, created_at, updated_at, revision FROM share_links WHERE token_hash = ?1 AND revoked_at IS NULL LIMIT 1", params![token_hash], row_to_share_link, ) .optional()?; Ok(link) } fn revoke_share_link(&self, link_id: &str) -> Result<(), ControlPlaneError> { let link_id = link_id.trim(); if link_id.is_empty() { return Err(ControlPlaneError::InvalidInput( "link_id 不能为空".to_string(), )); } let conn = self.lock_conn()?; let existed = conn.execute( "UPDATE share_links SET revoked_at = ?1, updated_at = ?2, revision = revision + 1 WHERE id = ?3 AND revoked_at IS NULL", params![now_text(), now_text(), link_id], )?; if existed == 0 { return Err(ControlPlaneError::NotFound(format!( "share link not found: {link_id}" ))); } Ok(()) } fn append_audit(&self, input: AppendAuditInput) -> Result<(), ControlPlaneError> { let action = input.action.trim(); let target_kind = input.target_kind.trim(); if action.is_empty() || target_kind.is_empty() { return Err(ControlPlaneError::InvalidInput( "audit action/target_kind 不能为空".to_string(), )); } let conn = self.lock_conn()?; conn.execute( "INSERT INTO audit_log (id, actor_user_id, action, target_kind, target_id, metadata_json, created_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", params![ new_id("audit"), input.actor_user_id, action, target_kind, input.target_id, input.metadata_json, now_text() ], )?; Ok(()) } fn list_audit_log(&self, limit: usize) -> Result, ControlPlaneError> { let conn = self.lock_conn()?; let mut stmt = conn.prepare( "SELECT id, actor_user_id, action, target_kind, target_id, metadata_json, created_at FROM audit_log ORDER BY created_at DESC LIMIT ?1", )?; let rows = stmt .query_map(params![limit as i64], row_to_audit_log)? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(rows) } fn append_outbox( &self, input: OutboxEventInput, ) -> Result { let conn = self.lock_conn()?; let record = OutboxEventRecord { id: new_id("evt"), topic: input.topic, event_type: input.event_type, payload_json: input.payload_json, created_at: now_text(), delivered_at: None, attempts: 0, }; conn.execute( "INSERT INTO outbox_events (id, topic, event_type, payload_json, created_at, delivered_at, attempts) VALUES (?1, ?2, ?3, ?4, ?5, NULL, 0)", params![ record.id, record.topic, record.event_type, record.payload_json, record.created_at ], )?; Ok(record) } fn drain_outbox(&self, limit: usize) -> Result, ControlPlaneError> { let conn = self.lock_conn()?; let mut stmt = conn.prepare( "SELECT id, topic, event_type, payload_json, created_at, delivered_at, attempts FROM outbox_events WHERE delivered_at IS NULL ORDER BY created_at ASC LIMIT ?1", )?; let rows = stmt .query_map(params![limit as i64], row_to_outbox)? .collect::, _>>()?; let delivered_at = now_text(); for row in &rows { conn.execute( "UPDATE outbox_events SET delivered_at = ?1, attempts = attempts + 1 WHERE id = ?2", params![delivered_at, row.id], )?; } Ok(rows) } fn mark_outbox_delivered(&self, event_id: &str) -> Result<(), ControlPlaneError> { let event_id = event_id.trim(); if event_id.is_empty() { return Err(ControlPlaneError::InvalidInput( "event_id 不能为空".to_string(), )); } let conn = self.lock_conn()?; let updated = conn.execute( "UPDATE outbox_events SET delivered_at = COALESCE(delivered_at, ?1), attempts = attempts + 1 WHERE id = ?2", params![now_text(), event_id], )?; if updated == 0 { return Err(ControlPlaneError::NotFound(format!( "outbox event not found: {event_id}" ))); } Ok(()) } fn upsert_sidebar_shortcut( &self, input: UpsertSidebarShortcutInput, ) -> Result { let user_id = input.user_id.trim().to_string(); let workspace_id = input.workspace_id.trim().to_string(); let root_uri = input .root_uri .map(|value| value.trim().to_string()) .filter(|value| !value.is_empty()); let kind = input.kind.trim().to_string(); let source_kind = input.source_kind.trim().to_string(); let target_id = input.target_id.trim().to_string(); let title = input.title.trim().to_string(); if user_id.is_empty() || workspace_id.is_empty() || kind.is_empty() || source_kind.is_empty() || target_id.is_empty() || title.is_empty() { return Err(ControlPlaneError::InvalidInput( "sidebar shortcut user/workspace/kind/target/title 不能为空".to_string(), )); } if kind != "page" && kind != "folder" { return Err(ControlPlaneError::InvalidInput( "sidebar shortcut kind 只能是 page 或 folder".to_string(), )); } let conn = self.lock_conn()?; let now = now_text(); conn.execute( "INSERT INTO sidebar_shortcuts ( id, user_id, workspace_id, root_uri, kind, source_kind, target_id, relative_path, document_id, title, icon, sort_order, status, metadata_json, created_at, updated_at, revision ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, 'active', ?13, ?14, ?15, 1) ON CONFLICT(user_id, workspace_id, kind, target_id) DO UPDATE SET root_uri = excluded.root_uri, relative_path = excluded.relative_path, document_id = excluded.document_id, title = excluded.title, icon = excluded.icon, sort_order = excluded.sort_order, status = 'active', metadata_json = excluded.metadata_json, updated_at = excluded.updated_at, revision = sidebar_shortcuts.revision + 1", params![ input.id.unwrap_or_else(|| new_id("shortcut")), user_id, workspace_id, root_uri, kind, source_kind, target_id, input.relative_path, input.document_id, title, input.icon, input.sort_order, input.metadata_json, now, now ], )?; let record = conn.query_row( "SELECT id, user_id, workspace_id, root_uri, kind, source_kind, target_id, relative_path, document_id, title, icon, sort_order, status, metadata_json, created_at, updated_at, revision FROM sidebar_shortcuts WHERE user_id = ?1 AND workspace_id = ?2 AND kind = ?3 AND target_id = ?4 LIMIT 1", params![user_id, workspace_id, kind, target_id], row_to_sidebar_shortcut, )?; Ok(record) } fn list_sidebar_shortcuts( &self, user_id: &str, workspace_id: &str, ) -> Result, ControlPlaneError> { let user_id = user_id.trim(); let workspace_id = workspace_id.trim(); if user_id.is_empty() || workspace_id.is_empty() { return Ok(Vec::new()); } let conn = self.lock_conn()?; let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, root_uri, kind, source_kind, target_id, relative_path, document_id, title, icon, sort_order, status, metadata_json, created_at, updated_at, revision FROM sidebar_shortcuts WHERE user_id = ?1 AND workspace_id = ?2 AND status = 'active' ORDER BY sort_order ASC, created_at ASC", )?; let rows = stmt .query_map(params![user_id, workspace_id], row_to_sidebar_shortcut)? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(rows) } fn list_sidebar_shortcuts_with_global_local( &self, user_id: &str, workspace_id: &str, ) -> Result, ControlPlaneError> { let user_id = user_id.trim(); let workspace_id = workspace_id.trim(); if user_id.is_empty() || workspace_id.is_empty() { return Ok(Vec::new()); } let conn = self.lock_conn()?; let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, root_uri, kind, source_kind, target_id, relative_path, document_id, title, icon, sort_order, status, metadata_json, created_at, updated_at, revision FROM sidebar_shortcuts WHERE user_id = ?1 AND status = 'active' AND (workspace_id = ?2 OR source_kind = 'local_folder') ORDER BY sort_order ASC, created_at ASC", )?; let rows = stmt .query_map(params![user_id, workspace_id], row_to_sidebar_shortcut)? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(rows) } fn delete_sidebar_shortcut( &self, user_id: &str, shortcut_id: &str, ) -> Result<(), ControlPlaneError> { let user_id = user_id.trim(); let shortcut_id = shortcut_id.trim(); if user_id.is_empty() || shortcut_id.is_empty() { return Err(ControlPlaneError::InvalidInput( "sidebar shortcut user_id/shortcut_id 不能为空".to_string(), )); } let conn = self.lock_conn()?; let changed = conn.execute( "UPDATE sidebar_shortcuts SET status = 'removed', updated_at = ?1, revision = revision + 1 WHERE id = ?2 AND user_id = ?3 AND status = 'active'", params![now_text(), shortcut_id, user_id], )?; if changed == 0 { return Err(ControlPlaneError::NotFound(format!( "sidebar shortcut not found: {shortcut_id}" ))); } Ok(()) } fn upsert_user_ui_preference( &self, input: UpsertUserUiPreferenceInput, ) -> Result { let user_id = input.user_id.trim().to_string(); let workspace_id = option_to_stored_text(input.workspace_id); let source_kind = option_to_stored_text(input.source_kind); let scope_kind = input.scope_kind.trim().to_string(); let scope_id = input.scope_id.trim().to_string(); let key = input.key.trim().to_string(); if user_id.is_empty() || scope_kind.is_empty() || scope_id.is_empty() || key.is_empty() { return Err(ControlPlaneError::InvalidInput( "user UI preference user/scope/key 不能为空".to_string(), )); } serde_json::from_str::(&input.value_json)?; let conn = self.lock_conn()?; let now = now_text(); conn.execute( "INSERT INTO user_ui_preferences ( id, user_id, workspace_id, source_kind, scope_kind, scope_id, key, value_json, status, created_at, updated_at, revision ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 'active', ?9, ?10, 1) ON CONFLICT(user_id, workspace_id, source_kind, scope_kind, scope_id, key) DO UPDATE SET value_json = excluded.value_json, status = 'active', updated_at = excluded.updated_at, revision = user_ui_preferences.revision + 1", params![ input.id.unwrap_or_else(|| new_id("ui_pref")), user_id, workspace_id, source_kind, scope_kind, scope_id, key, input.value_json, now, now ], )?; let record = conn.query_row( "SELECT id, user_id, workspace_id, source_kind, scope_kind, scope_id, key, value_json, status, created_at, updated_at, revision FROM user_ui_preferences WHERE user_id = ?1 AND workspace_id = ?2 AND source_kind = ?3 AND scope_kind = ?4 AND scope_id = ?5 AND key = ?6 LIMIT 1", params![ user_id, workspace_id, source_kind, scope_kind, scope_id, key ], row_to_user_ui_preference, )?; Ok(record) } fn list_user_ui_preferences( &self, user_id: &str, workspace_id: Option<&str>, source_kind: Option<&str>, ) -> Result, ControlPlaneError> { let user_id = user_id.trim(); if user_id.is_empty() { return Ok(Vec::new()); } let workspace_id = workspace_id.map(str::trim).unwrap_or_default(); let source_kind = source_kind.map(str::trim).unwrap_or_default(); let conn = self.lock_conn()?; let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, source_kind, scope_kind, scope_id, key, value_json, status, created_at, updated_at, revision FROM user_ui_preferences WHERE user_id = ?1 AND status = 'active' AND (workspace_id = '' OR workspace_id = ?2) AND (source_kind = '' OR source_kind = ?3) ORDER BY created_at ASC", )?; let rows = stmt .query_map( params![user_id, workspace_id, source_kind], row_to_user_ui_preference, )? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(rows) } fn list_user_ui_preferences_for_scope( &self, workspace_id: Option<&str>, source_kind: Option<&str>, scope_kind: &str, scope_id: &str, key: &str, ) -> Result, ControlPlaneError> { let workspace_id = workspace_id.map(str::trim).unwrap_or_default(); let source_kind = source_kind.map(str::trim).unwrap_or_default(); let scope_kind = scope_kind.trim(); let scope_id = scope_id.trim(); let key = key.trim(); if scope_kind.is_empty() || scope_id.is_empty() || key.is_empty() { return Ok(Vec::new()); } let conn = self.lock_conn()?; let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, source_kind, scope_kind, scope_id, key, value_json, status, created_at, updated_at, revision FROM user_ui_preferences WHERE status = 'active' AND workspace_id = ?1 AND source_kind = ?2 AND scope_kind = ?3 AND scope_id = ?4 AND key = ?5 ORDER BY updated_at ASC", )?; let rows = stmt .query_map( params![workspace_id, source_kind, scope_kind, scope_id, key], row_to_user_ui_preference, )? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(rows) } fn ensure_ai_agent_profile_policy( &self, user_id: &str, is_admin: bool, ) -> Result, ControlPlaneError> { let user_id = user_id.trim(); if user_id.is_empty() { return Err(ControlPlaneError::InvalidInput( "ai agent profile user_id 不能为空".to_string(), )); } let conn = self.lock_conn()?; let now = now_text(); conn.execute( "INSERT INTO ai_agent_profiles ( id, agent_id, profile_kind, owner_user_id, base_profile_name, isolated_profile_name, display_name, status, created_at, updated_at, revision ) VALUES ('shared_lite', 'hermes', 'shared', '', 'lite', 'lite', 'Lite', 'active', ?1, ?2, 1) ON CONFLICT(agent_id, profile_kind, owner_user_id, base_profile_name) DO UPDATE SET status = 'active', updated_at = excluded.updated_at", params![now, now], )?; conn.execute( "INSERT INTO ai_agent_profile_grants ( id, profile_id, user_id, role, can_run, can_manage_skills, can_manage_config, created_at, updated_at, revision ) VALUES ('grant_shared_lite_all', 'shared_lite', '', 'user', 1, 0, 0, ?1, ?2, 1) ON CONFLICT(profile_id, user_id, role) DO UPDATE SET can_run = 1, can_manage_skills = 0, can_manage_config = 0, updated_at = excluded.updated_at", params![now, now], )?; // 旧的泛化 OpenClaw 网页问答入口已拆成明确的三类 chat agent。 conn.execute( "UPDATE ai_agent_profiles SET status = 'retired', updated_at = ?1 WHERE id = 'shared_openclaw_webqa'", params![now], )?; let shared_chat_profiles = [ ( "shared_deepseek_chat", "deepseek-chat", "openclaw-deepseek-chat", "DeepSeek Chat", ), ( "shared_gemini_chat", "gemini-chat", "openclaw-gemini-chat", "Gemini Chat", ), ( "shared_doubao_chat", "doubao-chat", "openclaw-doubao-chat", "豆包 Chat", ), ( "shared_api_deepseek_flash_chat", "api-deepseek-flash-chat", "api-deepseek-flash-chat", "DeepSeek Flash Chat", ), ( "shared_api_deepseek_pro_chat", "api-deepseek-pro-chat", "api-deepseek-pro-chat", "DeepSeek Pro Chat", ), ( "shared_api_gpt_chat", "api-gpt-chat", "api-gpt-chat", "GPT Chat", ), ( "shared_api_kimi_chat", "api-kimi-chat", "api-kimi-chat", "Kimi Chat", ), ( "shared_api_gemini_chat", "api-gemini-chat", "api-gemini-chat", "Gemini API Chat", ), ( "shared_api_grok_chat", "api-grok-chat", "api-grok-chat", "Grok API Chat", ), ]; for (profile_id, base_profile, isolated_profile, display_name) in shared_chat_profiles { conn.execute( "INSERT INTO ai_agent_profiles ( id, agent_id, profile_kind, owner_user_id, base_profile_name, isolated_profile_name, display_name, status, created_at, updated_at, revision ) VALUES (?1, 'hermes', 'shared', '', ?2, ?3, ?4, 'active', ?5, ?6, 1) ON CONFLICT(agent_id, profile_kind, owner_user_id, base_profile_name) DO UPDATE SET status = 'active', display_name = excluded.display_name, isolated_profile_name = excluded.isolated_profile_name, updated_at = excluded.updated_at", params![profile_id, base_profile, isolated_profile, display_name, now, now], )?; let grant_id = format!("grant_{profile_id}_all"); conn.execute( "INSERT INTO ai_agent_profile_grants ( id, profile_id, user_id, role, can_run, can_manage_skills, can_manage_config, created_at, updated_at, revision ) VALUES (?1, ?2, '', 'user', 1, 0, 0, ?3, ?4, 1) ON CONFLICT(profile_id, user_id, role) DO UPDATE SET can_run = 1, can_manage_skills = 0, can_manage_config = 0, updated_at = excluded.updated_at", params![grant_id, profile_id, now, now], )?; } let user_part = sanitize_profile_component(user_id); let personal_profile_id = format!("usr_{user_part}_default"); let personal_profile_name = format!("mnote-u-{user_part}-default"); conn.execute( "INSERT INTO ai_agent_profiles ( id, agent_id, profile_kind, owner_user_id, base_profile_name, isolated_profile_name, display_name, status, created_at, updated_at, revision ) VALUES (?1, 'hermes', 'personal', ?2, 'default', ?3, '我的 Hermes', 'active', ?4, ?5, 1) ON CONFLICT(agent_id, profile_kind, owner_user_id, base_profile_name) DO UPDATE SET status = 'active', updated_at = excluded.updated_at", params![personal_profile_id, user_id, personal_profile_name, now, now], )?; let personal_grant_id = format!("grant_{personal_profile_id}_owner"); conn.execute( "INSERT INTO ai_agent_profile_grants ( id, profile_id, user_id, role, can_run, can_manage_skills, can_manage_config, created_at, updated_at, revision ) VALUES (?1, ?2, ?3, 'owner', 1, 1, 1, ?4, ?5, 1) ON CONFLICT(profile_id, user_id, role) DO UPDATE SET can_run = 1, can_manage_skills = 1, can_manage_config = 1, updated_at = excluded.updated_at", params![personal_grant_id, personal_profile_id, user_id, now, now], )?; if is_admin { let admin_grant_id = format!("grant_shared_lite_admin_{user_part}"); conn.execute( "INSERT INTO ai_agent_profile_grants ( id, profile_id, user_id, role, can_run, can_manage_skills, can_manage_config, created_at, updated_at, revision ) VALUES (?1, 'shared_lite', ?2, 'admin', 1, 1, 1, ?3, ?4, 1) ON CONFLICT(profile_id, user_id, role) DO UPDATE SET can_run = 1, can_manage_skills = 1, can_manage_config = 1, updated_at = excluded.updated_at", params![admin_grant_id, user_id, now, now], )?; for (profile_id, _, _, _) in shared_chat_profiles { let admin_chat_grant_id = format!("grant_{profile_id}_admin_{user_part}"); conn.execute( "INSERT INTO ai_agent_profile_grants ( id, profile_id, user_id, role, can_run, can_manage_skills, can_manage_config, created_at, updated_at, revision ) VALUES (?1, ?2, ?3, 'admin', 1, 0, 0, ?4, ?5, 1) ON CONFLICT(profile_id, user_id, role) DO UPDATE SET can_run = 1, can_manage_skills = 0, can_manage_config = 0, updated_at = excluded.updated_at", params![admin_chat_grant_id, profile_id, user_id, now, now], )?; } } list_ai_agent_profile_access_rows(&conn, user_id) } fn resolve_ai_agent_profile( &self, user_id: &str, is_admin: bool, profile_id: &str, ) -> Result, ControlPlaneError> { let profile_id = profile_id.trim(); if profile_id.is_empty() { return Ok(None); } let access = self.ensure_ai_agent_profile_policy(user_id, is_admin)?; Ok(access .into_iter() .find(|item| item.profile.id == profile_id && item.grant.can_run)) } fn upsert_navigation_recent( &self, input: UpsertNavigationRecentInput, ) -> Result { let user_id = input.user_id.trim().to_string(); let workspace_id = option_to_stored_text(input.workspace_id); let kind = input.kind.trim().to_string(); let source_kind = input.source_kind.trim().to_string(); let root_uri = input.root_uri.trim().to_string(); let relative_path = input .relative_path .map(|value| value.trim().trim_matches('/').to_string()) .filter(|value| !value.is_empty()); let document_id = input .document_id .map(|value| value.trim().to_string()) .filter(|value| !value.is_empty()); let title = input.title.trim().to_string(); if user_id.is_empty() || kind.is_empty() || source_kind.is_empty() || root_uri.is_empty() || title.is_empty() { return Err(ControlPlaneError::InvalidInput( "navigation recent user/kind/source/root/title 不能为空".to_string(), )); } serde_json::from_str::(&input.metadata_json)?; let target_key = navigation_recent_target_key( &kind, &root_uri, relative_path.as_deref(), document_id.as_deref(), )?; let conn = self.lock_conn()?; let now = now_text(); conn.execute( "INSERT INTO user_navigation_recent ( id, user_id, workspace_id, kind, source_kind, root_uri, target_key, relative_path, document_id, title, status, metadata_json, visited_at, created_at, updated_at, revision ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, 'active', ?11, ?12, ?13, ?14, 1) ON CONFLICT(user_id, kind, target_key) DO UPDATE SET workspace_id = excluded.workspace_id, source_kind = excluded.source_kind, root_uri = excluded.root_uri, relative_path = excluded.relative_path, document_id = excluded.document_id, title = excluded.title, status = 'active', metadata_json = excluded.metadata_json, visited_at = excluded.visited_at, updated_at = excluded.updated_at, revision = user_navigation_recent.revision + 1", params![ input.id.unwrap_or_else(|| new_id("nav_recent")), user_id, workspace_id, kind, source_kind, root_uri, target_key, relative_path, document_id, title, input.metadata_json, now, now, now, ], )?; prune_navigation_recent_for_kind(&conn, &user_id, &kind)?; let record = conn.query_row( "SELECT id, user_id, workspace_id, kind, source_kind, root_uri, target_key, relative_path, document_id, title, status, metadata_json, visited_at, created_at, updated_at, revision FROM user_navigation_recent WHERE user_id = ?1 AND kind = ?2 AND target_key = ?3 LIMIT 1", params![user_id, kind, target_key], row_to_navigation_recent, )?; Ok(record) } fn list_navigation_recent( &self, user_id: &str, kind: Option<&str>, limit: usize, ) -> Result, ControlPlaneError> { let user_id = user_id.trim(); if user_id.is_empty() { return Ok(Vec::new()); } let limit = limit.clamp(1, 100) as i64; let conn = self.lock_conn()?; let rows = if let Some(kind) = kind.map(str::trim).filter(|value| !value.is_empty()) { let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, kind, source_kind, root_uri, target_key, relative_path, document_id, title, status, metadata_json, visited_at, created_at, updated_at, revision FROM user_navigation_recent WHERE user_id = ?1 AND kind = ?2 AND status = 'active' ORDER BY visited_at DESC, updated_at DESC LIMIT ?3", )?; let rows = stmt .query_map(params![user_id, kind, limit], row_to_navigation_recent)? .collect::, _>>()?; rows } else { let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, kind, source_kind, root_uri, target_key, relative_path, document_id, title, status, metadata_json, visited_at, created_at, updated_at, revision FROM user_navigation_recent WHERE user_id = ?1 AND status = 'active' ORDER BY visited_at DESC, updated_at DESC LIMIT ?2", )?; let rows = stmt .query_map(params![user_id, limit], row_to_navigation_recent)? .collect::, _>>()?; rows }; Ok(rows) } fn upsert_ai_policy( &self, input: UpsertAiPolicyInput, ) -> Result { if input.user_id.is_none() && input.workspace_id.is_none() { return Err(ControlPlaneError::InvalidInput( "ai_policy 必须绑定 user_id 或 workspace_id".to_string(), )); } serde_json::from_str::(&input.allowed_roots_json)?; serde_json::from_str::(&input.model_policy_json)?; serde_json::from_str::(&input.quota_json)?; let conn = self.lock_conn()?; let now = now_text(); let existing = if let Some(workspace_id) = input.workspace_id.as_deref() { conn.query_row( "SELECT id, user_id, workspace_id, allowed_roots_json, model_policy_json, quota_json, created_at, updated_at, revision FROM ai_policies WHERE workspace_id = ?1 ORDER BY created_at DESC LIMIT 1", params![workspace_id], row_to_ai_policy, ) .optional()? } else if let Some(user_id) = input.user_id.as_deref() { conn.query_row( "SELECT id, user_id, workspace_id, allowed_roots_json, model_policy_json, quota_json, created_at, updated_at, revision FROM ai_policies WHERE user_id = ?1 AND workspace_id IS NULL ORDER BY created_at DESC LIMIT 1", params![user_id], row_to_ai_policy, ) .optional()? } else { None }; if let Some(existing) = existing { let revision = existing.revision + 1; conn.execute( "UPDATE ai_policies SET allowed_roots_json = ?1, model_policy_json = ?2, quota_json = ?3, updated_at = ?4, revision = ?5 WHERE id = ?6", params![ input.allowed_roots_json, input.model_policy_json, input.quota_json, now, revision, existing.id ], )?; return Ok(AiPolicyRecord { id: existing.id, user_id: existing.user_id, workspace_id: existing.workspace_id, allowed_roots_json: input.allowed_roots_json, model_policy_json: input.model_policy_json, quota_json: input.quota_json, created_at: existing.created_at, updated_at: now, revision, }); } let record = AiPolicyRecord { id: input.id.unwrap_or_else(|| new_id("aip")), user_id: input.user_id, workspace_id: input.workspace_id, allowed_roots_json: input.allowed_roots_json, model_policy_json: input.model_policy_json, quota_json: input.quota_json, created_at: now.clone(), updated_at: now, revision: 1, }; conn.execute( "INSERT INTO ai_policies (id, user_id, workspace_id, allowed_roots_json, model_policy_json, quota_json, created_at, updated_at, revision) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 1)", params![ record.id, record.user_id, record.workspace_id, record.allowed_roots_json, record.model_policy_json, record.quota_json, record.created_at, record.updated_at ], )?; Ok(record) } fn get_ai_policy( &self, actor_id: &str, workspace_id: Option<&str>, ) -> Result, ControlPlaneError> { let actor_id = actor_id.trim(); let workspace_id = workspace_id .map(str::trim) .filter(|value| !value.is_empty()); if actor_id.is_empty() && workspace_id.is_none() { return Ok(None); } let conn = self.lock_conn()?; if let Some(workspace_id) = workspace_id { let workspace_policy = conn .query_row( "SELECT id, user_id, workspace_id, allowed_roots_json, model_policy_json, quota_json, created_at, updated_at, revision FROM ai_policies WHERE workspace_id = ?1 ORDER BY updated_at DESC LIMIT 1", params![workspace_id], row_to_ai_policy, ) .optional()?; if workspace_policy.is_some() { return Ok(workspace_policy); } } if actor_id.is_empty() { return Ok(None); } conn.query_row( "SELECT id, user_id, workspace_id, allowed_roots_json, model_policy_json, quota_json, created_at, updated_at, revision FROM ai_policies WHERE user_id = ?1 AND workspace_id IS NULL ORDER BY updated_at DESC LIMIT 1", params![actor_id], row_to_ai_policy, ) .optional() .map_err(ControlPlaneError::from) } fn upsert_sync_state( &self, input: UpsertSyncStateInput, ) -> Result { let workspace_id = input.workspace_id.trim().to_string(); let remote_kind = input.remote_kind.trim().to_string(); let status = input.status.trim().to_string(); if workspace_id.is_empty() || remote_kind.is_empty() || status.is_empty() { return Err(ControlPlaneError::InvalidInput( "sync_state workspace_id/remote_kind/status 不能为空".to_string(), )); } if let Some(error_json) = &input.error_json { serde_json::from_str::(error_json)?; } let conn = self.lock_conn()?; let now = now_text(); let existing = conn .query_row( "SELECT id, workspace_id, remote_kind, remote_id, cursor, last_synced_at, status, error_json, updated_at, revision FROM sync_state WHERE workspace_id = ?1 AND remote_kind = ?2 LIMIT 1", params![workspace_id, remote_kind], row_to_sync_state, ) .optional()?; if let Some(existing) = existing { let revision = existing.revision + 1; conn.execute( "UPDATE sync_state SET remote_id = ?1, cursor = ?2, last_synced_at = ?3, status = ?4, error_json = ?5, updated_at = ?6, revision = ?7 WHERE id = ?8", params![ input.remote_id, input.cursor, input.last_synced_at, status, input.error_json, now, revision, existing.id ], )?; return Ok(SyncStateRecord { id: existing.id, workspace_id: existing.workspace_id, remote_kind: existing.remote_kind, remote_id: input.remote_id, cursor: input.cursor, last_synced_at: input.last_synced_at, status, error_json: input.error_json, updated_at: now, revision, }); } let record = SyncStateRecord { id: input.id.unwrap_or_else(|| new_id("sync")), workspace_id, remote_kind, remote_id: input.remote_id, cursor: input.cursor, last_synced_at: input.last_synced_at, status, error_json: input.error_json, updated_at: now, revision: 1, }; conn.execute( "INSERT INTO sync_state (id, workspace_id, remote_kind, remote_id, cursor, last_synced_at, status, error_json, updated_at, revision) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 1)", params![ record.id, record.workspace_id, record.remote_kind, record.remote_id, record.cursor, record.last_synced_at, record.status, record.error_json, record.updated_at ], )?; Ok(record) } fn get_sync_state( &self, workspace_id: &str, remote_kind: &str, ) -> Result, ControlPlaneError> { let workspace_id = workspace_id.trim(); let remote_kind = remote_kind.trim(); if workspace_id.is_empty() || remote_kind.is_empty() { return Ok(None); } let conn = self.lock_conn()?; conn.query_row( "SELECT id, workspace_id, remote_kind, remote_id, cursor, last_synced_at, status, error_json, updated_at, revision FROM sync_state WHERE workspace_id = ?1 AND remote_kind = ?2 LIMIT 1", params![workspace_id, remote_kind], row_to_sync_state, ) .optional() .map_err(ControlPlaneError::from) } fn upsert_ai_runtime_run( &self, input: UpsertAiRuntimeRunInput, ) -> Result { if input.user_id.trim().is_empty() || input.session_id.trim().is_empty() || input.run_id.trim().is_empty() { return Err(ControlPlaneError::InvalidInput( "ai runtime run user_id/session_id/run_id 不能为空".to_string(), )); } serde_json::from_str::(&input.runtime_json)?; serde_json::from_str::(&input.payload_json)?; let conn = self.lock_conn()?; let now = now_text(); let existing = conn .query_row( "SELECT id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision FROM ai_runtime_runs WHERE run_id = ?1 LIMIT 1", params![input.run_id], row_to_ai_runtime_run, ) .optional()?; if let Some(existing) = existing { let revision = existing.revision + 1; conn.execute( "UPDATE ai_runtime_runs SET title = ?1, profile = ?2, acp_runtime = ?3, trace_id = ?4, status = ?5, runtime_json = ?6, payload_json = ?7, deleted_at = NULL, updated_at = ?8, revision = ?9 WHERE id = ?10", params![ input.title, input.profile, input.acp_runtime, input.trace_id, input.status, input.runtime_json, input.payload_json, now, revision, existing.id ], )?; return Ok(AiRuntimeRunRecord { id: existing.id, user_id: existing.user_id, workspace_id: existing.workspace_id, document_id: existing.document_id, session_id: existing.session_id, run_id: existing.run_id, title: input.title, profile: input.profile, acp_runtime: input.acp_runtime, trace_id: input.trace_id, status: input.status, runtime_json: input.runtime_json, payload_json: input.payload_json, deleted_at: None, created_at: existing.created_at, updated_at: now, revision, }); } let record = AiRuntimeRunRecord { id: input.id.unwrap_or_else(|| new_id("acr")), user_id: input.user_id, workspace_id: input.workspace_id, document_id: input.document_id, session_id: input.session_id, run_id: input.run_id, title: input.title, profile: input.profile, acp_runtime: input.acp_runtime, trace_id: input.trace_id, status: input.status, runtime_json: input.runtime_json, payload_json: input.payload_json, deleted_at: None, created_at: now.clone(), updated_at: now, revision: 1, }; conn.execute( "INSERT INTO ai_runtime_runs (id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, NULL, ?14, ?15, 1)", params![ record.id, record.user_id, record.workspace_id, record.document_id, record.session_id, record.run_id, record.title, record.profile, record.acp_runtime, record.trace_id, record.status, record.runtime_json, record.payload_json, record.created_at, record.updated_at ], )?; Ok(record) } fn list_ai_runtime_runs( &self, user_id: &str, workspace_id: Option<&str>, document_id: Option<&str>, session_id: Option<&str>, limit: usize, ) -> Result, ControlPlaneError> { let conn = self.lock_conn()?; let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision FROM ai_runtime_runs WHERE user_id = ?1 AND (?2 IS NULL OR workspace_id = ?2) AND (?3 IS NULL OR document_id = ?3) AND (?4 IS NULL OR session_id = ?4) AND deleted_at IS NULL ORDER BY updated_at DESC LIMIT ?5", )?; let rows = stmt .query_map( params![user_id, workspace_id, document_id, session_id, limit as i64], row_to_ai_runtime_run, )? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(rows) } fn find_ai_runtime_run( &self, user_id: &str, run_id: &str, ) -> Result, ControlPlaneError> { let conn = self.lock_conn()?; conn.query_row( "SELECT id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision FROM ai_runtime_runs WHERE user_id = ?1 AND run_id = ?2 AND deleted_at IS NULL LIMIT 1", params![user_id, run_id], row_to_ai_runtime_run, ) .optional() .map_err(ControlPlaneError::from) } fn append_ai_runtime_event( &self, input: AppendAiRuntimeEventInput, ) -> Result { if input.user_id.trim().is_empty() || input.session_id.trim().is_empty() || input.run_id.trim().is_empty() { return Err(ControlPlaneError::InvalidInput( "ai runtime event user_id/session_id/run_id 不能为空".to_string(), )); } serde_json::from_str::(&input.payload_json)?; let conn = self.lock_conn()?; let record = AiRuntimeEventRecord { id: input.id.unwrap_or_else(|| new_id("are")), user_id: input.user_id, workspace_id: input.workspace_id, document_id: input.document_id, session_id: input.session_id, run_id: input.run_id, profile: input.profile, acp_runtime: input.acp_runtime, event_type: input.event_type, payload_json: input.payload_json, created_at: now_text(), }; conn.execute( "INSERT INTO ai_runtime_events (id, user_id, workspace_id, document_id, session_id, run_id, profile, acp_runtime, event_type, payload_json, created_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)", params![ record.id, record.user_id, record.workspace_id, record.document_id, record.session_id, record.run_id, record.profile, record.acp_runtime, record.event_type, record.payload_json, record.created_at ], )?; Ok(record) } fn delete_ai_runtime_events_for_run( &self, user_id: &str, run_id: &str, ) -> Result { let user_id = user_id.trim(); let run_id = run_id.trim(); if user_id.is_empty() || run_id.is_empty() { return Err(ControlPlaneError::InvalidInput( "user_id/run_id 不能为空".to_string(), )); } let conn = self.lock_conn()?; let rows = conn.execute( "DELETE FROM ai_runtime_events WHERE user_id = ?1 AND run_id = ?2", params![user_id, run_id], )?; Ok(rows) } fn list_ai_runtime_events( &self, user_id: &str, run_id: &str, limit: usize, ) -> Result, ControlPlaneError> { let conn = self.lock_conn()?; let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, document_id, session_id, run_id, profile, acp_runtime, event_type, payload_json, created_at FROM ( SELECT id, user_id, workspace_id, document_id, session_id, run_id, profile, acp_runtime, event_type, payload_json, created_at FROM ai_runtime_events WHERE user_id = ?1 AND run_id = ?2 ORDER BY created_at DESC, id DESC LIMIT ?3 ) ORDER BY created_at ASC, id ASC", )?; let rows = stmt .query_map( params![user_id, run_id, limit as i64], row_to_ai_runtime_event, )? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(rows) } fn list_ai_runtime_journal_events( &self, user_id: &str, run_id: &str, after_seq: i64, limit: usize, ) -> Result, ControlPlaneError> { let conn = self.lock_conn()?; let mut stmt = conn.prepare( "WITH ordered AS ( SELECT ROW_NUMBER() OVER (ORDER BY created_at ASC, id ASC) AS seq, id, user_id, workspace_id, document_id, session_id, run_id, profile, acp_runtime, event_type, payload_json, created_at FROM ai_runtime_events WHERE user_id = ?1 AND run_id = ?2 ) SELECT seq, id, user_id, workspace_id, document_id, session_id, run_id, profile, acp_runtime, event_type, payload_json, created_at FROM ordered WHERE seq > ?3 ORDER BY seq ASC LIMIT ?4", )?; let rows = stmt .query_map( params![user_id, run_id, after_seq.max(0), limit.max(1) as i64], row_to_ai_runtime_journal_event, )? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(rows) } fn upsert_ai_external_conversation_binding( &self, input: UpsertAiExternalConversationBindingInput, ) -> Result { let user_id = input.user_id.trim(); let mnote_session_id = input.mnote_session_id.trim(); let provider = input.provider.trim(); let remote_conversation_id = input.remote_conversation_id.trim(); if user_id.is_empty() || mnote_session_id.is_empty() || provider.is_empty() || remote_conversation_id.is_empty() { return Err(ControlPlaneError::InvalidInput( "external conversation binding user_id/session_id/provider/remote_id 不能为空" .to_string(), )); } serde_json::from_str::(&input.metadata_json)?; validate_ai_external_conversation_status(&input.status)?; let conn = self.lock_conn()?; let now = now_text(); let existing = conn .query_row( "SELECT id, user_id, workspace_id, mnote_session_id, acp_session_id, agent_id, profile, provider, remote_conversation_id, remote_url, status, metadata_json, created_at, updated_at, deleted_at, revision FROM ai_external_conversation_bindings WHERE user_id = ?1 AND mnote_session_id = ?2 AND provider = ?3 LIMIT 1", params![user_id, mnote_session_id, provider], row_to_ai_external_conversation_binding, ) .optional()?; if let Some(existing) = existing { let revision = existing.revision + 1; let deleted_at = if input.status == "active" { None } else { existing.deleted_at.or_else(|| Some(now.clone())) }; conn.execute( "UPDATE ai_external_conversation_bindings SET workspace_id = ?1, acp_session_id = ?2, agent_id = ?3, profile = ?4, remote_conversation_id = ?5, remote_url = ?6, status = ?7, metadata_json = ?8, updated_at = ?9, deleted_at = ?10, revision = ?11 WHERE id = ?12", params![ input.workspace_id, input.acp_session_id, input.agent_id, input.profile, remote_conversation_id, input.remote_url, input.status, input.metadata_json, now, deleted_at, revision, existing.id ], )?; return Ok(AiExternalConversationBindingRecord { id: existing.id, user_id: existing.user_id, workspace_id: input.workspace_id, mnote_session_id: existing.mnote_session_id, acp_session_id: input.acp_session_id, agent_id: input.agent_id, profile: input.profile, provider: existing.provider, remote_conversation_id: remote_conversation_id.to_string(), remote_url: input.remote_url, status: input.status, metadata_json: input.metadata_json, created_at: existing.created_at, updated_at: now, deleted_at, revision, }); } let deleted_at = if input.status == "active" { None } else { Some(now.clone()) }; let record = AiExternalConversationBindingRecord { id: input.id.unwrap_or_else(|| new_id("aecb")), user_id: user_id.to_string(), workspace_id: input.workspace_id, mnote_session_id: mnote_session_id.to_string(), acp_session_id: input.acp_session_id, agent_id: input.agent_id, profile: input.profile, provider: provider.to_string(), remote_conversation_id: remote_conversation_id.to_string(), remote_url: input.remote_url, status: input.status, metadata_json: input.metadata_json, created_at: now.clone(), updated_at: now, deleted_at, revision: 1, }; conn.execute( "INSERT INTO ai_external_conversation_bindings (id, user_id, workspace_id, mnote_session_id, acp_session_id, agent_id, profile, provider, remote_conversation_id, remote_url, status, metadata_json, created_at, updated_at, deleted_at, revision) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, 1)", params![ record.id, record.user_id, record.workspace_id, record.mnote_session_id, record.acp_session_id, record.agent_id, record.profile, record.provider, record.remote_conversation_id, record.remote_url, record.status, record.metadata_json, record.created_at, record.updated_at, record.deleted_at ], )?; Ok(record) } fn find_ai_external_conversation_binding( &self, user_id: &str, workspace_id: Option<&str>, mnote_session_id: &str, provider: &str, ) -> Result, ControlPlaneError> { let user_id = user_id.trim(); let mnote_session_id = mnote_session_id.trim(); let provider = provider.trim(); if user_id.is_empty() || mnote_session_id.is_empty() || provider.is_empty() { return Ok(None); } let conn = self.lock_conn()?; conn.query_row( "SELECT id, user_id, workspace_id, mnote_session_id, acp_session_id, agent_id, profile, provider, remote_conversation_id, remote_url, status, metadata_json, created_at, updated_at, deleted_at, revision FROM ai_external_conversation_bindings WHERE user_id = ?1 AND (?2 IS NULL OR workspace_id = ?2) AND mnote_session_id = ?3 AND provider = ?4 LIMIT 1", params![user_id, workspace_id, mnote_session_id, provider], row_to_ai_external_conversation_binding, ) .optional() .map_err(ControlPlaneError::from) } fn list_ai_external_conversation_bindings( &self, user_id: &str, workspace_id: Option<&str>, mnote_session_id: &str, limit: usize, ) -> Result, ControlPlaneError> { let user_id = user_id.trim(); let mnote_session_id = mnote_session_id.trim(); if user_id.is_empty() || mnote_session_id.is_empty() { return Ok(Vec::new()); } let conn = self.lock_conn()?; let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, mnote_session_id, acp_session_id, agent_id, profile, provider, remote_conversation_id, remote_url, status, metadata_json, created_at, updated_at, deleted_at, revision FROM ai_external_conversation_bindings WHERE user_id = ?1 AND (?2 IS NULL OR workspace_id = ?2) AND mnote_session_id = ?3 ORDER BY updated_at DESC LIMIT ?4", )?; let rows = stmt .query_map( params![user_id, workspace_id, mnote_session_id, limit.max(1) as i64], row_to_ai_external_conversation_binding, )? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(rows) } fn mark_ai_external_conversation_binding_status( &self, user_id: &str, workspace_id: Option<&str>, mnote_session_id: &str, provider: &str, status: &str, metadata_json: Option<&str>, ) -> Result { let user_id = user_id.trim(); let mnote_session_id = mnote_session_id.trim(); let provider = provider.trim(); validate_ai_external_conversation_status(status)?; let metadata_json = metadata_json.unwrap_or("{}"); serde_json::from_str::(metadata_json)?; if user_id.is_empty() || mnote_session_id.is_empty() || provider.is_empty() { return Ok(0); } let conn = self.lock_conn()?; let now = now_text(); let deleted_at = if status == "active" { None } else { Some(now.clone()) }; let changed = conn.execute( "UPDATE ai_external_conversation_bindings SET status = ?1, metadata_json = ?2, updated_at = ?3, deleted_at = COALESCE(?4, deleted_at), revision = revision + 1 WHERE user_id = ?5 AND (?6 IS NULL OR workspace_id = ?6) AND mnote_session_id = ?7 AND provider = ?8", params![ status, metadata_json, now, deleted_at, user_id, workspace_id, mnote_session_id, provider ], )?; Ok(changed) } fn rename_ai_runtime_session( &self, user_id: &str, session_id: &str, workspace_id: Option<&str>, title: &str, ) -> Result, ControlPlaneError> { let user_id = user_id.trim(); let session_id = session_id.trim(); let title = title.trim(); if user_id.is_empty() || session_id.is_empty() || title.is_empty() { return Err(ControlPlaneError::InvalidInput( "ai runtime session user_id/session_id/title 不能为空".to_string(), )); } let conn = self.lock_conn()?; let now = now_text(); conn.execute( "UPDATE ai_runtime_runs SET title = ?1, updated_at = ?2, revision = revision + 1 WHERE user_id = ?3 AND session_id = ?4 AND (?5 IS NULL OR workspace_id = ?5) AND deleted_at IS NULL", params![title, now, user_id, session_id, workspace_id], )?; let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision FROM ai_runtime_runs WHERE user_id = ?1 AND session_id = ?2 AND (?3 IS NULL OR workspace_id = ?3) AND deleted_at IS NULL ORDER BY updated_at DESC", )?; let rows = stmt .query_map( params![user_id, session_id, workspace_id], row_to_ai_runtime_run, )? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(rows) } fn auto_title_ai_runtime_session( &self, user_id: &str, session_id: &str, workspace_id: Option<&str>, ) -> Result, ControlPlaneError> { let user_id = user_id.trim(); let session_id = session_id.trim(); if user_id.is_empty() || session_id.is_empty() { return Err(ControlPlaneError::InvalidInput( "ai runtime session user_id/session_id 不能为空".to_string(), )); } let conn = self.lock_conn()?; let existing = conn .query_row( "SELECT id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision FROM ai_runtime_runs WHERE user_id = ?1 AND session_id = ?2 AND (?3 IS NULL OR workspace_id = ?3) AND deleted_at IS NULL ORDER BY updated_at DESC LIMIT 1", params![user_id, session_id, workspace_id], row_to_ai_runtime_run, ) .optional()?; let Some(existing) = existing else { return Ok(None); }; let title = existing .title .clone() .filter(|value| !value.trim().is_empty()) .unwrap_or_else(|| { derive_ai_runtime_title(&existing.payload_json, &existing.session_id) }); let now = now_text(); let revision = existing.revision + 1; conn.execute( "UPDATE ai_runtime_runs SET title = ?1, updated_at = ?2, revision = ?3 WHERE id = ?4", params![title, now, revision, existing.id], )?; Ok(Some(AiRuntimeRunRecord { title: Some(title), updated_at: now, revision, ..existing })) } fn delete_ai_runtime_session( &self, user_id: &str, session_id: &str, workspace_id: Option<&str>, ) -> Result { let user_id = user_id.trim(); let session_id = session_id.trim(); if user_id.is_empty() || session_id.is_empty() { return Err(ControlPlaneError::InvalidInput( "ai runtime session user_id/session_id 不能为空".to_string(), )); } let conn = self.lock_conn()?; let changed = conn.execute( "UPDATE ai_runtime_runs SET deleted_at = ?1, updated_at = ?1, revision = revision + 1 WHERE user_id = ?2 AND session_id = ?3 AND (?4 IS NULL OR workspace_id = ?4) AND deleted_at IS NULL", params![now_text(), user_id, session_id, workspace_id], )?; Ok(changed) } // ----------------------------------------------------------------------- // P2: AI tool events (receipts) — 7-71 // ----------------------------------------------------------------------- fn append_ai_tool_event( &self, input: AppendAiToolEventInput, ) -> Result { if input.user_id.trim().is_empty() || input.session_id.trim().is_empty() { return Err(ControlPlaneError::InvalidInput( "ai tool event user_id/session_id 不能为空".to_string(), )); } serde_json::from_str::(&input.payload_json)?; let conn = self.lock_conn()?; let record = AiToolEventRecord { id: input.id.unwrap_or_else(|| new_id("ate")), user_id: input.user_id, workspace_id: input.workspace_id, session_id: input.session_id, run_id: input.run_id, provider: input.provider, provider_session_id: input.provider_session_id, tool_name: input.tool_name, allowed: input.allowed, deny_reason: input.deny_reason, root_uri: input.root_uri, page_path: input.page_path, normalized_file_path: input.normalized_file_path, diff_summary: input.diff_summary, citation_count: input.citation_count, before_file_version: input.before_file_version, after_file_version: input.after_file_version, payload_json: input.payload_json, created_at: now_text(), deleted_at: None, }; conn.execute( "INSERT INTO ai_tool_events (id, user_id, workspace_id, session_id, run_id, provider, provider_session_id, tool_name, allowed, deny_reason, root_uri, page_path, normalized_file_path, diff_summary, citation_count, before_file_version, after_file_version, payload_json, created_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19)", params![ record.id, record.user_id, record.workspace_id, record.session_id, record.run_id, record.provider, record.provider_session_id, record.tool_name, record.allowed, record.deny_reason, record.root_uri, record.page_path, record.normalized_file_path, record.diff_summary, record.citation_count, record.before_file_version, record.after_file_version, record.payload_json, record.created_at ], )?; Ok(record) } fn list_ai_tool_events( &self, user_id: &str, session_id: Option<&str>, limit: usize, ) -> Result, ControlPlaneError> { let conn = self.lock_conn()?; let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, session_id, run_id, provider, provider_session_id, tool_name, allowed, deny_reason, root_uri, page_path, normalized_file_path, diff_summary, citation_count, before_file_version, after_file_version, payload_json, created_at, deleted_at FROM ai_tool_events WHERE user_id = ?1 AND deleted_at IS NULL AND (?2 IS NULL OR session_id = ?2) ORDER BY created_at DESC LIMIT ?3", )?; let rows = stmt .query_map( params![user_id, session_id, limit as i64], row_to_ai_tool_event, )? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(rows) } // ----------------------------------------------------------------------- // P2: AI file patches — 7-71 // ----------------------------------------------------------------------- fn append_ai_file_patch( &self, input: AppendAiFilePatchInput, ) -> Result { if input.user_id.trim().is_empty() || input.session_id.trim().is_empty() || input.tool_event_id.trim().is_empty() { return Err(ControlPlaneError::InvalidInput( "ai file patch user_id/session_id/tool_event_id 不能为空".to_string(), )); } serde_json::from_str::(&input.patch_summary_json)?; let conn = self.lock_conn()?; let record = AiFilePatchRecord { id: input.id.unwrap_or_else(|| new_id("afp")), user_id: input.user_id, workspace_id: input.workspace_id, session_id: input.session_id, run_id: input.run_id, tool_event_id: input.tool_event_id, root_uri: input.root_uri, relative_path: input.relative_path, before_file_version: input.before_file_version, after_file_version: input.after_file_version, patch_summary_json: input.patch_summary_json, created_at: now_text(), deleted_at: None, }; conn.execute( "INSERT INTO ai_file_patches (id, user_id, workspace_id, session_id, run_id, tool_event_id, root_uri, relative_path, before_file_version, after_file_version, patch_summary_json, created_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)", params![ record.id, record.user_id, record.workspace_id, record.session_id, record.run_id, record.tool_event_id, record.root_uri, record.relative_path, record.before_file_version, record.after_file_version, record.patch_summary_json, record.created_at ], )?; Ok(record) } fn list_ai_file_patches( &self, user_id: &str, session_id: Option<&str>, limit: usize, ) -> Result, ControlPlaneError> { let conn = self.lock_conn()?; let mut stmt = conn.prepare( "SELECT id, user_id, workspace_id, session_id, run_id, tool_event_id, root_uri, relative_path, before_file_version, after_file_version, patch_summary_json, created_at, deleted_at FROM ai_file_patches WHERE user_id = ?1 AND deleted_at IS NULL AND (?2 IS NULL OR session_id = ?2) ORDER BY created_at DESC LIMIT ?3", )?; let rows = stmt .query_map( params![user_id, session_id, limit as i64], row_to_ai_file_patch, )? .collect::, _>>() .map_err(ControlPlaneError::from)?; Ok(rows) } } fn max_permission<'a>(left: &'a str, right: &'a str) -> &'a str { if permission_rank(right) > permission_rank(left) { right } else { left } } fn permission_rank(permission: &str) -> u8 { match permission { "admin" => 3, "write" => 2, "read" => 1, _ => 0, } } // libSQL 和 rusqlite 都会初始化 SQLite C API;同一单元测试进程里混跑 // 两套 store 容易触发 libSQL 的线程模型断言。默认用独立 integration test // 覆盖 Turso/libSQL local 行为,这组镜像测试仅用于手动深度排查。 #[cfg(all(test, feature = "turso-unit-tests"))] mod tests { use super::*; use crate::model::{ password_hash_v1, session_token_hash, AppendAiRuntimeEventInput, AuthenticatePasswordInput, CreatePasswordIdentityInput, UpsertAiExternalConversationBindingInput, UpsertAiPolicyInput, UpsertAiRuntimeRunInput, UpsertNavigationRecentInput, UpsertSidebarShortcutInput, UpsertSyncStateInput, UpsertUserUiPreferenceInput, }; fn store() -> TursoControlPlaneStore { TursoControlPlaneStore::in_memory().expect("in-memory control plane") } fn create_user(store: &TursoControlPlaneStore, id: &str) -> UserRecord { store .upsert_user(UpsertUserInput { id: Some(id.to_string()), email: Some(format!("{id}@example.com")), username: id.to_string(), display_name: id.to_string(), role: None, password_hash: None, }) .expect("upsert user") } fn create_password_identity( store: &TursoControlPlaneStore, username: &str, email: &str, password: &str, ) -> UserRecord { let user = create_user(store, username); store .create_password_identity(CreatePasswordIdentityInput { user_id: user.id.clone(), email: Some(email.to_string()), username: username.to_string(), password: password.to_string(), }) .expect("create password identity"); user } #[test] fn upsert_user_inserts_and_updates_revision() { let store = store(); let user = create_user(&store, "alice"); assert_eq!(user.revision, 1); let updated = store .upsert_user(UpsertUserInput { id: Some("alice".to_string()), email: Some("alice2@example.com".to_string()), username: "alice".to_string(), display_name: "Alice".to_string(), role: Some("admin".to_string()), password_hash: None, }) .expect("update user"); assert_eq!(updated.email.as_deref(), Some("alice2@example.com")); assert_eq!(updated.role, "admin"); assert_eq!(updated.revision, 2); } #[test] fn list_users_returns_stable_ordering() { let store = store(); // Create 3 users in known order let u1 = create_user(&store, "bravo"); let u2 = create_user(&store, "alpha"); let u3 = create_user(&store, "charlie"); let users = store.list_users(10).expect("list users"); assert_eq!(users.len(), 3, "should return all three users"); // Order must be by created_at ASC, id ASC // u1 (bravo) created first, u2 (alpha) second, u3 (charlie) third assert_eq!(users[0].id, u1.id, "first created should be first"); assert_eq!(users[1].id, u2.id, "second created should be second"); assert_eq!(users[2].id, u3.id, "third created should be third"); } #[test] fn list_users_respects_limit() { let store = store(); for i in 0..5usize { create_user(&store, &format!("user{i}")); } let users = store.list_users(3).expect("list users with limit"); assert_eq!(users.len(), 3, "should respect limit=3"); } #[test] fn list_users_excludes_password_data() { let store = store(); create_user(&store, "nopassword"); let users = store.list_users(10).expect("list users"); assert_eq!(users.len(), 1); // UserRecord has no password_hash field — this is a compile-time // guarantee that list_users cannot leak password data. assert!(users[0].email.is_some()); } #[test] fn ensure_default_workspace_creates_owner_grant() { let store = store(); create_user(&store, "shujuan"); let workspace = store .ensure_default_workspace("shujuan") .expect("default workspace"); assert_eq!(workspace.name, "shujuan 的空间"); let access = store .resolve_access("shujuan", &workspace.root_uri) .expect("resolve access"); assert_eq!(access.permission, "write"); assert_eq!(access.grant_ids.len(), 1); } #[test] fn upsert_workspace_supports_explicit_local_root_for_dev_seed() { let store = store(); create_user(&store, "seed_owner"); let workspace = store .upsert_workspace(UpsertWorkspaceInput { id: Some("local-ws:seed-owner:custom".to_string()), owner_user_id: "seed_owner".to_string(), name: "Seed Workspace".to_string(), kind: Some("personal".to_string()), root_uri: "file:///tmp/mnote-seed-workspace".to_string(), root_path: "/tmp/mnote-seed-workspace".to_string(), source_kind: Some("local_folder".to_string()), }) .expect("upsert workspace"); assert_eq!(workspace.id, "local-ws:seed-owner:custom"); assert_eq!(workspace.status, "active"); assert_eq!(workspace.root_uri, "file:///tmp/mnote-seed-workspace"); let updated = store .upsert_workspace(UpsertWorkspaceInput { id: Some(workspace.id.clone()), owner_user_id: "seed_owner".to_string(), name: "Seed Workspace Renamed".to_string(), kind: Some("personal".to_string()), root_uri: "file:///tmp/mnote-seed-workspace".to_string(), root_path: "/tmp/mnote-seed-workspace".to_string(), source_kind: Some("local_folder".to_string()), }) .expect("update workspace"); assert_eq!(updated.name, "Seed Workspace Renamed"); assert_eq!(updated.revision, workspace.revision + 1); assert_eq!(updated.status, "active"); let missing_user = store .upsert_workspace(UpsertWorkspaceInput { id: Some("local-ws:missing:custom".to_string()), owner_user_id: "missing".to_string(), name: "Missing".to_string(), kind: Some("personal".to_string()), root_uri: "file:///tmp/missing".to_string(), root_path: "/tmp/missing".to_string(), source_kind: Some("local_folder".to_string()), }) .expect_err("missing owner should fail"); assert!(matches!(missing_user, ControlPlaneError::NotFound(_))); } #[test] fn explicit_write_grant_out_ranks_read_grant() { let store = store(); create_user(&store, "bob"); store .grant_directory_access(DirectoryGrantInput { user_id: "bob".to_string(), workspace_id: None, root_uri: "local://shared".to_string(), root_path: "/tmp/shared".to_string(), permission: "read".to_string(), recursive: true, capabilities: vec![], source: "admin".to_string(), created_by: None, }) .expect("read grant"); store .grant_directory_access(DirectoryGrantInput { user_id: "bob".to_string(), workspace_id: None, root_uri: "local://shared".to_string(), root_path: "/tmp/shared".to_string(), permission: "write".to_string(), recursive: true, capabilities: vec!["ai".to_string()], source: "admin".to_string(), created_by: None, }) .expect("write grant"); let access = store .resolve_access("bob", "local://shared/page.md") .expect("resolve access"); assert_eq!(access.permission, "write"); assert_eq!(access.grant_ids.len(), 2); } #[test] fn list_and_revoke_directory_grant_updates_resolved_access() { let store = store(); create_user(&store, "reader"); let grant = store .grant_directory_access(DirectoryGrantInput { user_id: "reader".to_string(), workspace_id: None, root_uri: "local://shared".to_string(), root_path: "/tmp/shared".to_string(), permission: "read".to_string(), recursive: true, capabilities: vec![], source: "admin".to_string(), created_by: None, }) .expect("read grant"); let grants = store .list_directory_grants_for_actor("reader") .expect("list grants"); assert_eq!(grants.len(), 1); assert_eq!(grants[0].id, grant.id); store .revoke_directory_grant(&grant.id, None) .expect("revoke grant"); let access = store .resolve_access("reader", "local://shared/page.md") .expect("resolve revoked access"); assert_eq!(access.permission, "none"); } #[test] fn resolve_access_supports_legacy_directory_grants_with_plain_path_root_uri() { let store = store(); create_user(&store, "legacy_reader"); store .grant_directory_access(DirectoryGrantInput { user_id: "legacy_reader".to_string(), workspace_id: None, root_uri: "/tmp/shared".to_string(), root_path: "/tmp/shared".to_string(), permission: "read".to_string(), recursive: true, capabilities: vec![], source: "legacy".to_string(), created_by: None, }) .expect("legacy grant"); let access = store .resolve_access("legacy_reader", "file:///tmp/shared/page.md") .expect("resolve access"); assert_eq!(access.permission, "read"); assert_eq!(access.grant_ids.len(), 1); } #[test] fn resolve_access_does_not_match_sibling_uri_prefix() { let store = store(); create_user(&store, "reader"); store .grant_directory_access(DirectoryGrantInput { user_id: "reader".to_string(), workspace_id: None, root_uri: "file:///tmp/mnote-ai-root".to_string(), root_path: "/tmp/mnote-ai-root".to_string(), permission: "write".to_string(), recursive: true, capabilities: vec!["ai".to_string()], source: "admin".to_string(), created_by: None, }) .expect("grant root"); let inside = store .resolve_access("reader", "file:///tmp/mnote-ai-root/page.md") .expect("resolve inside"); assert_eq!(inside.permission, "write"); let sibling = store .resolve_access("reader", "file:///tmp/mnote-ai-root-sibling/page.md") .expect("resolve sibling"); assert_eq!(sibling.permission, "none"); assert!(sibling.grant_ids.is_empty()); } #[test] fn share_link_token_is_hashed_and_revocable() { let store = store(); create_user(&store, "owner"); create_user(&store, "ws_owner"); let workspace = store .ensure_default_workspace("ws_owner") .expect("workspace"); let created = store .create_share_link(CreateShareLinkInput { id: None, workspace_id: workspace.id, resource_kind: "page".to_string(), resource_id: "page_1".to_string(), token: Some("plain-token".to_string()), permission: "read".to_string(), created_by: "ws_owner".to_string(), expires_at: None, }) .expect("create share link"); assert_eq!(created.token, "plain-token"); assert!(created.link.token_hash.starts_with("sha256-v1:")); assert_ne!(created.link.token_hash, "plain-token"); let resolved = store .resolve_share_link(&created.link.token_hash) .expect("resolve share link") .expect("active share link"); assert_eq!(resolved.id, created.link.id); let listed = store .list_share_links(&created.link.workspace_id) .expect("list links"); assert_eq!(listed.len(), 1); store .revoke_share_link(&created.link.id) .expect("revoke share"); assert!(store .resolve_share_link(&created.link.token_hash) .expect("resolve revoked share") .is_none()); } #[test] fn audit_log_appends_and_lists_latest_first() { let store = store(); create_user(&store, "owner"); store .append_audit(AppendAuditInput { actor_user_id: Some("owner".to_string()), action: "control.share.created".to_string(), target_kind: "share_link".to_string(), target_id: Some("share_1".to_string()), metadata_json: "{}".to_string(), }) .expect("append audit"); store .append_audit(AppendAuditInput { actor_user_id: Some("owner".to_string()), action: "control.share.revoked".to_string(), target_kind: "share_link".to_string(), target_id: Some("share_1".to_string()), metadata_json: "{\"reason\":\"test\"}".to_string(), }) .expect("append audit 2"); let rows = store.list_audit_log(10).expect("list audit"); assert_eq!(rows.len(), 2); assert_eq!(rows[0].action, "control.share.revoked"); assert_eq!(rows[1].action, "control.share.created"); } #[test] fn drain_outbox_marks_events_delivered() { let store = store(); store .append_outbox(OutboxEventInput { topic: "control".to_string(), event_type: "control.workspace.updated".to_string(), payload_json: "{}".to_string(), }) .expect("append event"); let first = store.drain_outbox(10).expect("first drain"); assert_eq!(first.len(), 1); let second = store.drain_outbox(10).expect("second drain"); assert!(second.is_empty()); } #[test] fn mark_outbox_delivered_marks_one_pending_event() { let store = store(); let first = store .append_outbox(OutboxEventInput { topic: "control".to_string(), event_type: "control.grant.created".to_string(), payload_json: "{}".to_string(), }) .expect("first event"); let second = store .append_outbox(OutboxEventInput { topic: "control".to_string(), event_type: "control.share.created".to_string(), payload_json: "{}".to_string(), }) .expect("second event"); store .mark_outbox_delivered(&first.id) .expect("mark first delivered"); let pending = store.drain_outbox(10).expect("drain pending"); assert_eq!(pending.len(), 1); assert_eq!(pending[0].id, second.id); } #[test] fn sidebar_shortcuts_are_user_scoped_and_upserted_by_target() { let store = store(); create_user(&store, "alice"); create_user(&store, "bob"); let workspace = store.ensure_default_workspace("alice").expect("workspace"); let first = store .upsert_sidebar_shortcut(UpsertSidebarShortcutInput { id: None, user_id: "alice".to_string(), workspace_id: workspace.id.clone(), root_uri: Some("file:///tmp/mnote".to_string()), kind: "folder".to_string(), source_kind: "local_folder".to_string(), target_id: "local-dir:design".to_string(), relative_path: Some("design".to_string()), document_id: None, title: "design".to_string(), icon: Some("folder_open".to_string()), sort_order: 0, metadata_json: "{}".to_string(), }) .expect("insert shortcut"); let updated = store .upsert_sidebar_shortcut(UpsertSidebarShortcutInput { id: None, user_id: "alice".to_string(), workspace_id: workspace.id.clone(), root_uri: Some("file:///tmp/mnote".to_string()), kind: "folder".to_string(), source_kind: "local_folder".to_string(), target_id: "local-dir:design".to_string(), relative_path: Some("design".to_string()), document_id: None, title: "Design".to_string(), icon: Some("folder_open".to_string()), sort_order: 4, metadata_json: "{\"scope\":\"filetree\"}".to_string(), }) .expect("upsert shortcut"); assert_eq!(updated.id, first.id); assert_eq!(updated.title, "Design"); assert_eq!(updated.root_uri.as_deref(), Some("file:///tmp/mnote")); assert_eq!(updated.sort_order, 4); assert_eq!(updated.revision, first.revision + 1); let alice_shortcuts = store .list_sidebar_shortcuts("alice", &workspace.id) .expect("list alice shortcuts"); assert_eq!(alice_shortcuts.len(), 1); assert_eq!(alice_shortcuts[0].target_id, "local-dir:design"); let alice_global_local_shortcuts = store .list_sidebar_shortcuts_with_global_local("alice", "cloud-workspace") .expect("list alice global local shortcuts"); assert_eq!(alice_global_local_shortcuts.len(), 1); assert_eq!(alice_global_local_shortcuts[0].workspace_id, workspace.id); let bob_shortcuts = store .list_sidebar_shortcuts("bob", &workspace.id) .expect("list bob shortcuts"); assert!(bob_shortcuts.is_empty()); store .delete_sidebar_shortcut("alice", &updated.id) .expect("delete shortcut"); let after_delete = store .list_sidebar_shortcuts("alice", &workspace.id) .expect("list deleted shortcuts"); assert!(after_delete.is_empty()); } #[test] fn user_ui_preferences_are_scoped_upserted_and_user_isolated() { let store = store(); create_user(&store, "alice"); create_user(&store, "bob"); let first = store .upsert_user_ui_preference(UpsertUserUiPreferenceInput { id: None, user_id: "alice".to_string(), workspace_id: Some("local:_mnt_Data1T_mnote".to_string()), source_kind: Some("local_folder".to_string()), scope_kind: "source_family".to_string(), scope_id: "external_local_folder".to_string(), key: "hideTitleHeader".to_string(), value_json: "false".to_string(), }) .expect("insert preference"); let updated = store .upsert_user_ui_preference(UpsertUserUiPreferenceInput { id: None, user_id: "alice".to_string(), workspace_id: Some("local:_mnt_Data1T_mnote".to_string()), source_kind: Some("local_folder".to_string()), scope_kind: "source_family".to_string(), scope_id: "external_local_folder".to_string(), key: "hideTitleHeader".to_string(), value_json: "true".to_string(), }) .expect("upsert preference"); assert_eq!(updated.id, first.id); assert_eq!(updated.value_json, "true"); assert_eq!(updated.revision, first.revision + 1); let alice_preferences = store .list_user_ui_preferences( "alice", Some("local:_mnt_Data1T_mnote"), Some("local_folder"), ) .expect("list alice preferences"); assert_eq!(alice_preferences.len(), 1); assert_eq!(alice_preferences[0].scope_kind, "source_family"); assert_eq!(alice_preferences[0].scope_id, "external_local_folder"); let bob_preferences = store .list_user_ui_preferences("bob", Some("local:_mnt_Data1T_mnote"), Some("local_folder")) .expect("list bob preferences"); assert!(bob_preferences.is_empty()); let sidebar_scope = "filetree:root-hash:design-hash"; let sidebar_value = serde_json::json!({ "schemaVersion": 1, "treeKind": "filetree", "rootUri": "file:///mnt/Data1T/mnote", "scope": "design", "expandedRelativePaths": ["design/05-editor-mainline"], "expandedIds": [], "selectedId": "", "focusedId": "", "activeId": "", "scrollTop": 0 }) .to_string(); let sidebar_first = store .upsert_user_ui_preference(UpsertUserUiPreferenceInput { id: None, user_id: "alice".to_string(), workspace_id: Some("local:_mnt_Data1T_mnote".to_string()), source_kind: Some("local_folder".to_string()), scope_kind: "sidebar_tree".to_string(), scope_id: sidebar_scope.to_string(), key: "sidebarTreeViewState.v1".to_string(), value_json: sidebar_value, }) .expect("insert sidebar tree view state"); let sidebar_updated = store .upsert_user_ui_preference(UpsertUserUiPreferenceInput { id: None, user_id: "alice".to_string(), workspace_id: Some("local:_mnt_Data1T_mnote".to_string()), source_kind: Some("local_folder".to_string()), scope_kind: "sidebar_tree".to_string(), scope_id: sidebar_scope.to_string(), key: "sidebarTreeViewState.v1".to_string(), value_json: serde_json::json!({ "schemaVersion": 1, "treeKind": "filetree", "rootUri": "file:///mnt/Data1T/mnote", "scope": "design", "expandedRelativePaths": ["design/05-editor-mainline", "design/05-editor-mainline/process"], "expandedIds": [], "selectedId": "", "focusedId": "", "activeId": "", "scrollTop": 0 }) .to_string(), }) .expect("upsert sidebar tree view state"); assert_eq!(sidebar_updated.id, sidebar_first.id); assert_eq!(sidebar_updated.revision, sidebar_first.revision + 1); let alice_sidebar_preferences = store .list_user_ui_preferences( "alice", Some("local:_mnt_Data1T_mnote"), Some("local_folder"), ) .expect("list alice sidebar preferences"); assert!(alice_sidebar_preferences.iter().any(|preference| { preference.scope_kind == "sidebar_tree" && preference.scope_id == sidebar_scope && preference.key == "sidebarTreeViewState.v1" })); let bob_sidebar_preferences = store .list_user_ui_preferences("bob", Some("local:_mnt_Data1T_mnote"), Some("local_folder")) .expect("list bob sidebar preferences"); assert!(bob_sidebar_preferences.is_empty()); } #[test] fn ai_agent_profile_policy_provisions_personal_and_shared_boundaries() { let store = store(); create_user(&store, "alice"); create_user(&store, "bob"); store .upsert_user(UpsertUserInput { id: Some("admin".to_string()), email: Some("admin@example.com".to_string()), username: "admin".to_string(), display_name: "admin".to_string(), role: Some("admin".to_string()), password_hash: None, }) .expect("admin user"); let alice_profiles = store .ensure_ai_agent_profile_policy("alice", false) .expect("alice profiles"); assert_eq!(alice_profiles.len(), 11); let alice_personal = alice_profiles .iter() .find(|item| item.profile.profile_kind == "personal") .expect("alice personal profile"); assert_eq!( alice_personal.profile.owner_user_id.as_deref(), Some("alice") ); assert!(alice_personal.grant.can_manage_skills); let alice_shared = alice_profiles .iter() .find(|item| item.profile.id == "shared_lite") .expect("alice shared profile"); assert!(alice_shared.grant.can_run); assert!(!alice_shared.grant.can_manage_skills); for (profile_id, isolated_profile, display_name) in [ ( "shared_deepseek_chat", "openclaw-deepseek-chat", "DeepSeek Chat", ), ("shared_gemini_chat", "openclaw-gemini-chat", "Gemini Chat"), ("shared_doubao_chat", "openclaw-doubao-chat", "豆包 Chat"), ( "shared_api_deepseek_flash_chat", "api-deepseek-flash-chat", "DeepSeek Flash Chat", ), ( "shared_api_deepseek_pro_chat", "api-deepseek-pro-chat", "DeepSeek Pro Chat", ), ("shared_api_gpt_chat", "api-gpt-chat", "GPT Chat"), ("shared_api_kimi_chat", "api-kimi-chat", "Kimi Chat"), ( "shared_api_gemini_chat", "api-gemini-chat", "Gemini API Chat", ), ("shared_api_grok_chat", "api-grok-chat", "Grok API Chat"), ] { let alice_chat = alice_profiles .iter() .find(|item| item.profile.id == profile_id) .expect("alice shared chat profile"); assert_eq!(alice_chat.profile.isolated_profile_name, isolated_profile); assert_eq!(alice_chat.profile.display_name, display_name); assert!(alice_chat.grant.can_run); assert!(!alice_chat.grant.can_manage_skills); assert!(!alice_chat.grant.can_manage_config); } let bob_profiles = store .ensure_ai_agent_profile_policy("bob", false) .expect("bob profiles"); let bob_personal = bob_profiles .iter() .find(|item| item.profile.profile_kind == "personal") .expect("bob personal profile"); assert_ne!(bob_personal.profile.id, alice_personal.profile.id); assert!(store .resolve_ai_agent_profile("bob", false, &alice_personal.profile.id) .expect("resolve cross user") .is_none()); let admin_shared = store .resolve_ai_agent_profile("admin", true, "shared_lite") .expect("admin shared") .expect("admin can see shared"); assert!(admin_shared.grant.can_manage_skills); assert!(admin_shared.grant.can_manage_config); for profile_id in [ "shared_deepseek_chat", "shared_gemini_chat", "shared_doubao_chat", "shared_api_deepseek_flash_chat", "shared_api_deepseek_pro_chat", "shared_api_gpt_chat", "shared_api_kimi_chat", "shared_api_gemini_chat", "shared_api_grok_chat", ] { let admin_chat = store .resolve_ai_agent_profile("admin", true, profile_id) .expect("admin chat") .expect("admin can see shared chat profile"); assert!(admin_chat.grant.can_run); assert!(!admin_chat.grant.can_manage_skills); assert!(!admin_chat.grant.can_manage_config); } } #[test] fn navigation_recent_is_user_scoped_upserted_and_limited_by_kind() { let store = store(); create_user(&store, "alice"); create_user(&store, "bob"); let first = store .upsert_navigation_recent(UpsertNavigationRecentInput { id: None, user_id: "alice".to_string(), workspace_id: Some("local:_tmp_mnote".to_string()), kind: "folder".to_string(), source_kind: "local_folder".to_string(), root_uri: "file:///tmp/mnote".to_string(), relative_path: Some("design".to_string()), document_id: None, title: "design".to_string(), metadata_json: "{}".to_string(), }) .expect("insert folder recent"); let updated = store .upsert_navigation_recent(UpsertNavigationRecentInput { id: None, user_id: "alice".to_string(), workspace_id: Some("local:_tmp_mnote".to_string()), kind: "folder".to_string(), source_kind: "local_folder".to_string(), root_uri: "file:///tmp/mnote".to_string(), relative_path: Some("design".to_string()), document_id: None, title: "Design".to_string(), metadata_json: "{\"from\":\"test\"}".to_string(), }) .expect("upsert folder recent"); assert_eq!(updated.id, first.id); assert_eq!(updated.title, "Design"); assert_eq!(updated.revision, first.revision + 1); store .upsert_navigation_recent(UpsertNavigationRecentInput { id: None, user_id: "alice".to_string(), workspace_id: Some("local:_tmp_mnote".to_string()), kind: "page".to_string(), source_kind: "local_folder".to_string(), root_uri: "file:///tmp/mnote".to_string(), relative_path: Some("design/Plan.md".to_string()), document_id: Some("local-md:design~2FPlan.md".to_string()), title: "Plan".to_string(), metadata_json: "{}".to_string(), }) .expect("insert page recent"); for index in 0..12 { store .upsert_navigation_recent(UpsertNavigationRecentInput { id: None, user_id: "alice".to_string(), workspace_id: Some("local:_tmp_mnote".to_string()), kind: "folder".to_string(), source_kind: "local_folder".to_string(), root_uri: "file:///tmp/mnote".to_string(), relative_path: Some(format!("folder-{index}")), document_id: None, title: format!("folder-{index}"), metadata_json: "{}".to_string(), }) .expect("insert bounded folder recent"); } let alice_folders = store .list_navigation_recent("alice", Some("folder"), 50) .expect("list alice folders"); assert_eq!(alice_folders.len(), 10); assert_eq!(alice_folders[0].kind, "folder"); assert_eq!(alice_folders[0].title, "folder-11"); assert!(alice_folders.iter().all(|record| record.user_id == "alice")); assert!(!alice_folders.iter().any(|record| record.title == "Design")); let alice_pages = store .list_navigation_recent("alice", Some("page"), 20) .expect("list alice pages"); assert_eq!(alice_pages.len(), 1); assert_eq!( alice_pages[0].document_id.as_deref(), Some("local-md:design~2FPlan.md") ); let bob_recent = store .list_navigation_recent("bob", None, 20) .expect("list bob recent"); assert!(bob_recent.is_empty()); } #[test] fn ai_policy_upsert_returns_workspace_policy_before_user_policy() { let store = store(); create_user(&store, "ai_user"); let workspace = store .ensure_default_workspace("ai_user") .expect("default workspace"); let user_policy = store .upsert_ai_policy(UpsertAiPolicyInput { id: None, user_id: Some("ai_user".to_string()), workspace_id: None, allowed_roots_json: "[\"file:///user-root\"]".to_string(), model_policy_json: "{\"default\":\"local\"}".to_string(), quota_json: "{}".to_string(), }) .expect("user policy"); assert_eq!(user_policy.revision, 1); let workspace_policy = store .upsert_ai_policy(UpsertAiPolicyInput { id: None, user_id: Some("ai_user".to_string()), workspace_id: Some(workspace.id.clone()), allowed_roots_json: "[\"file:///workspace-root\"]".to_string(), model_policy_json: "{\"default\":\"workspace\"}".to_string(), quota_json: "{}".to_string(), }) .expect("workspace policy"); let resolved = store .get_ai_policy("ai_user", Some(&workspace.id)) .expect("get workspace policy") .expect("workspace policy exists"); assert_eq!(resolved.id, workspace_policy.id); assert!(resolved.allowed_roots_json.contains("workspace-root")); let user_only = store .get_ai_policy("ai_user", None) .expect("get user policy") .expect("user policy exists"); assert_eq!(user_only.id, user_policy.id); } #[test] fn ai_policy_user_upsert_does_not_overwrite_workspace_policy() { let store = store(); create_user(&store, "ai_user"); let workspace = store .ensure_default_workspace("ai_user") .expect("default workspace"); let workspace_policy = store .upsert_ai_policy(UpsertAiPolicyInput { id: None, user_id: Some("ai_user".to_string()), workspace_id: Some(workspace.id.clone()), allowed_roots_json: "[\"file:///workspace-root\"]".to_string(), model_policy_json: "{\"default\":\"workspace\"}".to_string(), quota_json: "{}".to_string(), }) .expect("workspace policy"); let user_policy = store .upsert_ai_policy(UpsertAiPolicyInput { id: None, user_id: Some("ai_user".to_string()), workspace_id: None, allowed_roots_json: "[\"file:///user-root\"]".to_string(), model_policy_json: "{\"default\":\"user\"}".to_string(), quota_json: "{}".to_string(), }) .expect("user policy"); assert_ne!(user_policy.id, workspace_policy.id); let user_only = store .get_ai_policy("ai_user", None) .expect("get user policy") .expect("user policy exists"); assert_eq!(user_only.id, user_policy.id); assert!(user_only.allowed_roots_json.contains("user-root")); let workspace_only = store .get_ai_policy("ai_user", Some(&workspace.id)) .expect("get workspace policy") .expect("workspace policy exists"); assert_eq!(workspace_only.id, workspace_policy.id); assert!(workspace_only.allowed_roots_json.contains("workspace-root")); } #[test] fn sync_state_upsert_tracks_cursor_status_and_revision() { let store = store(); create_user(&store, "sync_user"); let workspace = store .ensure_default_workspace("sync_user") .expect("default workspace"); let first = store .upsert_sync_state(UpsertSyncStateInput { id: None, workspace_id: workspace.id.clone(), remote_kind: "local_worker".to_string(), remote_id: Some("worker-1".to_string()), cursor: Some("cursor-1".to_string()), last_synced_at: None, status: "idle".to_string(), error_json: None, }) .expect("first sync state"); assert_eq!(first.revision, 1); let second = store .upsert_sync_state(UpsertSyncStateInput { id: None, workspace_id: workspace.id.clone(), remote_kind: "local_worker".to_string(), remote_id: Some("worker-1".to_string()), cursor: Some("cursor-2".to_string()), last_synced_at: Some("2026-05-22T00:00:00Z".to_string()), status: "error".to_string(), error_json: Some("{\"code\":\"test\"}".to_string()), }) .expect("second sync state"); assert_eq!(second.id, first.id); assert_eq!(second.revision, 2); assert_eq!(second.cursor.as_deref(), Some("cursor-2")); let loaded = store .get_sync_state(&workspace.id, "local_worker") .expect("get sync state") .expect("sync state exists"); assert_eq!(loaded.status, "error"); assert_eq!(loaded.error_json.as_deref(), Some("{\"code\":\"test\"}")); } #[test] fn ai_runtime_run_upsert_lists_updates_and_events() { let store = store(); create_user(&store, "ai_runtime_user"); let first = store .upsert_ai_runtime_run(UpsertAiRuntimeRunInput { id: None, user_id: "ai_runtime_user".to_string(), workspace_id: Some("ws_1".to_string()), document_id: Some("doc_1".to_string()), session_id: "sess_1".to_string(), run_id: "run_1".to_string(), title: Some("初始标题".to_string()), profile: "reasonix".to_string(), acp_runtime: "reasonix".to_string(), trace_id: Some("trace_1".to_string()), status: "running".to_string(), runtime_json: "{\"status\":\"running\"}".to_string(), payload_json: "{\"message\":\"读取当前页面\"}".to_string(), }) .expect("insert runtime run"); assert_eq!(first.revision, 1); let updated = store .upsert_ai_runtime_run(UpsertAiRuntimeRunInput { id: None, user_id: "ai_runtime_user".to_string(), workspace_id: Some("ws_1".to_string()), document_id: Some("doc_1".to_string()), session_id: "sess_1".to_string(), run_id: "run_1".to_string(), title: Some("更新标题".to_string()), profile: "reasonix".to_string(), acp_runtime: "reasonix".to_string(), trace_id: Some("trace_2".to_string()), status: "completed".to_string(), runtime_json: "{\"status\":\"completed\"}".to_string(), payload_json: "{\"message\":\"完成\"}".to_string(), }) .expect("update runtime run"); assert_eq!(updated.id, first.id); assert_eq!(updated.revision, 2); let runs = store .list_ai_runtime_runs( "ai_runtime_user", Some("ws_1"), Some("doc_1"), Some("sess_1"), 10, ) .expect("list runtime runs"); assert_eq!(runs.len(), 1); assert_eq!(runs[0].title.as_deref(), Some("更新标题")); let event = store .append_ai_runtime_event(AppendAiRuntimeEventInput { id: None, user_id: "ai_runtime_user".to_string(), workspace_id: Some("ws_1".to_string()), document_id: Some("doc_1".to_string()), session_id: "sess_1".to_string(), run_id: "run_1".to_string(), profile: "reasonix".to_string(), acp_runtime: "reasonix".to_string(), event_type: "message.delta".to_string(), payload_json: "{\"text\":\"hello\"}".to_string(), }) .expect("append runtime event"); assert_eq!(event.event_type, "message.delta"); store .append_ai_runtime_event(AppendAiRuntimeEventInput { id: None, user_id: "ai_runtime_user".to_string(), workspace_id: Some("ws_1".to_string()), document_id: Some("doc_1".to_string()), session_id: "sess_1".to_string(), run_id: "run_1".to_string(), profile: "reasonix".to_string(), acp_runtime: "reasonix".to_string(), event_type: "run.completed".to_string(), payload_json: "{\"status\":\"completed\"}".to_string(), }) .expect("append second runtime event"); let events = store .list_ai_runtime_events("ai_runtime_user", "run_1", 10) .expect("list runtime events"); assert_eq!(events.len(), 2); assert_eq!(events[0].payload_json, "{\"text\":\"hello\"}"); let found = store .find_ai_runtime_run("ai_runtime_user", "run_1") .expect("find runtime run") .expect("runtime run exists"); assert_eq!(found.status, "completed"); let journal_events = store .list_ai_runtime_journal_events("ai_runtime_user", "run_1", 1, 10) .expect("list runtime journal events"); assert_eq!(journal_events.len(), 1); assert_eq!(journal_events[0].seq, 2); assert_eq!(journal_events[0].event.event_type, "run.completed"); let deleted = store .delete_ai_runtime_events_for_run("ai_runtime_user", "run_1") .expect("delete runtime events"); assert_eq!(deleted, 2); assert!(store .list_ai_runtime_events("ai_runtime_user", "run_1", 10) .expect("list after delete") .is_empty()); } #[test] fn ai_runtime_session_management_renames_auto_titles_and_soft_deletes() { let store = store(); create_user(&store, "ai_runtime_user"); store .upsert_ai_runtime_run(UpsertAiRuntimeRunInput { id: None, user_id: "ai_runtime_user".to_string(), workspace_id: Some("ws_1".to_string()), document_id: Some("doc_1".to_string()), session_id: "sess_1".to_string(), run_id: "run_1".to_string(), title: None, profile: "reasonix".to_string(), acp_runtime: "reasonix".to_string(), trace_id: None, status: "completed".to_string(), runtime_json: "{\"status\":\"completed\"}".to_string(), payload_json: "{\"message\":\"请总结这篇文档\"}".to_string(), }) .expect("insert runtime run"); let titled = store .auto_title_ai_runtime_session("ai_runtime_user", "sess_1", Some("ws_1")) .expect("auto title") .expect("titled run"); assert_eq!(titled.title.as_deref(), Some("请总结这篇文档")); let renamed = store .rename_ai_runtime_session("ai_runtime_user", "sess_1", Some("ws_1"), "人工标题") .expect("rename session"); assert_eq!(renamed.len(), 1); assert_eq!(renamed[0].title.as_deref(), Some("人工标题")); let deleted = store .delete_ai_runtime_session("ai_runtime_user", "sess_1", Some("ws_1")) .expect("delete session"); assert_eq!(deleted, 1); let runs = store .list_ai_runtime_runs("ai_runtime_user", Some("ws_1"), None, Some("sess_1"), 10) .expect("list after delete"); assert!(runs.is_empty()); } #[test] fn ai_external_conversation_binding_is_user_scoped_and_statused() { let store = store(); create_user(&store, "doubao_user_a"); create_user(&store, "doubao_user_b"); let binding = store .upsert_ai_external_conversation_binding(UpsertAiExternalConversationBindingInput { id: None, user_id: "doubao_user_a".to_string(), workspace_id: Some("ws_1".to_string()), mnote_session_id: "chatonly_session_1".to_string(), acp_session_id: Some("acp_session_1".to_string()), agent_id: "chat_only".to_string(), profile: "openclaw-doubao-chat".to_string(), provider: "doubao-web".to_string(), remote_conversation_id: "38428454119180290".to_string(), remote_url: Some("https://www.doubao.com/chat/38428454119180290".to_string()), status: "active".to_string(), metadata_json: "{\"source\":\"provider.conversation.bound\"}".to_string(), }) .expect("upsert external conversation binding"); assert_eq!(binding.status, "active"); assert_eq!(binding.remote_conversation_id, "38428454119180290"); let found = store .find_ai_external_conversation_binding( "doubao_user_a", Some("ws_1"), "chatonly_session_1", "doubao-web", ) .expect("find binding") .expect("binding exists"); assert_eq!(found.id, binding.id); assert_eq!(found.acp_session_id.as_deref(), Some("acp_session_1")); let other_user = store .find_ai_external_conversation_binding( "doubao_user_b", Some("ws_1"), "chatonly_session_1", "doubao-web", ) .expect("find other user binding"); assert!(other_user.is_none()); let changed = store .mark_ai_external_conversation_binding_status( "doubao_user_a", Some("ws_1"), "chatonly_session_1", "doubao-web", "local_deleted", Some("{\"reason\":\"mnote_session_deleted\"}"), ) .expect("mark local deleted"); assert_eq!(changed, 1); let deleted = store .find_ai_external_conversation_binding( "doubao_user_a", Some("ws_1"), "chatonly_session_1", "doubao-web", ) .expect("find deleted binding") .expect("binding remains auditable"); assert_eq!(deleted.status, "local_deleted"); assert!(deleted.deleted_at.is_some()); } #[test] fn session_lookup_resolves_active_user_by_token_hash() { let store = store(); create_user(&store, "shujuan"); let session = store .create_session(CreateSessionInput { id: None, user_id: "shujuan".to_string(), token_hash: session_token_hash("raw-session-token"), user_agent: Some("mnote-test".to_string()), ip_hash: None, expires_at: None, }) .expect("create session"); let resolved = store .get_session_by_token_hash(&session.token_hash) .expect("lookup session") .expect("active session"); assert_eq!(resolved.session.user_id, "shujuan"); assert_eq!(resolved.user.id, "shujuan"); assert_eq!(resolved.user.email.as_deref(), Some("shujuan@example.com")); } #[test] fn revoked_session_is_not_resolved() { let store = store(); create_user(&store, "shujuan"); let session = store .create_session(CreateSessionInput { id: None, user_id: "shujuan".to_string(), token_hash: session_token_hash("revoked-token"), user_agent: None, ip_hash: None, expires_at: None, }) .expect("create session"); store.revoke_session(&session.id).expect("revoke session"); let resolved = store .get_session_by_token_hash(&session.token_hash) .expect("lookup session"); assert!(resolved.is_none()); } #[test] fn password_hash_v1_is_sha256_hex_placeholder() { assert!(password_hash_v1("secret").starts_with("sha256-v1:")); } #[test] fn authenticate_password_accepts_email_login_and_creates_session() { let store = store(); let test_password = ["correct", "horse"].join(" "); create_password_identity(&store, "alice", "alice@example.com", &test_password); let resolved = store .authenticate_password(AuthenticatePasswordInput { account: "alice@example.com".to_string(), password: test_password, session_id: None, token_hash: session_token_hash("alice-email-session"), user_agent: Some("mnote-test".to_string()), ip_hash: None, expires_at: None, }) .expect("authenticate by email"); assert_eq!(resolved.user.username, "alice"); assert_eq!( resolved.session.token_hash, session_token_hash("alice-email-session") ); let lookup = store .get_session_by_token_hash(&resolved.session.token_hash) .expect("lookup created session") .expect("created session"); assert_eq!(lookup.user.id, resolved.user.id); } #[test] fn authenticate_password_accepts_username_login() { let store = store(); let test_password = ["sword", "fish"].join(""); create_password_identity(&store, "bob", "bob@example.com", &test_password); let resolved = store .authenticate_password(AuthenticatePasswordInput { account: "bob".to_string(), password: test_password, session_id: None, token_hash: session_token_hash("bob-username-session"), user_agent: None, ip_hash: None, expires_at: None, }) .expect("authenticate by username"); assert_eq!(resolved.user.email.as_deref(), Some("bob@example.com")); assert_eq!( resolved.session.token_hash, session_token_hash("bob-username-session") ); } #[test] fn authenticate_password_rejects_wrong_password() { let store = store(); let right_password = ["right", "password"].join("-"); let wrong_password = ["wrong", "password"].join("-"); create_password_identity(&store, "chris", "chris@example.com", &right_password); let err = store .authenticate_password(AuthenticatePasswordInput { account: "chris@example.com".to_string(), password: wrong_password, session_id: None, token_hash: session_token_hash("chris-failed-session"), user_agent: None, ip_hash: None, expires_at: None, }) .expect_err("wrong password should fail"); assert!(matches!(err, ControlPlaneError::Unauthorized(_))); assert!(store .get_session_by_token_hash(&session_token_hash("chris-failed-session")) .expect("lookup failed session") .is_none()); } // --- Fault injection tests --- #[test] fn invalid_local_path_returns_storage_error_not_panic() { let path = "/tmp/__mnote_nonexistent_dir__/control-plane-unknown.db"; let result = TursoControlPlaneStore::open_local(path); let Err(error) = result else { panic!("invalid path should return an error"); }; assert!( matches!(&error, ControlPlaneError::Storage(_)), "expected Storage error, got {error:?}" ); } #[test] fn turso_remote_config_rejects_missing_url() { let result = TursoControlPlaneStore::open_with_config(TursoControlPlaneConfig { mode: TursoControlPlaneMode::Remote, path: None, remote_url: None, auth_token: Some("dummy-token".to_string()), sync_interval: None, }); let Err(error) = result else { panic!("missing URL should return error"); }; assert!( matches!(&error, ControlPlaneError::InvalidInput(_)), "expected InvalidInput for missing URL, got {error:?}" ); } #[test] fn turso_remote_config_rejects_missing_token() { let result = TursoControlPlaneStore::open_with_config(TursoControlPlaneConfig { mode: TursoControlPlaneMode::Remote, path: None, remote_url: Some("libsql://example.turso.io".to_string()), auth_token: None, sync_interval: None, }); let Err(error) = result else { panic!("missing token should return error"); }; assert!( matches!(&error, ControlPlaneError::InvalidInput(_)), "expected InvalidInput for missing token, got {error:?}" ); } #[test] fn turso_remote_config_rejects_empty_url() { let result = TursoControlPlaneStore::open_with_config(TursoControlPlaneConfig { mode: TursoControlPlaneMode::Remote, path: None, remote_url: Some("".to_string()), auth_token: Some("dummy-token".to_string()), sync_interval: None, }); let Err(error) = result else { panic!("empty URL should return error"); }; assert!( matches!(&error, ControlPlaneError::InvalidInput(_)), "expected InvalidInput for empty URL, got {error:?}" ); } #[test] fn turso_remote_config_rejects_empty_token() { let result = TursoControlPlaneStore::open_with_config(TursoControlPlaneConfig { mode: TursoControlPlaneMode::Remote, path: None, remote_url: Some("libsql://example.turso.io".to_string()), auth_token: Some("".to_string()), sync_interval: None, }); let Err(error) = result else { panic!("empty token should return error"); }; assert!( matches!(&error, ControlPlaneError::InvalidInput(_)), "expected InvalidInput for empty token, got {error:?}" ); } #[test] fn turso_remote_replica_config_rejects_missing_path() { let result = TursoControlPlaneStore::open_with_config(TursoControlPlaneConfig { mode: TursoControlPlaneMode::RemoteReplica, path: None, remote_url: Some("libsql://example.turso.io".to_string()), auth_token: Some("dummy-token".to_string()), sync_interval: None, }); let Err(error) = result else { panic!("missing replica path should return error"); }; assert!( matches!(&error, ControlPlaneError::InvalidInput(_)), "expected InvalidInput for missing replica path, got {error:?}" ); } #[test] fn turso_synced_config_rejects_missing_path() { let result = TursoControlPlaneStore::open_with_config(TursoControlPlaneConfig { mode: TursoControlPlaneMode::Synced, path: None, remote_url: Some("libsql://example.turso.io".to_string()), auth_token: Some("dummy-token".to_string()), sync_interval: None, }); let Err(error) = result else { panic!("missing synced path should return error"); }; assert!( matches!(&error, ControlPlaneError::InvalidInput(_)), "expected InvalidInput for missing synced path, got {error:?}" ); } #[test] fn timeout_variant_is_propagated_correctly() { let err = ControlPlaneError::Timeout("connection timed out".to_string()); let display = err.to_string(); assert!( display.starts_with("Timeout:"), "expected Timeout: prefix, got {display}" ); } #[test] fn rate_limit_variant_is_propagated_correctly() { let err = ControlPlaneError::RateLimit("too many requests".to_string()); let display = err.to_string(); assert!( display.starts_with("RateLimit:"), "expected RateLimit: prefix, got {display}" ); } #[test] fn ai_tool_events_append_and_list() { let store = store(); create_user(&store, "tool_user"); let event = store .append_ai_tool_event(AppendAiToolEventInput { id: None, user_id: "tool_user".to_string(), workspace_id: Some("ws_1".to_string()), session_id: "sess_1".to_string(), run_id: Some("run_1".to_string()), provider: "doubao-web".to_string(), provider_session_id: Some("ps_1".to_string()), tool_name: "mnote.local_file.read".to_string(), allowed: true, deny_reason: None, root_uri: "file:///tmp".to_string(), page_path: Some("page.md".to_string()), normalized_file_path: Some("/tmp/page.md".to_string()), diff_summary: Some("read file".to_string()), citation_count: 0, before_file_version: None, after_file_version: None, payload_json: "{}".to_string(), }) .expect("append tool event"); assert_eq!(event.tool_name, "mnote.local_file.read"); assert!(event.allowed); assert_eq!(event.citation_count, 0); store .append_ai_tool_event(AppendAiToolEventInput { id: None, user_id: "tool_user".to_string(), workspace_id: Some("ws_1".to_string()), session_id: "sess_1".to_string(), run_id: Some("run_2".to_string()), provider: "reasonix".to_string(), provider_session_id: None, tool_name: "fs.write".to_string(), allowed: false, deny_reason: Some("path not in allowed roots".to_string()), root_uri: "file:///etc".to_string(), page_path: None, normalized_file_path: Some("/etc/passwd".to_string()), diff_summary: None, citation_count: 0, before_file_version: None, after_file_version: None, payload_json: "{}".to_string(), }) .expect("append denied event"); let events = store .list_ai_tool_events("tool_user", Some("sess_1"), 10) .expect("list events"); assert_eq!(events.len(), 2); assert_eq!(events[0].tool_name, "fs.write"); assert!(!events[0].allowed); let all_events = store .list_ai_tool_events("tool_user", None, 10) .expect("list all"); assert_eq!(all_events.len(), 2); } #[test] fn ai_file_patches_append_and_list() { let store = store(); create_user(&store, "patch_user"); let event = store .append_ai_tool_event(AppendAiToolEventInput { id: Some("ate_turso_patch_ref".to_string()), user_id: "patch_user".to_string(), workspace_id: None, session_id: "sess_p1".to_string(), run_id: None, provider: "openclaw".to_string(), provider_session_id: None, tool_name: "mnote.local_file.patch".to_string(), allowed: true, deny_reason: None, root_uri: "file:///tmp".to_string(), page_path: Some("test.md".to_string()), normalized_file_path: Some("/tmp/test.md".to_string()), diff_summary: Some("edit file".to_string()), citation_count: 0, before_file_version: Some("v1".to_string()), after_file_version: Some("v2".to_string()), payload_json: "{}".to_string(), }) .expect("append reference tool event"); let patch = store .append_ai_file_patch(AppendAiFilePatchInput { id: None, user_id: "patch_user".to_string(), workspace_id: None, session_id: "sess_p1".to_string(), run_id: None, tool_event_id: event.id.clone(), root_uri: "file:///tmp".to_string(), relative_path: "test.md".to_string(), before_file_version: Some("v1".to_string()), after_file_version: Some("v2".to_string()), patch_summary_json: r#"{"insertions":5}"#.to_string(), }) .expect("append file patch"); assert_eq!(patch.tool_event_id, event.id); assert!(patch.deleted_at.is_none()); let patches = store .list_ai_file_patches("patch_user", Some("sess_p1"), 10) .expect("list patches"); assert_eq!(patches.len(), 1); let all_patches = store .list_ai_file_patches("patch_user", None, 10) .expect("list all"); assert_eq!(all_patches.len(), 1); } }