feat(approval): approvals on the storage ports

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"

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" }

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.

crates/openhuman-core/src/security/approval/gate.rs

@@ -246,6 +246,9 @@ struct WaiterGuard<'a> {
     gate: &'a ApprovalGate,
     request_id: String,
     thread_id: Option<String>,
+    /// Storage scope of the parking call, kept for `Drop`: the acting agent's
+    /// task-local is gone by then.
+    docs: Result<Option<super::store_documents::Docs>, String>,
     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,
+            &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 {

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,
         };
 

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::{

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<T>(config: &Config, f: impl FnOnce(&Connection) -> Result<T>)
 /// 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<usize> {
+    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<usize> {
 /// 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<Vec<PendingApproval>> {
+    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<Vec<PendingApproval>> {
 /// where the TTL elapses concurrently with a committed approval
 /// (CodeRabbit review on PR #2367).
 pub fn get_decision(config: &Config, request_id: &str) -> Result<Option<ApprovalDecision>> {
+    if let Some(docs) = super::store_documents::current()? {
+        return docs.get_decision(request_id);
+    }
     with_connection(config, |conn| {
         let mut stmt = conn
             .prepare(
@@ -325,6 +347,33 @@ pub fn get_decision(config: &Config, request_id: &str) -> Result<Option<Approval
     })
 }
 
+/// The document store of the current call, for a caller that decides later
+/// from outside this call's task scope (a `Drop`), where the acting agent is
+/// 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<Option<super::store_documents::Docs>, String> {
+    super::store_documents::current().map_err(|error| error.to_string())
+}
+
+/// [`decide`] against a store captured by [`capture_docs`].
+pub(super) fn decide_captured(
+    config: &Config,
+    captured: &Result<Option<super::store_documents::Docs>, String>,
+    request_id: &str,
+    decision: ApprovalDecision,
+) -> Result<Option<PendingApproval>> {
+    match captured {
+        Ok(Some(docs)) => {
+            publish_expired(&docs.expire_stale(Utc::now())?);
+            docs.decide(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}"
+        )),
+    }
+}
+
 /// 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).
@@ -333,6 +382,10 @@ pub fn decide(
     request_id: &str,
     decision: ApprovalDecision,
 ) -> Result<Option<PendingApproval>> {
+    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 +442,9 @@ pub fn record_execution(
     outcome: ExecutionOutcome,
     error: Option<&str>,
 ) -> Result<bool> {
+    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 +490,9 @@ pub fn record_execution(
 /// List recently decided approval rows for durable audit views.
 pub fn list_recent_decisions(config: &Config, limit: usize) -> Result<Vec<ApprovalAuditEntry>> {
     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 +518,9 @@ pub fn list_recent_decisions(config: &Config, limit: usize) -> Result<Vec<Approv
 /// Drop all rows owned by `session_id` — called when the gate detects
 /// a session changeover so stale parked rows do not accumulate.
 pub fn purge_session(config: &Config, session_id: &str) -> Result<usize> {
+    if let Some(docs) = super::store_documents::current()? {
+        return docs.purge_session(session_id);
+    }
     with_connection(config, |conn| {
         let removed = conn
             .execute(
@@ -546,18 +608,25 @@ fn expire_stale_with_now(conn: &Connection, now: DateTime<Utc>) -> Result<Vec<Pe
         rows = updated,
         "[approval::store] lazily expired stale pending_approvals rows"
     );
-    for row in &about_to_expire {
+    publish_expired(&about_to_expire);
+    Ok(about_to_expire)
+}
+
+/// Tells the web channel each expired row's parked card is now stale — a
+/// sweep runs with no live `ApprovalGate` in scope, so the store is the only
+/// place that observes an expiry.
+fn publish_expired(expired: &[PendingApproval]) {
+    for row in expired {
         BUS.publish(DomainEvent::ApprovalDecided {
             request_id: row.request_id.clone(),
             tool_name: row.tool_name.clone(),
-            decision: deny.to_string(),
+            decision: ApprovalDecision::Deny.as_str().to_string(),
             thread_id: None,
             client_id: None,
             tool_call_id: row.tool_call_id.clone(),
             resolution: Some("expired".to_string()),
         });
     }
-    Ok(about_to_expire)
 }
 
 fn row_to_audit_entry(row: &rusqlite::Row<'_>) -> rusqlite::Result<ApprovalAuditEntry> {