mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-30 23:14:37 +00:00
feat: Memory subtabs + WebGL memory graph; drop Knowledge Vaults (#3040)
This commit is contained in:
Steven Enamakel
@@ -191,8 +191,6 @@ fn build_registered_controllers() -> Vec<RegisteredController> {
|
||||
controllers.extend(crate::openhuman::skills::all_skills_registered_controllers());
|
||||
// User workspace and file management
|
||||
controllers.extend(crate::openhuman::workspace::all_workspace_registered_controllers());
|
||||
// Knowledge vaults — folder-of-files mirrored into memory
|
||||
controllers.extend(crate::openhuman::vault::all_vault_registered_controllers());
|
||||
// Skill tool registry
|
||||
controllers.extend(crate::openhuman::tools::all_tools_registered_controllers());
|
||||
// Unified read-only registry across MCP stdio tools and controller-backed tools
|
||||
@@ -341,7 +339,6 @@ fn build_declared_controller_schemas() -> Vec<ControllerSchema> {
|
||||
schemas.extend(crate::openhuman::javascript::all_javascript_controller_schemas());
|
||||
schemas.extend(crate::openhuman::skills::all_skills_controller_schemas());
|
||||
schemas.extend(crate::openhuman::workspace::all_workspace_controller_schemas());
|
||||
schemas.extend(crate::openhuman::vault::all_vault_controller_schemas());
|
||||
schemas.extend(crate::openhuman::tools::all_tools_controller_schemas());
|
||||
schemas.extend(crate::openhuman::tool_registry::all_tool_registry_controller_schemas());
|
||||
schemas.extend(crate::openhuman::memory::all_memory_controller_schemas());
|
||||
|
||||
@@ -1242,9 +1242,10 @@ fn is_prompt_injection_blocked_message(lower: &str) -> bool {
|
||||
/// boundary when a user typed/picked a path that doesn't resolve to an
|
||||
/// existing directory:
|
||||
///
|
||||
/// - `"root_path is not a directory: <path>"` —
|
||||
/// [`crate::openhuman::vault::ops::vault_create`] when the chosen vault
|
||||
/// folder doesn't exist or points at a file (Sentry TAURI-RUST-4QH).
|
||||
/// - `"root_path is not a directory: <path>"` — historically emitted by the
|
||||
/// now-removed knowledge-vault `vault_create` path when the chosen folder
|
||||
/// didn't exist or pointed at a file (Sentry TAURI-RUST-4QH). Kept as a
|
||||
/// classifier fixture since the wire shape may recur from other callers.
|
||||
/// - `"hosted path is not a directory: <path>"` —
|
||||
/// [`crate::openhuman::http_host::path_utils`] when an HTTP host config
|
||||
/// references a missing directory. Not yet observed in Sentry but
|
||||
|
||||
@@ -352,10 +352,11 @@ fn does_not_classify_unrelated_messages_as_context_window_exceeded() {
|
||||
|
||||
#[test]
|
||||
fn classifies_vault_create_root_path_not_a_directory_as_filesystem_user_path_invalid() {
|
||||
// TAURI-RUST-4QH: verbatim wire shape from
|
||||
// `openhuman::vault::ops::vault_create` line 37 when the
|
||||
// user-picked vault folder doesn't resolve to an existing
|
||||
// directory. Bubbles up as the RPC dispatcher's
|
||||
// TAURI-RUST-4QH: verbatim wire shape historically emitted by the
|
||||
// now-removed knowledge-vault `vault_create` path when the
|
||||
// user-picked folder didn't resolve to an existing directory.
|
||||
// Retained as a classifier fixture in case the shape recurs from
|
||||
// another caller. Bubbles up as the RPC dispatcher's
|
||||
// `display_message` and reaches `report_error_or_expected` —
|
||||
// must classify so no Sentry event fires.
|
||||
assert_eq!(
|
||||
|
||||
@@ -310,16 +310,6 @@ pub(super) const CAPABILITIES: &[Capability] = &[
|
||||
status: CapabilityStatus::Beta,
|
||||
privacy: LOCAL_RAW,
|
||||
},
|
||||
Capability {
|
||||
id: "intelligence.vault_markdown_writes",
|
||||
name: "Vault Markdown Writes",
|
||||
domain: "intelligence",
|
||||
category: CapabilityCategory::Intelligence,
|
||||
description: "Show whether a user-added local vault is writable, and write explicitly approved markdown/wiki artifacts back into that vault without leaving the device.",
|
||||
how_to: "Intelligence > Memory > Knowledge vaults",
|
||||
status: CapabilityStatus::Beta,
|
||||
privacy: LOCAL_RAW,
|
||||
},
|
||||
Capability {
|
||||
id: "intelligence.embedding_provider_config",
|
||||
name: "Configure Embedding Provider",
|
||||
|
||||
@@ -103,7 +103,6 @@ fn catalog_includes_additional_user_facing_surfaces() {
|
||||
"intelligence.mcp_server",
|
||||
"intelligence.searxng_search",
|
||||
"intelligence.tool_registry",
|
||||
"intelligence.vault_markdown_writes",
|
||||
"intelligence.embedding_provider_config",
|
||||
"intelligence.embedding_provider_test",
|
||||
"conversation.subagent_mascots",
|
||||
|
||||
@@ -879,8 +879,9 @@ pub struct DeleteChunkResponse {
|
||||
|
||||
/// Which graph the UI is asking for.
|
||||
///
|
||||
/// `Tree` returns summary nodes connected by parent_id (current
|
||||
/// Obsidian-style summary tree). `Contacts` returns raw chunks
|
||||
/// `Tree` returns the summary tree (summary nodes connected by
|
||||
/// parent_id) plus the leaf chunks hanging off it, bounded to ~1000
|
||||
/// nodes with summaries prioritized. `Contacts` returns raw chunks
|
||||
/// connected to the person entities they mention via the inverted
|
||||
/// `mem_tree_entity_index` — i.e. the document↔contact graph.
|
||||
///
|
||||
@@ -1071,10 +1072,26 @@ pub async fn obsidian_vault_status_rpc(
|
||||
}
|
||||
|
||||
/// Tree mode: summary nodes joined to their owning tree for the
|
||||
/// human-readable scope. Edges are encoded implicitly via
|
||||
/// `GraphNode.parent_id`.
|
||||
/// human-readable scope, plus the leaf chunks that hang off them. Edges
|
||||
/// are encoded implicitly via `GraphNode.parent_id` (a chunk's
|
||||
/// `parent_id` is its `parent_summary_id`, which matches a summary node's
|
||||
/// `id`).
|
||||
///
|
||||
/// Budget: summary (tree) nodes are **always kept in full** — they are
|
||||
/// the skeleton of the graph — then leaf chunks fill the remaining budget
|
||||
/// up to [`MAX_TREE_NODES`], most-recent first. Without the leaves the UI
|
||||
/// graph showed only the handful of sealed summaries (e.g. ~20) while
|
||||
/// Obsidian, which renders every `.md` on disk, showed hundreds; the
|
||||
/// chunks are the bulk of the tree. Unsealed chunks have a null
|
||||
/// `parent_summary_id` and render as orphan nodes — matching Obsidian's
|
||||
/// `showOrphans` view.
|
||||
fn collect_tree_graph(cfg: &Config) -> Result<(Vec<GraphNode>, Vec<GraphEdge>)> {
|
||||
let nodes = with_connection(cfg, |conn| {
|
||||
/// Hard cap on total nodes returned in Tree mode. The custom
|
||||
/// force-simulation in the UI does all-pairs repulsion (O(n²)), so
|
||||
/// this bounds both the wire payload and the layout cost.
|
||||
const MAX_TREE_NODES: usize = 1000;
|
||||
|
||||
let mut nodes = with_connection(cfg, |conn| {
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT s.id, s.tree_id, s.tree_kind, t.scope, s.level, s.parent_id,
|
||||
s.child_ids_json, s.time_range_start_ms, s.time_range_end_ms
|
||||
@@ -1119,6 +1136,62 @@ fn collect_tree_graph(cfg: &Config) -> Result<(Vec<GraphNode>, Vec<GraphEdge>)>
|
||||
.context("collect tree-mode summary rows")?;
|
||||
Ok(rows)
|
||||
})?;
|
||||
|
||||
// Fill the remaining budget with leaf chunks, most-recent first.
|
||||
// Summaries are kept whole; only the chunk tail is truncated when a
|
||||
// very large workspace would blow past MAX_TREE_NODES.
|
||||
let chunk_budget = MAX_TREE_NODES.saturating_sub(nodes.len());
|
||||
if chunk_budget == 0 {
|
||||
return Ok((nodes, Vec::new()));
|
||||
}
|
||||
|
||||
let chunk_nodes = with_connection(cfg, |conn| {
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT id, parent_summary_id, content, time_range_start_ms, time_range_end_ms
|
||||
FROM mem_tree_chunks
|
||||
ORDER BY timestamp_ms DESC
|
||||
LIMIT ?1",
|
||||
)?;
|
||||
let rows = stmt
|
||||
.query_map(params![chunk_budget as i64], |row| {
|
||||
let id: String = row.get(0)?;
|
||||
let parent_id: Option<String> = row.get(1)?;
|
||||
let content: String = row.get(2)?;
|
||||
let time_range_start_ms: i64 = row.get(3)?;
|
||||
let time_range_end_ms: i64 = row.get(4)?;
|
||||
// First line, trimmed for graph hover legibility — same
|
||||
// treatment as Contacts mode chunk labels.
|
||||
let label = content
|
||||
.lines()
|
||||
.next()
|
||||
.unwrap_or("")
|
||||
.chars()
|
||||
.take(72)
|
||||
.collect::<String>();
|
||||
Ok(GraphNode {
|
||||
kind: "chunk".into(),
|
||||
id,
|
||||
label,
|
||||
tree_kind: None,
|
||||
tree_scope: None,
|
||||
tree_id: None,
|
||||
level: None,
|
||||
// parent_summary_id matches a summary node's `id`,
|
||||
// so the UI draws the chunk→summary edge directly.
|
||||
// Null for unsealed chunks → orphan node.
|
||||
parent_id: parent_id.filter(|s| !s.is_empty()),
|
||||
child_count: None,
|
||||
time_range_start_ms: Some(time_range_start_ms),
|
||||
time_range_end_ms: Some(time_range_end_ms),
|
||||
file_basename: None,
|
||||
entity_kind: None,
|
||||
})
|
||||
})?
|
||||
.collect::<rusqlite::Result<Vec<_>>>()
|
||||
.context("collect tree-mode leaf chunk rows")?;
|
||||
Ok(rows)
|
||||
})?;
|
||||
nodes.extend(chunk_nodes);
|
||||
Ok((nodes, Vec::new()))
|
||||
}
|
||||
|
||||
|
||||
@@ -720,6 +720,118 @@ fn clear_composio_sync_state_removes_only_target_namespace() {
|
||||
assert_eq!(other_count, 1);
|
||||
}
|
||||
|
||||
// ── tree-mode graph export (summaries + leaf chunks) ────────────────────
|
||||
|
||||
/// Insert a tree row and one summary node under it.
|
||||
fn insert_tree_summary(cfg: &Config, tree_id: &str, scope: &str, summary_id: &str, level: i64) {
|
||||
with_connection(cfg, |conn| {
|
||||
conn.execute(
|
||||
"INSERT OR IGNORE INTO mem_tree_trees (id, kind, scope, created_at_ms)
|
||||
VALUES (?1, 'source', ?2, 0)",
|
||||
params![tree_id, scope],
|
||||
)?;
|
||||
conn.execute(
|
||||
"INSERT INTO mem_tree_summaries (
|
||||
id, tree_id, tree_kind, level, child_ids_json, content, token_count,
|
||||
entities_json, topics_json, time_range_start_ms, time_range_end_ms,
|
||||
score, sealed_at_ms, deleted
|
||||
) VALUES (?1, ?2, 'source', ?3, '[]', 'summary body', 1, '[]', '[]', 0, 0, 0.0, 0, 0)",
|
||||
params![summary_id, tree_id, level],
|
||||
)?;
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
/// Insert a leaf chunk, optionally linked to a parent summary.
|
||||
fn insert_chunk_with_parent(
|
||||
cfg: &Config,
|
||||
id: &str,
|
||||
parent_summary_id: Option<&str>,
|
||||
timestamp_ms: i64,
|
||||
content: &str,
|
||||
) {
|
||||
with_connection(cfg, |conn| {
|
||||
conn.execute(
|
||||
"INSERT INTO mem_tree_chunks (
|
||||
id, source_kind, source_id, source_ref, owner, timestamp_ms,
|
||||
time_range_start_ms, time_range_end_ms, tags_json, content,
|
||||
token_count, seq_in_source, created_at_ms, lifecycle_status,
|
||||
content_path, parent_summary_id
|
||||
) VALUES (?1, 'chat', 'slack:#eng', NULL, 'tester', ?2, ?2, ?2, '[]', ?3, 1, 0, ?2, 'seeded', NULL, ?4)",
|
||||
params![id, timestamp_ms, content, parent_summary_id],
|
||||
)?;
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tree_graph_includes_leaf_chunks_linked_to_their_summary() {
|
||||
let (_tmp, cfg) = test_config();
|
||||
insert_tree_summary(&cfg, "tree-1", "slack:#eng", "summary:1:L1-aaa", 1);
|
||||
insert_chunk_with_parent(
|
||||
&cfg,
|
||||
"chunk-sealed",
|
||||
Some("summary:1:L1-aaa"),
|
||||
1_700_000_000_000,
|
||||
"first line of sealed chunk\nmore body",
|
||||
);
|
||||
insert_chunk_with_parent(
|
||||
&cfg,
|
||||
"chunk-orphan",
|
||||
None,
|
||||
1_700_000_000_001,
|
||||
"orphan chunk body",
|
||||
);
|
||||
|
||||
let resp = graph_export_rpc(&cfg, GraphMode::Tree).await.unwrap().value;
|
||||
|
||||
// Before this fix tree mode returned only the summary node; the
|
||||
// leaves were invisible in the UI while Obsidian showed them all.
|
||||
assert_eq!(resp.nodes.len(), 3, "summary + both leaf chunks");
|
||||
|
||||
let summary = resp.nodes.iter().find(|n| n.kind == "summary").unwrap();
|
||||
assert_eq!(summary.id, "summary:1:L1-aaa");
|
||||
|
||||
let sealed = resp.nodes.iter().find(|n| n.id == "chunk-sealed").unwrap();
|
||||
assert_eq!(sealed.kind, "chunk");
|
||||
// A chunk's parent_id == its parent summary's node id, so the UI's
|
||||
// parent_id edge logic draws the chunk→summary link directly.
|
||||
assert_eq!(sealed.parent_id.as_deref(), Some("summary:1:L1-aaa"));
|
||||
// Label is the trimmed first line of the chunk content.
|
||||
assert_eq!(sealed.label, "first line of sealed chunk");
|
||||
|
||||
let orphan = resp.nodes.iter().find(|n| n.id == "chunk-orphan").unwrap();
|
||||
assert!(
|
||||
orphan.parent_id.is_none(),
|
||||
"unsealed chunk has no parent → renders as an orphan node"
|
||||
);
|
||||
|
||||
// Tree mode encodes edges via parent_id; the explicit edges array
|
||||
// stays empty.
|
||||
assert!(resp.edges.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tree_graph_keeps_summaries_first_then_chunks() {
|
||||
let (_tmp, cfg) = test_config();
|
||||
insert_tree_summary(&cfg, "tree-1", "slack:#eng", "summary:1:L1-aaa", 1);
|
||||
insert_chunk_with_parent(
|
||||
&cfg,
|
||||
"chunk-1",
|
||||
Some("summary:1:L1-aaa"),
|
||||
1_700_000_000_000,
|
||||
"a chunk",
|
||||
);
|
||||
|
||||
let resp = graph_export_rpc(&cfg, GraphMode::Tree).await.unwrap().value;
|
||||
// Summaries are emitted ahead of chunks so a budget truncation drops
|
||||
// chunk tails, never the tree skeleton.
|
||||
assert_eq!(resp.nodes[0].kind, "summary");
|
||||
assert!(resp.nodes.iter().any(|n| n.kind == "chunk"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn obsidian_status_registered_when_override_config_lists_content_root() {
|
||||
let (_tmp, cfg) = test_config();
|
||||
|
||||
@@ -8,42 +8,42 @@
|
||||
"collapse-color-groups": false,
|
||||
"colorGroups": [
|
||||
{
|
||||
"query": "file:summary-L1",
|
||||
"query": "path:L1",
|
||||
"color": {
|
||||
"a": 1,
|
||||
"rgb": 14701138
|
||||
}
|
||||
},
|
||||
{
|
||||
"query": "file:summary-L2",
|
||||
"query": "path:L2",
|
||||
"color": {
|
||||
"a": 1,
|
||||
"rgb": 14725458
|
||||
}
|
||||
},
|
||||
{
|
||||
"query": "file:summary-L3",
|
||||
"query": "path:L3",
|
||||
"color": {
|
||||
"a": 1,
|
||||
"rgb": 11657298
|
||||
}
|
||||
},
|
||||
{
|
||||
"query": "file:summary-L4",
|
||||
"query": "path:L4",
|
||||
"color": {
|
||||
"a": 1,
|
||||
"rgb": 5420768
|
||||
}
|
||||
},
|
||||
{
|
||||
"query": "file:summary-L5",
|
||||
"query": "path:L5",
|
||||
"color": {
|
||||
"a": 1,
|
||||
"rgb": 5431504
|
||||
}
|
||||
},
|
||||
{
|
||||
"query": "file:summary-L6",
|
||||
"query": "path:L6",
|
||||
"color": {
|
||||
"a": 1,
|
||||
"rgb": 14701261
|
||||
|
||||
@@ -5,13 +5,13 @@
|
||||
//!
|
||||
//! | Submodule | Source | Notes |
|
||||
//! | --- | --- | --- |
|
||||
//! | `vault` | Files dropped into the Obsidian vault by the user | Watch + diff |
|
||||
//! | `folder` | Files under a user-added folder memory source | Watch + diff |
|
||||
//! | `harness` | Agent harness turns (memory_archivist's caller side) | Push-based |
|
||||
//! | `dictation` | Local audio capture transcripts | Push-based |
|
||||
//!
|
||||
//! ## Status
|
||||
//!
|
||||
//! Scaffold only. Today the vault watch lives in `vault/sync.rs`,
|
||||
//! harness capture in `agent_experience/`, and dictation in
|
||||
//! `dictation_hotkeys/`. Each will land here as a [`SyncPipeline`] impl
|
||||
//! in a follow-up.
|
||||
//! Scaffold only. Today folder ingestion lives in
|
||||
//! `memory_sources/readers/folder.rs`, harness capture in
|
||||
//! `agent_experience/`, and dictation in `dictation_hotkeys/`. Each will
|
||||
//! land here as a [`SyncPipeline`] impl in a follow-up.
|
||||
|
||||
@@ -107,7 +107,6 @@ pub mod tool_timeout;
|
||||
pub mod tools;
|
||||
pub mod update;
|
||||
pub mod util;
|
||||
pub mod vault;
|
||||
pub mod voice;
|
||||
pub mod wallet;
|
||||
pub mod web3;
|
||||
|
||||
@@ -11,7 +11,6 @@ mod read_diff;
|
||||
mod run_linter;
|
||||
mod run_tests;
|
||||
mod update_memory_md;
|
||||
mod vault_write_markdown;
|
||||
|
||||
pub use apply_patch::ApplyPatchTool;
|
||||
pub use csv_export::CsvExportTool;
|
||||
@@ -26,4 +25,3 @@ pub use read_diff::ReadDiffTool;
|
||||
pub use run_linter::RunLinterTool;
|
||||
pub use run_tests::RunTestsTool;
|
||||
pub use update_memory_md::UpdateMemoryMdTool;
|
||||
pub use vault_write_markdown::VaultWriteMarkdownTool;
|
||||
|
||||
@@ -1,177 +0,0 @@
|
||||
//! Tool: vault_write_markdown — approved markdown/wiki writes into user vaults.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use serde_json::json;
|
||||
|
||||
use crate::openhuman::config::Config;
|
||||
use crate::openhuman::security::{CommandClass, GateDecision, SecurityPolicy};
|
||||
use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult};
|
||||
|
||||
pub struct VaultWriteMarkdownTool {
|
||||
config: Arc<Config>,
|
||||
security: Arc<SecurityPolicy>,
|
||||
}
|
||||
|
||||
impl VaultWriteMarkdownTool {
|
||||
pub fn new(config: Arc<Config>, security: Arc<SecurityPolicy>) -> Self {
|
||||
Self { config, security }
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Tool for VaultWriteMarkdownTool {
|
||||
fn name(&self) -> &str {
|
||||
"vault_write_markdown"
|
||||
}
|
||||
|
||||
fn description(&self) -> &str {
|
||||
"Create or update an approved .md/.markdown wiki note inside a registered knowledge vault. \
|
||||
Use only after the user has approved writing generated content into that vault."
|
||||
}
|
||||
|
||||
fn parameters_schema(&self) -> serde_json::Value {
|
||||
json!({
|
||||
"type": "object",
|
||||
"required": ["vault_id", "rel_path", "content"],
|
||||
"properties": {
|
||||
"vault_id": {
|
||||
"type": "string",
|
||||
"description": "Identifier of the registered vault to write into."
|
||||
},
|
||||
"rel_path": {
|
||||
"type": "string",
|
||||
"description": "Relative path under the vault root. Must end with .md or .markdown."
|
||||
},
|
||||
"content": {
|
||||
"type": "string",
|
||||
"description": "Markdown content to write."
|
||||
},
|
||||
"overwrite": {
|
||||
"type": "boolean",
|
||||
"description": "Set true to update an existing markdown file."
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn permission_level(&self) -> PermissionLevel {
|
||||
PermissionLevel::Write
|
||||
}
|
||||
|
||||
fn external_effect_with_args(&self, _args: &serde_json::Value) -> bool {
|
||||
self.security.gate_decision(CommandClass::Write) == GateDecision::Prompt
|
||||
}
|
||||
|
||||
async fn execute(&self, args: serde_json::Value) -> anyhow::Result<ToolResult> {
|
||||
if !self.security.can_act() {
|
||||
return Ok(ToolResult::error(
|
||||
"[policy-blocked] Action blocked: autonomy is read-only",
|
||||
));
|
||||
}
|
||||
if self.security.is_rate_limited() {
|
||||
return Ok(ToolResult::error(
|
||||
"Rate limit exceeded: too many actions in the last hour",
|
||||
));
|
||||
}
|
||||
|
||||
let vault_id = args
|
||||
.get("vault_id")
|
||||
.and_then(|value| value.as_str())
|
||||
.ok_or_else(|| anyhow::anyhow!("Missing 'vault_id' parameter"))?;
|
||||
let rel_path = args
|
||||
.get("rel_path")
|
||||
.and_then(|value| value.as_str())
|
||||
.ok_or_else(|| anyhow::anyhow!("Missing 'rel_path' parameter"))?;
|
||||
let content = args
|
||||
.get("content")
|
||||
.and_then(|value| value.as_str())
|
||||
.ok_or_else(|| anyhow::anyhow!("Missing 'content' parameter"))?;
|
||||
let overwrite = args
|
||||
.get("overwrite")
|
||||
.and_then(|value| value.as_bool())
|
||||
.unwrap_or(false);
|
||||
|
||||
if !self.security.record_action() {
|
||||
return Ok(ToolResult::error(
|
||||
"Rate limit exceeded: action budget exhausted",
|
||||
));
|
||||
}
|
||||
|
||||
tracing::debug!(
|
||||
vault_id = %vault_id,
|
||||
rel_path = %rel_path,
|
||||
overwrite,
|
||||
content_bytes = content.len(),
|
||||
"[vault_write_markdown] execute"
|
||||
);
|
||||
match crate::openhuman::vault::ops::vault_write_markdown(
|
||||
&self.config,
|
||||
vault_id,
|
||||
rel_path,
|
||||
content,
|
||||
overwrite,
|
||||
true,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(outcome) => Ok(ToolResult::json(json!(outcome.value))),
|
||||
Err(err) => Ok(ToolResult::error(err)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::openhuman::security::AutonomyLevel;
|
||||
|
||||
fn security(workspace: std::path::PathBuf, autonomy: AutonomyLevel) -> Arc<SecurityPolicy> {
|
||||
Arc::new(SecurityPolicy {
|
||||
autonomy,
|
||||
workspace_dir: workspace,
|
||||
..SecurityPolicy::default()
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn vault_write_markdown_tool_schema_requires_core_fields() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let config = Arc::new(Config::default());
|
||||
let tool = VaultWriteMarkdownTool::new(
|
||||
config,
|
||||
security(dir.path().to_path_buf(), AutonomyLevel::Supervised),
|
||||
);
|
||||
|
||||
assert_eq!(tool.name(), "vault_write_markdown");
|
||||
assert_eq!(tool.permission_level(), PermissionLevel::Write);
|
||||
assert!(tool.external_effect_with_args(&json!({})));
|
||||
let schema = tool.parameters_schema();
|
||||
let required = schema["required"].as_array().unwrap();
|
||||
assert!(required.iter().any(|value| value == "vault_id"));
|
||||
assert!(required.iter().any(|value| value == "rel_path"));
|
||||
assert!(required.iter().any(|value| value == "content"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn vault_write_markdown_tool_blocks_read_only_autonomy() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let config = Arc::new(Config::default());
|
||||
let tool = VaultWriteMarkdownTool::new(
|
||||
config,
|
||||
security(dir.path().to_path_buf(), AutonomyLevel::ReadOnly),
|
||||
);
|
||||
|
||||
let result = tool
|
||||
.execute(json!({
|
||||
"vault_id": "v-1",
|
||||
"rel_path": "wiki/a.md",
|
||||
"content": "# A"
|
||||
}))
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(result.is_error);
|
||||
assert!(result.text().contains("read-only"));
|
||||
}
|
||||
}
|
||||
@@ -133,10 +133,6 @@ pub fn all_tools_with_runtime(
|
||||
shell,
|
||||
Box::new(FileReadTool::new(security.clone())),
|
||||
Box::new(FileWriteTool::new(security.clone())),
|
||||
Box::new(VaultWriteMarkdownTool::new(
|
||||
config.clone(),
|
||||
security.clone(),
|
||||
)),
|
||||
// Coding-harness baseline tools (issue #1205): file navigation
|
||||
// + atomic editing primitives. Use these instead of falling
|
||||
// through to `shell` for grep/find/sed work.
|
||||
|
||||
@@ -177,43 +177,6 @@ fn all_tools_includes_spawn_parallel_agents() {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn all_tools_includes_vault_write_markdown() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let security = Arc::new(SecurityPolicy::default());
|
||||
let mem_cfg = MemoryConfig {
|
||||
backend: "markdown".into(),
|
||||
..MemoryConfig::default()
|
||||
};
|
||||
let mem: Arc<dyn Memory> =
|
||||
Arc::from(crate::openhuman::memory_store::create_memory(&mem_cfg, tmp.path()).unwrap());
|
||||
let browser = BrowserConfig {
|
||||
enabled: false,
|
||||
allowed_domains: vec![],
|
||||
session_name: None,
|
||||
..BrowserConfig::default()
|
||||
};
|
||||
let http = crate::openhuman::config::HttpRequestConfig::default();
|
||||
let cfg = test_config(&tmp);
|
||||
|
||||
let tools = all_tools(
|
||||
Arc::new(cfg.clone()),
|
||||
&security,
|
||||
AuditLogger::disabled(),
|
||||
mem,
|
||||
&browser,
|
||||
&http,
|
||||
tmp.path(),
|
||||
&HashMap::new(),
|
||||
&cfg,
|
||||
);
|
||||
let names: Vec<&str> = tools.iter().map(|t| t.name()).collect();
|
||||
assert!(
|
||||
names.contains(&"vault_write_markdown"),
|
||||
"vault_write_markdown must be registered so agents can write approved markdown into user vaults; got: {names:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn all_tools_always_registers_curl() {
|
||||
// Regression guard: `curl` is always registered (gated only by
|
||||
@@ -419,7 +382,6 @@ fn all_tools_default_registry_contains_expected_baseline_surface() {
|
||||
"shell",
|
||||
"file_read",
|
||||
"file_write",
|
||||
"vault_write_markdown",
|
||||
"grep",
|
||||
"glob",
|
||||
"list",
|
||||
|
||||
@@ -9,10 +9,7 @@ const TOOL_ID_TO_RUST_NAMES: &[(&str, &[&str])] = &[
|
||||
("install_tool", &["install_tool"]),
|
||||
("git_operations", &["git_operations"]),
|
||||
("file_read", &["file_read", "read_diff", "csv_export"]),
|
||||
(
|
||||
"file_write",
|
||||
&["file_write", "update_memory_md", "vault_write_markdown"],
|
||||
),
|
||||
("file_write", &["file_write", "update_memory_md"]),
|
||||
("screenshot", &["screenshot"]),
|
||||
("image_info", &["image_info"]),
|
||||
("browser_open", &["browser_open"]),
|
||||
|
||||
@@ -1,95 +0,0 @@
|
||||
# vault
|
||||
|
||||
Knowledge vault domain — a NotebookLM-style "folder of files" mirrored into the memory-tree backend. A `Vault` points at a local directory; on sync the module walks that directory, routes supported files to a plain-UTF-8 extractor by extension, and feeds them through the memory ingestion pipeline under a vault-derived namespace. Per-file dedup uses `(rel_path, mtime, content hash)` so re-syncs only touch what actually changed, and vanished files have their memory rows deleted to keep retrieval in sync with disk.
|
||||
|
||||
## Responsibilities
|
||||
|
||||
- Register / list / fetch / remove user-owned local folders as "vaults" backed by SQLite.
|
||||
- Walk a vault's root directory, prune built-in noise dirs (`.git`, `node_modules`, `target`, …), apply user include/exclude substring patterns, and filter by supported extension + max file size (5 MiB).
|
||||
- Ingest new/changed files into the **memory-tree** backend (`mem_tree_chunks` / `mem_tree_ingested_sources`) via the ingest pipeline, keyed by a stable `source_id = vault:{vault_id}:{rel_path}`.
|
||||
- Two-tier dedup: fast-path mtime skip during discovery, secondary content-hash skip during concurrent ingestion.
|
||||
- For content updates, delete prior chunks before re-ingest (the pipeline's `already_ingested` gate is content-blind on a stable source_id).
|
||||
- Delete memory rows for files that vanished since the last sync; record per-file outcome in a ledger.
|
||||
- Run sync as a background tokio task; expose live progress + final outcome via an in-process state registry, polled by the frontend.
|
||||
- On vault removal, optionally purge all memory rows for the vault (memory-tree prefix delete + best-effort legacy `clear_namespace`).
|
||||
- Hide vaults that belong to an incompatible host OS / path shape (cross-machine config sharing safety).
|
||||
|
||||
## Key files
|
||||
|
||||
| File | Role |
|
||||
| --- | --- |
|
||||
| `src/openhuman/vault/mod.rs` | Export-focused module root: docstring, `mod`/`pub mod` decls, re-exports of types + controller-schema pair. |
|
||||
| `src/openhuman/vault/types.rs` | Serde domain types: `Vault`, `VaultFile`, `VaultFileStatus`, `VaultSyncState`, `VaultSyncStatus`, `VaultSyncReport`. |
|
||||
| `src/openhuman/vault/ops.rs` | RPC-facing business logic: `vault_create/list/get/files/remove/sync/sync_status`, namespace derivation, background-task spawn with panic guard, memory purge on remove. |
|
||||
| `src/openhuman/vault/schemas.rs` | Controller schemas + `handle_*` fns delegating to `ops`; `all_controller_schemas` / `all_registered_controllers`. |
|
||||
| `src/openhuman/vault/store.rs` | SQLite persistence (`vault.db`): vault + per-file ledger tables, migrations, host-OS compatibility filtering. |
|
||||
| `src/openhuman/vault/sync.rs` | The directory-walk + 4-phase ingest engine (discovery → concurrent ingest → ledger writes → deletions); supported-extension list; memory-tree source_id helper. Inline `sync_tests`. |
|
||||
| `src/openhuman/vault/state.rs` | Process-global in-memory sync-progress registry (`once_cell::Lazy` + `parking_lot::RwLock`), keyed by `vault_id`. |
|
||||
| `src/openhuman/vault/tests.rs` | `#[cfg(test)]` module (additional tests beyond the inline `sync.rs` suite). |
|
||||
|
||||
## Public surface
|
||||
|
||||
Re-exported from `mod.rs`:
|
||||
|
||||
- Types: `Vault`, `VaultFile`, `VaultFileStatus`, `VaultSyncReport`, `VaultSyncState`, `VaultSyncStatus`.
|
||||
- `all_vault_controller_schemas` / `all_vault_registered_controllers` (the controller-registry pair, wired in `src/core/all.rs`).
|
||||
- `ops` is `pub mod` — `vault_create`, `vault_list`, `vault_get`, `vault_files`, `vault_remove`, `vault_sync`, `vault_sync_status`.
|
||||
|
||||
## RPC / controllers
|
||||
|
||||
Namespace `vault` (invoked as `openhuman.vault_<function>`):
|
||||
|
||||
| Method | Inputs | Output |
|
||||
| --- | --- | --- |
|
||||
| `vault.create` | `name`, `root_path` (absolute dir), `include_globs?`, `exclude_globs?` | `Vault` |
|
||||
| `vault.list` | — | `Vec<Vault>` (current-host only) |
|
||||
| `vault.get` | `vault_id` | `Vault` |
|
||||
| `vault.files` | `vault_id` | `Vec<VaultFile>` (per-file ledger) |
|
||||
| `vault.remove` | `vault_id`, `purge_memory?` | `{ vault_id, removed, purged, memory_tree_chunks_deleted, purge_error? }` |
|
||||
| `vault.sync` | `vault_id` | `{ status: "started", vault_id }` (background; rejects if already running) |
|
||||
| `vault.sync_status` | `vault_id` | `VaultSyncState` (Idle / Running / Completed / Failed) |
|
||||
|
||||
`include_globs` / `exclude_globs` are substring matches (case-insensitive), not true globs despite the name.
|
||||
|
||||
## Agent tools
|
||||
|
||||
None — the module owns no `tools.rs`.
|
||||
|
||||
## Events
|
||||
|
||||
None — no `bus.rs`; the module publishes/subscribes to no `DomainEvent`s. Sync progress is surfaced via the in-memory `state` registry + `vault.sync_status` polling rather than the event bus.
|
||||
|
||||
## Persistence
|
||||
|
||||
- **SQLite** at `{workspace_dir}/vault/vault.db` (`store.rs`):
|
||||
- `vaults` — id, name, root_path, host_os, namespace (UNIQUE), include/exclude globs (JSON), created_at, last_synced_at.
|
||||
- `vault_files` — per-file ledger keyed by `(vault_id, rel_path)`, FK→`vaults` with `ON DELETE CASCADE`; stores `document_id` (= memory-tree source_id post-#2705), `content_hash`, `mtime_ms`, `bytes`, `ingested_at`, `status`. The dedup ledger.
|
||||
- Lazy schema init + an additive `host_os` column migration (cached per DB path).
|
||||
- **Process-global in-memory** (`state.rs`): `VaultSyncState` per vault, retained after completion until the next sync overwrites it. Lost on process restart.
|
||||
- **Memory-tree** (owned by the `memory` domain, written via the ingest pipeline): chunks under `source_id = vault:{vault_id}:{rel_path}`.
|
||||
|
||||
## Dependencies
|
||||
|
||||
- `crate::openhuman::config` (`Config`, `config::rpc::load_config_with_timeout`) — workspace dir for the DB, config passed into ingest/delete calls.
|
||||
- `crate::openhuman::memory::ingest_pipeline` — `ingest_document` / `IngestResult`: the canonical memory-tree ingest path.
|
||||
- `crate::openhuman::memory::ops` — `clear_namespace` (legacy purge on remove) and `doc_delete` (best-effort legacy UnifiedMemory cleanup for pre-#2705 ledger rows).
|
||||
- `crate::openhuman::memory_store::chunks` — `delete_chunks_by_source` / `delete_chunks_by_source_prefix` (chunk cleanup on re-ingest, file deletion, and vault purge); `SourceKind::Document`.
|
||||
- `crate::openhuman::memory_sync::canonicalize::document::DocumentInput` — the document shape handed to the ingest pipeline.
|
||||
- `crate::core::all` (`ControllerFuture`, `RegisteredController`) + `crate::core::{ControllerSchema, FieldSchema, TypeSchema}` — controller registration plumbing.
|
||||
- `crate::rpc::RpcOutcome` — RPC return contract.
|
||||
- External crates: `rusqlite`, `walkdir`, `sha2`, `uuid`, `chrono`, `futures` (`buffer_unordered`), `tokio`, `once_cell`, `parking_lot`.
|
||||
|
||||
## Used by
|
||||
|
||||
- `src/core/all.rs` — registers vault controllers + schemas into the global controller registry (the only wiring point; vault is controller-only, no CLI/JSON-RPC branches).
|
||||
- `src/core/observability.rs` (+ tests) — references `vault_create` / `openhuman.vault_create` as an example in path-boundary / autonomy-policy diagnostics, not a runtime dependency.
|
||||
|
||||
## Notes / gotchas
|
||||
|
||||
- **Memory-tree, not `memory_docs` (#2705).** Sync ingests via the memory-tree pipeline; the pre-#2705 path wrote to legacy `UnifiedMemory` (`memory_docs`), which the UI reported as "synced" but retrieval never saw. The ledger now stores the memory-tree `source_id` (`vault:{id}:{rel_path}`); deletion/purge logic keys off `document_id.starts_with("vault:")` to distinguish post- vs pre-#2705 rows and run the right cleanup.
|
||||
- **Namespace is a hashed digest, not the raw UUID.** `vault_namespace_for_id` derives an alphabet-only SHA-256 suffix (`vault-<24 a-z chars>`) because memory writes reject namespace/key values that resemble PII / strict identifier patterns.
|
||||
- **`include_globs`/`exclude_globs` are substring matches**, lowercased, not real glob patterns — the field name is misleading.
|
||||
- **Background sync + panic guard.** `vault_sync` spawns a tokio task and returns immediately; the work is wrapped in `catch_unwind` so a panic marks state `Failed` instead of leaving it stuck in `Running` (which would reject every future sync until restart). Concurrency is bounded to 4 (`buffer_unordered`).
|
||||
- **Host-OS gating.** `list_vaults`/`get_vault` hide vaults whose stamped `host_os` (or, for legacy rows, whose `root_path` shape) doesn't match the current machine — prevents surfacing another machine's folders from shared config.
|
||||
- **Ledger ↔ memory_tree desync is logged loudly.** If ingest returns `already_ingested && chunks_written == 0` after the delete-first guard, the code emits a `warn!` because it means new content never reached retrieval (the exact false-success class #2705 was meant to kill).
|
||||
- `vault_create` canonicalizes `root_path` (falling back to the trimmed input if canonicalization fails), so the stored id/path may differ from the literal input.
|
||||
@@ -1,25 +0,0 @@
|
||||
//! Knowledge vault — folder-of-files ingested into memory (NotebookLM-style).
|
||||
//!
|
||||
//! A `Vault` points at a local directory; on `vault.sync` we walk it, route
|
||||
//! files to extractors by extension, and feed them into the memory pipeline
|
||||
//! under a vault-derived namespace. Per-file dedup uses (path, mtime, content
|
||||
//! hash) so re-syncs only touch what changed.
|
||||
|
||||
pub mod ops;
|
||||
mod schemas;
|
||||
pub(crate) mod state;
|
||||
mod store;
|
||||
mod sync;
|
||||
mod types;
|
||||
|
||||
pub use schemas::{
|
||||
all_controller_schemas as all_vault_controller_schemas,
|
||||
all_registered_controllers as all_vault_registered_controllers,
|
||||
};
|
||||
pub use types::{
|
||||
Vault, VaultFile, VaultFileStatus, VaultSyncReport, VaultSyncState, VaultSyncStatus,
|
||||
VaultWriteMarkdownReport, VaultWriteState,
|
||||
};
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
@@ -1,492 +0,0 @@
|
||||
//! RPC-facing operations for the vault domain.
|
||||
|
||||
use chrono::Utc;
|
||||
use futures::FutureExt;
|
||||
use sha2::{Digest, Sha256};
|
||||
use std::path::{Component, Path, PathBuf};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::openhuman::config::Config;
|
||||
use crate::openhuman::memory::ops::{clear_namespace, ClearNamespaceParams};
|
||||
use crate::openhuman::memory_store::chunks::store::delete_chunks_by_source_prefix;
|
||||
use crate::openhuman::memory_store::chunks::types::SourceKind;
|
||||
use crate::rpc::RpcOutcome;
|
||||
|
||||
use super::state;
|
||||
use super::store;
|
||||
use super::sync;
|
||||
use super::types::{
|
||||
Vault, VaultFile, VaultSyncState, VaultSyncStatus, VaultWriteMarkdownReport, VaultWriteState,
|
||||
};
|
||||
|
||||
/// Derive a stable memory namespace for a vault without embedding the raw UUID.
|
||||
///
|
||||
/// Memory writes reject namespace/key values that resemble PII. Raw UUID hex can
|
||||
/// occasionally match strict alphanumeric identifier patterns, so vault
|
||||
/// namespaces use an alphabet-only digest suffix instead.
|
||||
pub(crate) fn vault_namespace_for_id(id: &str) -> String {
|
||||
let digest = Sha256::digest(id.as_bytes());
|
||||
let suffix: String = digest
|
||||
.iter()
|
||||
.take(24)
|
||||
.map(|byte| char::from(b'a' + (byte % 26)))
|
||||
.collect();
|
||||
format!("vault-{suffix}")
|
||||
}
|
||||
|
||||
/// Create a new vault pointing at a local folder.
|
||||
pub async fn vault_create(
|
||||
config: &Config,
|
||||
name: &str,
|
||||
root_path: &str,
|
||||
include_globs: Vec<String>,
|
||||
exclude_globs: Vec<String>,
|
||||
) -> Result<RpcOutcome<Vault>, String> {
|
||||
let trimmed_name = name.trim();
|
||||
if trimmed_name.is_empty() {
|
||||
return Err("vault name must not be empty".to_string());
|
||||
}
|
||||
let trimmed_root = root_path.trim();
|
||||
if trimmed_root.is_empty() {
|
||||
return Err("root_path must not be empty".to_string());
|
||||
}
|
||||
let root = std::path::Path::new(trimmed_root);
|
||||
if !root.is_absolute() {
|
||||
return Err(format!("root_path must be absolute: {trimmed_root}"));
|
||||
}
|
||||
if !root.is_dir() {
|
||||
return Err(format!("root_path is not a directory: {trimmed_root}"));
|
||||
}
|
||||
|
||||
let id = Uuid::new_v4().to_string();
|
||||
log::debug!(
|
||||
"[vault] create: name={trimmed_name:?} root={trimmed_root:?} id={id} \
|
||||
include_globs={} exclude_globs={}",
|
||||
include_globs.len(),
|
||||
exclude_globs.len(),
|
||||
);
|
||||
let namespace = vault_namespace_for_id(&id);
|
||||
let canonical_root = root
|
||||
.canonicalize()
|
||||
.map(|p| p.to_string_lossy().to_string())
|
||||
.unwrap_or_else(|_| trimmed_root.to_string());
|
||||
let (write_state, write_state_reason) = store::vault_write_state_for_root_path(&canonical_root);
|
||||
let vault = Vault {
|
||||
id: id.clone(),
|
||||
name: trimmed_name.to_string(),
|
||||
root_path: canonical_root,
|
||||
host_os: Some(store::current_host_os().to_string()),
|
||||
namespace,
|
||||
include_globs,
|
||||
exclude_globs,
|
||||
created_at: Utc::now(),
|
||||
last_synced_at: None,
|
||||
file_count: 0,
|
||||
write_state,
|
||||
write_state_reason,
|
||||
};
|
||||
|
||||
store::insert_vault(config, &vault).map_err(|e| e.to_string())?;
|
||||
Ok(RpcOutcome::single_log(
|
||||
vault,
|
||||
format!("vault created: {id}"),
|
||||
))
|
||||
}
|
||||
|
||||
pub async fn vault_list(config: &Config) -> Result<RpcOutcome<Vec<Vault>>, String> {
|
||||
let vaults = store::list_vaults(config).map_err(|e| e.to_string())?;
|
||||
log::debug!("[vault] list: count={}", vaults.len());
|
||||
Ok(RpcOutcome::single_log(vaults, "vaults listed"))
|
||||
}
|
||||
|
||||
pub async fn vault_get(config: &Config, id: &str) -> Result<RpcOutcome<Vault>, String> {
|
||||
let id = id.trim();
|
||||
if id.is_empty() {
|
||||
return Err("vault_id must not be empty".to_string());
|
||||
}
|
||||
let vault = store::get_vault(config, id)
|
||||
.map_err(|e| e.to_string())?
|
||||
.ok_or_else(|| format!("vault not found: {id}"))?;
|
||||
log::debug!("[vault] get: id={id} files={}", vault.file_count);
|
||||
Ok(RpcOutcome::single_log(vault, "vault loaded"))
|
||||
}
|
||||
|
||||
/// Write an explicitly approved markdown/wiki artifact into a registered vault.
|
||||
pub async fn vault_write_markdown(
|
||||
config: &Config,
|
||||
id: &str,
|
||||
rel_path: &str,
|
||||
content: &str,
|
||||
overwrite: bool,
|
||||
approved: bool,
|
||||
) -> Result<RpcOutcome<VaultWriteMarkdownReport>, String> {
|
||||
let id = id.trim();
|
||||
if id.is_empty() {
|
||||
return Err("vault_id must not be empty".to_string());
|
||||
}
|
||||
if !approved {
|
||||
log::debug!("[vault] write_markdown: rejected missing approval id={id}");
|
||||
return Err("vault markdown writes require explicit user approval".to_string());
|
||||
}
|
||||
|
||||
let vault = store::get_vault(config, id)
|
||||
.map_err(|e| e.to_string())?
|
||||
.ok_or_else(|| format!("vault not found: {id}"))?;
|
||||
let (write_state, write_reason) = store::vault_write_state_for_root_path(&vault.root_path);
|
||||
if write_state != VaultWriteState::Writable {
|
||||
log::debug!(
|
||||
"[vault] write_markdown: rejected non-writable id={id} state={write_state:?} reason={write_reason:?}"
|
||||
);
|
||||
return Err(store::vault_write_state_reason_message(write_reason.as_deref()).to_string());
|
||||
}
|
||||
|
||||
let rel = validate_markdown_rel_path(rel_path)?;
|
||||
let bytes = content.as_bytes().len() as u64;
|
||||
let root = std::fs::canonicalize(&vault.root_path)
|
||||
.map_err(|err| format!("failed to resolve vault folder: {err}"))?;
|
||||
ensure_existing_ancestors_stay_in_root(&root, &rel)?;
|
||||
let target = root.join(&rel);
|
||||
let parent = target
|
||||
.parent()
|
||||
.ok_or_else(|| "target path has no parent directory".to_string())?;
|
||||
std::fs::create_dir_all(parent)
|
||||
.map_err(|err| format!("failed to create vault note directory: {err}"))?;
|
||||
let parent_canon = std::fs::canonicalize(parent)
|
||||
.map_err(|err| format!("failed to resolve vault note directory: {err}"))?;
|
||||
if !parent_canon.starts_with(&root) {
|
||||
return Err("vault note path resolves outside the vault folder".to_string());
|
||||
}
|
||||
if let Ok(meta) = std::fs::symlink_metadata(&target) {
|
||||
if meta.file_type().is_symlink() {
|
||||
return Err("refusing to write through a symlink inside the vault".to_string());
|
||||
}
|
||||
}
|
||||
let created = !target.exists();
|
||||
if !created && !overwrite {
|
||||
return Err(
|
||||
"vault markdown file already exists; set overwrite=true to update it".to_string(),
|
||||
);
|
||||
}
|
||||
|
||||
log::debug!(
|
||||
"[vault] write_markdown: writing id={id} rel_path={} bytes={bytes} overwrite={overwrite} created={created}",
|
||||
rel.display()
|
||||
);
|
||||
std::fs::write(&target, content)
|
||||
.map_err(|err| format!("failed to write vault markdown file: {err}"))?;
|
||||
|
||||
Ok(RpcOutcome::single_log(
|
||||
VaultWriteMarkdownReport {
|
||||
vault_id: id.to_string(),
|
||||
rel_path: rel.to_string_lossy().replace('\\', "/"),
|
||||
bytes_written: bytes,
|
||||
created,
|
||||
},
|
||||
format!("vault markdown written: {id}"),
|
||||
))
|
||||
}
|
||||
|
||||
pub async fn vault_files(config: &Config, id: &str) -> Result<RpcOutcome<Vec<VaultFile>>, String> {
|
||||
let id = id.trim();
|
||||
if id.is_empty() {
|
||||
return Err("vault_id must not be empty".to_string());
|
||||
}
|
||||
store::get_vault(config, id)
|
||||
.map_err(|e| e.to_string())?
|
||||
.ok_or_else(|| format!("vault not found: {id}"))?;
|
||||
let files = store::list_files(config, id).map_err(|e| e.to_string())?;
|
||||
log::debug!("[vault] files: id={id} count={}", files.len());
|
||||
Ok(RpcOutcome::single_log(files, "vault files listed"))
|
||||
}
|
||||
|
||||
fn validate_markdown_rel_path(raw: &str) -> Result<PathBuf, String> {
|
||||
let trimmed = raw.trim();
|
||||
if trimmed.is_empty() {
|
||||
return Err("rel_path must not be empty".to_string());
|
||||
}
|
||||
let path = Path::new(trimmed);
|
||||
if path.is_absolute() {
|
||||
return Err("rel_path must be relative to the vault folder".to_string());
|
||||
}
|
||||
|
||||
let mut clean = PathBuf::new();
|
||||
for component in path.components() {
|
||||
match component {
|
||||
Component::Normal(part) => clean.push(part),
|
||||
Component::CurDir => {}
|
||||
Component::ParentDir => {
|
||||
return Err("rel_path must not contain '..' segments".to_string());
|
||||
}
|
||||
Component::RootDir | Component::Prefix(_) => {
|
||||
return Err("rel_path must stay inside the vault folder".to_string());
|
||||
}
|
||||
}
|
||||
}
|
||||
if clean.as_os_str().is_empty() {
|
||||
return Err("rel_path must name a markdown file".to_string());
|
||||
}
|
||||
let ext = clean
|
||||
.extension()
|
||||
.and_then(|value| value.to_str())
|
||||
.unwrap_or_default();
|
||||
if !matches!(ext.to_ascii_lowercase().as_str(), "md" | "markdown") {
|
||||
return Err("rel_path must end with .md or .markdown".to_string());
|
||||
}
|
||||
Ok(clean)
|
||||
}
|
||||
|
||||
fn ensure_existing_ancestors_stay_in_root(root: &Path, rel_path: &Path) -> Result<(), String> {
|
||||
let Some(parent) = rel_path.parent() else {
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
let mut current = root.to_path_buf();
|
||||
for component in parent.components() {
|
||||
let Component::Normal(part) = component else {
|
||||
continue;
|
||||
};
|
||||
let next = current.join(part);
|
||||
match std::fs::symlink_metadata(&next) {
|
||||
Ok(meta) if meta.file_type().is_symlink() => {
|
||||
let resolved = std::fs::canonicalize(&next)
|
||||
.map_err(|err| format!("failed to resolve vault note directory: {err}"))?;
|
||||
if !resolved.starts_with(root) {
|
||||
return Err(
|
||||
"vault note directory resolves outside the vault folder".to_string()
|
||||
);
|
||||
}
|
||||
current = resolved;
|
||||
}
|
||||
Ok(meta) if meta.is_dir() => {
|
||||
current = next;
|
||||
}
|
||||
Ok(_) => {
|
||||
return Err("vault note parent path is not a directory".to_string());
|
||||
}
|
||||
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
|
||||
break;
|
||||
}
|
||||
Err(err) => {
|
||||
return Err(format!("failed to inspect vault note directory: {err}"));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn vault_remove(
|
||||
config: &Config,
|
||||
id: &str,
|
||||
purge_memory: bool,
|
||||
) -> Result<RpcOutcome<serde_json::Value>, String> {
|
||||
let id = id.trim();
|
||||
if id.is_empty() {
|
||||
return Err("vault_id must not be empty".to_string());
|
||||
}
|
||||
let vault = store::get_vault(config, id).map_err(|e| e.to_string())?;
|
||||
let removed = store::remove_vault(config, id).map_err(|e| e.to_string())?;
|
||||
log::debug!("[vault] remove: id={id} removed={removed} purge_memory={purge_memory}");
|
||||
|
||||
let mut purged = false;
|
||||
let mut memory_tree_chunks_deleted: usize = 0;
|
||||
if removed && purge_memory {
|
||||
if let Some(v) = vault {
|
||||
// Memory-tree cleanup is the canonical path post-#2705: vault
|
||||
// sync writes to `mem_tree_chunks` / `mem_tree_ingested_sources`
|
||||
// keyed by `vault:{id}:{rel_path}`. A prefix delete with
|
||||
// `vault:{id}:` catches every per-file row for this vault. The
|
||||
// companion `clear_namespace` call below still drains any
|
||||
// pre-#2705 ledger rows that landed in the legacy
|
||||
// `memory_docs` table during the migration window.
|
||||
let cfg_for_blocking = config.clone();
|
||||
let prefix = format!("vault:{}:", v.id);
|
||||
let tree_result = tokio::task::spawn_blocking(move || {
|
||||
delete_chunks_by_source_prefix(&cfg_for_blocking, SourceKind::Document, &prefix)
|
||||
})
|
||||
.await;
|
||||
match tree_result {
|
||||
Ok(Ok(removed_chunks)) => {
|
||||
memory_tree_chunks_deleted = removed_chunks;
|
||||
log::debug!(
|
||||
"[vault] remove: id={id} memory_tree_chunks_deleted={removed_chunks}"
|
||||
);
|
||||
}
|
||||
Ok(Err(err)) => {
|
||||
log::warn!("[vault] remove: id={id} memory_tree_purge_failed err={err}");
|
||||
return Ok(RpcOutcome::single_log(
|
||||
serde_json::json!({
|
||||
"vault_id": id,
|
||||
"removed": removed,
|
||||
"purged": false,
|
||||
"purge_error": format!("memory_tree purge failed: {err}"),
|
||||
}),
|
||||
format!("vault removed with purge error: {id}"),
|
||||
));
|
||||
}
|
||||
Err(join_err) => {
|
||||
log::warn!(
|
||||
"[vault] remove: id={id} memory_tree_purge_join_failed err={join_err}"
|
||||
);
|
||||
return Ok(RpcOutcome::single_log(
|
||||
serde_json::json!({
|
||||
"vault_id": id,
|
||||
"removed": removed,
|
||||
"purged": false,
|
||||
"purge_error": format!("memory_tree purge join error: {join_err}"),
|
||||
}),
|
||||
format!("vault removed with purge error: {id}"),
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
// Best-effort legacy UnifiedMemory purge for pre-#2705 ledger
|
||||
// rows whose chunks still live in `memory_docs`. Failure here
|
||||
// doesn't undo the canonical memory_tree cleanup above.
|
||||
if let Err(err) = clear_namespace(ClearNamespaceParams {
|
||||
namespace: v.namespace.clone(),
|
||||
})
|
||||
.await
|
||||
{
|
||||
log::debug!(
|
||||
"[vault] remove: id={id} legacy_clear_namespace_failed (best-effort) err={err}"
|
||||
);
|
||||
}
|
||||
purged = true;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(RpcOutcome::single_log(
|
||||
serde_json::json!({
|
||||
"vault_id": id,
|
||||
"removed": removed,
|
||||
"purged": purged,
|
||||
"memory_tree_chunks_deleted": memory_tree_chunks_deleted,
|
||||
}),
|
||||
format!("vault removed: {id}"),
|
||||
))
|
||||
}
|
||||
|
||||
/// Trigger a vault sync as a background task and return immediately.
|
||||
///
|
||||
/// The caller should poll `vault_sync_status` to track progress and retrieve
|
||||
/// the final outcome. Returns an error if a sync is already running for this
|
||||
/// vault so the caller can surface a user-friendly message instead of silently
|
||||
/// queuing a duplicate.
|
||||
pub async fn vault_sync(
|
||||
config: &Config,
|
||||
id: &str,
|
||||
) -> Result<RpcOutcome<serde_json::Value>, String> {
|
||||
let id = id.trim();
|
||||
if id.is_empty() {
|
||||
return Err("vault_id must not be empty".to_string());
|
||||
}
|
||||
let vault = store::get_vault(config, id)
|
||||
.map_err(|e| e.to_string())?
|
||||
.ok_or_else(|| format!("vault not found: {id}"))?;
|
||||
|
||||
// Register in the state map; returns Err if already running.
|
||||
let started_at_ms = Utc::now().timestamp_millis();
|
||||
state::start(id, started_at_ms).map_err(|e| format!("sync already in progress: {e}"))?;
|
||||
|
||||
log::debug!(
|
||||
"[vault] sync: background task spawned id={id} root={:?}",
|
||||
vault.root_path,
|
||||
);
|
||||
|
||||
// Clone what the background task needs — Config is Clone (derives it).
|
||||
let config_clone = config.clone();
|
||||
let vault_id = id.to_string();
|
||||
|
||||
tokio::spawn(async move {
|
||||
log::debug!("[vault] sync: background task running id={vault_id}");
|
||||
|
||||
// Wrap the work in catch_unwind so a panic inside sync_vault cannot leave
|
||||
// the vault state permanently stuck in `Running`. Without this guard a
|
||||
// panic would unwind the task, the state map entry would never be updated,
|
||||
// and every subsequent sync attempt would be rejected with "already in progress"
|
||||
// until the app is restarted.
|
||||
let result =
|
||||
std::panic::AssertUnwindSafe(async { sync::sync_vault(&config_clone, &vault).await })
|
||||
.catch_unwind()
|
||||
.await;
|
||||
|
||||
match result {
|
||||
Ok(report) => {
|
||||
let success = report.failed == 0;
|
||||
let finished_at_ms = Utc::now().timestamp_millis();
|
||||
|
||||
// Write final counters back into the state map.
|
||||
state::update_progress(&vault_id, |s| {
|
||||
s.status = if success {
|
||||
VaultSyncStatus::Completed
|
||||
} else {
|
||||
VaultSyncStatus::Failed
|
||||
};
|
||||
s.finished_at_ms = Some(finished_at_ms);
|
||||
s.ingested = report.ingested;
|
||||
s.unchanged = report.unchanged;
|
||||
s.removed = report.removed;
|
||||
s.failed = report.failed;
|
||||
s.skipped_unsupported = report.skipped_unsupported;
|
||||
s.scanned = report.scanned;
|
||||
s.duration_ms = report.duration_ms;
|
||||
s.errors = report.errors.clone();
|
||||
});
|
||||
|
||||
log::debug!(
|
||||
"[vault] sync: background task done id={vault_id} ingested={} failed={} duration_ms={}",
|
||||
report.ingested,
|
||||
report.failed,
|
||||
report.duration_ms,
|
||||
);
|
||||
}
|
||||
Err(_) => {
|
||||
log::error!(
|
||||
"[vault] sync: background task panicked id={vault_id} — marking state as Failed"
|
||||
);
|
||||
state::update_progress(&vault_id, |s| {
|
||||
s.status = VaultSyncStatus::Failed;
|
||||
s.errors = vec!["sync task panicked unexpectedly".to_string()];
|
||||
});
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
Ok(RpcOutcome::single_log(
|
||||
serde_json::json!({ "status": "started", "vault_id": id }),
|
||||
format!("vault sync started in background: {id}"),
|
||||
))
|
||||
}
|
||||
|
||||
/// Return the current sync progress for a vault.
|
||||
///
|
||||
/// Returns an `Idle` state if no sync has ever run for this vault.
|
||||
pub async fn vault_sync_status(id: &str) -> Result<RpcOutcome<VaultSyncState>, String> {
|
||||
let id = id.trim();
|
||||
if id.is_empty() {
|
||||
return Err("vault_id must not be empty".to_string());
|
||||
}
|
||||
let st = state::get(id).unwrap_or_else(|| VaultSyncState {
|
||||
vault_id: id.to_string(),
|
||||
status: VaultSyncStatus::Idle,
|
||||
scanned: 0,
|
||||
ingested: 0,
|
||||
unchanged: 0,
|
||||
removed: 0,
|
||||
failed: 0,
|
||||
skipped_unsupported: 0,
|
||||
total: 0,
|
||||
started_at_ms: 0,
|
||||
finished_at_ms: None,
|
||||
duration_ms: 0,
|
||||
errors: vec![],
|
||||
});
|
||||
log::debug!(
|
||||
"[vault] sync_status: id={id} status={:?} ingested={} total={}",
|
||||
st.status,
|
||||
st.ingested,
|
||||
st.total,
|
||||
);
|
||||
Ok(RpcOutcome::single_log(st, "vault sync status"))
|
||||
}
|
||||
@@ -1,395 +0,0 @@
|
||||
use serde::de::DeserializeOwned;
|
||||
use serde_json::{Map, Value};
|
||||
|
||||
use crate::core::all::{ControllerFuture, RegisteredController};
|
||||
use crate::core::{ControllerSchema, FieldSchema, TypeSchema};
|
||||
use crate::openhuman::config::rpc as config_rpc;
|
||||
use crate::rpc::RpcOutcome;
|
||||
|
||||
fn vault_id_input(comment: &'static str) -> FieldSchema {
|
||||
FieldSchema {
|
||||
name: "vault_id",
|
||||
ty: TypeSchema::String,
|
||||
comment,
|
||||
required: true,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn all_controller_schemas() -> Vec<ControllerSchema> {
|
||||
vec![
|
||||
schemas("create"),
|
||||
schemas("list"),
|
||||
schemas("get"),
|
||||
schemas("files"),
|
||||
schemas("write_markdown"),
|
||||
schemas("remove"),
|
||||
schemas("sync"),
|
||||
schemas("sync_status"),
|
||||
]
|
||||
}
|
||||
|
||||
pub fn all_registered_controllers() -> Vec<RegisteredController> {
|
||||
vec![
|
||||
RegisteredController {
|
||||
schema: schemas("create"),
|
||||
handler: handle_create,
|
||||
},
|
||||
RegisteredController {
|
||||
schema: schemas("list"),
|
||||
handler: handle_list,
|
||||
},
|
||||
RegisteredController {
|
||||
schema: schemas("get"),
|
||||
handler: handle_get,
|
||||
},
|
||||
RegisteredController {
|
||||
schema: schemas("files"),
|
||||
handler: handle_files,
|
||||
},
|
||||
RegisteredController {
|
||||
schema: schemas("write_markdown"),
|
||||
handler: handle_write_markdown,
|
||||
},
|
||||
RegisteredController {
|
||||
schema: schemas("remove"),
|
||||
handler: handle_remove,
|
||||
},
|
||||
RegisteredController {
|
||||
schema: schemas("sync"),
|
||||
handler: handle_sync,
|
||||
},
|
||||
RegisteredController {
|
||||
schema: schemas("sync_status"),
|
||||
handler: handle_sync_status,
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
pub fn schemas(function: &str) -> ControllerSchema {
|
||||
match function {
|
||||
"create" => ControllerSchema {
|
||||
namespace: "vault",
|
||||
function: "create",
|
||||
description: "Register a new local folder as a knowledge vault.",
|
||||
inputs: vec![
|
||||
FieldSchema {
|
||||
name: "name",
|
||||
ty: TypeSchema::String,
|
||||
comment: "Display name for the vault.",
|
||||
required: true,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "root_path",
|
||||
ty: TypeSchema::String,
|
||||
comment: "Absolute path to the folder on disk.",
|
||||
required: true,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "include_globs",
|
||||
ty: TypeSchema::Option(Box::new(TypeSchema::Array(Box::new(
|
||||
TypeSchema::String,
|
||||
)))),
|
||||
comment: "Optional include patterns (substring match).",
|
||||
required: false,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "exclude_globs",
|
||||
ty: TypeSchema::Option(Box::new(TypeSchema::Array(Box::new(
|
||||
TypeSchema::String,
|
||||
)))),
|
||||
comment: "Optional exclude patterns (substring match, case-insensitive).",
|
||||
required: false,
|
||||
},
|
||||
],
|
||||
outputs: vec![FieldSchema {
|
||||
name: "vault",
|
||||
ty: TypeSchema::Ref("Vault"),
|
||||
comment: "The newly created vault.",
|
||||
required: true,
|
||||
}],
|
||||
},
|
||||
"list" => ControllerSchema {
|
||||
namespace: "vault",
|
||||
function: "list",
|
||||
description: "List all registered vaults.",
|
||||
inputs: vec![],
|
||||
outputs: vec![FieldSchema {
|
||||
name: "vaults",
|
||||
ty: TypeSchema::Array(Box::new(TypeSchema::Ref("Vault"))),
|
||||
comment: "All registered vaults.",
|
||||
required: true,
|
||||
}],
|
||||
},
|
||||
"get" => ControllerSchema {
|
||||
namespace: "vault",
|
||||
function: "get",
|
||||
description: "Fetch one vault by id.",
|
||||
inputs: vec![vault_id_input("Identifier of the vault to fetch.")],
|
||||
outputs: vec![FieldSchema {
|
||||
name: "vault",
|
||||
ty: TypeSchema::Ref("Vault"),
|
||||
comment: "The requested vault.",
|
||||
required: true,
|
||||
}],
|
||||
},
|
||||
"files" => ControllerSchema {
|
||||
namespace: "vault",
|
||||
function: "files",
|
||||
description: "List per-file ledger entries for a vault.",
|
||||
inputs: vec![vault_id_input("Identifier of the vault.")],
|
||||
outputs: vec![FieldSchema {
|
||||
name: "files",
|
||||
ty: TypeSchema::Array(Box::new(TypeSchema::Ref("VaultFile"))),
|
||||
comment: "Per-file ledger rows.",
|
||||
required: true,
|
||||
}],
|
||||
},
|
||||
"write_markdown" => ControllerSchema {
|
||||
namespace: "vault",
|
||||
function: "write_markdown",
|
||||
description: "Write an explicitly approved markdown/wiki file inside a writable vault.",
|
||||
inputs: vec![
|
||||
vault_id_input("Identifier of the vault to write into."),
|
||||
FieldSchema {
|
||||
name: "rel_path",
|
||||
ty: TypeSchema::String,
|
||||
comment: "Relative .md/.markdown path inside the vault.",
|
||||
required: true,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "content",
|
||||
ty: TypeSchema::String,
|
||||
comment: "Markdown content to write.",
|
||||
required: true,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "overwrite",
|
||||
ty: TypeSchema::Option(Box::new(TypeSchema::Bool)),
|
||||
comment: "When true, update an existing markdown file.",
|
||||
required: false,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "approved",
|
||||
ty: TypeSchema::Bool,
|
||||
comment: "Must be true after explicit user approval for arbitrary vault writes.",
|
||||
required: true,
|
||||
},
|
||||
],
|
||||
outputs: vec![FieldSchema {
|
||||
name: "report",
|
||||
ty: TypeSchema::Ref("VaultWriteMarkdownReport"),
|
||||
comment: "Write result with relative path and byte count.",
|
||||
required: true,
|
||||
}],
|
||||
},
|
||||
"remove" => ControllerSchema {
|
||||
namespace: "vault",
|
||||
function: "remove",
|
||||
description: "Remove a vault. Optionally purge its memory namespace.",
|
||||
inputs: vec![
|
||||
vault_id_input("Identifier of the vault to remove."),
|
||||
FieldSchema {
|
||||
name: "purge_memory",
|
||||
ty: TypeSchema::Option(Box::new(TypeSchema::Bool)),
|
||||
comment: "When true, also clear the vault's memory namespace.",
|
||||
required: false,
|
||||
},
|
||||
],
|
||||
outputs: vec![FieldSchema {
|
||||
name: "result",
|
||||
ty: TypeSchema::Object {
|
||||
fields: vec![
|
||||
FieldSchema {
|
||||
name: "vault_id",
|
||||
ty: TypeSchema::String,
|
||||
comment: "Identifier requested for removal.",
|
||||
required: true,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "removed",
|
||||
ty: TypeSchema::Bool,
|
||||
comment: "True when the vault row was deleted.",
|
||||
required: true,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "purged",
|
||||
ty: TypeSchema::Bool,
|
||||
comment: "True when the memory namespace was also cleared.",
|
||||
required: true,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "purge_error",
|
||||
ty: TypeSchema::Option(Box::new(TypeSchema::String)),
|
||||
comment: "Error detail when purge was requested but failed.",
|
||||
required: false,
|
||||
},
|
||||
],
|
||||
},
|
||||
comment: "Removal result payload.",
|
||||
required: true,
|
||||
}],
|
||||
},
|
||||
"sync" => ControllerSchema {
|
||||
namespace: "vault",
|
||||
function: "sync",
|
||||
description: "Start a background sync of a vault's root folder. Returns immediately. Poll `vault.sync_status` for progress.",
|
||||
inputs: vec![vault_id_input("Identifier of the vault to sync.")],
|
||||
outputs: vec![
|
||||
FieldSchema {
|
||||
name: "status",
|
||||
ty: TypeSchema::String,
|
||||
comment: "Always `\"started\"` on success.",
|
||||
required: true,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "vault_id",
|
||||
ty: TypeSchema::String,
|
||||
comment: "The vault that started syncing.",
|
||||
required: true,
|
||||
},
|
||||
],
|
||||
},
|
||||
"sync_status" => ControllerSchema {
|
||||
namespace: "vault",
|
||||
function: "sync_status",
|
||||
description: "Return the current sync progress and outcome for a vault.",
|
||||
inputs: vec![vault_id_input("Identifier of the vault to query.")],
|
||||
outputs: vec![FieldSchema {
|
||||
name: "state",
|
||||
ty: TypeSchema::Ref("VaultSyncState"),
|
||||
comment: "Current sync state (Idle / Running / Completed / Failed).",
|
||||
required: true,
|
||||
}],
|
||||
},
|
||||
_other => ControllerSchema {
|
||||
namespace: "vault",
|
||||
function: "unknown",
|
||||
description: "Unknown vault controller function.",
|
||||
inputs: vec![FieldSchema {
|
||||
name: "function",
|
||||
ty: TypeSchema::String,
|
||||
comment: "Unknown function requested for schema lookup.",
|
||||
required: true,
|
||||
}],
|
||||
outputs: vec![FieldSchema {
|
||||
name: "error",
|
||||
ty: TypeSchema::String,
|
||||
comment: "Lookup error details.",
|
||||
required: true,
|
||||
}],
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
fn handle_create(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let name = read_required::<String>(¶ms, "name")?;
|
||||
let root_path = read_required::<String>(¶ms, "root_path")?;
|
||||
let include_globs =
|
||||
read_optional::<Vec<String>>(¶ms, "include_globs")?.unwrap_or_default();
|
||||
let exclude_globs =
|
||||
read_optional::<Vec<String>>(¶ms, "exclude_globs")?.unwrap_or_default();
|
||||
to_json(
|
||||
crate::openhuman::vault::ops::vault_create(
|
||||
&config,
|
||||
&name,
|
||||
&root_path,
|
||||
include_globs,
|
||||
exclude_globs,
|
||||
)
|
||||
.await?,
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_list(_params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
to_json(crate::openhuman::vault::ops::vault_list(&config).await?)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_get(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let vault_id = read_required::<String>(¶ms, "vault_id")?;
|
||||
to_json(crate::openhuman::vault::ops::vault_get(&config, vault_id.trim()).await?)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_files(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let vault_id = read_required::<String>(¶ms, "vault_id")?;
|
||||
to_json(crate::openhuman::vault::ops::vault_files(&config, vault_id.trim()).await?)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_write_markdown(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let vault_id = read_required::<String>(¶ms, "vault_id")?;
|
||||
let rel_path = read_required::<String>(¶ms, "rel_path")?;
|
||||
let content = read_required::<String>(¶ms, "content")?;
|
||||
let overwrite = read_optional::<bool>(¶ms, "overwrite")?.unwrap_or(false);
|
||||
let approved = read_required::<bool>(¶ms, "approved")?;
|
||||
to_json(
|
||||
crate::openhuman::vault::ops::vault_write_markdown(
|
||||
&config, &vault_id, &rel_path, &content, overwrite, approved,
|
||||
)
|
||||
.await?,
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_remove(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let vault_id = read_required::<String>(¶ms, "vault_id")?;
|
||||
let purge_memory = read_optional::<bool>(¶ms, "purge_memory")?.unwrap_or(false);
|
||||
to_json(
|
||||
crate::openhuman::vault::ops::vault_remove(&config, vault_id.trim(), purge_memory)
|
||||
.await?,
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_sync(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let vault_id = read_required::<String>(¶ms, "vault_id")?;
|
||||
to_json(crate::openhuman::vault::ops::vault_sync(&config, vault_id.trim()).await?)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_sync_status(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let vault_id = read_required::<String>(¶ms, "vault_id")?;
|
||||
to_json(crate::openhuman::vault::ops::vault_sync_status(vault_id.trim()).await?)
|
||||
})
|
||||
}
|
||||
|
||||
fn read_required<T: DeserializeOwned>(params: &Map<String, Value>, key: &str) -> Result<T, String> {
|
||||
let value = params
|
||||
.get(key)
|
||||
.cloned()
|
||||
.ok_or_else(|| format!("missing required param '{key}'"))?;
|
||||
serde_json::from_value(value).map_err(|e| format!("invalid '{key}': {e}"))
|
||||
}
|
||||
|
||||
fn read_optional<T: DeserializeOwned>(
|
||||
params: &Map<String, Value>,
|
||||
key: &str,
|
||||
) -> Result<Option<T>, String> {
|
||||
match params.get(key) {
|
||||
None | Some(Value::Null) => Ok(None),
|
||||
Some(value) => serde_json::from_value(value.clone())
|
||||
.map(Some)
|
||||
.map_err(|e| format!("invalid '{key}': {e}")),
|
||||
}
|
||||
}
|
||||
|
||||
fn to_json<T: serde::Serialize>(outcome: RpcOutcome<T>) -> Result<Value, String> {
|
||||
outcome.into_cli_compatible_json()
|
||||
}
|
||||
@@ -1,80 +0,0 @@
|
||||
//! Global in-memory registry for vault sync progress state.
|
||||
//!
|
||||
//! State is keyed by `vault_id` and lives for the lifetime of the process.
|
||||
//! A completed/failed entry is retained until the next sync for the same
|
||||
//! vault overwrites it, so the frontend can always read the last outcome.
|
||||
//!
|
||||
//! Uses `once_cell::sync::Lazy` + `parking_lot::RwLock` — no heap allocation
|
||||
//! at import time, no `std::sync::Mutex` poisoning risk.
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
use once_cell::sync::Lazy;
|
||||
use parking_lot::RwLock;
|
||||
|
||||
use super::types::{VaultSyncState, VaultSyncStatus};
|
||||
|
||||
static SYNC_STATE: Lazy<RwLock<HashMap<String, VaultSyncState>>> =
|
||||
Lazy::new(|| RwLock::new(HashMap::new()));
|
||||
|
||||
/// Return the current sync state for a vault, or `None` if no sync has run.
|
||||
pub fn get(vault_id: &str) -> Option<VaultSyncState> {
|
||||
SYNC_STATE.read().get(vault_id).cloned()
|
||||
}
|
||||
|
||||
/// Replace the sync state for a vault (creates or overwrites the entry).
|
||||
pub fn set(state: VaultSyncState) {
|
||||
SYNC_STATE.write().insert(state.vault_id.clone(), state);
|
||||
}
|
||||
|
||||
/// Transition a vault to `Running`.
|
||||
///
|
||||
/// Returns `Err` if a sync for this vault is already in progress so the
|
||||
/// caller can reject duplicate requests.
|
||||
pub fn start(vault_id: &str, started_at_ms: i64) -> Result<(), String> {
|
||||
let mut map = SYNC_STATE.write();
|
||||
if let Some(s) = map.get(vault_id) {
|
||||
if s.status == VaultSyncStatus::Running {
|
||||
log::debug!(
|
||||
"[vault][state] start rejected: vault_id={vault_id} already running since={}",
|
||||
s.started_at_ms
|
||||
);
|
||||
return Err(format!("vault {vault_id} is already syncing"));
|
||||
}
|
||||
}
|
||||
log::debug!("[vault][state] start: vault_id={vault_id} started_at_ms={started_at_ms}");
|
||||
map.insert(
|
||||
vault_id.to_string(),
|
||||
VaultSyncState {
|
||||
vault_id: vault_id.to_string(),
|
||||
status: VaultSyncStatus::Running,
|
||||
scanned: 0,
|
||||
ingested: 0,
|
||||
unchanged: 0,
|
||||
removed: 0,
|
||||
failed: 0,
|
||||
skipped_unsupported: 0,
|
||||
total: 0,
|
||||
started_at_ms,
|
||||
finished_at_ms: None,
|
||||
duration_ms: 0,
|
||||
errors: vec![],
|
||||
},
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Apply `f` to the current `VaultSyncState` for `vault_id` in-place.
|
||||
///
|
||||
/// No-ops silently if no entry exists (e.g. if the registry was cleared or
|
||||
/// the vault_id is wrong — neither should happen in normal operation).
|
||||
pub fn update_progress(vault_id: &str, f: impl FnOnce(&mut VaultSyncState)) {
|
||||
let mut map = SYNC_STATE.write();
|
||||
if let Some(s) = map.get_mut(vault_id) {
|
||||
f(s);
|
||||
} else {
|
||||
log::debug!(
|
||||
"[vault][state] update_progress no-op: vault_id={vault_id} not found in state map"
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -1,404 +0,0 @@
|
||||
use std::collections::HashSet;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::{Mutex, OnceLock};
|
||||
|
||||
use anyhow::{anyhow, Context, Result};
|
||||
use chrono::{DateTime, Utc};
|
||||
use rusqlite::{params, Connection, OptionalExtension};
|
||||
|
||||
use crate::openhuman::config::Config;
|
||||
|
||||
use super::types::{Vault, VaultFile, VaultFileStatus, VaultWriteState};
|
||||
|
||||
static MIGRATED_VAULT_DBS: OnceLock<Mutex<HashSet<PathBuf>>> = OnceLock::new();
|
||||
|
||||
pub(crate) const VAULT_WRITE_REASON_EMPTY_PATH: &str = "empty_path";
|
||||
pub(crate) const VAULT_WRITE_REASON_UNAVAILABLE: &str = "unavailable";
|
||||
pub(crate) const VAULT_WRITE_REASON_NOT_DIRECTORY: &str = "not_directory";
|
||||
pub(crate) const VAULT_WRITE_REASON_READ_ONLY: &str = "read_only";
|
||||
pub(crate) const VAULT_WRITE_REASON_WRITABLE: &str = "writable";
|
||||
|
||||
pub(crate) fn with_connection<T>(
|
||||
config: &Config,
|
||||
f: impl FnOnce(&Connection) -> Result<T>,
|
||||
) -> Result<T> {
|
||||
let db_path = config.workspace_dir.join("vault").join("vault.db");
|
||||
if let Some(parent) = db_path.parent() {
|
||||
std::fs::create_dir_all(parent)
|
||||
.with_context(|| format!("Failed to create vault directory: {}", parent.display()))?;
|
||||
}
|
||||
|
||||
let conn = Connection::open(&db_path)
|
||||
.with_context(|| format!("Failed to open vault DB: {}", db_path.display()))?;
|
||||
|
||||
conn.execute_batch(
|
||||
"PRAGMA foreign_keys = ON;
|
||||
CREATE TABLE IF NOT EXISTS vaults (
|
||||
id TEXT PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
root_path TEXT NOT NULL,
|
||||
host_os TEXT,
|
||||
namespace TEXT NOT NULL UNIQUE,
|
||||
include_globs TEXT NOT NULL DEFAULT '[]',
|
||||
exclude_globs TEXT NOT NULL DEFAULT '[]',
|
||||
created_at TEXT NOT NULL,
|
||||
last_synced_at TEXT
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS vault_files (
|
||||
vault_id TEXT NOT NULL,
|
||||
rel_path TEXT NOT NULL,
|
||||
document_id TEXT NOT NULL,
|
||||
content_hash TEXT NOT NULL,
|
||||
mtime_ms INTEGER NOT NULL,
|
||||
bytes INTEGER NOT NULL,
|
||||
ingested_at TEXT NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'ok',
|
||||
PRIMARY KEY (vault_id, rel_path),
|
||||
FOREIGN KEY (vault_id) REFERENCES vaults(id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_vault_files_vault ON vault_files(vault_id);",
|
||||
)
|
||||
.context("Failed to initialize vault schema")?;
|
||||
|
||||
let migrated = MIGRATED_VAULT_DBS.get_or_init(|| Mutex::new(HashSet::new()));
|
||||
let mut migrated_paths = migrated
|
||||
.lock()
|
||||
.map_err(|_| anyhow!("Failed to lock vault migration cache"))?;
|
||||
if !migrated_paths.contains(&db_path) {
|
||||
ensure_host_os_column(&conn).context("Failed to migrate vault schema")?;
|
||||
migrated_paths.insert(db_path.clone());
|
||||
}
|
||||
|
||||
f(&conn)
|
||||
}
|
||||
|
||||
pub fn insert_vault(config: &Config, vault: &Vault) -> Result<()> {
|
||||
insert_vault_inner(config, vault, true)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn insert_vault_preserving_host_for_tests(config: &Config, vault: &Vault) -> Result<()> {
|
||||
insert_vault_inner(config, vault, false)
|
||||
}
|
||||
|
||||
fn insert_vault_inner(config: &Config, vault: &Vault, stamp_current_host: bool) -> Result<()> {
|
||||
with_connection(config, |conn| {
|
||||
let host_os = normalized_host_os(vault.host_os.as_deref()).or_else(|| {
|
||||
if stamp_current_host {
|
||||
Some(current_host_os())
|
||||
} else {
|
||||
None
|
||||
}
|
||||
});
|
||||
conn.execute(
|
||||
"INSERT INTO vaults (id, name, root_path, host_os, namespace, include_globs, exclude_globs, created_at, last_synced_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
|
||||
params![
|
||||
vault.id,
|
||||
vault.name,
|
||||
vault.root_path,
|
||||
host_os,
|
||||
vault.namespace,
|
||||
serde_json::to_string(&vault.include_globs)?,
|
||||
serde_json::to_string(&vault.exclude_globs)?,
|
||||
vault.created_at.to_rfc3339(),
|
||||
vault.last_synced_at.map(|t| t.to_rfc3339()),
|
||||
],
|
||||
)
|
||||
.context("Failed to insert vault")?;
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
pub fn list_vaults(config: &Config) -> Result<Vec<Vault>> {
|
||||
with_connection(config, |conn| {
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT v.id, v.name, v.root_path, v.host_os, v.namespace, v.include_globs, v.exclude_globs,
|
||||
v.created_at, v.last_synced_at,
|
||||
(SELECT COUNT(*) FROM vault_files vf WHERE vf.vault_id = v.id AND vf.status = 'ok')
|
||||
FROM vaults v
|
||||
ORDER BY v.created_at DESC",
|
||||
)?;
|
||||
let rows = stmt.query_map([], row_to_vault)?;
|
||||
let mut out = Vec::new();
|
||||
for row in rows {
|
||||
let vault = row?;
|
||||
if vault_belongs_to_current_host(&vault) {
|
||||
out.push(vault);
|
||||
} else {
|
||||
log::debug!(
|
||||
"[vault] hiding incompatible vault id={} host_os={:?}",
|
||||
vault.id,
|
||||
vault.host_os
|
||||
);
|
||||
}
|
||||
}
|
||||
Ok(out)
|
||||
})
|
||||
}
|
||||
|
||||
pub fn get_vault(config: &Config, id: &str) -> Result<Option<Vault>> {
|
||||
with_connection(config, |conn| {
|
||||
let vault = conn
|
||||
.query_row(
|
||||
"SELECT v.id, v.name, v.root_path, v.host_os, v.namespace, v.include_globs, v.exclude_globs,
|
||||
v.created_at, v.last_synced_at,
|
||||
(SELECT COUNT(*) FROM vault_files vf WHERE vf.vault_id = v.id AND vf.status = 'ok')
|
||||
FROM vaults v WHERE v.id = ?1",
|
||||
params![id],
|
||||
row_to_vault,
|
||||
)
|
||||
.optional()
|
||||
.context("Failed to read vault")?;
|
||||
Ok(vault.filter(vault_belongs_to_current_host))
|
||||
})
|
||||
}
|
||||
|
||||
pub fn remove_vault(config: &Config, id: &str) -> Result<bool> {
|
||||
with_connection(config, |conn| {
|
||||
let n = conn
|
||||
.execute("DELETE FROM vaults WHERE id = ?1", params![id])
|
||||
.context("Failed to delete vault")?;
|
||||
Ok(n > 0)
|
||||
})
|
||||
}
|
||||
|
||||
pub fn touch_last_synced(config: &Config, id: &str, when: DateTime<Utc>) -> Result<()> {
|
||||
with_connection(config, |conn| {
|
||||
conn.execute(
|
||||
"UPDATE vaults SET last_synced_at = ?2 WHERE id = ?1",
|
||||
params![id, when.to_rfc3339()],
|
||||
)?;
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
pub fn list_files(config: &Config, vault_id: &str) -> Result<Vec<VaultFile>> {
|
||||
with_connection(config, |conn| {
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT vault_id, rel_path, document_id, content_hash, mtime_ms, bytes, ingested_at, status
|
||||
FROM vault_files WHERE vault_id = ?1 ORDER BY rel_path",
|
||||
)?;
|
||||
let rows = stmt.query_map(params![vault_id], row_to_file)?;
|
||||
let mut out = Vec::new();
|
||||
for row in rows {
|
||||
out.push(row?);
|
||||
}
|
||||
Ok(out)
|
||||
})
|
||||
}
|
||||
|
||||
pub fn upsert_file(config: &Config, file: &VaultFile) -> Result<()> {
|
||||
with_connection(config, |conn| {
|
||||
conn.execute(
|
||||
"INSERT INTO vault_files (vault_id, rel_path, document_id, content_hash, mtime_ms, bytes, ingested_at, status)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
|
||||
ON CONFLICT(vault_id, rel_path) DO UPDATE SET
|
||||
document_id = excluded.document_id,
|
||||
content_hash = excluded.content_hash,
|
||||
mtime_ms = excluded.mtime_ms,
|
||||
bytes = excluded.bytes,
|
||||
ingested_at = excluded.ingested_at,
|
||||
status = excluded.status",
|
||||
params![
|
||||
file.vault_id,
|
||||
file.rel_path,
|
||||
file.document_id,
|
||||
file.content_hash,
|
||||
file.mtime_ms,
|
||||
file.bytes as i64,
|
||||
file.ingested_at.to_rfc3339(),
|
||||
file.status.as_str(),
|
||||
],
|
||||
)?;
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
pub fn delete_file(config: &Config, vault_id: &str, rel_path: &str) -> Result<()> {
|
||||
with_connection(config, |conn| {
|
||||
conn.execute(
|
||||
"DELETE FROM vault_files WHERE vault_id = ?1 AND rel_path = ?2",
|
||||
params![vault_id, rel_path],
|
||||
)?;
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
fn row_to_vault(row: &rusqlite::Row<'_>) -> rusqlite::Result<Vault> {
|
||||
let include_raw: String = row.get(5)?;
|
||||
let exclude_raw: String = row.get(6)?;
|
||||
let created_raw: String = row.get(7)?;
|
||||
let last_raw: Option<String> = row.get(8)?;
|
||||
let file_count: i64 = row.get(9)?;
|
||||
let root_path: String = row.get(2)?;
|
||||
let (write_state, write_state_reason) = vault_write_state_for_root_path(&root_path);
|
||||
Ok(Vault {
|
||||
id: row.get(0)?,
|
||||
name: row.get(1)?,
|
||||
root_path,
|
||||
host_os: row.get(3)?,
|
||||
namespace: row.get(4)?,
|
||||
include_globs: serde_json::from_str(&include_raw).unwrap_or_default(),
|
||||
exclude_globs: serde_json::from_str(&exclude_raw).unwrap_or_default(),
|
||||
created_at: parse_dt(&created_raw),
|
||||
last_synced_at: last_raw.as_deref().map(parse_dt),
|
||||
file_count: file_count.max(0) as u64,
|
||||
write_state,
|
||||
write_state_reason,
|
||||
})
|
||||
}
|
||||
|
||||
fn row_to_file(row: &rusqlite::Row<'_>) -> rusqlite::Result<VaultFile> {
|
||||
let ingested_raw: String = row.get(6)?;
|
||||
let status_raw: String = row.get(7)?;
|
||||
let bytes: i64 = row.get(5)?;
|
||||
Ok(VaultFile {
|
||||
vault_id: row.get(0)?,
|
||||
rel_path: row.get(1)?,
|
||||
document_id: row.get(2)?,
|
||||
content_hash: row.get(3)?,
|
||||
mtime_ms: row.get(4)?,
|
||||
bytes: bytes.max(0) as u64,
|
||||
ingested_at: parse_dt(&ingested_raw),
|
||||
status: VaultFileStatus::parse(&status_raw),
|
||||
})
|
||||
}
|
||||
|
||||
fn parse_dt(raw: &str) -> DateTime<Utc> {
|
||||
DateTime::parse_from_rfc3339(raw)
|
||||
.map(|t| t.with_timezone(&Utc))
|
||||
.unwrap_or_else(|_| Utc::now())
|
||||
}
|
||||
|
||||
fn ensure_host_os_column(conn: &Connection) -> Result<()> {
|
||||
let mut stmt = conn.prepare("PRAGMA table_info(vaults)")?;
|
||||
let columns = stmt.query_map([], |row| row.get::<_, String>(1))?;
|
||||
let mut has_host_os = false;
|
||||
for column in columns {
|
||||
if column?.eq_ignore_ascii_case("host_os") {
|
||||
has_host_os = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if !has_host_os {
|
||||
conn.execute("ALTER TABLE vaults ADD COLUMN host_os TEXT", [])?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn current_host_os() -> &'static str {
|
||||
std::env::consts::OS
|
||||
}
|
||||
|
||||
pub(crate) fn vault_write_state_for_root_path(
|
||||
root_path: &str,
|
||||
) -> (VaultWriteState, Option<String>) {
|
||||
let trimmed = root_path.trim();
|
||||
if trimmed.is_empty() {
|
||||
return (
|
||||
VaultWriteState::Unavailable,
|
||||
Some(VAULT_WRITE_REASON_EMPTY_PATH.to_string()),
|
||||
);
|
||||
}
|
||||
|
||||
let path = std::path::Path::new(trimmed);
|
||||
let metadata = match std::fs::metadata(path) {
|
||||
Ok(metadata) => metadata,
|
||||
Err(_) => {
|
||||
return (
|
||||
VaultWriteState::Unavailable,
|
||||
Some(VAULT_WRITE_REASON_UNAVAILABLE.to_string()),
|
||||
)
|
||||
}
|
||||
};
|
||||
|
||||
if !metadata.is_dir() {
|
||||
return (
|
||||
VaultWriteState::Unavailable,
|
||||
Some(VAULT_WRITE_REASON_NOT_DIRECTORY.to_string()),
|
||||
);
|
||||
}
|
||||
|
||||
if metadata.permissions().readonly() {
|
||||
return (
|
||||
VaultWriteState::ReadOnly,
|
||||
Some(VAULT_WRITE_REASON_READ_ONLY.to_string()),
|
||||
);
|
||||
}
|
||||
|
||||
(
|
||||
VaultWriteState::Writable,
|
||||
Some(VAULT_WRITE_REASON_WRITABLE.to_string()),
|
||||
)
|
||||
}
|
||||
|
||||
pub(crate) fn vault_write_state_reason_message(reason_code: Option<&str>) -> &'static str {
|
||||
match reason_code {
|
||||
Some(VAULT_WRITE_REASON_EMPTY_PATH) => "Vault folder path is empty.",
|
||||
Some(VAULT_WRITE_REASON_UNAVAILABLE) => "Vault folder is not available on this device.",
|
||||
Some(VAULT_WRITE_REASON_NOT_DIRECTORY) => "Vault path is not a directory.",
|
||||
Some(VAULT_WRITE_REASON_READ_ONLY) => "Vault folder is read-only on this device.",
|
||||
Some(VAULT_WRITE_REASON_WRITABLE) => {
|
||||
"Approved markdown/wiki writes can be saved in this vault."
|
||||
}
|
||||
_ => "Vault write state is unknown.",
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn path_looks_compatible_with_host_os(raw_path: &str, host_os: &str) -> bool {
|
||||
let path = raw_path.trim();
|
||||
if path.is_empty() {
|
||||
return false;
|
||||
}
|
||||
|
||||
if is_windows_host_os(host_os) {
|
||||
return looks_like_windows_absolute_path(path);
|
||||
}
|
||||
|
||||
looks_like_unix_absolute_path(path)
|
||||
}
|
||||
|
||||
fn vault_belongs_to_current_host(vault: &Vault) -> bool {
|
||||
let current = current_host_os();
|
||||
let Some(host_os) = normalized_host_os(vault.host_os.as_deref()) else {
|
||||
return path_looks_compatible_with_host_os(&vault.root_path, current);
|
||||
};
|
||||
|
||||
host_os.eq_ignore_ascii_case(current)
|
||||
&& path_looks_compatible_with_host_os(&vault.root_path, current)
|
||||
}
|
||||
|
||||
fn normalized_host_os(raw: Option<&str>) -> Option<&str> {
|
||||
raw.map(str::trim).filter(|host_os| !host_os.is_empty())
|
||||
}
|
||||
|
||||
fn is_windows_host_os(host_os: &str) -> bool {
|
||||
host_os.eq_ignore_ascii_case("windows") || host_os.eq_ignore_ascii_case("win32")
|
||||
}
|
||||
|
||||
fn looks_like_windows_absolute_path(path: &str) -> bool {
|
||||
looks_like_windows_drive_path(path) || looks_like_windows_unc_path(path)
|
||||
}
|
||||
|
||||
fn looks_like_windows_drive_path(path: &str) -> bool {
|
||||
let bytes = path.as_bytes();
|
||||
bytes.len() >= 3
|
||||
&& bytes[0].is_ascii_alphabetic()
|
||||
&& bytes[1] == b':'
|
||||
&& matches!(bytes[2], b'\\' | b'/')
|
||||
}
|
||||
|
||||
/// Only backslash-style UNC (`\\server\share`). Forward-slash `//…` is
|
||||
/// POSIX-legal and must not be classified as Windows.
|
||||
fn looks_like_windows_unc_path(path: &str) -> bool {
|
||||
let bytes = path.as_bytes();
|
||||
bytes.len() >= 3 && bytes[0] == b'\\' && bytes[1] == b'\\' && !matches!(bytes[2], b'\\' | b'/')
|
||||
}
|
||||
|
||||
fn looks_like_unix_absolute_path(path: &str) -> bool {
|
||||
path.starts_with('/')
|
||||
&& !looks_like_windows_drive_path(path)
|
||||
&& !looks_like_windows_unc_path(path)
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,643 +0,0 @@
|
||||
//! Unit tests for the vault domain. Hits a real SQLite db in a tempdir,
|
||||
//! but skips memory ingestion (covered in higher-level integration tests).
|
||||
|
||||
use std::path::PathBuf;
|
||||
use tempfile::TempDir;
|
||||
|
||||
use crate::openhuman::config::Config;
|
||||
|
||||
use super::ops;
|
||||
use super::state;
|
||||
use super::store;
|
||||
use super::sync::supported_extension;
|
||||
use super::types::{
|
||||
Vault, VaultFile, VaultFileStatus, VaultSyncState, VaultSyncStatus, VaultWriteState,
|
||||
};
|
||||
|
||||
fn make_config(tmp: &TempDir) -> Config {
|
||||
let mut config = Config::default();
|
||||
config.workspace_dir = tmp.path().to_path_buf();
|
||||
config
|
||||
}
|
||||
|
||||
fn sample_vault(root: PathBuf) -> Vault {
|
||||
Vault {
|
||||
id: "vault-test-1".to_string(),
|
||||
name: "Test".to_string(),
|
||||
root_path: root.to_string_lossy().to_string(),
|
||||
host_os: None,
|
||||
namespace: "vault:vault-test-1".to_string(),
|
||||
include_globs: vec![],
|
||||
exclude_globs: vec![],
|
||||
created_at: chrono::Utc::now(),
|
||||
last_synced_at: None,
|
||||
file_count: 0,
|
||||
write_state: VaultWriteState::Writable,
|
||||
write_state_reason: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn incompatible_path_for_current_host() -> &'static str {
|
||||
if cfg!(windows) {
|
||||
"/home/leigh/OHvault"
|
||||
} else {
|
||||
r"C:\Users\leigh\OHvault"
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn path_compatibility_rejects_cross_platform_absolute_paths() {
|
||||
assert!(store::path_looks_compatible_with_host_os(
|
||||
r"C:\Users\leigh\OHvault",
|
||||
"windows"
|
||||
));
|
||||
assert!(store::path_looks_compatible_with_host_os(
|
||||
r"\\server\share\OHvault",
|
||||
"windows"
|
||||
));
|
||||
// Forward-slash `//…` is POSIX-legal, not Windows UNC.
|
||||
assert!(!store::path_looks_compatible_with_host_os(
|
||||
"//server/share/OHvault",
|
||||
"windows"
|
||||
));
|
||||
assert!(!store::path_looks_compatible_with_host_os(
|
||||
"/home/leigh/OHvault",
|
||||
"windows"
|
||||
));
|
||||
|
||||
assert!(store::path_looks_compatible_with_host_os(
|
||||
"/home/leigh/OHvault",
|
||||
"linux"
|
||||
));
|
||||
assert!(store::path_looks_compatible_with_host_os(
|
||||
"/Users/leigh/OHvault",
|
||||
"macos"
|
||||
));
|
||||
assert!(!store::path_looks_compatible_with_host_os(
|
||||
r"C:\Users\leigh\OHvault",
|
||||
"linux"
|
||||
));
|
||||
assert!(!store::path_looks_compatible_with_host_os(
|
||||
r"\\server\share\OHvault",
|
||||
"macos"
|
||||
));
|
||||
// Forward-slash `//…` is POSIX-legal — compatible with Unix hosts.
|
||||
assert!(store::path_looks_compatible_with_host_os(
|
||||
"//server/share/OHvault",
|
||||
"macos"
|
||||
));
|
||||
assert!(store::path_looks_compatible_with_host_os(
|
||||
"//server/share/OHvault",
|
||||
"linux"
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn store_stamps_new_vaults_with_current_host_os() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = make_config(&tmp);
|
||||
let vault = sample_vault(tmp.path().to_path_buf());
|
||||
|
||||
store::insert_vault(&config, &vault).unwrap();
|
||||
|
||||
let listed = store::list_vaults(&config).unwrap();
|
||||
assert_eq!(listed.len(), 1);
|
||||
assert_eq!(listed[0].host_os.as_deref(), Some(std::env::consts::OS));
|
||||
assert_eq!(listed[0].write_state, VaultWriteState::Writable);
|
||||
assert_eq!(
|
||||
listed[0].write_state_reason.as_deref(),
|
||||
Some(store::VAULT_WRITE_REASON_WRITABLE)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn store_marks_missing_vault_folder_unavailable_instead_of_hiding_it() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = make_config(&tmp);
|
||||
let vault_root = tmp.path().join("vault-root");
|
||||
std::fs::create_dir_all(&vault_root).unwrap();
|
||||
let vault = sample_vault(vault_root.clone());
|
||||
|
||||
store::insert_vault(&config, &vault).unwrap();
|
||||
std::fs::remove_dir_all(&vault_root).unwrap();
|
||||
|
||||
let listed = store::list_vaults(&config).unwrap();
|
||||
assert_eq!(listed.len(), 1);
|
||||
assert_eq!(listed[0].write_state, VaultWriteState::Unavailable);
|
||||
assert_eq!(
|
||||
listed[0].write_state_reason.as_deref(),
|
||||
Some(store::VAULT_WRITE_REASON_UNAVAILABLE)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn store_filters_legacy_vaults_whose_path_belongs_to_another_host_family() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = make_config(&tmp);
|
||||
let mut vault = sample_vault(PathBuf::from(incompatible_path_for_current_host()));
|
||||
vault.host_os = None;
|
||||
|
||||
store::insert_vault_preserving_host_for_tests(&config, &vault).unwrap();
|
||||
|
||||
assert!(store::list_vaults(&config).unwrap().is_empty());
|
||||
assert!(store::get_vault(&config, &vault.id).unwrap().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn store_filters_vaults_created_on_a_different_host_os() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = make_config(&tmp);
|
||||
let mut vault = sample_vault(tmp.path().to_path_buf());
|
||||
vault.host_os = Some(if cfg!(windows) { "linux" } else { "windows" }.to_string());
|
||||
|
||||
store::insert_vault_preserving_host_for_tests(&config, &vault).unwrap();
|
||||
|
||||
assert!(store::list_vaults(&config).unwrap().is_empty());
|
||||
assert!(store::get_vault(&config, &vault.id).unwrap().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn supported_extension_accepts_md_and_code() {
|
||||
assert!(supported_extension("md"));
|
||||
assert!(supported_extension("MD"));
|
||||
assert!(supported_extension("rs"));
|
||||
assert!(supported_extension("tsx"));
|
||||
assert!(!supported_extension("png"));
|
||||
assert!(!supported_extension("zip"));
|
||||
assert!(!supported_extension(""));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn store_insert_get_list_remove_roundtrip() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = make_config(&tmp);
|
||||
let vault = sample_vault(tmp.path().to_path_buf());
|
||||
|
||||
store::insert_vault(&config, &vault).unwrap();
|
||||
|
||||
let listed = store::list_vaults(&config).unwrap();
|
||||
assert_eq!(listed.len(), 1);
|
||||
assert_eq!(listed[0].id, vault.id);
|
||||
assert_eq!(listed[0].namespace, vault.namespace);
|
||||
assert_eq!(listed[0].file_count, 0);
|
||||
|
||||
let fetched = store::get_vault(&config, &vault.id).unwrap();
|
||||
assert!(fetched.is_some());
|
||||
assert_eq!(fetched.unwrap().name, "Test");
|
||||
|
||||
let removed = store::remove_vault(&config, &vault.id).unwrap();
|
||||
assert!(removed);
|
||||
assert!(store::list_vaults(&config).unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn store_files_upsert_and_delete() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = make_config(&tmp);
|
||||
let vault = sample_vault(tmp.path().to_path_buf());
|
||||
store::insert_vault(&config, &vault).unwrap();
|
||||
|
||||
let file = VaultFile {
|
||||
vault_id: vault.id.clone(),
|
||||
rel_path: "notes/one.md".to_string(),
|
||||
document_id: "doc-1".to_string(),
|
||||
content_hash: "h1".to_string(),
|
||||
mtime_ms: 100,
|
||||
bytes: 42,
|
||||
ingested_at: chrono::Utc::now(),
|
||||
status: VaultFileStatus::Ok,
|
||||
};
|
||||
store::upsert_file(&config, &file).unwrap();
|
||||
|
||||
let listed = store::list_files(&config, &vault.id).unwrap();
|
||||
assert_eq!(listed.len(), 1);
|
||||
assert_eq!(listed[0].document_id, "doc-1");
|
||||
|
||||
// Re-upsert with same key should update, not duplicate.
|
||||
let mut updated = file.clone();
|
||||
updated.content_hash = "h2".to_string();
|
||||
updated.mtime_ms = 200;
|
||||
store::upsert_file(&config, &updated).unwrap();
|
||||
let listed = store::list_files(&config, &vault.id).unwrap();
|
||||
assert_eq!(listed.len(), 1);
|
||||
assert_eq!(listed[0].content_hash, "h2");
|
||||
assert_eq!(listed[0].mtime_ms, 200);
|
||||
|
||||
// File count on vault list should reflect 1 OK row.
|
||||
let vaults = store::list_vaults(&config).unwrap();
|
||||
assert_eq!(vaults[0].file_count, 1);
|
||||
|
||||
store::delete_file(&config, &vault.id, "notes/one.md").unwrap();
|
||||
assert!(store::list_files(&config, &vault.id).unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn remove_vault_cascades_files() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = make_config(&tmp);
|
||||
let vault = sample_vault(tmp.path().to_path_buf());
|
||||
store::insert_vault(&config, &vault).unwrap();
|
||||
|
||||
let file = VaultFile {
|
||||
vault_id: vault.id.clone(),
|
||||
rel_path: "a.md".to_string(),
|
||||
document_id: "doc-a".to_string(),
|
||||
content_hash: "h".to_string(),
|
||||
mtime_ms: 1,
|
||||
bytes: 1,
|
||||
ingested_at: chrono::Utc::now(),
|
||||
status: VaultFileStatus::Ok,
|
||||
};
|
||||
store::upsert_file(&config, &file).unwrap();
|
||||
|
||||
store::remove_vault(&config, &vault.id).unwrap();
|
||||
// Cascade should have wiped vault_files rows for this id.
|
||||
assert!(store::list_files(&config, &vault.id).unwrap().is_empty());
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// state.rs — in-memory sync state registry
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
fn make_state(vault_id: &str, status: VaultSyncStatus) -> VaultSyncState {
|
||||
VaultSyncState {
|
||||
vault_id: vault_id.to_string(),
|
||||
status,
|
||||
scanned: 0,
|
||||
ingested: 0,
|
||||
unchanged: 0,
|
||||
removed: 0,
|
||||
failed: 0,
|
||||
skipped_unsupported: 0,
|
||||
total: 0,
|
||||
started_at_ms: 100,
|
||||
finished_at_ms: None,
|
||||
duration_ms: 0,
|
||||
errors: vec![],
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn state_get_returns_none_for_unknown() {
|
||||
// Use a unique ID so parallel tests can't collide via the global map.
|
||||
assert!(state::get("__test_unknown_99z__").is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn state_set_and_get_roundtrip() {
|
||||
let id = "__test_set_1__";
|
||||
state::set(make_state(id, VaultSyncStatus::Completed));
|
||||
let st = state::get(id).unwrap();
|
||||
assert_eq!(st.status, VaultSyncStatus::Completed);
|
||||
assert_eq!(st.vault_id, id);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn state_start_creates_running_entry() {
|
||||
let id = "__test_start_1__";
|
||||
state::start(id, 12345).unwrap();
|
||||
let st = state::get(id).unwrap();
|
||||
assert_eq!(st.status, VaultSyncStatus::Running);
|
||||
assert_eq!(st.started_at_ms, 12345);
|
||||
assert_eq!(st.ingested, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn state_start_rejects_duplicate_running() {
|
||||
let id = "__test_start_dup__";
|
||||
state::start(id, 1).unwrap();
|
||||
let err = state::start(id, 2).unwrap_err();
|
||||
assert!(err.contains("already syncing"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn state_start_allowed_after_completed() {
|
||||
let id = "__test_start_after_completed__";
|
||||
state::start(id, 1).unwrap();
|
||||
// Mark as completed, then start again — must succeed.
|
||||
state::update_progress(id, |s| s.status = VaultSyncStatus::Completed);
|
||||
state::start(id, 2).unwrap();
|
||||
assert_eq!(state::get(id).unwrap().status, VaultSyncStatus::Running);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn state_start_allowed_after_failed() {
|
||||
let id = "__test_start_after_failed__";
|
||||
state::start(id, 1).unwrap();
|
||||
state::update_progress(id, |s| s.status = VaultSyncStatus::Failed);
|
||||
state::start(id, 2).unwrap();
|
||||
assert_eq!(state::get(id).unwrap().status, VaultSyncStatus::Running);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn state_update_progress_mutates_entry() {
|
||||
let id = "__test_update_1__";
|
||||
state::start(id, 1).unwrap();
|
||||
state::update_progress(id, |s| {
|
||||
s.ingested = 7;
|
||||
s.scanned = 10;
|
||||
s.total = 10;
|
||||
});
|
||||
let st = state::get(id).unwrap();
|
||||
assert_eq!(st.ingested, 7);
|
||||
assert_eq!(st.scanned, 10);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn state_update_progress_noop_on_missing() {
|
||||
// Must not panic when vault_id is absent from the map.
|
||||
state::update_progress("__test_noop_xyz__", |s| {
|
||||
s.ingested = 999; // should never execute
|
||||
});
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// ops.rs — vault_sync_status RPC operation
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[tokio::test]
|
||||
async fn vault_create_returns_current_host_os() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = make_config(&tmp);
|
||||
|
||||
let outcome = ops::vault_create(
|
||||
&config,
|
||||
"Test",
|
||||
tmp.path().to_str().unwrap(),
|
||||
vec![],
|
||||
vec![],
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(outcome.value.host_os.as_deref(), Some(std::env::consts::OS));
|
||||
assert_eq!(outcome.value.write_state, VaultWriteState::Writable);
|
||||
assert_eq!(
|
||||
outcome.value.write_state_reason.as_deref(),
|
||||
Some(store::VAULT_WRITE_REASON_WRITABLE)
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn vault_create_uses_pii_safe_memory_namespace() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = make_config(&tmp);
|
||||
|
||||
let outcome = ops::vault_create(
|
||||
&config,
|
||||
"Test",
|
||||
tmp.path().to_str().unwrap(),
|
||||
vec![],
|
||||
vec![],
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let namespace = &outcome.value.namespace;
|
||||
assert!(namespace.starts_with("vault-"));
|
||||
assert!(!namespace.contains(&outcome.value.id));
|
||||
assert!(!crate::openhuman::memory_store::safety::has_likely_secret(
|
||||
namespace
|
||||
));
|
||||
assert!(!crate::openhuman::memory_store::safety::pii::has_likely_pii(namespace));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn vault_namespace_derivation_does_not_embed_pii_like_ids() {
|
||||
let namespace = ops::vault_namespace_for_id("VECJ880326XK4");
|
||||
|
||||
assert!(namespace.starts_with("vault-"));
|
||||
assert!(!namespace.contains("VECJ880326XK4"));
|
||||
assert!(!crate::openhuman::memory_store::safety::has_likely_secret(
|
||||
&namespace
|
||||
));
|
||||
assert!(!crate::openhuman::memory_store::safety::pii::has_likely_pii(&namespace));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn vault_sync_status_returns_idle_for_unknown_vault() {
|
||||
let outcome = ops::vault_sync_status("__ops_status_unknown__")
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(outcome.value.status, VaultSyncStatus::Idle);
|
||||
assert_eq!(outcome.value.vault_id, "__ops_status_unknown__");
|
||||
assert_eq!(outcome.value.ingested, 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn vault_write_markdown_requires_explicit_approval() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = make_config(&tmp);
|
||||
|
||||
let vault = ops::vault_create(
|
||||
&config,
|
||||
"Test",
|
||||
tmp.path().to_str().unwrap(),
|
||||
vec![],
|
||||
vec![],
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.value;
|
||||
|
||||
let err = ops::vault_write_markdown(
|
||||
&config,
|
||||
&vault.id,
|
||||
"wiki/summary.md",
|
||||
"# Summary\n",
|
||||
false,
|
||||
false,
|
||||
)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(err.contains("explicit user approval"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn vault_write_markdown_creates_and_updates_relative_markdown() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = make_config(&tmp);
|
||||
let vault_root = tmp.path().join("vault-root");
|
||||
std::fs::create_dir_all(&vault_root).unwrap();
|
||||
|
||||
let vault = ops::vault_create(
|
||||
&config,
|
||||
"Test",
|
||||
vault_root.to_str().unwrap(),
|
||||
vec![],
|
||||
vec![],
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.value;
|
||||
|
||||
let first = ops::vault_write_markdown(
|
||||
&config,
|
||||
&vault.id,
|
||||
"wiki/summary.md",
|
||||
"# Summary\n\nInitial.",
|
||||
false,
|
||||
true,
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.value;
|
||||
assert!(first.created);
|
||||
assert_eq!(first.rel_path, "wiki/summary.md");
|
||||
assert_eq!(first.bytes_written, "# Summary\n\nInitial.".len() as u64);
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(vault_root.join("wiki/summary.md")).unwrap(),
|
||||
"# Summary\n\nInitial."
|
||||
);
|
||||
|
||||
let duplicate = ops::vault_write_markdown(
|
||||
&config,
|
||||
&vault.id,
|
||||
"wiki/summary.md",
|
||||
"# Summary\n\nUpdated.",
|
||||
false,
|
||||
true,
|
||||
)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(duplicate.contains("already exists"));
|
||||
|
||||
let updated = ops::vault_write_markdown(
|
||||
&config,
|
||||
&vault.id,
|
||||
"wiki/summary.md",
|
||||
"# Summary\n\nUpdated.",
|
||||
true,
|
||||
true,
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.value;
|
||||
assert!(!updated.created);
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(vault_root.join("wiki/summary.md")).unwrap(),
|
||||
"# Summary\n\nUpdated."
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn vault_write_markdown_rejects_escape_paths_and_non_markdown() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = make_config(&tmp);
|
||||
let vault = ops::vault_create(
|
||||
&config,
|
||||
"Test",
|
||||
tmp.path().to_str().unwrap(),
|
||||
vec![],
|
||||
vec![],
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.value;
|
||||
|
||||
let traversal = ops::vault_write_markdown(&config, &vault.id, "../x.md", "x", false, true)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(traversal.contains(".."));
|
||||
|
||||
let non_markdown =
|
||||
ops::vault_write_markdown(&config, &vault.id, "notes/out.txt", "x", false, true)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(non_markdown.contains(".md"));
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn vault_write_markdown_rejects_symlink_escape() {
|
||||
use std::os::unix::fs::symlink;
|
||||
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = make_config(&tmp);
|
||||
let vault_root = tmp.path().join("vault-root");
|
||||
let outside = tmp.path().join("outside");
|
||||
std::fs::create_dir_all(&vault_root).unwrap();
|
||||
std::fs::create_dir_all(&outside).unwrap();
|
||||
symlink(&outside, vault_root.join("linked")).unwrap();
|
||||
|
||||
let vault = ops::vault_create(
|
||||
&config,
|
||||
"Test",
|
||||
vault_root.to_str().unwrap(),
|
||||
vec![],
|
||||
vec![],
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.value;
|
||||
|
||||
let err = ops::vault_write_markdown(&config, &vault.id, "linked/escape.md", "x", false, true)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(err.contains("outside the vault"));
|
||||
assert!(!outside.join("escape.md").exists());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn vault_sync_status_returns_state_when_present() {
|
||||
let id = "__ops_status_running__";
|
||||
let mut st = make_state(id, VaultSyncStatus::Running);
|
||||
st.scanned = 10;
|
||||
st.ingested = 5;
|
||||
st.total = 10;
|
||||
state::set(st);
|
||||
|
||||
let outcome = ops::vault_sync_status(id).await.unwrap();
|
||||
assert_eq!(outcome.value.status, VaultSyncStatus::Running);
|
||||
assert_eq!(outcome.value.scanned, 10);
|
||||
assert_eq!(outcome.value.ingested, 5);
|
||||
assert_eq!(outcome.value.total, 10);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn vault_sync_status_returns_completed_state() {
|
||||
let id = "__ops_status_completed__";
|
||||
let mut st = make_state(id, VaultSyncStatus::Completed);
|
||||
st.ingested = 12;
|
||||
st.failed = 1;
|
||||
st.duration_ms = 500;
|
||||
st.errors = vec!["file.txt: too large".to_string()];
|
||||
state::set(st);
|
||||
|
||||
let outcome = ops::vault_sync_status(id).await.unwrap();
|
||||
assert_eq!(outcome.value.status, VaultSyncStatus::Completed);
|
||||
assert_eq!(outcome.value.ingested, 12);
|
||||
assert_eq!(outcome.value.failed, 1);
|
||||
assert_eq!(outcome.value.errors.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn vault_sync_status_rejects_empty_id() {
|
||||
let err = ops::vault_sync_status("").await.unwrap_err();
|
||||
assert!(err.contains("vault_id must not be empty"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn vault_sync_panic_guard_marks_state_failed_and_allows_retry() {
|
||||
// Simulate the panic-recovery path that the catch_unwind guard in
|
||||
// ops::vault_sync triggers: vault goes Running -> Failed (with a panic
|
||||
// message), then can be restarted. This verifies the invariant that no
|
||||
// panic can permanently lock the state in `Running`.
|
||||
let id = "__test_panic_guard_recovery__";
|
||||
state::start(id, 1_000).unwrap();
|
||||
assert_eq!(state::get(id).unwrap().status, VaultSyncStatus::Running);
|
||||
|
||||
// Simulate what the Err(_) branch of the catch_unwind match does.
|
||||
state::update_progress(id, |s| {
|
||||
s.status = VaultSyncStatus::Failed;
|
||||
s.errors = vec!["sync task panicked unexpectedly".to_string()];
|
||||
});
|
||||
|
||||
let st = state::get(id).unwrap();
|
||||
assert_eq!(st.status, VaultSyncStatus::Failed);
|
||||
assert!(st.errors[0].contains("panicked"));
|
||||
|
||||
// A subsequent sync attempt must not be blocked by the old Running entry.
|
||||
state::start(id, 2_000).unwrap();
|
||||
assert_eq!(state::get(id).unwrap().status, VaultSyncStatus::Running);
|
||||
}
|
||||
@@ -1,138 +0,0 @@
|
||||
use chrono::{DateTime, Utc};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// Status of a running or completed vault sync operation.
|
||||
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
pub enum VaultSyncStatus {
|
||||
#[default]
|
||||
Idle,
|
||||
Running,
|
||||
Completed,
|
||||
Failed,
|
||||
}
|
||||
|
||||
/// Live progress for a vault sync operation.
|
||||
///
|
||||
/// Held in the global registry while a sync is in flight, and retained after
|
||||
/// completion so the frontend can poll the final outcome without timing
|
||||
/// concerns. The next `vault.sync` call for the same vault overwrites this.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct VaultSyncState {
|
||||
pub vault_id: String,
|
||||
pub status: VaultSyncStatus,
|
||||
/// Files seen by the directory walker (updated after discovery phase).
|
||||
pub scanned: u64,
|
||||
/// Files successfully ingested so far.
|
||||
pub ingested: u64,
|
||||
/// Files skipped because hash+mtime was unchanged.
|
||||
pub unchanged: u64,
|
||||
/// Files removed from the vault (source file gone).
|
||||
pub removed: u64,
|
||||
/// Files that failed ingestion.
|
||||
pub failed: u64,
|
||||
/// Files skipped (unsupported extension or too large).
|
||||
pub skipped_unsupported: u64,
|
||||
/// Total files queued for ingestion (set after discovery; 0 while walking).
|
||||
pub total: u64,
|
||||
/// Unix milliseconds when this sync started.
|
||||
pub started_at_ms: i64,
|
||||
/// Unix milliseconds when this sync finished; `None` while still running.
|
||||
pub finished_at_ms: Option<i64>,
|
||||
/// Wall-clock duration in ms; 0 while running; set from VaultSyncReport on completion.
|
||||
pub duration_ms: i64,
|
||||
/// Accumulated error strings.
|
||||
pub errors: Vec<String>,
|
||||
}
|
||||
|
||||
/// A user-registered local folder whose files are mirrored into memory.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
pub struct Vault {
|
||||
pub id: String,
|
||||
pub name: String,
|
||||
pub root_path: String,
|
||||
#[serde(default)]
|
||||
pub host_os: Option<String>,
|
||||
pub namespace: String,
|
||||
pub include_globs: Vec<String>,
|
||||
pub exclude_globs: Vec<String>,
|
||||
pub created_at: DateTime<Utc>,
|
||||
pub last_synced_at: Option<DateTime<Utc>>,
|
||||
pub file_count: u64,
|
||||
#[serde(default)]
|
||||
pub write_state: VaultWriteState,
|
||||
#[serde(default)]
|
||||
pub write_state_reason: Option<String>,
|
||||
}
|
||||
|
||||
/// Whether OpenHuman can write approved markdown artifacts back into a vault.
|
||||
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum VaultWriteState {
|
||||
Writable,
|
||||
ReadOnly,
|
||||
#[default]
|
||||
Unavailable,
|
||||
}
|
||||
|
||||
/// Per-file ledger entry used for dedup on re-sync.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
pub struct VaultFile {
|
||||
pub vault_id: String,
|
||||
pub rel_path: String,
|
||||
pub document_id: String,
|
||||
pub content_hash: String,
|
||||
pub mtime_ms: i64,
|
||||
pub bytes: u64,
|
||||
pub ingested_at: DateTime<Utc>,
|
||||
pub status: VaultFileStatus,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
pub enum VaultFileStatus {
|
||||
Ok,
|
||||
Skipped,
|
||||
Failed,
|
||||
}
|
||||
|
||||
impl VaultFileStatus {
|
||||
pub(crate) fn as_str(&self) -> &'static str {
|
||||
match self {
|
||||
Self::Ok => "ok",
|
||||
Self::Skipped => "skipped",
|
||||
Self::Failed => "failed",
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn parse(raw: &str) -> Self {
|
||||
match raw {
|
||||
"skipped" => Self::Skipped,
|
||||
"failed" => Self::Failed,
|
||||
_ => Self::Ok,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Summary returned from `vault.sync`.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
|
||||
pub struct VaultSyncReport {
|
||||
pub vault_id: String,
|
||||
pub scanned: u64,
|
||||
pub ingested: u64,
|
||||
pub unchanged: u64,
|
||||
pub removed: u64,
|
||||
pub failed: u64,
|
||||
pub skipped_unsupported: u64,
|
||||
pub duration_ms: i64,
|
||||
pub errors: Vec<String>,
|
||||
}
|
||||
|
||||
/// Result returned after writing an approved markdown/wiki artifact into a vault.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
pub struct VaultWriteMarkdownReport {
|
||||
pub vault_id: String,
|
||||
pub rel_path: String,
|
||||
pub bytes_written: u64,
|
||||
pub created: bool,
|
||||
}
|
||||
Reference in New Issue
Block a user