From e7ea429f175a7fcae8b037c11f8c5459360574b2 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 09:34:22 +0300 Subject: [PATCH 01/14] feat(storage): add scoped storage helpers and sync bridge Add current_scope and scope_from to resolve the acting agent's storage scope, refusing a shared scope in SaaS mode when no agent is present, and current_scoped to bind the installed backend to it. Add block_on, a shared-runtime bridge letting synchronous domain stores call the async storage ports without block_in_place, which panics on current-thread runtimes. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/openhuman-core/Cargo.toml | 2 +- crates/openhuman-core/src/storage/mod.rs | 73 ++++++++++++++++++- .../openhuman-core/src/storage/mod_tests.rs | 25 +++++++ 3 files changed, 97 insertions(+), 3 deletions(-) diff --git a/crates/openhuman-core/Cargo.toml b/crates/openhuman-core/Cargo.toml index daedb2d542a..2cf4e2bc3f6 100644 --- a/crates/openhuman-core/Cargo.toml +++ b/crates/openhuman-core/Cargo.toml @@ -152,7 +152,7 @@ tinyagents-registry = { path = "../../vendor/tinyagents/crates/tinyagents-regist tinyagents-session = { path = "../../vendor/tinyagents/crates/tinyagents-session", features = ["storage-drivers"] } # The storage ports and the URL-to-backend facade (`storage` domain). No driver # by default: the `storage-*` features below add them. -tinystoragedrivers = { path = "../../vendor/tinyagents/vendor/tinystoragedrivers/crates/tinystoragedrivers", default-features = false } +tinystoragedrivers = { path = "../../vendor/tinyagents/vendor/tinystoragedrivers/crates/tinystoragedrivers", default-features = false, features = ["blocking"] } tinyagents-runtime = { path = "../../vendor/tinyagents/crates/tinyagents-runtime" } tinyagents-orchestration = { path = "../../vendor/tinyagents/crates/tinyagents-orchestration" } tinyinference-core = { path = "../../vendor/tinyagents/vendor/tinyinference/crates/tinyinference-core" } diff --git a/crates/openhuman-core/src/storage/mod.rs b/crates/openhuman-core/src/storage/mod.rs index a603f9a17c1..891f5b92def 100644 --- a/crates/openhuman-core/src/storage/mod.rs +++ b/crates/openhuman-core/src/storage/mod.rs @@ -18,10 +18,12 @@ //! driver the build does not carry fails at [`open`] naming the feature, so a //! misconfigured deployment stops at boot instead of at its first write. -use std::sync::{Arc, LazyLock, RwLock}; +use std::future::Future; +use std::sync::{Arc, LazyLock, OnceLock, RwLock}; pub use tinystoragedrivers::{ - Scope, ScopedStorage, StorageBackend, StorageConfig as StorageUrl, StorageError, + Blocking, DocumentStore, DocumentStoreExt, Scope, ScopedStorage, StorageBackend, + StorageConfig as StorageUrl, StorageError, }; use crate::config::schema::storage::redact_url; @@ -91,6 +93,73 @@ pub fn clear() -> bool { .is_some() } +/// The storage scope of the current call: the acting agent's +/// ([`scope_for_agent`]) when the dispatch carries one +/// (`CoreContext::session_agent`), else [`Scope::local`] — except in SaaS +/// mode, where a call with no acting agent is refused rather than given a +/// bucket every user would share. +/// +/// # Errors +/// +/// In SaaS mode, when the current context names no agent. +pub fn current_scope() -> Result { + let agent = crate::core::runtime::CoreContext::current() + .and_then(|context| context.session_agent().map(str::to_string)); + scope_from(agent.as_deref(), crate::core::runtime::mode::is_saas()) +} + +/// [`current_scope`] with its two inputs made explicit, so the rule is +/// testable without a booted context or a locked mode. +/// +/// # Errors +/// +/// When `saas` and there is no `agent`. +pub fn scope_from(agent: Option<&str>, saas: bool) -> Result { + match agent { + Some(agent) => Ok(scope_for_agent(agent)), + None if saas => Err(StorageError::invalid_input( + "no acting agent in SaaS mode; refusing a shared storage scope", + )), + None => Ok(Scope::local()), + } +} + +/// The installed backend bound to [`current_scope`], or `None` when the host +/// configured no backend (the classic on-disk layout). +/// +/// # Errors +/// +/// When the scope cannot be resolved ([`current_scope`]) or the backend +/// refuses it. +pub fn current_scoped() -> Result, StorageError> { + installed() + .map(|backend| backend.for_scope(¤t_scope()?)) + .transpose() +} + +/// Runs `future` to completion from synchronous code, on one dedicated +/// runtime thread shared by every caller in the process. +/// +/// For domain stores whose API is synchronous (most of the core's), so they +/// can call the async storage ports without `block_in_place` — which would +/// panic on a current-thread runtime. +/// +/// # Errors +/// +/// When the bridge cannot start, or `future` called back into it. +pub fn block_on(future: F) -> Result +where + F: Future> + Send + 'static, + T: Send + 'static, +{ + static BRIDGE: OnceLock> = OnceLock::new(); + let bridge = BRIDGE + .get_or_init(|| Blocking::new().map_err(|error| error.to_string())) + .as_ref() + .map_err(|error| StorageError::backend(error.clone()))?; + bridge.run(future)? +} + /// The storage scope agent `agent_id`'s records live under — the same /// mapping the session store uses, so every domain agrees on it. pub fn scope_for_agent(agent_id: &str) -> Scope { diff --git a/crates/openhuman-core/src/storage/mod_tests.rs b/crates/openhuman-core/src/storage/mod_tests.rs index a9535025e40..764b2ec2b01 100644 --- a/crates/openhuman-core/src/storage/mod_tests.rs +++ b/crates/openhuman-core/src/storage/mod_tests.rs @@ -57,3 +57,28 @@ fn agent_scopes_match_the_session_store() { .as_str() .starts_with("sha256:")); } + +#[test] +fn the_scope_follows_the_acting_agent_and_fails_closed_in_saas() { + assert_eq!(scope_from(Some("agent-7"), false).unwrap().as_str(), "agent-7"); + assert_eq!(scope_from(Some("agent-7"), true).unwrap().as_str(), "agent-7"); + assert!(scope_from(None, false).unwrap().is_local()); + assert!(scope_from(None, true).is_err(), "SaaS never falls back to a shared scope"); +} + +#[test] +fn block_on_runs_a_future_from_sync_code() { + let answer = block_on(async { Ok::<_, StorageError>(42) }).unwrap(); + assert_eq!(answer, 42); + let error = block_on(async { Err::<(), _>(StorageError::conflict("raced")) }).unwrap_err(); + assert_eq!(error.message(), "raced"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn block_on_works_inside_a_runtime_too() { + let answer = tokio::task::spawn_blocking(|| block_on(async { Ok::<_, StorageError>(7) })) + .await + .unwrap() + .unwrap(); + assert_eq!(answer, 7); +} From 59951282109cddd535be58fe942a3fab68a850a1 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 09:35:16 +0300 Subject: [PATCH 02/14] feat(security): add document approval store Add a store for document approval records so pending and resolved approvals can be persisted and queried. This provides the backing storage needed for the document approval workflow. Auto-committed-on: dragonfly Co-authored-by: Medulla --- .../src/security/approval/store_documents.rs | 468 ++++++++++++++++++ 1 file changed, 468 insertions(+) create mode 100644 crates/openhuman-core/src/security/approval/store_documents.rs diff --git a/crates/openhuman-core/src/security/approval/store_documents.rs b/crates/openhuman-core/src/security/approval/store_documents.rs new file mode 100644 index 00000000000..097396ce142 --- /dev/null +++ b/crates/openhuman-core/src/security/approval/store_documents.rs @@ -0,0 +1,468 @@ +//! The approval store on the `tinystoragedrivers` document port. +//! +//! Used instead of `approval/approval.db` when the host configured a storage +//! backend ([`crate::storage`]); the public functions in `store.rs` pick it +//! per call. Records live under the current call's storage scope (the acting +//! agent, `local` on a single-user host), so in a multi-tenant process one +//! user's approvals are invisible to another's. +//! +//! # Layout +//! +//! | Collection | Document id | Holds | +//! | --- | --- | --- | +//! | `approvals` | `request_id` | one approval request and its audit trail | +//! | `approval_flow_trust` | `(flow_id, tool_name)` | one "approve always for this flow" grant | +//! +//! The SQL guards become preconditions: a request is inserted only if absent, +//! and deciding, expiring and recording execution are compare-and-swap on the +//! document's version, each applied only while the row is still in the state +//! the SQL `WHERE` clause required (`decided_at IS NULL`, and so on). Two +//! processes sharing one database therefore decide a request at most once. +//! `args_redacted` and `source_context` are stored as JSON strings, so a +//! MongoDB-backed port never meets an arbitrary key. + +use std::sync::Arc; + +use anyhow::{Context, Result, anyhow}; +use chrono::{DateTime, Utc}; +use serde_json::{Value, json}; + +use crate::storage::{ + DocumentStore, DocumentStoreExt, ScopedStorage, StorageError, block_on, current_scoped, +}; +use tinystoragedrivers::{ + CollectionSpec, ErrorKind, Filter, IndexSpec, Precondition, Query, Sort, Versioned, +}; + +use super::types::{ + ApprovalAuditEntry, ApprovalDecision, ApprovalSourceContext, ExecutionOutcome, PendingApproval, +}; + +const APPROVALS: &str = "approvals"; +const FLOW_TRUST: &str = "approval_flow_trust"; + +/// Compare-and-swap attempts before a contended update gives up. +const CAS_ATTEMPTS: usize = 32; + +/// The approval store over one scoped document handle. +#[derive(Clone)] +pub(super) struct Docs { + docs: Arc, +} + +/// The document store to use for this call, when the host configured one. +/// +/// # Errors +/// +/// When the storage scope cannot be resolved — in SaaS mode with no acting +/// agent — so the call fails rather than reading a shared bucket. +pub(super) fn current() -> Result> { + Ok(current_scoped() + .context("[approval::store] resolve the storage scope")? + .map(|scoped| Docs::new(&scoped))) +} + +fn storage(error: StorageError) -> anyhow::Error { + anyhow!("[approval::store] storage: {error}") +} + +fn rfc3339(at: DateTime) -> String { + at.to_rfc3339() +} + +fn parse_time(raw: Option<&str>) -> Option> { + raw.and_then(|value| DateTime::parse_from_rfc3339(value).ok()) + .map(|at| at.with_timezone(&Utc)) +} + +fn text<'a>(doc: &'a Value, field: &str) -> Option<&'a str> { + doc.get(field).and_then(Value::as_str) +} + +/// A stored request as a [`PendingApproval`]. +fn to_pending(stored: &Versioned) -> PendingApproval { + let doc = &stored.doc; + PendingApproval { + request_id: stored.id.clone(), + tool_name: text(doc, "tool_name").unwrap_or_default().to_string(), + action_summary: text(doc, "action_summary").unwrap_or_default().to_string(), + args_redacted: text(doc, "args_redacted") + .and_then(|raw| serde_json::from_str(raw).ok()) + .unwrap_or(Value::Null), + created_at: parse_time(text(doc, "created_at")).unwrap_or_else(Utc::now), + expires_at: parse_time(text(doc, "expires_at")), + source_context: text(doc, "source_context").and_then(|raw| { + serde_json::from_str::(raw) + .map_err(|error| { + tracing::warn!( + %error, + "[approval::store] failed to decode source_context JSON — treating as absent" + ); + }) + .ok() + }), + tool_call_id: text(doc, "tool_call_id").map(str::to_string), + } +} + +/// A decided request as an [`ApprovalAuditEntry`]; `None` if it is not +/// decided or its decision is unknown. +fn to_audit(stored: &Versioned) -> Option { + let pending = to_pending(stored); + let decided_at = parse_time(text(&stored.doc, "decided_at"))?; + let decision = ApprovalDecision::from_str(text(&stored.doc, "decision")?)?; + Some(ApprovalAuditEntry { + request_id: pending.request_id, + tool_name: pending.tool_name, + action_summary: pending.action_summary, + args_redacted: pending.args_redacted, + created_at: pending.created_at, + expires_at: pending.expires_at, + decided_at, + decision, + }) +} + +/// `error` sanitized for the durable audit row, then capped at 512 chars — +/// sanitize first, so truncation can never split a secret past the redactor. +fn audit_error(error: Option<&str>) -> Option { + error.map(|raw| { + let sanitized = crate::security::scrub::sanitize_text(raw).value; + if sanitized.chars().count() > 512 { + let head: String = sanitized.chars().take(511).collect(); + format!("{head}…") + } else { + sanitized + } + }) +} + +impl Docs { + pub(super) fn new(scoped: &ScopedStorage) -> Self { + Self { + docs: Arc::clone(scoped.documents()), + } + } + + /// Runs `op` against the store from synchronous code. + fn run(&self, op: F) -> Result + where + F: FnOnce(Arc) -> Fut, + Fut: std::future::Future> + Send + 'static, + T: Send + 'static, + { + let docs = Arc::clone(&self.docs); + let future = op(Arc::clone(&docs)); + block_on(async move { + declare(&docs).await?; + future.await + }) + .map_err(storage) + } + + pub(super) fn insert_pending(&self, pending: &PendingApproval, session_id: &str) -> Result<()> { + let doc = json!({ + "tool_name": pending.tool_name, + "action_summary": pending.action_summary, + "args_redacted": serde_json::to_string(&pending.args_redacted) + .context("[approval::store] serialize args_redacted")?, + "session_id": session_id, + "created_at": rfc3339(pending.created_at), + "expires_at": pending.expires_at.map(rfc3339), + "expires_ts": pending.expires_at.map(|at| at.timestamp()), + "source_context": pending + .source_context + .as_ref() + .map(serde_json::to_string) + .transpose() + .context("[approval::store] serialize source_context")?, + "tool_call_id": pending.tool_call_id, + "pending": true, + }); + let id = pending.request_id.clone(); + self.run(|docs| async move { + docs.put(APPROVALS, &id, doc, Precondition::Absent) + .await + .map(|_| ()) + }) + } + + /// Inserts a request that is decided from birth (a flow + /// pre-authorization): it never appears pending, only in the audit + /// trail. + pub(super) fn insert_decided( + &self, + tool_name: &str, + action_summary: &str, + session_id: &str, + source_context: &ApprovalSourceContext, + decision: ApprovalDecision, + ) -> Result<()> { + let now = rfc3339(Utc::now()); + let doc = json!({ + "tool_name": tool_name, + "action_summary": action_summary, + "args_redacted": "{}", + "session_id": session_id, + "created_at": now, + "source_context": serde_json::to_string(source_context) + .context("[approval::store] serialize preauthorization source_context")?, + "pending": false, + "decided_at": now, + "decision": decision.as_str(), + }); + let id = uuid::Uuid::new_v4().to_string(); + self.run(|docs| async move { + docs.put(APPROVALS, &id, doc, Precondition::Absent) + .await + .map(|_| ()) + }) + } + + /// Moves every undecided request past its expiry to a terminal `deny` + /// and returns the ones this call moved, so the caller can announce + /// them. + pub(super) fn expire_stale(&self, now: DateTime) -> Result> { + let deny = ApprovalDecision::Deny.as_str(); + let decided_at = rfc3339(now); + let now_ts = now.timestamp(); + self.run(|docs| async move { + let query = Query::filter( + Filter::eq("pending", true).and(Filter::lte("expires_ts", now_ts)), + ); + let mut expired = Vec::new(); + for stale in docs.query_all(APPROVALS, &query).await? { + let moved = update(&docs, &stale.id, |doc| { + if doc.get("pending") != Some(&json!(true)) { + return None; + } + let mut next = doc.clone(); + next["pending"] = json!(false); + next["decided_at"] = json!(decided_at); + next["decision"] = json!(deny); + Some(next) + }) + .await?; + if moved.is_some() { + expired.push(to_pending(&stale)); + } + } + Ok(expired) + }) + } + + pub(super) fn list_pending(&self) -> Result> { + self.run(|docs| async move { + let query = Query::filter(Filter::eq("pending", true)).sort(Sort::asc("created_at")); + Ok(docs + .query_all(APPROVALS, &query) + .await? + .iter() + .map(to_pending) + .collect()) + }) + } + + pub(super) fn get_decision(&self, request_id: &str) -> Result> { + let id = request_id.to_string(); + self.run(|docs| async move { + Ok(docs.get(APPROVALS, &id).await?.and_then(|stored| { + text(&stored.doc, "decided_at")?; + ApprovalDecision::from_str(text(&stored.doc, "decision")?) + })) + }) + } + + pub(super) fn decide( + &self, + request_id: &str, + decision: ApprovalDecision, + ) -> Result> { + let id = request_id.to_string(); + let decided_at = rfc3339(Utc::now()); + self.run(|docs| async move { + let decided = update(&docs, &id, |doc| { + if doc.get("pending") != Some(&json!(true)) { + return None; + } + let mut next = doc.clone(); + next["pending"] = json!(false); + next["decided_at"] = json!(decided_at); + next["decision"] = json!(decision.as_str()); + Some(next) + }) + .await?; + Ok(decided.as_ref().map(to_pending)) + }) + } + + pub(super) fn record_execution( + &self, + request_id: &str, + outcome: ExecutionOutcome, + error: Option<&str>, + ) -> Result { + let id = request_id.to_string(); + let executed_at = rfc3339(Utc::now()); + let error = audit_error(error); + self.run(|docs| async move { + let recorded = update(&docs, &id, |doc| { + // Decided first, and the first outcome wins. + if text(doc, "decided_at").is_none() || text(doc, "executed_at").is_some() { + return None; + } + let mut next = doc.clone(); + next["executed_at"] = json!(executed_at); + next["execution_outcome"] = json!(outcome.as_str()); + next["execution_error"] = json!(error); + Some(next) + }) + .await?; + Ok(recorded.is_some()) + }) + } + + pub(super) fn list_recent_decisions(&self, limit: usize) -> Result> { + self.run(|docs| async move { + let query = Query::filter(Filter::eq("pending", false).and(Filter::exists("decision", true))) + .sort(Sort::desc("decided_at")) + .limit(limit); + let page = docs.query(APPROVALS, &query).await?; + Ok(page.items.iter().filter_map(to_audit).collect()) + }) + } + + pub(super) fn purge_session(&self, session_id: &str) -> Result { + let session = session_id.to_string(); + self.run(|docs| async move { + let removed = docs + .delete_where( + APPROVALS, + &Filter::eq("session_id", session).and(Filter::eq("pending", true)), + ) + .await?; + Ok(usize::try_from(removed).unwrap_or(usize::MAX)) + }) + } + + pub(super) fn insert_flow_trust(&self, flow_id: &str, tool_name: &str) -> Result<()> { + let id = trust_id(flow_id, tool_name); + let doc = json!({ + "flow_id": flow_id, + "tool_name": tool_name, + "created_at": rfc3339(Utc::now()), + }); + self.run(|docs| async move { + // `INSERT OR IGNORE`: re-granting is a no-op. + match docs.put(FLOW_TRUST, &id, doc, Precondition::Absent).await { + Ok(_) => Ok(()), + Err(error) if error.kind() == ErrorKind::Conflict => Ok(()), + Err(error) => Err(error), + } + }) + } + + pub(super) fn list_flow_trust(&self, flow_id: &str) -> Result> { + let flow = flow_id.to_string(); + self.run(|docs| async move { + let query = Query::filter(Filter::eq("flow_id", flow)).sort(Sort::asc("tool_name")); + Ok(docs + .query_all(FLOW_TRUST, &query) + .await? + .iter() + .filter_map(|grant| text(&grant.doc, "tool_name").map(str::to_string)) + .collect()) + }) + } + + pub(super) fn delete_flow_trust( + &self, + flow_id: &str, + tool_names: Option<&[String]>, + ) -> Result { + let flow = flow_id.to_string(); + let names = tool_names.map(<[String]>::to_vec); + self.run(|docs| async move { + let removed = match names { + None => docs.delete_where(FLOW_TRUST, &Filter::eq("flow_id", flow)).await?, + Some(names) => { + let mut removed = 0; + for name in names { + if docs + .delete(FLOW_TRUST, &trust_id(&flow, &name), Precondition::None) + .await? + { + removed += 1; + } + } + removed + } + }; + Ok(usize::try_from(removed).unwrap_or(usize::MAX)) + }) + } + + pub(super) fn is_flow_tool_trusted(&self, flow_id: &str, tool_name: &str) -> Result { + let id = trust_id(flow_id, tool_name); + self.run(|docs| async move { Ok(docs.get(FLOW_TRUST, &id).await?.is_some()) }) + } +} + +/// The id of a `(flow_id, tool_name)` grant: length-prefixed, so no two +/// pairs collide. +fn trust_id(flow_id: &str, tool_name: &str) -> String { + format!("{}:{flow_id}/{tool_name}", flow_id.len()) +} + +async fn declare(docs: &Arc) -> Result<(), StorageError> { + docs.ensure_collection( + &CollectionSpec::new(APPROVALS) + .index(IndexSpec::new("by_pending", ["pending", "created_at"])) + .index(IndexSpec::new("by_decided", ["pending", "decided_at"])) + .index(IndexSpec::new("by_session", ["session_id", "pending"])), + ) + .await?; + docs.ensure_collection( + &CollectionSpec::new(FLOW_TRUST).index(IndexSpec::new("by_flow", ["flow_id", "tool_name"])), + ) + .await +} + +/// Applies `change` to the request `id` under compare-and-swap and returns +/// the stored result, or `None` when the request is missing or `change` +/// declines (the request is no longer in the state the change requires). +async fn update( + docs: &Arc, + id: &str, + change: impl Fn(&Value) -> Option, +) -> Result>, StorageError> { + for _ in 0..CAS_ATTEMPTS { + let Some(stored) = docs.get(APPROVALS, id).await? else { + return Ok(None); + }; + let Some(next) = change(&stored.doc) else { + return Ok(None); + }; + match docs + .put(APPROVALS, id, next.clone(), stored.unchanged()) + .await + { + Ok(version) => { + return Ok(Some(Versioned { + id: id.to_string(), + version, + doc: next, + })); + } + Err(error) if error.kind() == ErrorKind::Conflict => {} + Err(error) => return Err(error), + } + } + Err(StorageError::conflict(format!( + "approval {id} kept changing under {CAS_ATTEMPTS} attempts" + ))) +} + +#[cfg(test)] +#[path = "store_documents_tests.rs"] +mod tests; From dac5539ef9b5e79028e475f93a18b0e295c76452 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 09:37:50 +0300 Subject: [PATCH 03/14] feat(approval): route approval store through document backend When a storage backend is configured, approval operations now go through the tinystoragedrivers document port under the current storage scope instead of the local SQLite database, falling back to approval.db when no backend is set. Expiry sweeps also publish stale-card events from the store since no live gate is in scope during a sweep. Auto-committed-on: dragonfly Co-authored-by: Medulla --- .../src/security/approval/mod.rs | 1 + .../src/security/approval/store.rs | 48 +++++++++++++++++-- .../approval/store_documents_tests.rs | 0 .../src/security/approval/store_flow_trust.rs | 30 +++++++++++- 4 files changed, 75 insertions(+), 4 deletions(-) create mode 100644 crates/openhuman-core/src/security/approval/store_documents_tests.rs diff --git a/crates/openhuman-core/src/security/approval/mod.rs b/crates/openhuman-core/src/security/approval/mod.rs index 142017a42ca..4acf392c898 100644 --- a/crates/openhuman-core/src/security/approval/mod.rs +++ b/crates/openhuman-core/src/security/approval/mod.rs @@ -18,6 +18,7 @@ pub mod redact; pub mod rpc; pub mod schemas; pub mod store; +mod store_documents; pub mod types; pub use gate::{ diff --git a/crates/openhuman-core/src/security/approval/store.rs b/crates/openhuman-core/src/security/approval/store.rs index 2e3cf4054d1..a8d5742ef8e 100644 --- a/crates/openhuman-core/src/security/approval/store.rs +++ b/crates/openhuman-core/src/security/approval/store.rs @@ -22,6 +22,13 @@ //! Follows the same `with_connection` shape as `notifications/store.rs` //! and `cron/store.rs`: synchronous `rusqlite::Connection` opened per //! call, schema applied idempotently. +//! +//! When the host configured a storage backend ([`crate::storage`]), every +//! function here uses `store_documents` instead: the same operations on the +//! `tinystoragedrivers` document port, under the current call's storage +//! scope (the acting agent; `local` on a single-user host; refused in SaaS +//! mode without one). With no backend — the desktop default — nothing +//! changes and `approval/approval.db` is used as before. use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; @@ -224,6 +231,9 @@ fn with_connection(config: &Config, f: impl FnOnce(&Connection) -> Result) /// internal correlation only and is never re-exposed on /// [`PendingApproval`] (see that type's doc-comment). pub fn insert_pending(config: &Config, pending: &PendingApproval, session_id: &str) -> Result<()> { + if let Some(docs) = super::store_documents::current()? { + return docs.insert_pending(pending, session_id); + } with_connection(config, |conn| { let args = serde_json::to_string(&pending.args_redacted) .context("[approval::store] serialize args_redacted")?; @@ -266,6 +276,11 @@ pub fn insert_pending(config: &Config, pending: &PendingApproval, session_id: &s /// (`decided_at` + `decision`) without leaving expired rows pending /// forever. pub fn expire_stale(config: &Config) -> Result { + if let Some(docs) = super::store_documents::current()? { + let expired = docs.expire_stale(Utc::now())?; + publish_expired(&expired); + return Ok(expired.len()); + } with_connection(config, |conn| { Ok(expire_stale_with_now(conn, Utc::now())?.len()) }) @@ -275,6 +290,10 @@ pub fn expire_stale(config: &Config) -> Result { /// which launch queued them. Orphan rows from prior sessions remain /// visible until they are explicitly decided or expire. pub fn list_pending(config: &Config) -> Result> { + if let Some(docs) = super::store_documents::current()? { + publish_expired(&docs.expire_stale(Utc::now())?); + return docs.list_pending(); + } with_connection(config, |conn| { expire_stale_with_now(conn, Utc::now())?; @@ -304,6 +323,9 @@ pub fn list_pending(config: &Config) -> Result> { /// where the TTL elapses concurrently with a committed approval /// (CodeRabbit review on PR #2367). pub fn get_decision(config: &Config, request_id: &str) -> Result> { + if let Some(docs) = super::store_documents::current()? { + return docs.get_decision(request_id); + } with_connection(config, |conn| { let mut stmt = conn .prepare( @@ -333,6 +355,10 @@ pub fn decide( request_id: &str, decision: ApprovalDecision, ) -> Result> { + if let Some(docs) = super::store_documents::current()? { + publish_expired(&docs.expire_stale(Utc::now())?); + return docs.decide(request_id, decision); + } with_connection(config, |conn| { expire_stale_with_now(conn, Utc::now())?; @@ -389,6 +415,9 @@ pub fn record_execution( outcome: ExecutionOutcome, error: Option<&str>, ) -> Result { + if let Some(docs) = super::store_documents::current()? { + return docs.record_execution(request_id, outcome, error); + } with_connection(config, |conn| { let now = Utc::now().to_rfc3339(); // Sanitize before truncation so the durable audit row can't @@ -434,6 +463,9 @@ pub fn record_execution( /// List recently decided approval rows for durable audit views. pub fn list_recent_decisions(config: &Config, limit: usize) -> Result> { let limit = limit.clamp(1, 500); + if let Some(docs) = super::store_documents::current()? { + return docs.list_recent_decisions(limit); + } with_connection(config, |conn| { let mut stmt = conn .prepare( @@ -459,6 +491,9 @@ pub fn list_recent_decisions(config: &Config, limit: usize) -> Result Result { + if let Some(docs) = super::store_documents::current()? { + return docs.purge_session(session_id); + } with_connection(config, |conn| { let removed = conn .execute( @@ -546,18 +581,25 @@ fn expire_stale_with_now(conn: &Connection, now: DateTime) -> Result) -> rusqlite::Result { diff --git a/crates/openhuman-core/src/security/approval/store_documents_tests.rs b/crates/openhuman-core/src/security/approval/store_documents_tests.rs new file mode 100644 index 00000000000..e69de29bb2d diff --git a/crates/openhuman-core/src/security/approval/store_flow_trust.rs b/crates/openhuman-core/src/security/approval/store_flow_trust.rs index 60cfbfefb5e..02f5edf1b3a 100644 --- a/crates/openhuman-core/src/security/approval/store_flow_trust.rs +++ b/crates/openhuman-core/src/security/approval/store_flow_trust.rs @@ -15,6 +15,9 @@ use crate::config::Config; use super::super::types::{ApprovalDecision, ApprovalSourceContext}; use super::with_connection; +/// The action summary of a flow pre-authorization audit row. +const PREAUTHORIZATION_SUMMARY: &str = "Pre-authorized for this flow when it was saved and enabled"; + /// Record a save-time flow pre-authorization in the durable audit trail as a /// born-decided row (`decided_at = created_at`, decision /// `approve_always_for_flow`): it never appears in `list_pending` (which @@ -29,6 +32,19 @@ pub fn record_flow_preauthorization( tool_name: &str, session_id: &str, ) -> Result<()> { + if let Some(docs) = super::super::store_documents::current()? { + return docs.insert_decided( + tool_name, + PREAUTHORIZATION_SUMMARY, + session_id, + &ApprovalSourceContext::Flow { + flow_id: flow_id.to_string(), + run_id: String::new(), + node_id: None, + }, + ApprovalDecision::ApproveAlwaysForFlow, + ); + } with_connection(config, |conn| { let now = Utc::now().to_rfc3339(); let source_context = serde_json::to_string(&ApprovalSourceContext::Flow { @@ -46,7 +62,7 @@ pub fn record_flow_preauthorization( params![ uuid::Uuid::new_v4().to_string(), tool_name, - "Pre-authorized for this flow when it was saved and enabled", + PREAUTHORIZATION_SUMMARY, "{}", session_id, now, @@ -64,6 +80,9 @@ pub fn record_flow_preauthorization( /// flow-origin park. `INSERT OR IGNORE` makes re-granting an already-trusted /// pair a harmless no-op rather than a primary-key error. pub fn insert_flow_trust(config: &Config, flow_id: &str, tool_name: &str) -> Result<()> { + if let Some(docs) = super::super::store_documents::current()? { + return docs.insert_flow_trust(flow_id, tool_name); + } with_connection(config, |conn| { conn.execute( "INSERT OR IGNORE INTO flow_tool_trust (flow_id, tool_name, created_at) @@ -80,6 +99,9 @@ pub fn insert_flow_trust(config: &Config, flow_id: &str, tool_name: &str) -> Res /// save-time pre-authorization manifest (`flows_approval_manifest`) to diff /// "what the graph needs" against "what is already granted". pub fn list_flow_trust(config: &Config, flow_id: &str) -> Result> { + if let Some(docs) = super::super::store_documents::current()? { + return docs.list_flow_trust(flow_id); + } with_connection(config, |conn| { let mut stmt = conn .prepare( @@ -105,6 +127,9 @@ pub fn delete_flow_trust( flow_id: &str, tool_names: Option<&[String]>, ) -> Result { + if let Some(docs) = super::super::store_documents::current()? { + return docs.delete_flow_trust(flow_id, tool_names); + } with_connection(config, |conn| { let removed = match tool_names { None => conn @@ -135,6 +160,9 @@ pub fn delete_flow_trust( /// this flow" trust. Consulted by [`super::gate::ApprovalGate::intercept_audited`] /// before parking a `Workflow`-origin tool call. pub fn is_flow_tool_trusted(config: &Config, flow_id: &str, tool_name: &str) -> Result { + if let Some(docs) = super::super::store_documents::current()? { + return docs.is_flow_tool_trusted(flow_id, tool_name); + } with_connection(config, |conn| { let exists: bool = conn .query_row( From bddc1a386463e4811c9897f79d7f5731251e2b5f Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 09:38:22 +0300 Subject: [PATCH 04/14] test(security): cover approval store document persistence Add tests for the approval store's document handling, exercising the persistence paths that were previously untested. Auto-committed-on: dragonfly Co-authored-by: Medulla --- .../approval/store_documents_tests.rs | 265 ++++++++++++++++++ 1 file changed, 265 insertions(+) diff --git a/crates/openhuman-core/src/security/approval/store_documents_tests.rs b/crates/openhuman-core/src/security/approval/store_documents_tests.rs index e69de29bb2d..d8c36897816 100644 --- a/crates/openhuman-core/src/security/approval/store_documents_tests.rs +++ b/crates/openhuman-core/src/security/approval/store_documents_tests.rs @@ -0,0 +1,265 @@ +use super::*; +use chrono::Duration; +use tinystoragedrivers::{MemoryStorage, Scope, StorageBackend}; + +fn docs_in(storage: &MemoryStorage, scope: &str) -> Docs { + Docs::new(&storage.for_scope(&Scope::new(scope).unwrap()).unwrap()) +} + +fn docs() -> Docs { + docs_in(&MemoryStorage::new(), "local") +} + +fn pending(id: &str, created_secs_ago: i64, expires_in: Option) -> PendingApproval { + let now = Utc::now(); + PendingApproval { + request_id: id.to_string(), + tool_name: "shell".to_string(), + action_summary: format!("run {id}"), + args_redacted: json!({ "file.name": "x", "cmd": "ls" }), + created_at: now - Duration::seconds(created_secs_ago), + expires_at: expires_in.map(|secs| now + Duration::seconds(secs)), + source_context: Some(ApprovalSourceContext::Flow { + flow_id: "flow-1".to_string(), + run_id: "run-1".to_string(), + node_id: None, + }), + tool_call_id: Some(format!("call-{id}")), + } +} + +#[test] +fn pending_requests_round_trip_oldest_first() { + let store = docs(); + store.insert_pending(&pending("b", 10, None), "s1").unwrap(); + store.insert_pending(&pending("a", 20, None), "s1").unwrap(); + let listed = store.list_pending().unwrap(); + let ids: Vec<&str> = listed.iter().map(|p| p.request_id.as_str()).collect(); + assert_eq!(ids, ["a", "b"]); + assert_eq!(listed[0].args_redacted["file.name"], json!("x")); + assert_eq!(listed[0].tool_call_id.as_deref(), Some("call-a")); + assert!(matches!( + listed[0].source_context, + Some(ApprovalSourceContext::Flow { .. }) + )); + assert!( + store.insert_pending(&pending("a", 0, None), "s1").is_err(), + "a request id is inserted once" + ); +} + +#[test] +fn a_request_is_decided_exactly_once() { + let store = docs(); + store.insert_pending(&pending("r", 0, None), "s").unwrap(); + assert_eq!(store.get_decision("r").unwrap(), None); + let decided = store.decide("r", ApprovalDecision::ApproveOnce).unwrap(); + assert_eq!(decided.map(|p| p.request_id), Some("r".to_string())); + assert!(store.decide("r", ApprovalDecision::Deny).unwrap().is_none()); + assert_eq!( + store.get_decision("r").unwrap(), + Some(ApprovalDecision::ApproveOnce) + ); + assert!(store.decide("missing", ApprovalDecision::Deny).unwrap().is_none()); + assert!(store.list_pending().unwrap().is_empty()); +} + +#[test] +fn concurrent_deciders_never_both_win() { + let storage = MemoryStorage::new(); + let store = docs_in(&storage, "local"); + store.insert_pending(&pending("race", 0, None), "s").unwrap(); + let winners: usize = (0..8) + .map(|_| { + let store = store.clone(); + std::thread::spawn(move || { + store + .decide("race", ApprovalDecision::ApproveOnce) + .unwrap() + .is_some() + }) + }) + .collect::>() + .into_iter() + .map(|handle| usize::from(handle.join().unwrap())) + .sum(); + assert_eq!(winners, 1); +} + +#[test] +fn stale_requests_expire_to_deny_once() { + let store = docs(); + store.insert_pending(&pending("old", 0, Some(-5)), "s").unwrap(); + store.insert_pending(&pending("fresh", 0, Some(600)), "s").unwrap(); + store.insert_pending(&pending("forever", 0, None), "s").unwrap(); + let expired = store.expire_stale(Utc::now()).unwrap(); + assert_eq!( + expired.iter().map(|p| p.request_id.as_str()).collect::>(), + ["old"] + ); + assert!(store.expire_stale(Utc::now()).unwrap().is_empty(), "only once"); + assert_eq!( + store.get_decision("old").unwrap(), + Some(ApprovalDecision::Deny) + ); + let left: Vec = store + .list_pending() + .unwrap() + .into_iter() + .map(|p| p.request_id) + .collect(); + assert_eq!(left.len(), 2); + assert!(!left.contains(&"old".to_string())); +} + +#[test] +fn execution_is_recorded_after_a_decision_and_only_once() { + let store = docs(); + store.insert_pending(&pending("r", 0, None), "s").unwrap(); + assert!( + !store + .record_execution("r", ExecutionOutcome::Success, None) + .unwrap(), + "not before a decision" + ); + store.decide("r", ApprovalDecision::ApproveOnce).unwrap(); + let long = format!("token=sk-abcdefghijklmnopqrstuvwxyz {}", "e".repeat(800)); + assert!( + store + .record_execution("r", ExecutionOutcome::Failure, Some(&long)) + .unwrap() + ); + assert!( + !store + .record_execution("r", ExecutionOutcome::Success, None) + .unwrap(), + "the first outcome wins" + ); + assert!( + !store + .record_execution("missing", ExecutionOutcome::Success, None) + .unwrap() + ); + let stored = store + .run(|docs| async move { docs.get(APPROVALS, "r").await }) + .unwrap() + .unwrap(); + assert_eq!(stored.doc["execution_outcome"], json!("failure")); + let error = stored.doc["execution_error"].as_str().unwrap(); + assert!(error.chars().count() <= 512, "capped"); +} + +#[test] +fn recent_decisions_are_newest_first_and_capped() { + let store = docs(); + for id in ["a", "b", "c"] { + store.insert_pending(&pending(id, 0, None), "s").unwrap(); + store.decide(id, ApprovalDecision::ApproveOnce).unwrap(); + std::thread::sleep(std::time::Duration::from_millis(5)); + } + store.insert_pending(&pending("open", 0, None), "s").unwrap(); + let recent = store.list_recent_decisions(2).unwrap(); + assert_eq!( + recent.iter().map(|e| e.request_id.as_str()).collect::>(), + ["c", "b"] + ); + assert_eq!(recent[0].decision, ApprovalDecision::ApproveOnce); +} + +#[test] +fn purging_a_session_drops_only_its_undecided_requests() { + let store = docs(); + store.insert_pending(&pending("mine", 0, None), "s1").unwrap(); + store.insert_pending(&pending("decided", 0, None), "s1").unwrap(); + store.decide("decided", ApprovalDecision::Deny).unwrap(); + store.insert_pending(&pending("theirs", 0, None), "s2").unwrap(); + assert_eq!(store.purge_session("s1").unwrap(), 1); + assert!(store.get_decision("decided").unwrap().is_some()); + let left: Vec = store + .list_pending() + .unwrap() + .into_iter() + .map(|p| p.request_id) + .collect(); + assert_eq!(left, ["theirs"]); +} + +#[test] +fn a_preauthorization_is_audit_only() { + let store = docs(); + store + .insert_decided( + "http", + "pre-authorized", + "s", + &ApprovalSourceContext::Flow { + flow_id: "flow-1".to_string(), + run_id: String::new(), + node_id: None, + }, + ApprovalDecision::ApproveAlwaysForFlow, + ) + .unwrap(); + assert!(store.list_pending().unwrap().is_empty()); + let audit = store.list_recent_decisions(10).unwrap(); + assert_eq!(audit.len(), 1); + assert_eq!(audit[0].decision, ApprovalDecision::ApproveAlwaysForFlow); +} + +#[test] +fn flow_trust_grants_list_and_revoke() { + let store = docs(); + store.insert_flow_trust("flow-1", "shell").unwrap(); + store.insert_flow_trust("flow-1", "shell").unwrap(); + store.insert_flow_trust("flow-1", "http").unwrap(); + store.insert_flow_trust("flow-2", "shell").unwrap(); + assert_eq!(store.list_flow_trust("flow-1").unwrap(), ["http", "shell"]); + assert!(store.is_flow_tool_trusted("flow-1", "shell").unwrap()); + assert!(!store.is_flow_tool_trusted("flow-1", "mail").unwrap()); + assert_eq!( + store + .delete_flow_trust("flow-1", Some(&["shell".to_string(), "nope".to_string()])) + .unwrap(), + 1 + ); + assert_eq!(store.delete_flow_trust("flow-1", None).unwrap(), 1); + assert!(store.list_flow_trust("flow-1").unwrap().is_empty()); + assert!(store.is_flow_tool_trusted("flow-2", "shell").unwrap()); + assert_ne!(trust_id("a/b", "c"), trust_id("a", "b/c")); +} + +#[test] +fn scopes_keep_users_apart() { + let storage = MemoryStorage::new(); + let alice = docs_in(&storage, "alice"); + let bob = docs_in(&storage, "bob"); + alice.insert_pending(&pending("r", 0, None), "s").unwrap(); + alice.insert_flow_trust("flow-1", "shell").unwrap(); + assert!(bob.list_pending().unwrap().is_empty()); + assert!(bob.decide("r", ApprovalDecision::ApproveOnce).unwrap().is_none()); + assert!(!bob.is_flow_tool_trusted("flow-1", "shell").unwrap()); + assert_eq!(alice.list_pending().unwrap().len(), 1); +} + +#[test] +fn an_unreadable_source_context_reads_as_absent() { + let store = docs(); + store.insert_pending(&pending("r", 0, None), "s").unwrap(); + store + .run(|docs| async move { + let mut stored = docs.get(APPROVALS, "r").await?.unwrap(); + stored.doc["source_context"] = json!("not json"); + docs.put(APPROVALS, "r", stored.doc, Precondition::None) + .await + .map(|_| ()) + }) + .unwrap(); + assert!(store.list_pending().unwrap()[0].source_context.is_none()); +} + +#[test] +fn no_backend_means_the_classic_store() { + crate::storage::clear(); + assert!(current().unwrap().is_none()); + assert!(audit_error(None).is_none()); +} From a33a26fc7e1cda1694fb0233d8f2a8a4db9121b1 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 09:44:12 +0300 Subject: [PATCH 05/14] refactor(storage): isolate backend slot for tests Extract the process-wide backend holder into a Slot type so tests can install and clear a private backend without rerouting storage calls for other tests running in the same process. The approval store tests were also adjusted to drop a test that mutated the shared slot and to cover the missing-error path instead. Auto-committed-on: dragonfly Co-authored-by: Medulla --- .../src/security/approval/store_documents.rs | 23 +++--- .../approval/store_documents_tests.rs | 78 ++++++++++++------- crates/openhuman-core/src/storage/mod.rs | 49 ++++++++---- .../openhuman-core/src/storage/mod_tests.rs | 36 ++++++--- 4 files changed, 124 insertions(+), 62 deletions(-) diff --git a/crates/openhuman-core/src/security/approval/store_documents.rs b/crates/openhuman-core/src/security/approval/store_documents.rs index 097396ce142..04233fea621 100644 --- a/crates/openhuman-core/src/security/approval/store_documents.rs +++ b/crates/openhuman-core/src/security/approval/store_documents.rs @@ -23,12 +23,12 @@ use std::sync::Arc; -use anyhow::{Context, Result, anyhow}; +use anyhow::{anyhow, Context, Result}; use chrono::{DateTime, Utc}; -use serde_json::{Value, json}; +use serde_json::{json, Value}; use crate::storage::{ - DocumentStore, DocumentStoreExt, ScopedStorage, StorageError, block_on, current_scoped, + block_on, current_scoped, DocumentStore, DocumentStoreExt, ScopedStorage, StorageError, }; use tinystoragedrivers::{ CollectionSpec, ErrorKind, Filter, IndexSpec, Precondition, Query, Sort, Versioned, @@ -227,9 +227,8 @@ impl Docs { let decided_at = rfc3339(now); let now_ts = now.timestamp(); self.run(|docs| async move { - let query = Query::filter( - Filter::eq("pending", true).and(Filter::lte("expires_ts", now_ts)), - ); + let query = + Query::filter(Filter::eq("pending", true).and(Filter::lte("expires_ts", now_ts))); let mut expired = Vec::new(); for stale in docs.query_all(APPROVALS, &query).await? { let moved = update(&docs, &stale.id, |doc| { @@ -324,9 +323,10 @@ impl Docs { pub(super) fn list_recent_decisions(&self, limit: usize) -> Result> { self.run(|docs| async move { - let query = Query::filter(Filter::eq("pending", false).and(Filter::exists("decision", true))) - .sort(Sort::desc("decided_at")) - .limit(limit); + let query = + Query::filter(Filter::eq("pending", false).and(Filter::exists("decision", true))) + .sort(Sort::desc("decided_at")) + .limit(limit); let page = docs.query(APPROVALS, &query).await?; Ok(page.items.iter().filter_map(to_audit).collect()) }) @@ -384,7 +384,10 @@ impl Docs { let names = tool_names.map(<[String]>::to_vec); self.run(|docs| async move { let removed = match names { - None => docs.delete_where(FLOW_TRUST, &Filter::eq("flow_id", flow)).await?, + None => { + docs.delete_where(FLOW_TRUST, &Filter::eq("flow_id", flow)) + .await? + } Some(names) => { let mut removed = 0; for name in names { diff --git a/crates/openhuman-core/src/security/approval/store_documents_tests.rs b/crates/openhuman-core/src/security/approval/store_documents_tests.rs index d8c36897816..8a17d7308da 100644 --- a/crates/openhuman-core/src/security/approval/store_documents_tests.rs +++ b/crates/openhuman-core/src/security/approval/store_documents_tests.rs @@ -60,7 +60,10 @@ fn a_request_is_decided_exactly_once() { store.get_decision("r").unwrap(), Some(ApprovalDecision::ApproveOnce) ); - assert!(store.decide("missing", ApprovalDecision::Deny).unwrap().is_none()); + assert!(store + .decide("missing", ApprovalDecision::Deny) + .unwrap() + .is_none()); assert!(store.list_pending().unwrap().is_empty()); } @@ -68,7 +71,9 @@ fn a_request_is_decided_exactly_once() { fn concurrent_deciders_never_both_win() { let storage = MemoryStorage::new(); let store = docs_in(&storage, "local"); - store.insert_pending(&pending("race", 0, None), "s").unwrap(); + store + .insert_pending(&pending("race", 0, None), "s") + .unwrap(); let winners: usize = (0..8) .map(|_| { let store = store.clone(); @@ -89,15 +94,27 @@ fn concurrent_deciders_never_both_win() { #[test] fn stale_requests_expire_to_deny_once() { let store = docs(); - store.insert_pending(&pending("old", 0, Some(-5)), "s").unwrap(); - store.insert_pending(&pending("fresh", 0, Some(600)), "s").unwrap(); - store.insert_pending(&pending("forever", 0, None), "s").unwrap(); + store + .insert_pending(&pending("old", 0, Some(-5)), "s") + .unwrap(); + store + .insert_pending(&pending("fresh", 0, Some(600)), "s") + .unwrap(); + store + .insert_pending(&pending("forever", 0, None), "s") + .unwrap(); let expired = store.expire_stale(Utc::now()).unwrap(); assert_eq!( - expired.iter().map(|p| p.request_id.as_str()).collect::>(), + expired + .iter() + .map(|p| p.request_id.as_str()) + .collect::>(), ["old"] ); - assert!(store.expire_stale(Utc::now()).unwrap().is_empty(), "only once"); + assert!( + store.expire_stale(Utc::now()).unwrap().is_empty(), + "only once" + ); assert_eq!( store.get_decision("old").unwrap(), Some(ApprovalDecision::Deny) @@ -124,22 +141,18 @@ fn execution_is_recorded_after_a_decision_and_only_once() { ); store.decide("r", ApprovalDecision::ApproveOnce).unwrap(); let long = format!("token=sk-abcdefghijklmnopqrstuvwxyz {}", "e".repeat(800)); - assert!( - store - .record_execution("r", ExecutionOutcome::Failure, Some(&long)) - .unwrap() - ); + assert!(store + .record_execution("r", ExecutionOutcome::Failure, Some(&long)) + .unwrap()); assert!( !store .record_execution("r", ExecutionOutcome::Success, None) .unwrap(), "the first outcome wins" ); - assert!( - !store - .record_execution("missing", ExecutionOutcome::Success, None) - .unwrap() - ); + assert!(!store + .record_execution("missing", ExecutionOutcome::Success, None) + .unwrap()); let stored = store .run(|docs| async move { docs.get(APPROVALS, "r").await }) .unwrap() @@ -157,10 +170,15 @@ fn recent_decisions_are_newest_first_and_capped() { store.decide(id, ApprovalDecision::ApproveOnce).unwrap(); std::thread::sleep(std::time::Duration::from_millis(5)); } - store.insert_pending(&pending("open", 0, None), "s").unwrap(); + store + .insert_pending(&pending("open", 0, None), "s") + .unwrap(); let recent = store.list_recent_decisions(2).unwrap(); assert_eq!( - recent.iter().map(|e| e.request_id.as_str()).collect::>(), + recent + .iter() + .map(|e| e.request_id.as_str()) + .collect::>(), ["c", "b"] ); assert_eq!(recent[0].decision, ApprovalDecision::ApproveOnce); @@ -169,10 +187,16 @@ fn recent_decisions_are_newest_first_and_capped() { #[test] fn purging_a_session_drops_only_its_undecided_requests() { let store = docs(); - store.insert_pending(&pending("mine", 0, None), "s1").unwrap(); - store.insert_pending(&pending("decided", 0, None), "s1").unwrap(); + store + .insert_pending(&pending("mine", 0, None), "s1") + .unwrap(); + store + .insert_pending(&pending("decided", 0, None), "s1") + .unwrap(); store.decide("decided", ApprovalDecision::Deny).unwrap(); - store.insert_pending(&pending("theirs", 0, None), "s2").unwrap(); + store + .insert_pending(&pending("theirs", 0, None), "s2") + .unwrap(); assert_eq!(store.purge_session("s1").unwrap(), 1); assert!(store.get_decision("decided").unwrap().is_some()); let left: Vec = store @@ -236,7 +260,10 @@ fn scopes_keep_users_apart() { alice.insert_pending(&pending("r", 0, None), "s").unwrap(); alice.insert_flow_trust("flow-1", "shell").unwrap(); assert!(bob.list_pending().unwrap().is_empty()); - assert!(bob.decide("r", ApprovalDecision::ApproveOnce).unwrap().is_none()); + assert!(bob + .decide("r", ApprovalDecision::ApproveOnce) + .unwrap() + .is_none()); assert!(!bob.is_flow_tool_trusted("flow-1", "shell").unwrap()); assert_eq!(alice.list_pending().unwrap().len(), 1); } @@ -258,8 +285,7 @@ fn an_unreadable_source_context_reads_as_absent() { } #[test] -fn no_backend_means_the_classic_store() { - crate::storage::clear(); - assert!(current().unwrap().is_none()); +fn a_missing_error_stays_missing() { assert!(audit_error(None).is_none()); + assert_eq!(audit_error(Some("plain")).as_deref(), Some("plain")); } diff --git a/crates/openhuman-core/src/storage/mod.rs b/crates/openhuman-core/src/storage/mod.rs index 891f5b92def..08d736f5158 100644 --- a/crates/openhuman-core/src/storage/mod.rs +++ b/crates/openhuman-core/src/storage/mod.rs @@ -32,8 +32,37 @@ use crate::config::Config; /// The environment variable that overrides `[storage] url`. pub const STORAGE_URL_VAR: &str = "OPENHUMAN_STORAGE_URL"; -static BACKEND: LazyLock>>> = - LazyLock::new(|| RwLock::new(None)); +/// A holder for one backend. The process has exactly one ([`BACKEND`]); +/// tests make their own, so they never change what other tests in the same +/// process see. +#[derive(Default)] +struct Slot(RwLock>>); + +impl Slot { + fn install(&self, backend: Arc) -> Option> { + self.0 + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .replace(backend) + } + + fn installed(&self) -> Option> { + self.0 + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } + + fn clear(&self) -> bool { + self.0 + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take() + .is_some() + } +} + +static BACKEND: LazyLock = LazyLock::new(Slot::default); /// The storage URL in effect: [`STORAGE_URL_VAR`], else `config`'s /// `[storage] url`. Blank values count as unset. `None` keeps the classic @@ -70,27 +99,17 @@ pub async fn open(url: &str) -> Result, StorageError> { /// Makes `backend` the process's storage backend; returns the previous one. pub fn install(backend: Arc) -> Option> { - let mut slot = BACKEND - .write() - .unwrap_or_else(std::sync::PoisonError::into_inner); - slot.replace(backend) + BACKEND.install(backend) } /// The installed backend, when the host configured one. pub fn installed() -> Option> { - BACKEND - .read() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .clone() + BACKEND.installed() } /// Removes the installed backend; returns whether there was one. pub fn clear() -> bool { - BACKEND - .write() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .take() - .is_some() + BACKEND.clear() } /// The storage scope of the current call: the acting agent's diff --git a/crates/openhuman-core/src/storage/mod_tests.rs b/crates/openhuman-core/src/storage/mod_tests.rs index 764b2ec2b01..55db3f39152 100644 --- a/crates/openhuman-core/src/storage/mod_tests.rs +++ b/crates/openhuman-core/src/storage/mod_tests.rs @@ -39,15 +39,20 @@ async fn opens_memory_and_refuses_garbage() { #[test] fn install_installed_and_clear() { - // One test owns the process slot, so it cannot race another. - clear(); - assert!(installed().is_none()); + // A private slot: installing into the process's would reroute every other + // test's storage calls while this one runs. + let slot = Slot::default(); + assert!(slot.installed().is_none()); let backend: Arc = Arc::new(tinystoragedrivers::MemoryStorage::new()); - assert!(install(Arc::clone(&backend)).is_none()); - assert!(installed().is_some_and(|got| Arc::ptr_eq(&got, &backend))); - assert!(install(Arc::new(tinystoragedrivers::MemoryStorage::new())).is_some()); - assert!(clear()); - assert!(!clear()); + assert!(slot.install(Arc::clone(&backend)).is_none()); + assert!(slot + .installed() + .is_some_and(|got| Arc::ptr_eq(&got, &backend))); + assert!(slot + .install(Arc::new(tinystoragedrivers::MemoryStorage::new())) + .is_some()); + assert!(slot.clear()); + assert!(!slot.clear()); } #[test] @@ -60,10 +65,19 @@ fn agent_scopes_match_the_session_store() { #[test] fn the_scope_follows_the_acting_agent_and_fails_closed_in_saas() { - assert_eq!(scope_from(Some("agent-7"), false).unwrap().as_str(), "agent-7"); - assert_eq!(scope_from(Some("agent-7"), true).unwrap().as_str(), "agent-7"); + assert_eq!( + scope_from(Some("agent-7"), false).unwrap().as_str(), + "agent-7" + ); + assert_eq!( + scope_from(Some("agent-7"), true).unwrap().as_str(), + "agent-7" + ); assert!(scope_from(None, false).unwrap().is_local()); - assert!(scope_from(None, true).is_err(), "SaaS never falls back to a shared scope"); + assert!( + scope_from(None, true).is_err(), + "SaaS never falls back to a shared scope" + ); } #[test] From 6696ee48b3dba06c0d88ce451a1c8a17b452ec30 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 09:44:34 +0300 Subject: [PATCH 06/14] test(cli): register storage approvals e2e test target Add a Cargo test target for the new storage approvals end-to-end suite. It installs a backend into the process-wide storage slot, so it must run in its own process to avoid rerouting other suites' approvals. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/openhuman-cli/Cargo.toml | 6 ++++ tests/storage_approvals_e2e.rs | 60 +++++++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+) create mode 100644 tests/storage_approvals_e2e.rs diff --git a/crates/openhuman-cli/Cargo.toml b/crates/openhuman-cli/Cargo.toml index e4be4568c93..5341e1c4296 100644 --- a/crates/openhuman-cli/Cargo.toml +++ b/crates/openhuman-cli/Cargo.toml @@ -116,6 +116,12 @@ required-features = ["rss-bench"] name = "cost_cap_removed" path = "../../tests/cost_cap_removed.rs" +[[test]] +# Installs a backend into the process-wide storage slot, so it needs its own +# process: in a shared one it would reroute other suites' approvals. +name = "storage_approvals_e2e" +path = "../../tests/storage_approvals_e2e.rs" + [[test]] name = "embedded_server_shutdown_e2e" path = "../../tests/embedded_server_shutdown_e2e.rs" diff --git a/tests/storage_approvals_e2e.rs b/tests/storage_approvals_e2e.rs new file mode 100644 index 00000000000..dc60e11b865 --- /dev/null +++ b/tests/storage_approvals_e2e.rs @@ -0,0 +1,60 @@ +//! The approval store on a configured storage backend, end to end through +//! its public functions. +//! +//! Its own test binary because it installs a backend into the process-wide +//! storage slot, which would reroute every other suite's approvals in a +//! shared process. + +use std::sync::Arc; + +use chrono::Utc; +use openhuman_core::config::Config; +use openhuman_core::security::approval::store; +use openhuman_core::security::approval::types::{ApprovalDecision, PendingApproval}; + +fn pending(id: &str) -> PendingApproval { + PendingApproval { + request_id: id.to_string(), + tool_name: "shell".to_string(), + action_summary: format!("run {id}"), + args_redacted: serde_json::json!({ "cmd": "ls" }), + created_at: Utc::now(), + expires_at: None, + source_context: None, + tool_call_id: None, + } +} + +#[test] +fn a_configured_backend_holds_the_approvals_instead_of_approval_db() { + let workspace = tempfile::tempdir().unwrap(); + let config = Config { + workspace_dir: workspace.path().to_path_buf(), + ..Config::default() + }; + openhuman_core::storage::install(Arc::new(openhuman_core::storage::MemoryStorage::new())); + + store::insert_pending(&config, &pending("r1"), "session").unwrap(); + store::insert_flow_trust(&config, "flow-1", "shell").unwrap(); + assert_eq!(store::list_pending(&config).unwrap().len(), 1); + assert!(store::is_flow_tool_trusted(&config, "flow-1", "shell").unwrap()); + let decided = store::decide(&config, "r1", ApprovalDecision::ApproveOnce).unwrap(); + assert!(decided.is_some()); + assert_eq!( + store::get_decision(&config, "r1").unwrap(), + Some(ApprovalDecision::ApproveOnce) + ); + assert_eq!(store::list_recent_decisions(&config, 10).unwrap().len(), 1); + assert_eq!(store::expire_stale(&config).unwrap(), 0); + assert_eq!(store::purge_session(&config, "session").unwrap(), 0); + + assert!( + !workspace.path().join("approval").join("approval.db").exists(), + "nothing was written to the classic database" + ); + + // Without a backend the classic database is back in use. + assert!(openhuman_core::storage::clear()); + assert!(store::list_pending(&config).unwrap().is_empty()); + assert!(workspace.path().join("approval").join("approval.db").exists()); +} From 659272bf854d0c09c8d51f79c77bf132eb6aaa10 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 09:47:43 +0300 Subject: [PATCH 07/14] test(storage): cover approvals fallback to classic database Re-export MemoryStorage from the storage module and extend the approvals end-to-end test to assert the classic database is used once the configured backend is cleared. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/openhuman-core/src/storage/mod.rs | 2 +- tests/storage_approvals_e2e.rs | 12 ++++++++++-- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/crates/openhuman-core/src/storage/mod.rs b/crates/openhuman-core/src/storage/mod.rs index 08d736f5158..84fa4819a26 100644 --- a/crates/openhuman-core/src/storage/mod.rs +++ b/crates/openhuman-core/src/storage/mod.rs @@ -22,7 +22,7 @@ use std::future::Future; use std::sync::{Arc, LazyLock, OnceLock, RwLock}; pub use tinystoragedrivers::{ - Blocking, DocumentStore, DocumentStoreExt, Scope, ScopedStorage, StorageBackend, + Blocking, DocumentStore, DocumentStoreExt, MemoryStorage, Scope, ScopedStorage, StorageBackend, StorageConfig as StorageUrl, StorageError, }; diff --git a/tests/storage_approvals_e2e.rs b/tests/storage_approvals_e2e.rs index dc60e11b865..9be48067fb2 100644 --- a/tests/storage_approvals_e2e.rs +++ b/tests/storage_approvals_e2e.rs @@ -49,12 +49,20 @@ fn a_configured_backend_holds_the_approvals_instead_of_approval_db() { assert_eq!(store::purge_session(&config, "session").unwrap(), 0); assert!( - !workspace.path().join("approval").join("approval.db").exists(), + !workspace + .path() + .join("approval") + .join("approval.db") + .exists(), "nothing was written to the classic database" ); // Without a backend the classic database is back in use. assert!(openhuman_core::storage::clear()); assert!(store::list_pending(&config).unwrap().is_empty()); - assert!(workspace.path().join("approval").join("approval.db").exists()); + assert!(workspace + .path() + .join("approval") + .join("approval.db") + .exists()); } From f7aba9ef7e27e3dae4896a5cf847479aaa4893e6 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 09:48:25 +0300 Subject: [PATCH 08/14] test(storage): use PendingApproval::new in approvals e2e helper The pending helper now builds its fixture through PendingApproval::new instead of constructing the struct literal directly, dropping the chrono import that was only needed for the created_at field. This keeps the test aligned with the constructor's defaults and avoids duplicating field setup. Auto-committed-on: dragonfly Co-authored-by: Medulla --- tests/storage_approvals_e2e.rs | 18 +++++++----------- 1 file changed, 7 insertions(+), 11 deletions(-) diff --git a/tests/storage_approvals_e2e.rs b/tests/storage_approvals_e2e.rs index 9be48067fb2..d51112f8de7 100644 --- a/tests/storage_approvals_e2e.rs +++ b/tests/storage_approvals_e2e.rs @@ -7,22 +7,18 @@ use std::sync::Arc; -use chrono::Utc; use openhuman_core::config::Config; use openhuman_core::security::approval::store; use openhuman_core::security::approval::types::{ApprovalDecision, PendingApproval}; fn pending(id: &str) -> PendingApproval { - PendingApproval { - request_id: id.to_string(), - tool_name: "shell".to_string(), - action_summary: format!("run {id}"), - args_redacted: serde_json::json!({ "cmd": "ls" }), - created_at: Utc::now(), - expires_at: None, - source_context: None, - tool_call_id: None, - } + PendingApproval::new( + id, + "shell", + format!("run {id}"), + serde_json::json!({ "cmd": "ls" }), + None, + ) } #[test] From eae46d3623f8ea4572e20da4ffb8ee5e2b23e488 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 09:52:15 +0300 Subject: [PATCH 09/14] docs(security): document approval storage backend behaviour Explain that when a storage backend is configured, approval store operations run against the document port instead of approval.db, with SQL guards expressed as preconditions and the desktop default unchanged. Auto-committed-on: dragonfly Co-authored-by: Medulla --- .../src/security/approval/README.md | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/crates/openhuman-core/src/security/approval/README.md b/crates/openhuman-core/src/security/approval/README.md index 6ba78374cae..580209dd102 100644 --- a/crates/openhuman-core/src/security/approval/README.md +++ b/crates/openhuman-core/src/security/approval/README.md @@ -71,6 +71,21 @@ No `bus.rs` in this module. It only publishes; the subscriber (`ApprovalSurfaceS SQLite DB at `{workspace_dir}/approval/approval.db`, table `pending_approvals` (opened per-call via `with_connection`, schema + column migration applied idempotently). Columns: `request_id` (PK), `tool_name`, `action_summary`, `args_redacted` (JSON), `session_id`, `created_at`, `expires_at`, `decided_at`, `decision`, plus the after-action audit columns `executed_at`, `execution_outcome`, `execution_error` (added by `migrate_columns` for v1 DBs). Pending rows survive restart; expired rows are lazily transitioned to a terminal `deny` decision; `record_execution` is write-once (`executed_at IS NULL` guard) and sanitizes/caps error text to 512 chars to keep secrets/PII out of the durable log. +### On a storage backend + +When the host configured a storage backend (`OPENHUMAN_STORAGE_URL` / +`[storage] url`, see `crate::storage`), every `store` function uses +`store_documents.rs` instead of `approval.db`: the same operations on the +`tinystoragedrivers` document port, under the current call's storage scope +(the acting agent; `local` on a single-user host; refused in SaaS mode with +no acting agent). Collections `approvals` (one document per `request_id`) +and `approval_flow_trust` (one per `(flow_id, tool_name)`). The SQL guards +become preconditions: insert-only for new requests, compare-and-swap for +decide, expiry and `record_execution`, so two processes on one database +decide a request at most once. `args_redacted` and `source_context` are +stored as JSON strings. With no backend configured (the desktop default) +`approval.db` is used exactly as above. + ## Dependencies - `crate::core::bus::BUS` + `crate::core::events::DomainEvent` to surface approval prompts/decisions. From e22bd34896b22b8799dddb9c66b35927b7b0caa3 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 12:29:28 +0300 Subject: [PATCH 10/14] feat(security): add document approval store and gate interception Adds a document-backed approval store and wires the approval gate to intercept requests before they proceed. This lets approvals be persisted and checked consistently across the security layer. Auto-committed-on: dragonfly Co-authored-by: Medulla --- .../src/security/approval/gate.rs | 10 +++++++- .../src/security/approval/gate_intercept.rs | 1 + .../src/security/approval/store.rs | 23 ++++++++++++++++++ .../src/security/approval/store_documents.rs | 24 ++++++++++++++----- 4 files changed, 51 insertions(+), 7 deletions(-) diff --git a/crates/openhuman-core/src/security/approval/gate.rs b/crates/openhuman-core/src/security/approval/gate.rs index 56dade16b7c..d3e698b0ba9 100644 --- a/crates/openhuman-core/src/security/approval/gate.rs +++ b/crates/openhuman-core/src/security/approval/gate.rs @@ -246,6 +246,9 @@ struct WaiterGuard<'a> { gate: &'a ApprovalGate, request_id: String, thread_id: Option, + /// Storage scope of the parking call, kept for `Drop`: the acting agent's + /// task-local is gone by then. + docs: Option, armed: bool, } @@ -277,7 +280,12 @@ impl Drop for WaiterGuard<'_> { self.gate .clear_thread_route_if_owned(thread_id, &self.request_id); } - let decided = store::decide(&self.gate.config, &self.request_id, ApprovalDecision::Deny); + let decided = store::decide_captured( + &self.gate.config, + self.docs.as_ref(), + &self.request_id, + ApprovalDecision::Deny, + ); if let Ok(Some(row)) = decided { let route = self.gate.take_request_route(&self.request_id); BUS.publish(DomainEvent::ApprovalDecided { diff --git a/crates/openhuman-core/src/security/approval/gate_intercept.rs b/crates/openhuman-core/src/security/approval/gate_intercept.rs index 03b1937ebbb..6c9b55b9a7f 100644 --- a/crates/openhuman-core/src/security/approval/gate_intercept.rs +++ b/crates/openhuman-core/src/security/approval/gate_intercept.rs @@ -591,6 +591,7 @@ impl ApprovalGate { gate: self, request_id: request_id.clone(), thread_id: chat_thread_id.clone(), + docs: store::capture_docs(), armed: true, }; diff --git a/crates/openhuman-core/src/security/approval/store.rs b/crates/openhuman-core/src/security/approval/store.rs index a8d5742ef8e..49ec012c091 100644 --- a/crates/openhuman-core/src/security/approval/store.rs +++ b/crates/openhuman-core/src/security/approval/store.rs @@ -347,6 +347,29 @@ pub fn get_decision(config: &Config, request_id: &str) -> Result Option { + super::store_documents::current().ok().flatten() +} + +/// [`decide`] against a store captured by [`capture_docs`] when there is one. +pub(super) fn decide_captured( + config: &Config, + captured: Option<&super::store_documents::Docs>, + request_id: &str, + decision: ApprovalDecision, +) -> Result> { + match captured { + Some(docs) => { + publish_expired(&docs.expire_stale(Utc::now())?); + docs.decide(request_id, decision) + } + None => decide(config, request_id, decision), + } +} + /// Mark a pending row as decided and return the now-decided row. /// Returns `Ok(None)` if no row matched (already decided, expired, or /// unknown id). diff --git a/crates/openhuman-core/src/security/approval/store_documents.rs b/crates/openhuman-core/src/security/approval/store_documents.rs index 04233fea621..43825a12cde 100644 --- a/crates/openhuman-core/src/security/approval/store_documents.rs +++ b/crates/openhuman-core/src/security/approval/store_documents.rs @@ -66,8 +66,9 @@ fn storage(error: StorageError) -> anyhow::Error { anyhow!("[approval::store] storage: {error}") } +/// Fixed-width (nanosecond, `Z`) so the strings sort in time order. fn rfc3339(at: DateTime) -> String { - at.to_rfc3339() + at.to_rfc3339_opts(chrono::SecondsFormat::Nanos, true) } fn parse_time(raw: Option<&str>) -> Option> { @@ -169,7 +170,7 @@ impl Docs { "session_id": session_id, "created_at": rfc3339(pending.created_at), "expires_at": pending.expires_at.map(rfc3339), - "expires_ts": pending.expires_at.map(|at| at.timestamp()), + "expires_ts": pending.expires_at.map(|at| at.timestamp_millis()), "source_context": pending .source_context .as_ref() @@ -225,7 +226,7 @@ impl Docs { pub(super) fn expire_stale(&self, now: DateTime) -> Result> { let deny = ApprovalDecision::Deny.as_str(); let decided_at = rfc3339(now); - let now_ts = now.timestamp(); + let now_ts = now.timestamp_millis(); self.run(|docs| async move { let query = Query::filter(Filter::eq("pending", true).and(Filter::lte("expires_ts", now_ts))); @@ -242,8 +243,8 @@ impl Docs { Some(next) }) .await?; - if moved.is_some() { - expired.push(to_pending(&stale)); + if let Some(moved) = moved { + expired.push(to_pending(&moved)); } } Ok(expired) @@ -278,7 +279,18 @@ impl Docs { decision: ApprovalDecision, ) -> Result> { let id = request_id.to_string(); - let decided_at = rfc3339(Utc::now()); + self.decide_at(request_id, decision, Utc::now()) + } + + /// [`decide`](Self::decide) stamped `at`, so ordering can be pinned. + pub(super) fn decide_at( + &self, + request_id: &str, + decision: ApprovalDecision, + at: DateTime, + ) -> Result> { + let id = request_id.to_string(); + let decided_at = rfc3339(at); self.run(|docs| async move { let decided = update(&docs, &id, |doc| { if doc.get("pending") != Some(&json!(true)) { From 5a052033985386496ffb55d36d0806e3172b114c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 12:29:47 +0300 Subject: [PATCH 11/14] chore(security): remove unused variable in approval store Dropped an unused `id` binding in the approval decision helper, which was left over from an earlier implementation and triggered a compiler warning. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/openhuman-core/src/security/approval/store_documents.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/crates/openhuman-core/src/security/approval/store_documents.rs b/crates/openhuman-core/src/security/approval/store_documents.rs index 43825a12cde..5708c5fbdcc 100644 --- a/crates/openhuman-core/src/security/approval/store_documents.rs +++ b/crates/openhuman-core/src/security/approval/store_documents.rs @@ -278,7 +278,6 @@ impl Docs { request_id: &str, decision: ApprovalDecision, ) -> Result> { - let id = request_id.to_string(); self.decide_at(request_id, decision, Utc::now()) } From 52d142828fa17d84e9007a05149da0e4f35e5dde Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 12:39:03 +0300 Subject: [PATCH 12/14] test(approval): pin timestamp ordering, expiry precision and captured-scope decide Co-authored-by: Medulla --- .../approval/store_documents_tests.rs | 62 +++++++++++++++++-- 1 file changed, 58 insertions(+), 4 deletions(-) diff --git a/crates/openhuman-core/src/security/approval/store_documents_tests.rs b/crates/openhuman-core/src/security/approval/store_documents_tests.rs index 8a17d7308da..69127d05b36 100644 --- a/crates/openhuman-core/src/security/approval/store_documents_tests.rs +++ b/crates/openhuman-core/src/security/approval/store_documents_tests.rs @@ -1,5 +1,5 @@ use super::*; -use chrono::Duration; +use chrono::{Duration, Timelike}; use tinystoragedrivers::{MemoryStorage, Scope, StorageBackend}; fn docs_in(storage: &MemoryStorage, scope: &str) -> Docs { @@ -160,15 +160,23 @@ fn execution_is_recorded_after_a_decision_and_only_once() { assert_eq!(stored.doc["execution_outcome"], json!("failure")); let error = stored.doc["execution_error"].as_str().unwrap(); assert!(error.chars().count() <= 512, "capped"); + assert!( + !error.contains("sk-abcdefghijklmnopqrstuvwxyz"), + "the secret is redacted before storage: {error}" + ); } #[test] fn recent_decisions_are_newest_first_and_capped() { let store = docs(); - for id in ["a", "b", "c"] { + let base = Utc::now(); + for (n, id) in ["a", "b", "c"].into_iter().enumerate() { store.insert_pending(&pending(id, 0, None), "s").unwrap(); - store.decide(id, ApprovalDecision::ApproveOnce).unwrap(); - std::thread::sleep(std::time::Duration::from_millis(5)); + // Distinct sub-second stamps (the fixed-width format must order them). + let at = base + Duration::milliseconds(n as i64 * 100); + store + .decide_at(id, ApprovalDecision::ApproveOnce, at) + .unwrap(); } store .insert_pending(&pending("open", 0, None), "s") @@ -289,3 +297,49 @@ fn a_missing_error_stays_missing() { assert!(audit_error(None).is_none()); assert_eq!(audit_error(Some("plain")).as_deref(), Some("plain")); } + +#[test] +fn sub_second_timestamps_order_and_expire_on_time() { + let store = docs(); + let base = Utc::now(); + // Whole-second and sub-second values in the same second must sort by time. + let whole = base.with_nanosecond(0).unwrap(); + for (id, at) in [("x", whole + Duration::milliseconds(500)), ("w", whole)] { + let mut p = pending(id, 0, None); + p.created_at = at; + store.insert_pending(&p, "s").unwrap(); + } + let ids: Vec = store + .list_pending() + .unwrap() + .into_iter() + .map(|p| p.request_id) + .collect(); + assert_eq!(ids, ["w", "x"]); + + // An approval due 900 ms into a second is not expired at 100 ms. + let due = whole + Duration::milliseconds(900); + let mut p = pending("late", 0, None); + p.expires_at = Some(due); + store.insert_pending(&p, "s").unwrap(); + assert!(store + .expire_stale(whole + Duration::milliseconds(100)) + .unwrap() + .is_empty()); + assert_eq!(store.expire_stale(due).unwrap().len(), 1); +} + +#[test] +fn a_captured_store_decides_without_the_task_scope() { + let store = docs(); + store.insert_pending(&pending("r", 0, None), "s").unwrap(); + let config = crate::config::Config::default(); + let decided = crate::security::approval::store::decide_captured( + &config, + Some(&store), + "r", + ApprovalDecision::Deny, + ) + .unwrap(); + assert_eq!(decided.map(|p| p.request_id), Some("r".to_string())); +} From 1660a6eecf60cb79298d30b68db7c0338d2e7547 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 12:58:00 +0300 Subject: [PATCH 13/14] fix(approval): fail captured decisions when the storage scope is unresolved The captured document store now carries the scope-resolution error instead of silently falling back to SQLite, so a decision made after the task scope is gone fails loudly rather than writing to the wrong store. Expiry timestamps are also compared at nanosecond resolution, with a millisecond fallback beyond the i64-nanosecond range, so approvals expiring within the same millisecond are ordered correctly. Auto-committed-on: dragonfly Co-authored-by: Medulla --- .../src/security/approval/gate.rs | 4 +-- .../src/security/approval/store.rs | 18 +++++++----- .../src/security/approval/store_documents.rs | 11 +++++-- .../approval/store_documents_tests.rs | 29 +++++++++++++++++++ 4 files changed, 51 insertions(+), 11 deletions(-) diff --git a/crates/openhuman-core/src/security/approval/gate.rs b/crates/openhuman-core/src/security/approval/gate.rs index d3e698b0ba9..d6abbb42b48 100644 --- a/crates/openhuman-core/src/security/approval/gate.rs +++ b/crates/openhuman-core/src/security/approval/gate.rs @@ -248,7 +248,7 @@ struct WaiterGuard<'a> { thread_id: Option, /// Storage scope of the parking call, kept for `Drop`: the acting agent's /// task-local is gone by then. - docs: Option, + docs: Result, String>, armed: bool, } @@ -282,7 +282,7 @@ impl Drop for WaiterGuard<'_> { } let decided = store::decide_captured( &self.gate.config, - self.docs.as_ref(), + &self.docs, &self.request_id, ApprovalDecision::Deny, ); diff --git a/crates/openhuman-core/src/security/approval/store.rs b/crates/openhuman-core/src/security/approval/store.rs index 49ec012c091..e952d5f6bba 100644 --- a/crates/openhuman-core/src/security/approval/store.rs +++ b/crates/openhuman-core/src/security/approval/store.rs @@ -349,24 +349,28 @@ pub fn get_decision(config: &Config, request_id: &str) -> Result Option { - super::store_documents::current().ok().flatten() +/// no longer installed. A scope that cannot be resolved is kept as an error, +/// so the later decision fails instead of switching to the SQLite store. +pub(super) fn capture_docs() -> Result, String> { + super::store_documents::current().map_err(|error| error.to_string()) } -/// [`decide`] against a store captured by [`capture_docs`] when there is one. +/// [`decide`] against a store captured by [`capture_docs`]. pub(super) fn decide_captured( config: &Config, - captured: Option<&super::store_documents::Docs>, + captured: &Result, String>, request_id: &str, decision: ApprovalDecision, ) -> Result> { match captured { - Some(docs) => { + Ok(Some(docs)) => { publish_expired(&docs.expire_stale(Utc::now())?); docs.decide(request_id, decision) } - None => decide(config, request_id, decision), + Ok(None) => decide(config, request_id, decision), + Err(error) => Err(anyhow::anyhow!( + "[approval::store] storage scope was unresolved when the request parked: {error}" + )), } } diff --git a/crates/openhuman-core/src/security/approval/store_documents.rs b/crates/openhuman-core/src/security/approval/store_documents.rs index 5708c5fbdcc..a5a2e2b7fa3 100644 --- a/crates/openhuman-core/src/security/approval/store_documents.rs +++ b/crates/openhuman-core/src/security/approval/store_documents.rs @@ -71,6 +71,13 @@ fn rfc3339(at: DateTime) -> String { at.to_rfc3339_opts(chrono::SecondsFormat::Nanos, true) } +/// The comparable expiry key: nanoseconds since the epoch, falling back to +/// milliseconds for instants beyond the i64-nanosecond range (year 2262). +fn expiry_key(at: DateTime) -> i64 { + at.timestamp_nanos_opt() + .unwrap_or_else(|| at.timestamp_millis().saturating_mul(1_000_000)) +} + fn parse_time(raw: Option<&str>) -> Option> { raw.and_then(|value| DateTime::parse_from_rfc3339(value).ok()) .map(|at| at.with_timezone(&Utc)) @@ -170,7 +177,7 @@ impl Docs { "session_id": session_id, "created_at": rfc3339(pending.created_at), "expires_at": pending.expires_at.map(rfc3339), - "expires_ts": pending.expires_at.map(|at| at.timestamp_millis()), + "expires_ts": pending.expires_at.map(expiry_key), "source_context": pending .source_context .as_ref() @@ -226,7 +233,7 @@ impl Docs { pub(super) fn expire_stale(&self, now: DateTime) -> Result> { let deny = ApprovalDecision::Deny.as_str(); let decided_at = rfc3339(now); - let now_ts = now.timestamp_millis(); + let now_ts = expiry_key(now); self.run(|docs| async move { let query = Query::filter(Filter::eq("pending", true).and(Filter::lte("expires_ts", now_ts))); diff --git a/crates/openhuman-core/src/security/approval/store_documents_tests.rs b/crates/openhuman-core/src/security/approval/store_documents_tests.rs index 69127d05b36..b6c92d0ba00 100644 --- a/crates/openhuman-core/src/security/approval/store_documents_tests.rs +++ b/crates/openhuman-core/src/security/approval/store_documents_tests.rs @@ -343,3 +343,32 @@ fn a_captured_store_decides_without_the_task_scope() { .unwrap(); assert_eq!(decided.map(|p| p.request_id), Some("r".to_string())); } + +#[test] +fn an_unresolved_scope_fails_the_captured_decide_instead_of_using_sqlite() { + let config = crate::config::Config::default(); + let captured = Err("no acting agent".to_string()); + let error = crate::security::approval::store::decide_captured( + &config, + &captured, + "r", + ApprovalDecision::Deny, + ) + .unwrap_err(); + assert!(error.to_string().contains("unresolved"), "{error}"); +} + +#[test] +fn expiry_is_compared_below_a_millisecond() { + let store = docs(); + let whole = Utc::now().with_nanosecond(0).unwrap(); + let due = whole + Duration::nanoseconds(900_000); + let mut p = pending("tight", 0, None); + p.expires_at = Some(due); + store.insert_pending(&p, "s").unwrap(); + assert!(store + .expire_stale(whole + Duration::nanoseconds(100_000)) + .unwrap() + .is_empty()); + assert_eq!(store.expire_stale(due).unwrap().len(), 1); +} From 9b893a70781cf76873347469450d65a73a721a6f Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 13:04:05 +0300 Subject: [PATCH 14/14] test(security): update captured store decision call Adjust the approval store test to pass the captured store wrapped in a result, matching the updated decide_captured signature. Auto-committed-on: dragonfly Co-authored-by: Medulla --- .../src/security/approval/store_documents_tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/openhuman-core/src/security/approval/store_documents_tests.rs b/crates/openhuman-core/src/security/approval/store_documents_tests.rs index b6c92d0ba00..d1732b8493d 100644 --- a/crates/openhuman-core/src/security/approval/store_documents_tests.rs +++ b/crates/openhuman-core/src/security/approval/store_documents_tests.rs @@ -336,7 +336,7 @@ fn a_captured_store_decides_without_the_task_scope() { let config = crate::config::Config::default(); let decided = crate::security::approval::store::decide_captured( &config, - Some(&store), + &Ok(Some(store.clone())), "r", ApprovalDecision::Deny, )