feat(approval): gate external-effect tool calls until user approves (#1339) (#2149)

src/core/all.rs

@@ -127,6 +127,8 @@ fn build_registered_controllers() -> Vec<RegisteredController> {
     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<ControllerSchema> {
     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());

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

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

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<String> = 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)

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,
+                                    "<tool_result name=\"{}\">\n{reason}\n</tool_result>",
+                                    call.name
+                                );
+                                continue;
+                            }
+                        }
+                    }
+                }
+            }
+
             let result = if let Some(tool) = tool_opt {
                 let tool_deadline =
                     crate::openhuman::tool_timeout::tool_execution_timeout_duration();

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!(

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<Arc<ApprovalGate>> = OnceLock::new();
+
+/// Coordinator for pending approvals.
+pub struct ApprovalGate {
+    config: Config,
+    session_id: String,
+    ttl: Duration,
+    waiters: Mutex<HashMap<String, oneshot::Sender<ApprovalDecision>>>,
+    always_allowlist: Mutex<HashSet<String>>,
+}
+
+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<String>) -> Arc<ApprovalGate> {
+        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<Arc<ApprovalGate>> {
+        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::<ApprovalDecision>();
+        {
+            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<Option<PendingApproval>> {
+        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<Vec<PendingApproval>> {
+        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<oneshot::Sender<ApprovalDecision>> {
+        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());
+    }
+}

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

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/<name>/`
+//! 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 `"<redacted: <kind> (<n> 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<String, Value>) -> Map<String, Value> {
+    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!("<redacted: string ({} chars)>", s.chars().count()))
+        }
+        Value::Array(items) => Value::String(format!("<redacted: array ({} items)>", items.len())),
+        Value::Object(map) => Value::String(format!("<redacted: object ({} keys)>", map.len())),
+        Value::Number(_) => Value::String("<redacted: number>".to_string()),
+        Value::Bool(_) => Value::String("<redacted: bool>".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/<name>/…`, `/home/<name>/…`) and
+/// Windows (`C:\Users\<name>\…`) 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("<HOME>");
+            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<usize> {
+    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<String> = 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("<redacted: string ("),
+            "got {:?}",
+            red["body"]
+        );
+    }
+
+    #[test]
+    fn nested_sensitive_object_fields_are_redacted() {
+        let args = json!({
+            "action": "execute",
+            "params": {
+                "message": "secret",
+                "channel_id": "C123",
+                "tool_slug": "SLACK_SEND",
+            }
+        });
+        let red = redact_args(&args);
+        let params = red.get("params").unwrap().as_object().unwrap();
+        assert!(params["message"]
+            .as_str()
+            .unwrap()
+            .starts_with("<redacted: string"));
+        assert_eq!(params["channel_id"], json!("C123"));
+        assert_eq!(params["tool_slug"], json!("SLACK_SEND"));
+    }
+
+    #[test]
+    fn case_insensitive_match_on_sensitive_keys() {
+        let args = json!({ "Body": "x", "TOKEN": "y" });
+        let red = redact_args(&args);
+        assert!(red["Body"].as_str().unwrap().starts_with("<redacted"));
+        assert!(red["TOKEN"].as_str().unwrap().starts_with("<redacted"));
+    }
+
+    #[test]
+    fn array_field_redacts_to_count_marker() {
+        let args = json!({ "recipients": ["a@x", "b@y", "c@z"] });
+        let red = redact_args(&args);
+        assert_eq!(
+            red["recipients"].as_str().unwrap(),
+            "<redacted: array (3 items)>"
+        );
+    }
+
+    #[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("<HOME>"));
+        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("<HOME>"));
+        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("<HOME>"));
+        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("<HOME>").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"));
+    }
+}

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<RpcOutcome<Vec<PendingApproval>>> {
+    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<RpcOutcome<PendingApproval>> {
+    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))
+}

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<ControllerSchema> {
+    vec![schemas("list_pending"), schemas("decide")]
+}
+
+pub fn all_registered_controllers() -> Vec<RegisteredController> {
+    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<String, Value>) -> 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<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let request_id = read_required_string(&params, "request_id")?;
+        let decision_str = read_required_string(&params, "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<String, Value>, key: &str) -> Result<String, String> {
+    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<T: serde::Serialize>(outcome: RpcOutcome<T>) -> Result<Value, String> {
+    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(&params, "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(&params, "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"));
+    }
+}

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<T>(config: &Config, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
+    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<Vec<PendingApproval>> {
+    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<Option<PendingApproval>> {
+    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<usize> {
+    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<PendingApproval> {
+    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<String> = 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<Utc> {
+    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");
+    }
+}

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<Utc>,
+    pub expires_at: Option<DateTime<Utc>>,
+}
+
+/// 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<Self> {
+        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\"");
+    }
+}

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!(

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<ToolResult> {
         let action = args
             .get("action")

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<ToolResult> {
         let sender = args
             .get("sender")

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<ToolResult> {
         if !self.security.can_act() {
             return Ok(ToolResult::error("Action blocked: autonomy is read-only"));

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]

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;