From 34bcf343dc8aef468b64fb4a738818be93ddf388 Mon Sep 17 00:00:00 2001 From: oxoxDev <164490987+oxoxDev@users.noreply.github.com> Date: Wed, 20 May 2026 08:28:25 +0530 Subject: [PATCH] feat(approval): gate external-effect tool calls until user approves (#1339) (#2149) --- src/core/all.rs | 3 + src/core/event_bus/events.rs | 30 ++ src/core/jsonrpc.rs | 41 ++ .../agent/harness/subagent_runner/ops.rs | 53 ++- src/openhuman/agent/harness/tool_loop.rs | 37 ++ src/openhuman/agent/triage/escalation.rs | 43 ++ src/openhuman/approval/gate.rs | 387 ++++++++++++++++++ src/openhuman/approval/mod.rs | 26 +- src/openhuman/approval/redact.rs | 326 +++++++++++++++ src/openhuman/approval/rpc.rs | 85 ++++ src/openhuman/approval/schemas.rs | 198 +++++++++ src/openhuman/approval/store.rs | 324 +++++++++++++++ src/openhuman/approval/types.rs | 107 +++++ src/openhuman/channels/proactive.rs | 33 ++ src/openhuman/tools/impl/network/composio.rs | 23 ++ .../tools/impl/network/gmail_unsubscribe.rs | 14 + src/openhuman/tools/impl/system/pushover.rs | 16 + src/openhuman/tools/traits.rs | 36 ++ src/openhuman/voice/reply_speech.rs | 7 + 19 files changed, 1778 insertions(+), 11 deletions(-) create mode 100644 src/openhuman/approval/gate.rs create mode 100644 src/openhuman/approval/redact.rs create mode 100644 src/openhuman/approval/rpc.rs create mode 100644 src/openhuman/approval/schemas.rs create mode 100644 src/openhuman/approval/store.rs create mode 100644 src/openhuman/approval/types.rs diff --git a/src/core/all.rs b/src/core/all.rs index 3d4a662d2..1dc5942e3 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -127,6 +127,8 @@ fn build_registered_controllers() -> Vec { controllers.extend(crate::openhuman::encryption::all_encryption_registered_controllers()); // Security policy metadata controllers.extend(crate::openhuman::security::all_security_registered_controllers()); + // Interactive approval workflow (#1339 — gate external-effect tool calls) + controllers.extend(crate::openhuman::approval::all_approval_registered_controllers()); // Background heartbeat loop controls controllers.extend(crate::openhuman::heartbeat::all_heartbeat_registered_controllers()); // Ad-hoc static directory HTTP hosting for local file sharing / previews @@ -276,6 +278,7 @@ fn build_declared_controller_schemas() -> Vec { schemas.extend(crate::openhuman::doctor::all_doctor_controller_schemas()); schemas.extend(crate::openhuman::encryption::all_encryption_controller_schemas()); schemas.extend(crate::openhuman::security::all_security_controller_schemas()); + schemas.extend(crate::openhuman::approval::all_approval_controller_schemas()); schemas.extend(crate::openhuman::heartbeat::all_heartbeat_controller_schemas()); schemas.extend(crate::openhuman::http_host::all_http_host_controller_schemas()); schemas.extend(crate::openhuman::cost::all_cost_controller_schemas()); diff --git a/src/core/event_bus/events.rs b/src/core/event_bus/events.rs index 18087217e..33d161b73 100644 --- a/src/core/event_bus/events.rs +++ b/src/core/event_bus/events.rs @@ -202,6 +202,34 @@ pub enum DomainEvent { elapsed_ms: u64, }, + // ── Approval ──────────────────────────────────────────────────────── + /// Agent attempted a tool call that produces an external side + /// effect; awaiting user approval. Published by `ApprovalGate` + /// before parking the tool-call future. Issue #1339. + ApprovalRequested { + /// Unique id used to correlate the decision back to the + /// parked future. + request_id: String, + /// Tool name being gated (e.g. `"composio"`, `"pushover"`). + tool_name: String, + /// Short human-readable summary of the action, redacted of + /// PII/secrets/message bodies (counts/shape only). + action_summary: String, + /// Redacted JSON arguments — also stripped of raw user content. + args_redacted: serde_json::Value, + /// Session id binding the request to the current core launch + /// so stale approvals cannot be replayed after restart. + session_id: String, + }, + /// User decided a pending approval. Published by `approval_decide` + /// RPC handler after the gate's parked future resolves. + ApprovalDecided { + request_id: String, + tool_name: String, + /// `"approve_once"`, `"approve_always_for_tool"`, or `"deny"`. + decision: String, + }, + // ── Webhooks ──────────────────────────────────────────────────────── /// An incoming webhook request from the transport layer, ready for routing. WebhookIncomingRequest { @@ -539,6 +567,8 @@ impl DomainEvent { | Self::HealthRestarted { .. } => "system", Self::SessionExpired { .. } => "auth", + + Self::ApprovalRequested { .. } | Self::ApprovalDecided { .. } => "approval", } } } diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 133b1fcbe..f60901152 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -1256,6 +1256,47 @@ pub async fn bootstrap_core_runtime(embedded_core: bool) { ); } + // --- Approval gate (#1339) --- + // Opt-in via `OPENHUMAN_APPROVAL_GATE=1`. When enabled, tool calls + // with `external_effect() == true` (composio, pushover, gmail + // unsubscribe, proactive external sends, triage React/Escalate) + // route through `ApprovalGate::intercept` and park until the UI + // dispatches `approval_decide` (or the 10-minute TTL elapses and + // the call is denied). Off by default until the React UI + // (toast + settings panel) lands — otherwise gated tool calls + // would block the agent loop with nothing to release them. + if std::env::var("OPENHUMAN_APPROVAL_GATE") + .map(|v| matches!(v.trim(), "1" | "true" | "TRUE")) + .unwrap_or(false) + { + let (session_id, ephemeral) = match std::env::var("OPENHUMAN_CORE_TOKEN") + .ok() + .filter(|s| !s.is_empty()) + { + Some(token) => (token, false), + None => (format!("session-{}", uuid::Uuid::new_v4()), true), + }; + if ephemeral { + log::debug!( + "[runtime] OPENHUMAN_CORE_TOKEN unset; generated ephemeral session_id={session_id} \ + for approval gate — `approval_list_pending` is session-agnostic so pending rows \ + from prior launches will still be visible, but per-session audit grouping will not \ + correlate across restarts" + ); + } + let _ = + crate::openhuman::approval::ApprovalGate::init_global(cfg.clone(), session_id.clone()); + log::info!( + "[runtime] approval gate installed (OPENHUMAN_APPROVAL_GATE=1, session_id={session_id}) — \ + external-effect tool calls will block until approval_decide" + ); + } else { + log::debug!( + "[runtime] approval gate disabled (OPENHUMAN_APPROVAL_GATE unset) — \ + external-effect tool calls run unsupervised" + ); + } + // --- Session storage layout migration ------------------------------- // One-shot move from `session_raw/{DDMMYYYY}/` (≤ 0.53.4) to the new // flat `session_raw/{stem}.jsonl` layout, plus DDMMYYYY → YYYY_MM_DD diff --git a/src/openhuman/agent/harness/subagent_runner/ops.rs b/src/openhuman/agent/harness/subagent_runner/ops.rs index 123995126..eee38e4e5 100644 --- a/src/openhuman/agent/harness/subagent_runner/ops.rs +++ b/src/openhuman/agent/harness/subagent_runner/ops.rs @@ -1464,17 +1464,52 @@ async fn run_inner_loop( { let args = parse_tool_arguments(&call.arguments); let timeout = crate::openhuman::tool_timeout::tool_execution_timeout_duration(); - match tokio::time::timeout(timeout, tool.execute(args)).await { - Ok(Ok(result)) => { - let raw = result.output(); - if result.is_error { - format!("Error: {raw}") - } else { - raw + // ── External-effect approval gate (#1339) ───── + // Subagents share the same gate as the parent loop; + // see `tool_loop.rs` for the rationale. + let gate_denial: Option = if tool.external_effect_with_args(&args) { + if let Some(gate) = crate::openhuman::approval::ApprovalGate::try_global() { + let summary = + crate::openhuman::approval::summarize_action(&call.name, &args); + let redacted = crate::openhuman::approval::redact_args(&args); + match gate.intercept(&call.name, &summary, redacted).await { + crate::openhuman::approval::GateOutcome::Allow => None, + crate::openhuman::approval::GateOutcome::Deny { reason } => { + tracing::warn!( + tool = call.name.as_str(), + reason = %reason, + "[subagent_runner] approval gate denied tool call" + ); + Some(reason) + } } + } else { + None + } + } else { + None + }; + + if let Some(reason) = gate_denial { + // Prefix as Error so the downstream `call_success` + // computation (`!result_text.starts_with("Error")`) + // marks the denial as a failed tool call in + // progress events and tool_result blocks. + // (CodeRabbit review on PR #2149.) + format!("Error: {reason}") + } else { + match tokio::time::timeout(timeout, tool.execute(args)).await { + Ok(Ok(result)) => { + let raw = result.output(); + if result.is_error { + format!("Error: {raw}") + } else { + raw + } + } + Ok(Err(err)) => format!("Error executing {}: {err}", call.name), + Err(_) => format!("Error: tool '{}' timed out", call.name), } - Ok(Err(err)) => format!("Error executing {}: {err}", call.name), - Err(_) => format!("Error: tool '{}' timed out", call.name), } } else { format!("Unknown tool: {}", call.name) diff --git a/src/openhuman/agent/harness/tool_loop.rs b/src/openhuman/agent/harness/tool_loop.rs index 7ddbef8f0..015ad5d11 100644 --- a/src/openhuman/agent/harness/tool_loop.rs +++ b/src/openhuman/agent/harness/tool_loop.rs @@ -672,6 +672,43 @@ pub(crate) async fn run_tool_call_loop( } } + // ── External-effect approval gate (#1339) ───────── + // Tools whose `external_effect()` returns true route + // through the process-global `ApprovalGate` so the UI + // can prompt the user before `execute()` runs. The gate + // is `None` when supervised mode is disabled or in test + // envs — behavior matches the pre-#1339 path. + if let Some(tool) = tool_opt { + if tool.external_effect_with_args(&call.arguments) { + if let Some(gate) = crate::openhuman::approval::ApprovalGate::try_global() { + let summary = crate::openhuman::approval::summarize_action( + &call.name, + &call.arguments, + ); + let redacted = crate::openhuman::approval::redact_args(&call.arguments); + match gate.intercept(&call.name, &summary, redacted).await { + crate::openhuman::approval::GateOutcome::Allow => {} + crate::openhuman::approval::GateOutcome::Deny { reason } => { + tracing::warn!( + iteration, + tool = call.name.as_str(), + reason = %reason, + "[agent_loop] approval gate denied tool call" + ); + emit_failed_completion(&reason).await; + individual_results.push(reason.clone()); + let _ = writeln!( + tool_results, + "\n{reason}\n", + call.name + ); + continue; + } + } + } + } + } + let result = if let Some(tool) = tool_opt { let tool_deadline = crate::openhuman::tool_timeout::tool_execution_timeout_duration(); diff --git a/src/openhuman/agent/triage/escalation.rs b/src/openhuman/agent/triage/escalation.rs index 0a0bf6702..0a79cf875 100644 --- a/src/openhuman/agent/triage/escalation.rs +++ b/src/openhuman/agent/triage/escalation.rs @@ -85,6 +85,49 @@ pub async fn apply_decision(run: TriageRun, envelope: &TriggerEnvelope) -> anyho "[triage::escalation] dispatching sub-agent" ); + // ── External-effect approval gate (#1339) ───────── + // React / Escalate fire a sub-agent that may call + // external-effect tools on the user's behalf. Catching + // here as well as at tool-loop level lets the user + // decline the whole escalation up-front instead of one + // tool call at a time. The per-tool gate further down + // still applies — defense in depth, not duplication + // (each gate is short-circuited by the session + // allowlist after the first approval). + if let Some(gate) = crate::openhuman::approval::ApprovalGate::try_global() { + let summary = format!( + "triage::{} target={} prompt_chars={}", + action_str, + target, + prompt.chars().count() + ); + let redacted = serde_json::json!({ + "action": action_str, + "target_agent": target, + "external_id": envelope.external_id, + "label": envelope.display_label, + "prompt_chars": prompt.chars().count(), + }); + let tool_key = format!("triage.{}", run.decision.action.as_str()); + match gate.intercept(&tool_key, &summary, redacted).await { + crate::openhuman::approval::GateOutcome::Allow => {} + crate::openhuman::approval::GateOutcome::Deny { reason } => { + tracing::warn!( + action = %action_str, + target_agent = %target, + external_id = %envelope.external_id, + reason = %reason, + "[triage::escalation] approval gate denied dispatch" + ); + events::publish_failed( + envelope, + &format!("approval denied for `{target}`: {reason}"), + ); + return Ok(()); + } + } + } + match dispatch_target_agent(target, prompt).await { Ok(output) => { tracing::info!( diff --git a/src/openhuman/approval/gate.rs b/src/openhuman/approval/gate.rs new file mode 100644 index 000000000..8a99af51b --- /dev/null +++ b/src/openhuman/approval/gate.rs @@ -0,0 +1,387 @@ +//! `ApprovalGate` — middleware between the agent and any tool whose +//! [`crate::openhuman::tools::Tool::external_effect`] returns `true`. +//! +//! Flow (issue #1339): +//! 1. Agent harness calls [`ApprovalGate::intercept`] with the tool +//! name, a redacted JSON of the arguments, and a short summary. +//! 2. Gate checks the session-scoped allowlist (built from prior +//! `ApproveAlwaysForTool` decisions). Hit → `Allow` immediately. +//! 3. Otherwise: persist a row in `pending_approvals`, publish a +//! [`DomainEvent::ApprovalRequested`] event so the UI can pop a +//! toast, and park the call on a `oneshot::Sender` keyed by +//! `request_id`. +//! 4. UI calls `approval_decide` (RPC) which routes through +//! [`ApprovalGate::decide`] → sends the decision on the oneshot. +//! 5. The parked future wakes with the decision and translates it +//! into [`GateOutcome::Allow`] / `Deny`. +//! +//! Sessions: the gate is keyed by a per-launch `session_id` (the +//! per-launch hex bearer the core hands out) for audit grouping. +//! Rows from prior launches are intentionally preserved on init — +//! the issue #1339 acceptance criterion requires they survive +//! restart so the UI can show / dismiss orphans. Decisions on +//! orphan rows update the DB but cannot resume a parked future +//! across processes — no side effect can fire across launches, so +//! the security invariant is preserved without auto-purging. + +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, OnceLock}; +use std::time::Duration; + +use parking_lot::Mutex; +use tokio::sync::oneshot; + +use crate::core::event_bus::{publish_global, DomainEvent}; +use crate::openhuman::config::Config; + +use super::store; +use super::types::{ApprovalDecision, GateOutcome, PendingApproval}; + +/// How long the gate will park a future before timing out and +/// returning `Deny`. 10 minutes matches the default `expires_at` +/// written into the persisted row. +const DEFAULT_APPROVAL_TTL: Duration = Duration::from_secs(60 * 10); + +static GLOBAL_GATE: OnceLock> = OnceLock::new(); + +/// Coordinator for pending approvals. +pub struct ApprovalGate { + config: Config, + session_id: String, + ttl: Duration, + waiters: Mutex>>, + always_allowlist: Mutex>, +} + +impl ApprovalGate { + /// Install the process-global gate. Returns the existing gate if + /// one was already installed (re-install is a no-op so repeated + /// `bootstrap_core_runtime` calls in tests don't panic). + /// + /// Rows from prior launches are intentionally NOT purged on + /// install — the issue #1339 acceptance criterion requires they + /// survive restart so the UI can show / dismiss them. Orphan + /// rows have no live parked future, so a `decide` is a DB-only + /// audit update; no side effect can fire across processes. + pub fn init_global(config: Config, session_id: impl Into) -> Arc { + let session_id = session_id.into(); + if let Some(existing) = GLOBAL_GATE.get() { + return existing.clone(); + } + let gate = Arc::new(ApprovalGate::new(config, session_id, DEFAULT_APPROVAL_TTL)); + let _ = GLOBAL_GATE.set(gate.clone()); + GLOBAL_GATE.get().cloned().unwrap_or(gate) + } + + /// Returns the global gate when installed; tools and harness + /// branches that don't care about supervised mode treat `None` + /// as "no gating". + pub fn try_global() -> Option> { + GLOBAL_GATE.get().cloned() + } + + fn new(config: Config, session_id: String, ttl: Duration) -> Self { + Self { + config, + session_id, + ttl, + waiters: Mutex::new(HashMap::new()), + always_allowlist: Mutex::new(HashSet::new()), + } + } + + /// Intercept a tool call. Blocks until the user decides or the + /// TTL elapses (timeout → `Deny`). + pub async fn intercept( + &self, + tool_name: &str, + action_summary: &str, + args_redacted: serde_json::Value, + ) -> GateOutcome { + // Session-scoped allowlist shortcut — set by prior + // ApproveAlwaysForTool decisions in this launch. + { + let list = self.always_allowlist.lock(); + if list.contains(tool_name) { + tracing::debug!( + tool = tool_name, + "[approval::gate] session-allowlist hit, skipping prompt" + ); + return GateOutcome::Allow; + } + } + + let request_id = uuid::Uuid::new_v4().to_string(); + let now = chrono::Utc::now(); + let expires_at = Some(now + chrono::Duration::from_std(self.ttl).unwrap_or_default()); + let pending = PendingApproval { + request_id: request_id.clone(), + tool_name: tool_name.to_string(), + action_summary: action_summary.to_string(), + args_redacted: args_redacted.clone(), + session_id: self.session_id.clone(), + created_at: now, + expires_at, + }; + + // Register the waiter BEFORE persisting the row so a fast + // `approval_decide` cannot mark the request approved while + // no waiter exists — would otherwise leave the parked call + // to time out and return `Deny` incorrectly. (CodeRabbit + // review on PR #2149.) + let (tx, rx) = oneshot::channel::(); + { + let mut waiters = self.waiters.lock(); + waiters.insert(request_id.clone(), tx); + } + + if let Err(err) = store::insert_pending(&self.config, &pending) { + self.evict_waiter(&request_id); + tracing::error!( + error = %err, + tool = tool_name, + "[approval::gate] failed to persist pending row — failing closed" + ); + return GateOutcome::Deny { + reason: format!( + "Approval gate could not persist the request — denying for safety: {err}" + ), + }; + } + + publish_global(DomainEvent::ApprovalRequested { + request_id: request_id.clone(), + tool_name: tool_name.to_string(), + action_summary: action_summary.to_string(), + args_redacted, + session_id: self.session_id.clone(), + }); + + tracing::info!( + request_id = %request_id, + tool = tool_name, + "[approval::gate] tool call parked, waiting for decision" + ); + + match tokio::time::timeout(self.ttl, rx).await { + Ok(Ok(decision)) => { + tracing::info!( + request_id = %request_id, + tool = tool_name, + decision = decision.as_str(), + "[approval::gate] decision received" + ); + if decision.is_approve() { + GateOutcome::Allow + } else { + GateOutcome::Deny { + reason: format!("User denied '{tool_name}' execution."), + } + } + } + Ok(Err(_canceled)) => { + // Sender dropped — treat as denial so the agent does + // not silently no-op. + tracing::warn!( + request_id = %request_id, + tool = tool_name, + "[approval::gate] decision channel dropped — denying" + ); + let _ = store::decide(&self.config, &request_id, ApprovalDecision::Deny); + GateOutcome::Deny { + reason: format!( + "Approval channel for '{tool_name}' closed before a decision was made." + ), + } + } + Err(_elapsed) => { + self.evict_waiter(&request_id); + let _ = store::decide(&self.config, &request_id, ApprovalDecision::Deny); + tracing::warn!( + request_id = %request_id, + tool = tool_name, + ttl_secs = self.ttl.as_secs(), + "[approval::gate] approval timed out, denying" + ); + GateOutcome::Deny { + reason: format!( + "Approval for '{tool_name}' timed out after {}s.", + self.ttl.as_secs() + ), + } + } + } + } + + /// Apply a user decision. Returns the now-decided + /// [`PendingApproval`] row when one was found. + pub fn decide( + &self, + request_id: &str, + decision: ApprovalDecision, + ) -> anyhow::Result> { + let decided = store::decide(&self.config, request_id, decision)?; + if let Some(row) = &decided { + if decision == ApprovalDecision::ApproveAlwaysForTool { + let mut list = self.always_allowlist.lock(); + list.insert(row.tool_name.clone()); + } + if let Some(tx) = self.take_waiter(request_id) { + let _ = tx.send(decision); + } + publish_global(DomainEvent::ApprovalDecided { + request_id: row.request_id.clone(), + tool_name: row.tool_name.clone(), + decision: decision.as_str().to_string(), + }); + } + Ok(decided) + } + + /// List all undecided rows, including orphans from prior launches. + /// Orphan rows have no live parked future so a `decide` on them + /// updates the DB but cannot resume an action — see [`store::list_pending`]. + pub fn list_pending(&self) -> anyhow::Result> { + store::list_pending(&self.config) + } + + /// Return the session id this gate was installed with (used by + /// RPC handlers for diagnostics). + pub fn session_id(&self) -> &str { + &self.session_id + } + + fn take_waiter(&self, request_id: &str) -> Option> { + let mut waiters = self.waiters.lock(); + waiters.remove(request_id) + } + + fn evict_waiter(&self, request_id: &str) { + let mut waiters = self.waiters.lock(); + waiters.remove(request_id); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + fn test_gate() -> (ApprovalGate, TempDir) { + let dir = TempDir::new().unwrap(); + let config = Config { + workspace_dir: dir.path().to_path_buf(), + ..Config::default() + }; + let session = format!("test-session-{}", uuid::Uuid::new_v4()); + let gate = ApprovalGate::new(config, session, Duration::from_millis(500)); + (gate, dir) + } + + #[tokio::test] + async fn approve_once_returns_allow() { + let (gate, _dir) = test_gate(); + let gate = Arc::new(gate); + + let g = gate.clone(); + let handle = tokio::spawn(async move { + g.intercept("composio", "send slack", serde_json::json!({})) + .await + }); + + // Wait for pending row to land. + let mut tries = 0; + let pending = loop { + let list = gate.list_pending().unwrap(); + if let Some(p) = list.into_iter().next() { + break p; + } + tries += 1; + assert!(tries < 50, "pending row never appeared"); + tokio::time::sleep(Duration::from_millis(10)).await; + }; + + gate.decide(&pending.request_id, ApprovalDecision::ApproveOnce) + .unwrap(); + + let outcome = handle.await.unwrap(); + assert!(matches!(outcome, GateOutcome::Allow)); + } + + #[tokio::test] + async fn deny_returns_deny_with_reason() { + let (gate, _dir) = test_gate(); + let gate = Arc::new(gate); + + let g = gate.clone(); + let handle = tokio::spawn(async move { + g.intercept("pushover", "send push", serde_json::json!({})) + .await + }); + + let pending = loop { + if let Some(p) = gate.list_pending().unwrap().into_iter().next() { + break p; + } + tokio::time::sleep(Duration::from_millis(10)).await; + }; + + gate.decide(&pending.request_id, ApprovalDecision::Deny) + .unwrap(); + + let outcome = handle.await.unwrap(); + match outcome { + GateOutcome::Deny { reason } => assert!(reason.contains("pushover")), + other => panic!("expected deny, got {other:?}"), + } + } + + #[tokio::test] + async fn approve_always_for_tool_short_circuits_future_calls() { + let (gate, _dir) = test_gate(); + let gate = Arc::new(gate); + + let g = gate.clone(); + let first = tokio::spawn(async move { + g.intercept("composio", "first", serde_json::json!({})) + .await + }); + let pending = loop { + if let Some(p) = gate.list_pending().unwrap().into_iter().next() { + break p; + } + tokio::time::sleep(Duration::from_millis(10)).await; + }; + gate.decide(&pending.request_id, ApprovalDecision::ApproveAlwaysForTool) + .unwrap(); + assert!(matches!(first.await.unwrap(), GateOutcome::Allow)); + + // Second call to the same tool must NOT block — allowlist hit. + let outcome = gate + .intercept("composio", "second", serde_json::json!({})) + .await; + assert!(matches!(outcome, GateOutcome::Allow)); + } + + #[tokio::test] + async fn timeout_returns_deny() { + let (gate, _dir) = test_gate(); // TTL = 500ms + let gate = Arc::new(gate); + let outcome = gate + .intercept("composio", "timed out", serde_json::json!({})) + .await; + match outcome { + GateOutcome::Deny { reason } => assert!(reason.contains("timed out")), + other => panic!("expected deny, got {other:?}"), + } + } + + #[tokio::test] + async fn decide_unknown_id_is_noop() { + let (gate, _dir) = test_gate(); + let decided = gate + .decide("does-not-exist", ApprovalDecision::ApproveOnce) + .unwrap(); + assert!(decided.is_none()); + } +} diff --git a/src/openhuman/approval/mod.rs b/src/openhuman/approval/mod.rs index 7de86e67a..317f1d247 100644 --- a/src/openhuman/approval/mod.rs +++ b/src/openhuman/approval/mod.rs @@ -1,7 +1,29 @@ //! Interactive approval workflow for supervised mode. //! -//! Provides a pre-execution hook that prompts the user before tool calls, -//! with session-scoped "Always" allowlists and audit logging. +//! Two layers: +//! +//! - [`ApprovalManager`] (legacy, in [`ops`]) — CLI-only synchronous +//! prompt + in-memory session allowlist + audit log. Still used by +//! the agent harness when running under `--channel cli`. +//! - [`ApprovalGate`] (new, in [`gate`]) — async middleware between the +//! agent and any tool whose [`crate::openhuman::tools::Tool::external_effect`] +//! returns `true`. Persists pending rows in SQLite, parks the +//! tool-call future on a oneshot, and resumes when the UI dispatches +//! `approval_decide`. Introduced for issue #1339 so external-channel +//! writes (Slack post, email send, calendar create, …) cannot fire +//! without explicit user consent. +pub mod gate; pub mod ops; +pub mod redact; +pub mod rpc; +pub mod schemas; +pub mod store; +pub mod types; + +pub use gate::ApprovalGate; pub use ops::*; +pub use redact::{redact_args, summarize_action}; +pub use schemas::all_controller_schemas as all_approval_controller_schemas; +pub use schemas::all_registered_controllers as all_approval_registered_controllers; +pub use types::{ApprovalDecision, GateOutcome, PendingApproval}; diff --git a/src/openhuman/approval/redact.rs b/src/openhuman/approval/redact.rs new file mode 100644 index 000000000..f72a0467f --- /dev/null +++ b/src/openhuman/approval/redact.rs @@ -0,0 +1,326 @@ +//! Argument redaction for approval prompts. +//! +//! Anything written to `pending_approvals` or broadcast on the event +//! bus must be scrubbed first — per +//! `feedback_redact_paths_and_ids_in_public.md` (no `/Users//` +//! paths, no openhuman user_ids) and `feedback_pr_no_chat_content.md` +//! (no raw message bodies, contact names, subjects, addresses — only +//! counts/shape). +//! +//! Approach: walk the JSON value tree and replace any field whose +//! name matches a known PII / chat-content key with a redacted +//! marker `" ( chars)>"`. Unknown fields pass +//! through unchanged so the UI can still show useful context +//! (action slug, tool name, integration id). + +use serde_json::{Map, Value}; + +/// Field names whose values are assumed to contain raw user content +/// or PII and MUST be redacted. Matching is case-insensitive. +const SENSITIVE_KEYS: &[&str] = &[ + "body", + "content", + "text", + "message", + "messages", + "html", + "html_body", + "snippet", + "subject", + "title", + "recipient", + "recipients", + "to", + "cc", + "bcc", + "from", + "sender", + "address", + "email", + "phone", + "contact", + "contacts", + "name", + "first_name", + "last_name", + "full_name", + "channel_name", + "user", + "user_id", + "userid", + "username", + "thread_id", + "thread_ts", + "conversation_id", + "token", + "api_key", + "secret", + "password", + "authorization", + "auth", +]; + +/// Produce a redacted clone of `args` suitable for persistence / +/// broadcast / display. +pub fn redact_args(args: &Value) -> Value { + walk(args) +} + +fn walk(value: &Value) -> Value { + match value { + Value::Object(map) => Value::Object(walk_object(map)), + Value::Array(items) => Value::Array(items.iter().map(walk).collect()), + Value::String(s) => Value::String(scrub_paths(s)), + other => other.clone(), + } +} + +fn walk_object(map: &Map) -> Map { + let mut out = Map::with_capacity(map.len()); + for (k, v) in map { + if is_sensitive_key(k) { + out.insert(k.clone(), redact_value(v)); + } else { + out.insert(k.clone(), walk(v)); + } + } + out +} + +fn is_sensitive_key(key: &str) -> bool { + let lower = key.to_ascii_lowercase(); + SENSITIVE_KEYS.iter().any(|s| s == &lower.as_str()) +} + +fn redact_value(value: &Value) -> Value { + match value { + Value::String(s) => { + Value::String(format!("", s.chars().count())) + } + Value::Array(items) => Value::String(format!("", items.len())), + Value::Object(map) => Value::String(format!("", map.len())), + Value::Number(_) => Value::String("".to_string()), + Value::Bool(_) => Value::String("".to_string()), + Value::Null => Value::Null, + } +} + +/// Strip absolute home paths so the action summary cannot leak the +/// user's username on multi-tenant log shipping. +/// +/// Handles both Unix (`/Users//…`, `/home//…`) and +/// Windows (`C:\Users\\…`) shapes — `MAIN_SEPARATOR` alone +/// would miss the Windows case in a Unix-built artifact looking at +/// log payloads that originated on Windows, or vice versa. +fn scrub_paths(input: &str) -> String { + if !input.contains("Users") && !input.contains("home") { + return input.to_string(); + } + let mut out = String::with_capacity(input.len()); + let mut i = 0; + while i < input.len() { + if let Some(prefix_len) = match_home_prefix(&input[i..]) { + out.push_str(""); + i += prefix_len; + // Skip past the username segment up to the next path + // separator (or end of input). + let rest = &input[i..]; + match rest.find(|c: char| c == '/' || c == '\\') { + Some(end) => i += end, + None => i = input.len(), + } + } else { + // Push one char and advance — char-safe so we don't + // split a multi-byte UTF-8 codepoint. + let ch = input[i..].chars().next().unwrap(); + out.push(ch); + i += ch.len_utf8(); + } + } + out +} + +/// Detects the start of an absolute home path at the front of `s`. +/// Returns the byte length of the marker (so `s[len..]` is the +/// username's first character) when matched, `None` otherwise. +fn match_home_prefix(s: &str) -> Option { + let bytes = s.as_bytes(); + let starts_with_ci = |needle: &str| -> bool { + bytes.len() >= needle.len() && bytes[..needle.len()].eq_ignore_ascii_case(needle.as_bytes()) + }; + if starts_with_ci("/Users/") { + return Some(7); + } + if starts_with_ci("/home/") { + return Some(6); + } + // Windows — accept any drive letter + `:\Users\` + if bytes.len() >= 9 + && bytes[0].is_ascii_alphabetic() + && bytes[1] == b':' + && bytes[2] == b'\\' + && bytes[3..9].eq_ignore_ascii_case(b"Users\\") + { + return Some(9); + } + None +} + +/// Build a short human-readable summary of an approval-bound tool +/// call. Pulls a handful of safe fields (`action`, `tool_slug`, +/// `integration`, etc.) and tacks on a redacted-byte-count hint so +/// the user knows *what* the agent wants to do without exposing the +/// content. +pub fn summarize_action(tool_name: &str, args: &Value) -> String { + let safe_fields: &[&str] = &[ + "action", + "tool_slug", + "action_name", + "integration", + "app", + "provider", + "channel", + "method", + "endpoint", + ]; + let mut parts: Vec = Vec::new(); + if let Value::Object(map) = args { + for key in safe_fields { + if let Some(v) = map.get(*key) { + if let Some(s) = v.as_str() { + parts.push(format!("{key}={s}")); + } + } + } + } + let bytes = serde_json::to_vec(args).map(|b| b.len()).unwrap_or(0); + if parts.is_empty() { + format!("{tool_name} ({bytes} bytes of arguments)") + } else { + format!("{tool_name}({}, {bytes} bytes)", parts.join(", ")) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn sensitive_string_field_is_replaced_with_marker() { + let args = json!({ "body": "hello world", "action": "execute" }); + let red = redact_args(&args); + assert_eq!(red["action"], json!("execute")); + assert!( + red["body"] + .as_str() + .unwrap() + .starts_with("" + ); + } + + #[test] + fn home_path_in_unredacted_string_is_scrubbed() { + let args = json!({ "action": "list", "cwd": "/Users/oxoxdev/work/openhuman" }); + let red = redact_args(&args); + let cwd = red["cwd"].as_str().unwrap(); + assert!(!cwd.contains("oxoxdev"), "got {cwd}"); + assert!(cwd.contains("")); + assert!(cwd.ends_with("/work/openhuman")); + } + + #[test] + fn windows_home_path_is_scrubbed() { + let args = json!({ "action": "list", "cwd": "C:\\Users\\oxoxdev\\work\\openhuman" }); + let red = redact_args(&args); + let cwd = red["cwd"].as_str().unwrap(); + assert!(!cwd.contains("oxoxdev"), "got {cwd}"); + assert!(cwd.contains("")); + assert!(cwd.ends_with("\\work\\openhuman")); + } + + #[test] + fn linux_home_path_is_scrubbed() { + let args = json!({ "action": "list", "cwd": "/home/jane/project" }); + let red = redact_args(&args); + let cwd = red["cwd"].as_str().unwrap(); + assert!(!cwd.contains("jane"), "got {cwd}"); + assert!(cwd.contains("")); + assert!(cwd.ends_with("/project")); + } + + #[test] + fn multiple_home_paths_in_same_string_all_scrubbed() { + let args = json!({ + "action": "list", + "summary": "from /Users/alice/a.txt to /Users/bob/b.txt", + }); + let red = redact_args(&args); + let summary = red["summary"].as_str().unwrap(); + assert!(!summary.contains("alice")); + assert!(!summary.contains("bob")); + assert_eq!(summary.matches("").count(), 2); + } + + #[test] + fn summarize_action_pulls_safe_fields() { + let args = json!({ + "action": "execute", + "tool_slug": "SLACK_SEND", + "params": { "body": "hi" } + }); + let summary = summarize_action("composio", &args); + assert!(summary.contains("composio")); + assert!(summary.contains("action=execute")); + assert!(summary.contains("tool_slug=SLACK_SEND")); + assert!(!summary.contains("hi")); + } + + #[test] + fn summarize_action_falls_back_to_size_only() { + let args = json!({}); + let summary = summarize_action("pushover", &args); + assert!(summary.contains("pushover")); + assert!(summary.contains("bytes")); + } +} diff --git a/src/openhuman/approval/rpc.rs b/src/openhuman/approval/rpc.rs new file mode 100644 index 000000000..591539e56 --- /dev/null +++ b/src/openhuman/approval/rpc.rs @@ -0,0 +1,85 @@ +//! Approval RPC operations. +//! +//! Exposed as `approval_list_pending` and `approval_decide` through +//! the controller registry (see [`super::schemas`]). + +use anyhow::anyhow; + +use crate::rpc::RpcOutcome; + +use super::gate::ApprovalGate; +use super::types::{ApprovalDecision, PendingApproval}; + +/// List rows still awaiting a user decision in the current session. +/// +/// Returns an empty list (not an error) when the gate is not +/// installed — supervised mode may be disabled, in which case there +/// is nothing pending by definition. +pub async fn approval_list_pending() -> anyhow::Result>> { + tracing::debug!("[rpc:approval_list_pending] entry"); + let Some(gate) = ApprovalGate::try_global() else { + tracing::debug!("[rpc:approval_list_pending] gate not installed, returning empty"); + return Ok(RpcOutcome::new(Vec::new(), vec![])); + }; + let rows = match gate.list_pending() { + Ok(rows) => rows, + Err(err) => { + tracing::error!(error = %err, "[rpc:approval_list_pending] store error"); + return Err(err); + } + }; + tracing::debug!(rows = rows.len(), "[rpc:approval_list_pending] exit"); + let log = format!("[approval] list_pending returned {} row(s)", rows.len()); + Ok(RpcOutcome::single_log(rows, log)) +} + +/// Apply a decision to a pending row. Errors when the request id is +/// unknown / already decided / belongs to a different session. +pub async fn approval_decide( + request_id: &str, + decision: ApprovalDecision, +) -> anyhow::Result> { + tracing::debug!( + request_id = request_id, + decision = decision.as_str(), + "[rpc:approval_decide] entry" + ); + let gate = ApprovalGate::try_global().ok_or_else(|| { + tracing::warn!( + request_id = request_id, + "[rpc:approval_decide] gate not installed" + ); + anyhow!("approval gate is not installed; supervised mode disabled") + })?; + let decided = match gate.decide(request_id, decision) { + Ok(row) => row, + Err(err) => { + tracing::error!( + request_id = request_id, + error = %err, + "[rpc:approval_decide] gate decide failed" + ); + return Err(err); + } + }; + let row = decided.ok_or_else(|| { + tracing::warn!( + request_id = request_id, + "[rpc:approval_decide] no pending approval found" + ); + anyhow!("no pending approval found for request_id '{request_id}'") + })?; + tracing::info!( + request_id = row.request_id.as_str(), + tool = row.tool_name.as_str(), + decision = decision.as_str(), + "[rpc:approval_decide] exit" + ); + let log = format!( + "[approval] decided request_id={} tool={} decision={}", + row.request_id, + row.tool_name, + decision.as_str() + ); + Ok(RpcOutcome::single_log(row, log)) +} diff --git a/src/openhuman/approval/schemas.rs b/src/openhuman/approval/schemas.rs new file mode 100644 index 000000000..799d1059b --- /dev/null +++ b/src/openhuman/approval/schemas.rs @@ -0,0 +1,198 @@ +//! Controller schemas + handlers for the `approval` namespace. +//! +//! Wires `approval_list_pending` and `approval_decide` into the +//! global registry consumed by `src/core/all.rs`. + +use serde_json::{Map, Value}; + +use crate::core::all::{ControllerFuture, RegisteredController}; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; +use crate::rpc::RpcOutcome; + +use super::rpc as approval_rpc; +use super::types::ApprovalDecision; + +pub fn all_controller_schemas() -> Vec { + vec![schemas("list_pending"), schemas("decide")] +} + +pub fn all_registered_controllers() -> Vec { + vec![ + RegisteredController { + schema: schemas("list_pending"), + handler: handle_list_pending, + }, + RegisteredController { + schema: schemas("decide"), + handler: handle_decide, + }, + ] +} + +pub fn schemas(function: &str) -> ControllerSchema { + match function { + "list_pending" => ControllerSchema { + namespace: "approval", + function: "list_pending", + description: + "List pending approval requests awaiting a user decision in the current session.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "pending", + ty: TypeSchema::Array(Box::new(TypeSchema::Ref("PendingApproval"))), + comment: "Pending approval rows.", + required: true, + }], + }, + "decide" => ControllerSchema { + namespace: "approval", + function: "decide", + description: + "Apply a decision to a pending approval (approve_once / approve_always_for_tool / deny).", + inputs: vec![ + FieldSchema { + name: "request_id", + ty: TypeSchema::String, + comment: "Identifier of the pending approval to decide.", + required: true, + }, + FieldSchema { + name: "decision", + ty: TypeSchema::String, + comment: + "One of \"approve_once\", \"approve_always_for_tool\", or \"deny\".", + required: true, + }, + ], + outputs: vec![FieldSchema { + name: "decided", + ty: TypeSchema::Ref("PendingApproval"), + comment: "The pending row after the decision was applied.", + required: true, + }], + }, + _ => ControllerSchema { + namespace: "approval", + function: "unknown", + description: "Unknown approval function.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "error", + ty: TypeSchema::String, + comment: "Schema not defined for the requested function.", + required: true, + }], + }, + } +} + +fn handle_list_pending(_params: Map) -> ControllerFuture { + Box::pin(async move { + let outcome = approval_rpc::approval_list_pending() + .await + .map_err(|e| e.to_string())?; + to_json(outcome) + }) +} + +fn handle_decide(params: Map) -> ControllerFuture { + Box::pin(async move { + let request_id = read_required_string(¶ms, "request_id")?; + let decision_str = read_required_string(¶ms, "decision")?; + let decision = ApprovalDecision::from_str(decision_str.trim()).ok_or_else(|| { + format!( + "invalid 'decision': expected approve_once|approve_always_for_tool|deny, got '{decision_str}'" + ) + })?; + let outcome = approval_rpc::approval_decide(request_id.trim(), decision) + .await + .map_err(|e| e.to_string())?; + to_json(outcome) + }) +} + +fn read_required_string(params: &Map, key: &str) -> Result { + match params.get(key) { + Some(Value::String(s)) => Ok(s.clone()), + Some(other) => Err(format!( + "invalid '{key}': expected string, got {}", + type_name(other) + )), + None => Err(format!("missing required param '{key}'")), + } +} + +fn to_json(outcome: RpcOutcome) -> Result { + outcome.into_cli_compatible_json() +} + +fn type_name(value: &Value) -> &'static str { + match value { + Value::Null => "null", + Value::Bool(_) => "bool", + Value::Number(_) => "number", + Value::String(_) => "string", + Value::Array(_) => "array", + Value::Object(_) => "object", + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn schemas_list_pending_has_no_inputs() { + let s = schemas("list_pending"); + assert_eq!(s.namespace, "approval"); + assert_eq!(s.function, "list_pending"); + assert!(s.inputs.is_empty()); + } + + #[test] + fn schemas_decide_requires_request_id_and_decision() { + let s = schemas("decide"); + let names: Vec<_> = s.inputs.iter().map(|f| f.name).collect(); + assert!(names.contains(&"request_id")); + assert!(names.contains(&"decision")); + assert!(s.inputs.iter().all(|f| f.required)); + } + + #[test] + fn schemas_unknown_returns_placeholder() { + let s = schemas("nope"); + assert_eq!(s.function, "unknown"); + assert_eq!(s.outputs[0].name, "error"); + } + + #[test] + fn all_registered_controllers_has_handler_per_schema() { + let controllers = all_registered_controllers(); + assert_eq!(controllers.len(), 2); + let names: Vec<_> = controllers.iter().map(|c| c.schema.function).collect(); + assert_eq!(names, vec!["list_pending", "decide"]); + } + + #[test] + fn read_required_string_returns_value_for_present_key() { + let mut params = Map::new(); + params.insert("request_id".into(), json!("abc")); + let got = read_required_string(¶ms, "request_id").unwrap(); + assert_eq!(got, "abc"); + } + + #[test] + fn read_required_string_rejects_wrong_type() { + let mut params = Map::new(); + params.insert("decision".into(), json!(42)); + let err = read_required_string(¶ms, "decision").unwrap_err(); + assert!(err.contains("expected string")); + } + + #[test] + fn read_required_string_missing_key_errors() { + let err = read_required_string(&Map::new(), "request_id").unwrap_err(); + assert!(err.contains("missing required")); + } +} diff --git a/src/openhuman/approval/store.rs b/src/openhuman/approval/store.rs new file mode 100644 index 000000000..b0dd5188c --- /dev/null +++ b/src/openhuman/approval/store.rs @@ -0,0 +1,324 @@ +//! SQLite persistence for pending approval requests. +//! +//! Pending rows survive core restart so a queued approval is not lost +//! when the user quits before deciding. Each row carries the +//! `session_id` of the launch that queued it (informational — +//! `list_pending` returns every undecided row regardless of session +//! so the UI can audit / dismiss orphans after restart, per the +//! issue #1339 acceptance criterion). +//! +//! Replay safety: a `decide` on an orphan row (process that queued it +//! is gone) updates the DB but cannot resume the parked future — no +//! side effect can fire across processes. `purge_session` is a +//! best-effort cleanup helper kept for an explicit RPC in a follow-up. +//! +//! Follows the same `with_connection` shape as `notifications/store.rs` +//! and `cron/store.rs` — synchronous `rusqlite::Connection` opened per +//! call, schema applied idempotently. + +use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; +use rusqlite::{params, Connection}; + +use crate::openhuman::config::Config; + +use super::types::{ApprovalDecision, PendingApproval}; + +/// SQL schema applied on every `with_connection` call. +const SCHEMA: &str = " +PRAGMA foreign_keys = ON; + +CREATE TABLE IF NOT EXISTS pending_approvals ( + request_id TEXT PRIMARY KEY, + tool_name TEXT NOT NULL, + action_summary TEXT NOT NULL, + args_redacted TEXT NOT NULL, + session_id TEXT NOT NULL, + created_at TEXT NOT NULL, + expires_at TEXT, + decided_at TEXT, + decision TEXT +); +CREATE INDEX IF NOT EXISTS idx_pending_approvals_pending + ON pending_approvals(decided_at); +CREATE INDEX IF NOT EXISTS idx_pending_approvals_session + ON pending_approvals(session_id); +"; + +/// Open (and migrate) the approval DB, then call `f` with a live +/// connection. Mirrors `notifications/store.rs::with_connection`. +fn with_connection(config: &Config, f: impl FnOnce(&Connection) -> Result) -> Result { + let db_path = config.workspace_dir.join("approval").join("approval.db"); + + tracing::trace!( + path = %db_path.display(), + "[approval::store] opening DB connection" + ); + + if let Some(parent) = db_path.parent() { + std::fs::create_dir_all(parent).with_context(|| { + format!( + "[approval::store] failed to create dir {}", + parent.display() + ) + })?; + } + + let conn = Connection::open(&db_path).with_context(|| { + format!( + "[approval::store] failed to open DB at {}", + db_path.display() + ) + })?; + + conn.execute_batch(SCHEMA) + .context("[approval::store] schema migration failed")?; + + f(&conn) +} + +/// Insert a pending row. Caller supplies the `request_id` and +/// `session_id` so the gate can correlate the parked future. +pub fn insert_pending(config: &Config, pending: &PendingApproval) -> Result<()> { + with_connection(config, |conn| { + let args = serde_json::to_string(&pending.args_redacted) + .context("[approval::store] serialize args_redacted")?; + let created = pending.created_at.to_rfc3339(); + let expires = pending.expires_at.map(|t| t.to_rfc3339()); + conn.execute( + "INSERT INTO pending_approvals + (request_id, tool_name, action_summary, args_redacted, + session_id, created_at, expires_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + params![ + pending.request_id, + pending.tool_name, + pending.action_summary, + args, + pending.session_id, + created, + expires, + ], + ) + .context("[approval::store] insert pending row")?; + Ok(()) + }) +} + +/// List all rows with no `decided_at` (still awaiting user input) +/// regardless of which launch queued them. Orphan rows (the gate's +/// in-memory waiter has been dropped — process died between +/// `intercept` and the user's decision) stay visible so the UI can +/// audit / dismiss them after restart, satisfying the issue #1339 +/// acceptance criterion "pending rows survive app restart". +/// +/// `decide` on an orphan row updates the DB and returns the row but +/// the parked tool call is gone — no side effect ever fires, which +/// matches the security invariant. +pub fn list_pending(config: &Config) -> Result> { + with_connection(config, |conn| { + let mut stmt = conn + .prepare( + "SELECT request_id, tool_name, action_summary, args_redacted, + session_id, created_at, expires_at + FROM pending_approvals + WHERE decided_at IS NULL + ORDER BY created_at ASC", + ) + .context("[approval::store] prepare list_pending")?; + let rows = stmt + .query_map(params![], |row| Ok(row_to_pending(row))) + .context("[approval::store] query list_pending")?; + let mut out = Vec::new(); + for r in rows { + out.push(r.context("[approval::store] row decode")??); + } + Ok(out) + }) +} + +/// 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). +pub fn decide( + config: &Config, + request_id: &str, + decision: ApprovalDecision, +) -> Result> { + with_connection(config, |conn| { + let decision_str = decision.as_str(); + let now = Utc::now().to_rfc3339(); + let updated = conn + .execute( + "UPDATE pending_approvals + SET decided_at = ?1, decision = ?2 + WHERE request_id = ?3 AND decided_at IS NULL", + params![now, decision_str, request_id], + ) + .context("[approval::store] update decided")?; + if updated == 0 { + return Ok(None); + } + let mut stmt = conn + .prepare( + "SELECT request_id, tool_name, action_summary, args_redacted, + session_id, created_at, expires_at + FROM pending_approvals WHERE request_id = ?1", + ) + .context("[approval::store] prepare select decided")?; + let mut rows = stmt + .query(params![request_id]) + .context("[approval::store] query decided row")?; + if let Some(row) = rows.next().context("[approval::store] decided row next")? { + Ok(Some(row_to_pending(row)?)) + } else { + Ok(None) + } + }) +} + +/// 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 { + with_connection(config, |conn| { + let removed = conn + .execute( + "DELETE FROM pending_approvals + WHERE session_id = ?1 AND decided_at IS NULL", + params![session_id], + ) + .context("[approval::store] purge_session")?; + Ok(removed) + }) +} + +fn row_to_pending(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let args_str: String = row.get(3)?; + let args_redacted: serde_json::Value = serde_json::from_str(&args_str) + .unwrap_or_else(|_| serde_json::json!({ "_error": "args_redacted not valid JSON" })); + let created_str: String = row.get(5)?; + let expires_opt: Option = row.get(6)?; + Ok(PendingApproval { + request_id: row.get(0)?, + tool_name: row.get(1)?, + action_summary: row.get(2)?, + args_redacted, + session_id: row.get(4)?, + created_at: parse_rfc3339(&created_str), + expires_at: expires_opt.as_deref().map(parse_rfc3339), + }) +} + +fn parse_rfc3339(input: &str) -> DateTime { + DateTime::parse_from_rfc3339(input) + .map(|t| t.with_timezone(&Utc)) + .unwrap_or_else(|_| Utc::now()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::approval::types::{ApprovalDecision, PendingApproval}; + use chrono::Duration; + use serde_json::json; + use tempfile::TempDir; + + fn test_config() -> (Config, TempDir) { + let dir = TempDir::new().unwrap(); + let config = Config { + workspace_dir: dir.path().to_path_buf(), + ..Config::default() + }; + (config, dir) + } + + fn sample(request_id: &str, session_id: &str) -> PendingApproval { + PendingApproval { + request_id: request_id.to_string(), + tool_name: "composio".to_string(), + action_summary: "send slack message (12 chars)".to_string(), + args_redacted: json!({ "action": "execute", "tool_slug": "SLACK_SEND" }), + session_id: session_id.to_string(), + created_at: Utc::now(), + expires_at: Some(Utc::now() + Duration::minutes(10)), + } + } + + #[test] + fn insert_then_list_returns_pending_row() { + let (config, _dir) = test_config(); + insert_pending(&config, &sample("req-1", "sess-A")).unwrap(); + let rows = list_pending(&config).unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].request_id, "req-1"); + assert_eq!(rows[0].tool_name, "composio"); + } + + #[test] + fn list_pending_returns_rows_from_every_session() { + let (config, _dir) = test_config(); + insert_pending(&config, &sample("a", "sess-A")).unwrap(); + insert_pending(&config, &sample("b", "sess-B")).unwrap(); + let rows = list_pending(&config).unwrap(); + assert_eq!( + rows.len(), + 2, + "orphan rows from other sessions must remain visible" + ); + } + + #[test] + fn decide_marks_row_and_excludes_from_pending_list() { + let (config, _dir) = test_config(); + insert_pending(&config, &sample("req-9", "sess-A")).unwrap(); + let decided = decide(&config, "req-9", ApprovalDecision::ApproveOnce) + .unwrap() + .expect("decided row"); + assert_eq!(decided.request_id, "req-9"); + let rows = list_pending(&config).unwrap(); + assert!(rows.is_empty(), "decided rows should not appear in pending"); + } + + #[test] + fn decide_second_time_returns_none() { + let (config, _dir) = test_config(); + insert_pending(&config, &sample("dupe", "sess-A")).unwrap(); + decide(&config, "dupe", ApprovalDecision::Deny).unwrap(); + let again = decide(&config, "dupe", ApprovalDecision::ApproveOnce).unwrap(); + assert!(again.is_none(), "second decide should be a no-op"); + } + + #[test] + fn decide_unknown_id_is_noop() { + let (config, _dir) = test_config(); + let res = decide(&config, "never-existed", ApprovalDecision::Deny).unwrap(); + assert!(res.is_none()); + } + + #[test] + fn purge_session_removes_only_undecided_rows_for_session() { + let (config, _dir) = test_config(); + insert_pending(&config, &sample("p1", "sess-A")).unwrap(); + insert_pending(&config, &sample("p2", "sess-A")).unwrap(); + insert_pending(&config, &sample("p3", "sess-B")).unwrap(); + decide(&config, "p2", ApprovalDecision::ApproveOnce).unwrap(); + let removed = purge_session(&config, "sess-A").unwrap(); + assert_eq!(removed, 1, "only undecided sess-A row should be purged"); + // p2 stays because it is decided; sess-B untouched. + let remaining = list_pending(&config).unwrap(); + assert_eq!(remaining.len(), 1); + assert_eq!(remaining[0].request_id, "p3"); + } + + #[test] + fn pending_row_survives_connection_close() { + let (config, _dir) = test_config(); + insert_pending(&config, &sample("survives", "sess-A")).unwrap(); + // Each `with_connection` opens a fresh handle — re-reading + // proves the row persisted to disk (acceptance criterion: + // pending rows survive app restart). + let rows = list_pending(&config).unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].request_id, "survives"); + } +} diff --git a/src/openhuman/approval/types.rs b/src/openhuman/approval/types.rs new file mode 100644 index 000000000..95695b416 --- /dev/null +++ b/src/openhuman/approval/types.rs @@ -0,0 +1,107 @@ +//! Types shared between the `ApprovalGate`, the SQLite store, and the +//! RPC layer. Kept narrow so the gate, the store, and the RPC ops can +//! evolve independently without circular imports through `mod.rs`. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +/// A tool call that has been intercepted and is awaiting a user +/// decision. Persisted in `pending_approvals` and surfaced to the UI +/// via `approval_list_pending`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PendingApproval { + pub request_id: String, + pub tool_name: String, + /// Short human-readable summary (scrubbed of PII / chat content + /// per `feedback_redact_paths_and_ids_in_public.md`). + pub action_summary: String, + /// Redacted JSON arguments — counts/shape only, no raw message + /// bodies, per `feedback_pr_no_chat_content.md`. + pub args_redacted: serde_json::Value, + pub session_id: String, + pub created_at: DateTime, + pub expires_at: Option>, +} + +/// User's decision on a pending approval. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ApprovalDecision { + /// Run the call this once; future calls of the same tool will be + /// gated again. + ApproveOnce, + /// Run the call AND add the tool to the session-scoped allowlist + /// so subsequent calls of the same tool skip the gate until the + /// session ends or the core restarts. + ApproveAlwaysForTool, + /// Reject the call. The agent receives a structured error string. + Deny, +} + +impl ApprovalDecision { + pub fn as_str(self) -> &'static str { + match self { + Self::ApproveOnce => "approve_once", + Self::ApproveAlwaysForTool => "approve_always_for_tool", + Self::Deny => "deny", + } + } + + #[allow(clippy::should_implement_trait)] + pub fn from_str(s: &str) -> Option { + match s { + "approve_once" => Some(Self::ApproveOnce), + "approve_always_for_tool" => Some(Self::ApproveAlwaysForTool), + "deny" => Some(Self::Deny), + _ => None, + } + } + + pub fn is_approve(self) -> bool { + matches!(self, Self::ApproveOnce | Self::ApproveAlwaysForTool) + } +} + +/// Outcome of routing a tool call through `ApprovalGate::intercept`. +#[derive(Debug, Clone)] +pub enum GateOutcome { + /// Proceed with `tool.execute(args)`. + Allow, + /// Abort the call. The agent sees `reason` in place of a tool + /// result. + Deny { reason: String }, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn approval_decision_round_trips() { + for d in [ + ApprovalDecision::ApproveOnce, + ApprovalDecision::ApproveAlwaysForTool, + ApprovalDecision::Deny, + ] { + assert_eq!(ApprovalDecision::from_str(d.as_str()), Some(d)); + } + } + + #[test] + fn from_str_unknown_decision_is_none() { + assert!(ApprovalDecision::from_str("maybe").is_none()); + } + + #[test] + fn is_approve_true_for_approval_variants_only() { + assert!(ApprovalDecision::ApproveOnce.is_approve()); + assert!(ApprovalDecision::ApproveAlwaysForTool.is_approve()); + assert!(!ApprovalDecision::Deny.is_approve()); + } + + #[test] + fn approval_decision_serializes_as_snake_case() { + let s = serde_json::to_string(&ApprovalDecision::ApproveAlwaysForTool).unwrap(); + assert_eq!(s, "\"approve_always_for_tool\""); + } +} diff --git a/src/openhuman/channels/proactive.rs b/src/openhuman/channels/proactive.rs index f11e5a6e4..1f41f5873 100644 --- a/src/openhuman/channels/proactive.rs +++ b/src/openhuman/channels/proactive.rs @@ -163,6 +163,39 @@ impl EventHandler for ProactiveMessageSubscriber { channel = %key, "[proactive] delivering to active external channel" ); + + // ── External-effect approval gate (#1339) ───── + // Proactive sends to Telegram/Discord/Slack/etc. + // are outbound writes — route through the gate + // before handing off to the channel implementation. + // Web delivery above is internal and exempt. + if let Some(gate) = crate::openhuman::approval::ApprovalGate::try_global() { + let summary = format!( + "proactive-send to {key} ({} chars)", + message.chars().count() + ); + let redacted = serde_json::json!({ + "channel": key, + "source": source.to_string(), + "message_chars": message.chars().count(), + }); + match gate + .intercept("channels.proactive_send", &summary, redacted) + .await + { + crate::openhuman::approval::GateOutcome::Allow => {} + crate::openhuman::approval::GateOutcome::Deny { reason } => { + tracing::warn!( + source = %source, + channel = %key, + reason = %reason, + "[proactive] approval gate denied external delivery" + ); + return; + } + } + } + match ch.send(&SendMessage::new(message, "")).await { Ok(()) => { tracing::debug!( diff --git a/src/openhuman/tools/impl/network/composio.rs b/src/openhuman/tools/impl/network/composio.rs index 0a7ede0b2..b015e74c2 100644 --- a/src/openhuman/tools/impl/network/composio.rs +++ b/src/openhuman/tools/impl/network/composio.rs @@ -554,6 +554,29 @@ impl Tool for ComposioTool { ToolCategory::Skill } + fn external_effect(&self) -> bool { + // Conservative default for the arg-less path: assume any + // composio call is a write so callers that don't reach the + // args-aware override still get gated. The harness uses + // `external_effect_with_args` (below) which inspects + // `action` and lets read-only branches through. + true + } + + fn external_effect_with_args(&self, args: &serde_json::Value) -> bool { + // `action="list"` enumerates available Composio actions — + // a read-only catalog call. `action="connect"` only returns + // an OAuth URL the user then visits manually; the + // subsequent OAuth handoff is its own consent flow so the + // tool call itself has no outbound side effect to gate. + // `action="execute"` (or anything unknown / missing) is the + // write path and routes through the approval gate. + match args.get("action").and_then(|v| v.as_str()) { + Some("list") | Some("connect") => false, + _ => true, + } + } + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { let action = args .get("action") diff --git a/src/openhuman/tools/impl/network/gmail_unsubscribe.rs b/src/openhuman/tools/impl/network/gmail_unsubscribe.rs index 098cf0eb0..7eb82a773 100644 --- a/src/openhuman/tools/impl/network/gmail_unsubscribe.rs +++ b/src/openhuman/tools/impl/network/gmail_unsubscribe.rs @@ -34,6 +34,20 @@ impl Tool for GmailUnsubscribeTool { ToolCategory::Skill } + // Intentionally NOT marked external_effect=true in v1. + // + // `execute()` below does NOT perform the unsubscribe itself — + // it returns a `pending_approval` JSON payload that the React + // UI intercepts and gates with its own legacy confirmation + // flow. Marking this tool external_effect=true would route the + // call through the new `ApprovalGate` AND the legacy UI prompt, + // so users would see two consecutive approval dialogs while the + // real side effect still lived outside core enforcement. + // + // Follow-up #1339-v3: move the actual outbound unsubscribe + // into `execute()`, retire the legacy UI prompt, and then flip + // `external_effect = true` here. + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { let sender = args .get("sender") diff --git a/src/openhuman/tools/impl/system/pushover.rs b/src/openhuman/tools/impl/system/pushover.rs index dac67b65d..ec44d9c2a 100644 --- a/src/openhuman/tools/impl/system/pushover.rs +++ b/src/openhuman/tools/impl/system/pushover.rs @@ -111,6 +111,22 @@ impl Tool for PushoverTool { }) } + // Intentionally NOT marked external_effect=true in v1. + // + // `execute()` below already enforces local policy via + // `security.can_act()` (read-only autonomy block) and + // `security.record_action()` (rate limit). The gate runs + // BEFORE `execute()`, so flipping external_effect would prompt + // the user even for calls that the policy will immediately + // reject — a dead-end approval flow. Leaving this tool at the + // trait default (false) means the local policy stays the + // single approval surface until the gate learns to consult + // policy before parking. + // + // Follow-up #1339-v3: thread the SecurityPolicy into the gate + // pre-check so external_effect can stay true here without + // generating ghost approvals. + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { if !self.security.can_act() { return Ok(ToolResult::error("Action blocked: autonomy is read-only")); diff --git a/src/openhuman/tools/traits.rs b/src/openhuman/tools/traits.rs index 620eeec8b..ab1f7d131 100644 --- a/src/openhuman/tools/traits.rs +++ b/src/openhuman/tools/traits.rs @@ -195,6 +195,36 @@ pub trait Tool: Send + Sync { false } + /// Whether this tool produces an externally-observable side effect + /// (outbound Slack/Telegram/email/calendar/webhook write, etc.). + /// + /// When `true`, the agent harness routes the call through the + /// `ApprovalGate` before `execute()` runs. Local file writes, + /// memory writes, and TTS `reply_speech` stay `false` — they are + /// either reversible inside the user's machine or considered + /// internal per issue #1339. + /// + /// Default: `false`. Override on tools that talk to external + /// services on the user's behalf. + fn external_effect(&self) -> bool { + false + } + + /// Args-aware version of [`Self::external_effect`]. Tools whose + /// classification depends on the call arguments (e.g. the + /// `composio` tool gates `action="execute"` but lets + /// `action="list"` / `action="connect"` flow through unprompted) + /// override this method to peek at `args`. + /// + /// The harness calls this method (not the arg-less variant) at + /// the gate-decision point, so most tools that need per-call + /// gating should override here rather than [`Self::external_effect`]. + /// Default: defer to the arg-less classification so existing + /// overrides keep working without changes. + fn external_effect_with_args(&self, _args: &serde_json::Value) -> bool { + self.external_effect() + } + /// Per-tool cap on the character length of the result body sent /// back to the model. /// @@ -323,6 +353,12 @@ mod tests { assert!(tool.max_result_size_chars().is_none()); } + #[test] + fn default_external_effect_is_false() { + let tool = DummyTool; + assert!(!tool.external_effect()); + } + // ── PermissionLevel ordering ─────────────────────────────────── #[test] diff --git a/src/openhuman/voice/reply_speech.rs b/src/openhuman/voice/reply_speech.rs index a5ca01a9b..b7383f479 100644 --- a/src/openhuman/voice/reply_speech.rs +++ b/src/openhuman/voice/reply_speech.rs @@ -7,6 +7,13 @@ //! Lives in the voice domain because the response is consumed by the //! mascot's lipsync pipeline (`useHumanMascot` → `findActiveFrame` → //! `oculusVisemeToShape`). +//! +//! Approval gate (#1339) classification: **internal**. Reply-speech is +//! the user's own assistant speaking through the user's own speakers +//! — there is no outbound side effect visible to a third party. +//! Coordinate with #1206 voice work: if `reply_speech` is ever wrapped +//! in a `Tool` impl, the `external_effect()` method MUST stay `false` +//! (the trait's default) so the approval gate never prompts on TTS. use log::debug; use reqwest::Method;