From fb43f2f60478ff905b44a46d6354fcce7375e8b0 Mon Sep 17 00:00:00 2001 From: "cyrus@tinyhumans.ai" Date: Wed, 8 Jul 2026 02:25:06 +0530 Subject: [PATCH 1/2] =?UTF-8?q?feat(flows):=20backend=20=E2=80=94=20surfac?= =?UTF-8?q?e=20flow-run=20approvals=20+=20per-flow=20trust=20(correlation,?= =?UTF-8?q?=20notify,=20socket)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A flow run parks a tool call (flow require_approval OR the Supervised-tier network/code escalation) with no thread_id/client_id, so nothing surfaces → silent deadlock. This makes the pending approval correlatable + resolvable, adds per-flow "approve always" trust, and bridges the park to a notification + a socket event the frontend can act on. - approval/types.rs + store.rs: PendingApproval gains source_context {kind:"flow",flow_id, run_id,node_id?}; new source_context column; list_pending_for_flow_run. - flows/ops.rs: FlowRunContext task-local (flow_id,run_id) around the engine run; gate reads it to populate source_context. - approval/gate.rs: new ApproveAlwaysForFlow decision; a flow_tool_trust (flow_id,tool_name) check before parking a Workflow-origin call → auto-Allow if trusted. - approval/store.rs: flow_tool_trust table + insert_flow_trust/is_flow_tool_trusted; approval_decide branches on source_context → insert_flow_trust for ApproveAlwaysForFlow. - gate.rs + socketio.rs + events.rs + channels/host/adapters.rs: on a flow-origin park, emit a `flow-gate-approval` CoreNotification + a `flow_approval_request` socket event (no thread/client needed). Chat/other-origin approval unchanged. Contract for the frontend: approval_decide accepts "approve_once"|"approve_always_for_flow"| "deny"; pending-approval JSON carries source_context; socket `flow_approval_request` {request_id,flow_id,run_id,tool_name,summary}; notification kind `flow-gate-approval`. cargo check/fmt clean; approval:: tests pass. --- src/core/event_bus/events.rs | 32 +- src/core/socketio.rs | 29 ++ src/openhuman/approval/gate.rs | 524 +++++++++++++++++++++++- src/openhuman/approval/mod.rs | 7 +- src/openhuman/approval/rpc.rs | 56 ++- src/openhuman/approval/schemas.rs | 9 +- src/openhuman/approval/store.rs | 222 +++++++++- src/openhuman/approval/types.rs | 137 ++++++- src/openhuman/channels/host/adapters.rs | 9 + src/openhuman/flows/ops.rs | 64 ++- 10 files changed, 1051 insertions(+), 38 deletions(-) diff --git a/src/core/event_bus/events.rs b/src/core/event_bus/events.rs index dc3ace350a..01c2a9294b 100644 --- a/src/core/event_bus/events.rs +++ b/src/core/event_bus/events.rs @@ -563,9 +563,35 @@ pub enum DomainEvent { ApprovalDecided { request_id: String, tool_name: String, - /// `"approve_once"`, `"approve_always_for_tool"`, or `"deny"`. + /// `"approve_once"`, `"approve_always_for_tool"`, + /// `"approve_always_for_flow"`, or `"deny"`. decision: String, }, + /// A `Workflow`-origin tool call parked in the `ApprovalGate` (issue + /// flow-approval-surface, PR2/PR3). Unlike `ApprovalRequested`, this + /// event carries no `thread_id`/`client_id` — a flow run has neither, so + /// the generic chat-routed socket bridge + /// (`channels::providers::web::event_bus::ApprovalSurfaceSubscriber`) + /// silently drops it (that gap was the original silent-deadlock bug). + /// Published by `ApprovalGate::intercept_audited` alongside the existing + /// `ApprovalRequested`, bridged by `core::socketio` directly to a + /// broadcast (not per-room) `flow_approval_request` Socket.IO event so + /// the Workflows UI can surface and resolve the park without polling. + FlowApprovalRequested { + /// Unique id used to correlate the decision back to the parked + /// future — pass to `approval_decide` unchanged. + request_id: String, + /// The `flows::Flow` id whose run parked this call. + flow_id: String, + /// The run's stable identifier (== the tinyflows checkpointer + /// thread id). + run_id: String, + /// Tool name being gated (e.g. `"composio"`). + tool_name: String, + /// Short human-readable summary of the action (redacted, same as + /// `ApprovalRequested::action_summary`). + summary: String, + }, // ── Plan review (interactive plan-mode gate) ──────────────────────── /// An interactive turn parked on a thread-scoped plan the user must @@ -1408,7 +1434,8 @@ impl DomainEvent { Self::ApprovalRequested { .. } | Self::ApprovalDecided { .. } | Self::ApprovalGateOverrideIgnored { .. } - | Self::ApprovalGateDisabled { .. } => "approval", + | Self::ApprovalGateDisabled { .. } + | Self::FlowApprovalRequested { .. } => "approval", Self::PlanReviewRequested { .. } | Self::PlanReviewDecided { .. } => "plan_review", @@ -1545,6 +1572,7 @@ impl DomainEvent { Self::SessionExpired { .. } => "SessionExpired", Self::ApprovalRequested { .. } => "ApprovalRequested", Self::ApprovalDecided { .. } => "ApprovalDecided", + Self::FlowApprovalRequested { .. } => "FlowApprovalRequested", Self::PlanReviewRequested { .. } => "PlanReviewRequested", Self::PlanReviewDecided { .. } => "PlanReviewDecided", Self::ApprovalGateOverrideIgnored { .. } => "ApprovalGateOverrideIgnored", diff --git a/src/core/socketio.rs b/src/core/socketio.rs index fb3c5b7689..82b01dc1e3 100644 --- a/src/core/socketio.rs +++ b/src/core/socketio.rs @@ -1072,6 +1072,35 @@ pub fn spawn_web_channel_bridge(io: SocketIo) { let _ = io_memory_sync.emit("flow:run_progress", &payload); let _ = io_memory_sync.emit("flow_run_progress", &payload); } + // A Workflow-origin tool call parked in the `ApprovalGate` + // (flow-approval-surface, PR2/PR3). Broadcast — not + // room-scoped like `ApprovalRequested`'s `approval_request` + // bridge — because a flow run has no chat thread/client to + // target; the Workflows UI listens process-wide and filters + // by `flow_id`/`run_id` client-side. + crate::core::event_bus::DomainEvent::FlowApprovalRequested { + request_id, + flow_id, + run_id, + tool_name, + summary, + } => { + let payload = serde_json::json!({ + "request_id": request_id, + "flow_id": flow_id, + "run_id": run_id, + "tool_name": tool_name, + "summary": summary, + }); + log::info!( + "[socketio] broadcast flow_approval_request request_id={} flow_id={} run_id={} tool={}", + request_id, + flow_id, + run_id, + tool_name + ); + let _ = io_memory_sync.emit("flow_approval_request", &payload); + } _ => {} } } diff --git a/src/openhuman/approval/gate.rs b/src/openhuman/approval/gate.rs index 14f438b3a5..2aa5dbd28a 100644 --- a/src/openhuman/approval/gate.rs +++ b/src/openhuman/approval/gate.rs @@ -43,7 +43,9 @@ use crate::openhuman::config::Config; use crate::openhuman::security::POLICY_DENIED_MARKER; use super::store; -use super::types::{ApprovalDecision, ExecutionOutcome, GateOutcome, PendingApproval}; +use super::types::{ + ApprovalDecision, ApprovalSourceContext, ExecutionOutcome, GateOutcome, PendingApproval, +}; /// Disambiguates why [`ApprovalGate::decide`] returned `Ok(None)`. See /// [`ApprovalGate::classify_decide_miss`] for the lookup that produces this. @@ -104,6 +106,25 @@ tokio::task_local! { pub static APPROVAL_IN_CALL_CONTEXT: InCallApprovalContext; } +/// Per-run flow context (flow-approval-surface, PR2 of the tinyflows +/// approval-surfacing design). `flows::ops::flows_run` / `flows_resume` +/// scope this around the engine invocation, alongside the existing +/// `Workflow` [`AgentTurnOrigin`](crate::openhuman::agent::turn_origin::AgentTurnOrigin), +/// so a tool call parked from that run can correlate +/// [`PendingApproval::source_context`](super::types::PendingApproval) back to +/// the exact flow + run (the origin alone only carries `flow_id`, not +/// `run_id`). Absent for every non-flow caller — chat, cron, subconscious, +/// CLI never scope this. +#[derive(Clone, Debug)] +pub struct FlowRunContext { + pub flow_id: String, + pub run_id: String, +} + +tokio::task_local! { + pub static APPROVAL_FLOW_RUN_CONTEXT: FlowRunContext; +} + /// Parse a chat reply to a parked approval into a binary decision (v1). Only an /// explicit yes/no answer maps to a decision; anything else returns `None` — the /// web channel treats `None` as "not an answer", cancels the parked turn, and @@ -317,6 +338,44 @@ impl ApprovalGate { // external_effect tool from an unlabelled call site. let origin = turn_origin::current().unwrap_or(AgentTurnOrigin::Unknown); + // Per-flow tool trust shortcut (flow-approval-surface, PR2): a prior + // `ApproveAlwaysForFlow` decision on this exact `(flow_id, tool_name)` + // pair short-circuits to `Allow` for every future Workflow-origin call + // of that tool from that flow — including a `require_approval: true` + // flow and a Supervised-tier `caps.rs::gate_call_for_tier` escalation, + // both of which otherwise force the park below. The trust is scoped to + // the *flow*, never the tool alone, so it cannot leak into a different + // workflow that happens to call the same tool (that stays gated, or + // uses the separate global `autonomy.auto_approve` allowlist). Checked + // before any other origin branching so it wins regardless of which + // arm of the match below would otherwise fire. + if let AgentTurnOrigin::TrustedAutomation { + source: TrustedAutomationSource::Workflow { .. }, + job_id: flow_id, + } = &origin + { + match store::is_flow_tool_trusted(&self.config, flow_id, tool_name) { + Ok(true) => { + tracing::debug!( + tool = tool_name, + flow_id = %flow_id, + "[approval::gate] flow_tool_trust hit — auto-allowing without prompt" + ); + return (GateOutcome::Allow, None); + } + Ok(false) => {} + Err(err) => { + tracing::warn!( + tool = tool_name, + flow_id = %flow_id, + error = %err, + "[approval::gate] flow_tool_trust lookup failed — falling through to \ + normal gating (fail-safe: still gated, not silently allowed)" + ); + } + } + } + // An autonomous goal continuation runs with no user present, so an // irreversible external action must never be auto-allowed — not even via // the `autonomy.auto_approve` allowlist. Skip the shortcut for that @@ -523,6 +582,30 @@ impl ApprovalGate { let now = chrono::Utc::now(); let expires_at = Some(now + chrono::Duration::from_std(self.effective_ttl()).unwrap_or_default()); + + // Correlation context (flow-approval-surface, PR2): a Workflow-origin + // park carries the flow id on the origin itself, but not the run id — + // that comes from the `APPROVAL_FLOW_RUN_CONTEXT` task-local + // `flows::ops::flows_run`/`flows_resume` scope alongside `with_origin`. + // `try_with` returns `Err` for every non-flow caller (chat, cron, + // subconscious, CLI, and even a Workflow origin reached without the + // flows module's scope, which "should never happen" but must not + // panic), so `source_context` stays `None` there — unchanged chat + // behavior. + let source_context = match &origin { + AgentTurnOrigin::TrustedAutomation { + source: TrustedAutomationSource::Workflow { .. }, + job_id: flow_id, + } => APPROVAL_FLOW_RUN_CONTEXT + .try_with(|ctx| ApprovalSourceContext::Flow { + flow_id: flow_id.clone(), + run_id: ctx.run_id.clone(), + node_id: None, + }) + .ok(), + _ => None, + }; + let pending = PendingApproval { request_id: request_id.clone(), tool_name: tool_name.to_string(), @@ -530,6 +613,7 @@ impl ApprovalGate { args_redacted: args_redacted.clone(), created_at: now, expires_at, + source_context: source_context.clone(), }; // Register the waiter BEFORE persisting the row so a fast @@ -592,6 +676,37 @@ impl ApprovalGate { client_id: chat_client_id.clone(), }); + // Flow-origin surface bridge (flow-approval-surface, PR3): a flow run + // has no chat thread/client to route the generic `ApprovalRequested` + // through (both are `None` above, so the web-channel bridge silently + // drops it — see `channels::providers::web::event_bus`'s + // `ApprovalSurfaceSubscriber`), which is exactly the silent-deadlock + // bug this correlation fixes. Broadcast a dedicated + // `flow_approval_request` socket event (no thread/client required, + // unlike the chat path) plus a `CoreNotification` with the three + // flow-scoped decision actions, so the Workflows UI can surface and + // resolve the park without polling. + if let Some(ApprovalSourceContext::Flow { + flow_id, run_id, .. + }) = &source_context + { + tracing::info!( + request_id = %request_id, + flow_id = %flow_id, + run_id = %run_id, + tool = tool_name, + "[approval::gate] flow-origin park — surfacing flow_approval_request + notification" + ); + publish_global(DomainEvent::FlowApprovalRequested { + request_id: request_id.clone(), + flow_id: flow_id.clone(), + run_id: run_id.clone(), + tool_name: tool_name.to_string(), + summary: action_summary.to_string(), + }); + publish_flow_gate_notification(&request_id, flow_id, run_id, tool_name, action_summary); + } + // Voice channel (issue #3513): tell the meeting bus to speak the // approval prompt into the call. if let Some(ic) = in_call_ctx.as_ref() { @@ -826,6 +941,34 @@ impl ApprovalGate { store::list_recent_decisions(&self.config, limit) } + /// List undecided rows correlated with a specific flow run (issue + /// flow-approval-surface, PR2) — lets a dedicated Workflows review + /// surface fetch just the gates blocking one run instead of filtering + /// [`Self::list_pending`] client-side. + pub fn list_pending_for_flow_run( + &self, + flow_id: &str, + run_id: &str, + ) -> anyhow::Result> { + store::list_pending_for_flow_run(&self.config, flow_id, run_id) + } + + /// Grant "approve always for this flow" trust to `(flow_id, tool_name)`. + /// Called by the `approval_decide` RPC handler after an + /// [`ApprovalDecision::ApproveAlwaysForFlow`] decides a flow-origin row — + /// mirrors the RPC-owns-persistence split documented on + /// [`Self::decide`] for `ApproveAlwaysForTool`. + pub fn insert_flow_trust(&self, flow_id: &str, tool_name: &str) -> anyhow::Result<()> { + store::insert_flow_trust(&self.config, flow_id, tool_name) + } + + /// Whether `(flow_id, tool_name)` currently holds "approve always for + /// this flow" trust. Exposed for tests and diagnostics; `intercept_audited` + /// consults [`store::is_flow_tool_trusted`] directly. + pub fn is_flow_tool_trusted(&self, flow_id: &str, tool_name: &str) -> anyhow::Result { + store::is_flow_tool_trusted(&self.config, flow_id, tool_name) + } + /// Return the session id this gate was installed with (used by /// RPC handlers for diagnostics). pub fn session_id(&self) -> &str { @@ -870,6 +1013,78 @@ impl ApprovalGate { } } +/// Wall-clock milliseconds since the Unix epoch, for `CoreNotificationEvent::timestamp_ms`. +fn now_ms() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +/// Surfaces a flow-origin park as a `CoreNotification` (category `Agents`, +/// `kind: "flow-gate-approval"`) with three action buttons matching the +/// [`ApprovalDecision`] variants a flow-scoped approval accepts: +/// `approve_once` / `approve_always_for_flow` / `deny`. Each action's payload +/// carries the same `{kind, request_id, flow_id, tool_name, summary}` shape +/// (plus `run_id`, additive) so the frontend can dispatch straight to +/// `approval_decide` without a second round-trip to fetch the pending row. +/// +/// Mirrors `flows::ops::notify_pending_approval` (the tinyflows-native +/// per-node HITL gate's notification) but is a distinct surface: this one +/// fires from the *tool-call* `ApprovalGate`, not the graph's own +/// `require_approval` gate node. +fn publish_flow_gate_notification( + request_id: &str, + flow_id: &str, + run_id: &str, + tool_name: &str, + summary: &str, +) { + use crate::openhuman::notifications::bus::publish_core_notification; + use crate::openhuman::notifications::types::{ + CoreNotificationAction, CoreNotificationCategory, CoreNotificationEvent, + }; + + const KIND: &str = "flow-gate-approval"; + let base_payload = |action: ApprovalDecision| { + serde_json::json!({ + "kind": KIND, + "request_id": request_id, + "flow_id": flow_id, + "run_id": run_id, + "tool_name": tool_name, + "summary": summary, + "decision": action.as_str(), + }) + }; + + publish_core_notification(CoreNotificationEvent { + id: format!("{KIND}:{request_id}"), + category: CoreNotificationCategory::Agents, + title: "Workflow needs approval".to_string(), + body: format!("\"{tool_name}\" — {summary}"), + deep_link: None, + timestamp_ms: now_ms(), + actions: Some(vec![ + CoreNotificationAction { + action_id: "approve_once".to_string(), + label: "Approve once".to_string(), + payload: Some(base_payload(ApprovalDecision::ApproveOnce)), + }, + CoreNotificationAction { + action_id: "approve_always_for_flow".to_string(), + label: "Always allow for this workflow".to_string(), + payload: Some(base_payload(ApprovalDecision::ApproveAlwaysForFlow)), + }, + CoreNotificationAction { + action_id: "deny".to_string(), + label: "Deny".to_string(), + payload: Some(base_payload(ApprovalDecision::Deny)), + }, + ]), + }); +} + #[cfg(test)] mod tests { use super::*; @@ -1676,4 +1891,311 @@ mod tests { "the prompting call still persists a row" ); } + + // ── flow-approval-surface (source_context, flow_tool_trust, surfacing) ── + + /// A `Workflow`-origin turn for the flow-correlation tests below. + fn flow_origin(flow_id: &str, require_approval: bool) -> AgentTurnOrigin { + AgentTurnOrigin::TrustedAutomation { + job_id: flow_id.to_string(), + source: TrustedAutomationSource::Workflow { require_approval }, + } + } + + #[tokio::test] + async fn flow_origin_park_populates_source_context_with_flow_and_run_id() { + // A `require_approval: true` flow still parks (same shape as before + // this change) but the persisted row must now carry the flow/run + // correlation the `APPROVAL_FLOW_RUN_CONTEXT` task-local supplies — + // the origin alone only carries `flow_id`, not `run_id`. + let (gate, _dir) = test_gate(); + let gate = Arc::new(gate); + + let g = gate.clone(); + let handle = tokio::spawn(async move { + turn_origin::with_origin( + flow_origin("flow-1", true), + APPROVAL_FLOW_RUN_CONTEXT.scope( + FlowRunContext { + flow_id: "flow-1".to_string(), + run_id: "run-1".to_string(), + }, + g.intercept_audited("composio", "post to slack", 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; + }; + + match &pending.source_context { + Some(super::super::types::ApprovalSourceContext::Flow { + flow_id, + run_id, + node_id, + }) => { + assert_eq!(flow_id, "flow-1"); + assert_eq!(run_id, "run-1"); + assert!( + node_id.is_none(), + "node_id is not yet threaded down to the gate" + ); + } + other => panic!("expected Flow source_context, got {other:?}"), + } + + gate.decide(&pending.request_id, ApprovalDecision::Deny) + .unwrap(); + let _ = handle.await.unwrap(); + } + + #[tokio::test] + async fn chat_origin_park_has_no_source_context() { + // Regression guard: the plain chat-routed path (unaffected by this + // change) must never gain a `source_context` — only Workflow-origin + // parks populate it. + let (gate, _dir) = test_gate(); + let gate = Arc::new(gate); + + let g = gate.clone(); + let handle = tokio::spawn(async move { + turn_origin::with_origin( + web_origin(), + APPROVAL_CHAT_CONTEXT.scope( + chat_ctx(), + g.intercept_audited("composio", "send slack", 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; + }; + assert!( + pending.source_context.is_none(), + "chat-origin parks must not carry a source_context" + ); + + gate.decide(&pending.request_id, ApprovalDecision::ApproveOnce) + .unwrap(); + let (outcome, _id) = handle.await.unwrap(); + assert!(matches!(outcome, GateOutcome::Allow)); + } + + #[tokio::test] + async fn flow_tool_trust_auto_allows_before_parking() { + // A prior `ApproveAlwaysForFlow` grant for (flow_id, tool_name) must + // short-circuit to `Allow` even for a `require_approval: true` flow — + // that is the whole point of "approve always for this workflow": no + // pending row is created and the call never parks. + let (gate, _dir) = test_gate(); + store::insert_flow_trust(&gate.config, "flow-trusted", "composio").unwrap(); + + let outcome = turn_origin::with_origin( + flow_origin("flow-trusted", true), + APPROVAL_FLOW_RUN_CONTEXT.scope( + FlowRunContext { + flow_id: "flow-trusted".to_string(), + run_id: "run-1".to_string(), + }, + gate.intercept("composio", "post to slack", serde_json::json!({})), + ), + ) + .await; + + assert!(matches!(outcome, GateOutcome::Allow)); + assert!( + gate.list_pending().unwrap().is_empty(), + "a trusted (flow, tool) pair must not persist a pending row" + ); + + // A different tool on the same trusted flow is unaffected — it still + // parks (TTL-denies on the 2s test gate). + let untrusted_outcome = turn_origin::with_origin( + flow_origin("flow-trusted", true), + APPROVAL_FLOW_RUN_CONTEXT.scope( + FlowRunContext { + flow_id: "flow-trusted".to_string(), + run_id: "run-1".to_string(), + }, + gate.intercept("pushover", "send push", serde_json::json!({})), + ), + ) + .await; + assert!( + matches!(untrusted_outcome, GateOutcome::Deny { .. }), + "trust must be scoped to the exact tool granted, not the whole flow" + ); + } + + #[tokio::test] + async fn decide_approve_always_for_flow_then_insert_flow_trust_composes_to_auto_allow() { + // Exercises the two building blocks the `approval_decide` RPC handler + // composes for `ApproveAlwaysForFlow` (see `approval::rpc`): the gate + // resolves the parked call and returns the decided row (carrying + // `source_context`), and the RPC layer then calls + // `ApprovalGate::insert_flow_trust` using that row's flow id. This + // test exercises both steps directly against a local (non-global) + // gate — the RPC handler itself reads the process-wide + // `ApprovalGate::try_global()` singleton, which tests must not touch + // (it would leak state into every other test in this binary). + let (gate, _dir) = test_gate(); + let gate = Arc::new(gate); + + let g = gate.clone(); + let handle = tokio::spawn(async move { + turn_origin::with_origin( + flow_origin("flow-2", true), + APPROVAL_FLOW_RUN_CONTEXT.scope( + FlowRunContext { + flow_id: "flow-2".to_string(), + run_id: "run-2".to_string(), + }, + g.intercept_audited("composio", "post to slack", 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; + }; + + let decided = gate + .decide(&pending.request_id, ApprovalDecision::ApproveAlwaysForFlow) + .unwrap() + .expect("decided row"); + + assert!(!gate.is_flow_tool_trusted("flow-2", "composio").unwrap()); + + match &decided.source_context { + Some(super::super::types::ApprovalSourceContext::Flow { flow_id, .. }) => { + gate.insert_flow_trust(flow_id, &decided.tool_name).unwrap(); + } + other => panic!("expected Flow source_context, got {other:?}"), + } + + assert!(gate.is_flow_tool_trusted("flow-2", "composio").unwrap()); + + let (outcome, _id) = handle.await.unwrap(); + assert!(matches!(outcome, GateOutcome::Allow)); + } + + #[tokio::test] + async fn flow_origin_park_publishes_flow_approval_request_and_notification() { + // The silent-deadlock bug this whole PR fixes: a flow-origin park has + // no chat thread/client, so the generic `ApprovalRequested` event's + // web-channel bridge silently drops it. This test asserts the two new + // surfaces fire instead — the `flow_approval_request` DomainEvent + // (bridged to a broadcast Socket.IO event by `core::socketio`) and + // the `flow-gate-approval` CoreNotification with its three actions. + crate::core::event_bus::init_global(crate::core::event_bus::DEFAULT_CAPACITY); + let mut event_rx = crate::core::event_bus::global() + .expect("event bus initialized above") + .raw_receiver(); + let mut notif_rx = crate::openhuman::notifications::bus::subscribe_core_notifications(); + + let (gate, _dir) = test_gate(); + let gate = Arc::new(gate); + + let g = gate.clone(); + let handle = tokio::spawn(async move { + turn_origin::with_origin( + flow_origin("flow-9", true), + APPROVAL_FLOW_RUN_CONTEXT.scope( + FlowRunContext { + flow_id: "flow-9".to_string(), + run_id: "run-9".to_string(), + }, + g.intercept_audited("composio", "post to slack", serde_json::json!({})), + ), + ) + .await + }); + + let (request_id, run_id, tool_name) = tokio::time::timeout( + Duration::from_secs(5), + find_flow_approval_requested(&mut event_rx, "flow-9"), + ) + .await + .expect("timed out waiting for FlowApprovalRequested"); + assert_eq!(run_id, "run-9"); + assert_eq!(tool_name, "composio"); + + let notif = tokio::time::timeout( + Duration::from_secs(5), + find_flow_gate_notification(&mut notif_rx, &request_id), + ) + .await + .expect("timed out waiting for the flow-gate-approval notification"); + assert_eq!(notif.id, format!("flow-gate-approval:{request_id}")); + let actions = notif.actions.expect("notification must declare actions"); + let action_ids: Vec<_> = actions.iter().map(|a| a.action_id.as_str()).collect(); + assert_eq!( + action_ids, + vec!["approve_once", "approve_always_for_flow", "deny"] + ); + + gate.decide(&request_id, ApprovalDecision::Deny).unwrap(); + let _ = handle.await.unwrap(); + } + + /// Drain `rx` (tolerating broadcast lag from unrelated concurrent + /// publishers on the process-wide bus) until a `FlowApprovalRequested` + /// arrives. + async fn find_flow_approval_requested( + rx: &mut tokio::sync::broadcast::Receiver, + ) -> (String, String, String, String) { + loop { + match rx.recv().await { + Ok(crate::core::event_bus::DomainEvent::FlowApprovalRequested { + request_id, + flow_id, + run_id, + tool_name, + .. + }) => return (request_id, flow_id, run_id, tool_name), + Ok(_) => continue, + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + panic!("event bus closed before FlowApprovalRequested arrived") + } + } + } + } + + /// Drain `rx` until the `flow-gate-approval` notification for + /// `request_id` arrives — the notification bus is process-wide, so + /// unrelated notifications from other concurrently-running tests are + /// tolerated and skipped. + async fn find_flow_gate_notification( + rx: &mut tokio::sync::broadcast::Receiver< + crate::openhuman::notifications::types::CoreNotificationEvent, + >, + request_id: &str, + ) -> crate::openhuman::notifications::types::CoreNotificationEvent { + let expected_id = format!("flow-gate-approval:{request_id}"); + loop { + match rx.recv().await { + Ok(event) if event.id == expected_id => return event, + Ok(_) => continue, + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + panic!("notification bus closed before the flow-gate-approval notification arrived") + } + } + } + } } diff --git a/src/openhuman/approval/mod.rs b/src/openhuman/approval/mod.rs index 55472a34f4..b575f525fb 100644 --- a/src/openhuman/approval/mod.rs +++ b/src/openhuman/approval/mod.rs @@ -21,12 +21,13 @@ pub mod store; pub mod types; pub use gate::{ - parse_approval_reply, ApprovalChatContext, ApprovalGate, InCallApprovalContext, - APPROVAL_CHAT_CONTEXT, APPROVAL_IN_CALL_CONTEXT, + parse_approval_reply, ApprovalChatContext, ApprovalGate, FlowRunContext, InCallApprovalContext, + APPROVAL_CHAT_CONTEXT, APPROVAL_FLOW_RUN_CONTEXT, APPROVAL_IN_CALL_CONTEXT, }; 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::{ - ApprovalAuditEntry, ApprovalDecision, ExecutionOutcome, GateOutcome, PendingApproval, + ApprovalAuditEntry, ApprovalDecision, ApprovalSourceContext, ExecutionOutcome, GateOutcome, + PendingApproval, }; diff --git a/src/openhuman/approval/rpc.rs b/src/openhuman/approval/rpc.rs index 4298adef12..a73fb5f161 100644 --- a/src/openhuman/approval/rpc.rs +++ b/src/openhuman/approval/rpc.rs @@ -8,7 +8,7 @@ use anyhow::anyhow; use crate::rpc::RpcOutcome; use super::gate::{try_boot_state, ApprovalGate, ApprovalGateBootState, DecideMiss}; -use super::types::{ApprovalAuditEntry, ApprovalDecision, PendingApproval}; +use super::types::{ApprovalAuditEntry, ApprovalDecision, ApprovalSourceContext, PendingApproval}; /// Read the host-aware approval-gate boot decision so the UI banner can /// render the right state on first paint (rather than waiting for a @@ -189,6 +189,60 @@ pub async fn approval_decide( } } + // "Approve always for this flow" (flow-approval-surface, PR2): unlike + // `ApproveAlwaysForTool` above, this grant is scoped to one workflow — + // persisted into `flow_tool_trust`, NOT the global `autonomy.auto_approve` + // allowlist, so it cannot leak into a different flow that happens to call + // the same tool. Only meaningful on a row whose `source_context` names a + // flow; a row without one (decision picked on a non-flow park, which the + // UI should never offer but the RPC must not trust blindly) logs a + // warning and no-ops the persistence — the decision itself still + // resolved the parked call above, so the RPC does not fail. + if decision == ApprovalDecision::ApproveAlwaysForFlow { + match &row.source_context { + Some(ApprovalSourceContext::Flow { flow_id, .. }) => { + match gate.insert_flow_trust(flow_id, &row.tool_name) { + Ok(()) => { + tracing::info!( + flow_id = flow_id.as_str(), + tool = row.tool_name.as_str(), + "[rpc:approval_decide] tool persisted to flow_tool_trust" + ); + logs.push(format!( + "[approval] '{}' added to '{}'s Always-allow list", + row.tool_name, flow_id + )); + } + Err(err) => { + tracing::warn!( + flow_id = flow_id.as_str(), + tool = row.tool_name.as_str(), + error = %err, + "[rpc:approval_decide] failed to persist flow_tool_trust; tool will prompt again next run" + ); + logs.push(format!( + "[approval] WARNING: could not save 'Always allow' for '{}' on flow '{flow_id}': {err}", + row.tool_name + )); + } + } + } + None => { + tracing::warn!( + request_id = row.request_id.as_str(), + tool = row.tool_name.as_str(), + "[rpc:approval_decide] ApproveAlwaysForFlow decided a row with no flow \ + source_context — nothing to persist, decision still resolved" + ); + logs.push(format!( + "[approval] WARNING: '{}' has no flow context — 'Always allow for this workflow' \ + was not persisted", + row.tool_name + )); + } + } + } + tracing::info!( request_id = row.request_id.as_str(), tool = row.tool_name.as_str(), diff --git a/src/openhuman/approval/schemas.rs b/src/openhuman/approval/schemas.rs index bc21dd378a..3eea77e6d7 100644 --- a/src/openhuman/approval/schemas.rs +++ b/src/openhuman/approval/schemas.rs @@ -91,7 +91,7 @@ pub fn schemas(function: &str) -> ControllerSchema { namespace: "approval", function: "decide", description: - "Apply a decision to a pending approval (approve_once / approve_always_for_tool / deny).", + "Apply a decision to a pending approval (approve_once / approve_always_for_tool / approve_always_for_flow / deny).", inputs: vec![ FieldSchema { name: "request_id", @@ -103,7 +103,9 @@ pub fn schemas(function: &str) -> ControllerSchema { name: "decision", ty: TypeSchema::String, comment: - "One of \"approve_once\", \"approve_always_for_tool\", or \"deny\".", + "One of \"approve_once\", \"approve_always_for_tool\", \ + \"approve_always_for_flow\" (only meaningful when the row's \ + source_context names a flow), or \"deny\".", required: true, }, ], @@ -163,7 +165,8 @@ fn handle_decide(params: Map) -> ControllerFuture { 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}'" + "invalid 'decision': expected \ + approve_once|approve_always_for_tool|approve_always_for_flow|deny, got '{decision_str}'" ) })?; let outcome = approval_rpc::approval_decide(request_id.trim(), decision) diff --git a/src/openhuman/approval/store.rs b/src/openhuman/approval/store.rs index 06ef8a9427..f19454a715 100644 --- a/src/openhuman/approval/store.rs +++ b/src/openhuman/approval/store.rs @@ -30,7 +30,9 @@ use rusqlite::{params, types::Type, Connection}; use crate::openhuman::config::Config; use crate::openhuman::memory_store::safety::sanitize_text; -use super::types::{ApprovalAuditEntry, ApprovalDecision, ExecutionOutcome, PendingApproval}; +use super::types::{ + ApprovalAuditEntry, ApprovalDecision, ApprovalSourceContext, ExecutionOutcome, PendingApproval, +}; /// SQL schema applied on every `with_connection` call. /// @@ -55,12 +57,25 @@ CREATE TABLE IF NOT EXISTS pending_approvals ( decision TEXT, executed_at TEXT, execution_outcome TEXT, - execution_error TEXT + execution_error TEXT, + source_context 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); + +-- Per-flow tool trust (flow-approval-surface, issue B-flows-approval PR2): +-- an `ApproveAlwaysForFlow` decision inserts a row here instead of the +-- global `autonomy.auto_approve` allowlist, so the grant is scoped to one +-- workflow's runs (including scheduled/triggered ones) rather than every +-- flow that happens to call the same tool. +CREATE TABLE IF NOT EXISTS flow_tool_trust ( + flow_id TEXT NOT NULL, + tool_name TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (flow_id, tool_name) +); "; /// Idempotently add the post-execution audit columns to an existing @@ -94,6 +109,10 @@ fn migrate_columns(conn: &Connection) -> Result<()> { "execution_error", "ALTER TABLE pending_approvals ADD COLUMN execution_error TEXT", ), + ( + "source_context", + "ALTER TABLE pending_approvals ADD COLUMN source_context TEXT", + ), ] { if !have.contains(col) { conn.execute(ddl, params![]) @@ -187,11 +206,17 @@ pub fn insert_pending(config: &Config, pending: &PendingApproval, session_id: &s .context("[approval::store] serialize args_redacted")?; let created = pending.created_at.to_rfc3339(); let expires = pending.expires_at.map(|t| t.to_rfc3339()); + let source_context = pending + .source_context + .as_ref() + .map(serde_json::to_string) + .transpose() + .context("[approval::store] serialize source_context")?; 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)", + session_id, created_at, expires_at, source_context) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", params![ pending.request_id, pending.tool_name, @@ -200,6 +225,7 @@ pub fn insert_pending(config: &Config, pending: &PendingApproval, session_id: &s session_id, created, expires, + source_context, ], ) .context("[approval::store] insert pending row")?; @@ -229,7 +255,7 @@ pub fn list_pending(config: &Config) -> Result> { let mut stmt = conn .prepare( "SELECT request_id, tool_name, action_summary, args_redacted, - session_id, created_at, expires_at + session_id, created_at, expires_at, source_context FROM pending_approvals WHERE decided_at IS NULL ORDER BY created_at ASC", @@ -300,7 +326,7 @@ pub fn decide( let mut stmt = conn .prepare( "SELECT request_id, tool_name, action_summary, args_redacted, - session_id, created_at, expires_at + session_id, created_at, expires_at, source_context FROM pending_approvals WHERE request_id = ?1", ) .context("[approval::store] prepare select decided")?; @@ -419,6 +445,63 @@ pub fn purge_session(config: &Config, session_id: &str) -> Result { }) } +/// Filter [`list_pending`] down to the rows correlated with a specific flow +/// run (`source_context == Flow { flow_id, run_id, .. }`). The table has no +/// dedicated index for this — pending rows are always few (parked, awaiting +/// a live decision), so a full scan + JSON-decode filter in Rust is simpler +/// than a JSON1 SQL predicate and avoids a SQLite extension dependency. +pub fn list_pending_for_flow_run( + config: &Config, + flow_id: &str, + run_id: &str, +) -> Result> { + let all = list_pending(config)?; + Ok(all + .into_iter() + .filter(|row| { + matches!( + &row.source_context, + Some(ApprovalSourceContext::Flow { flow_id: f, run_id: r, .. }) + if f == flow_id && r == run_id + ) + }) + .collect()) +} + +/// Grant "approve always for this flow" trust to a `(flow_id, tool_name)` +/// pair — inserted when the user picks `ApproveAlwaysForFlow` on a +/// flow-origin park. `INSERT OR IGNORE` makes re-granting an already-trusted +/// pair a harmless no-op rather than a primary-key error. +pub fn insert_flow_trust(config: &Config, flow_id: &str, tool_name: &str) -> Result<()> { + with_connection(config, |conn| { + conn.execute( + "INSERT OR IGNORE INTO flow_tool_trust (flow_id, tool_name, created_at) + VALUES (?1, ?2, ?3)", + params![flow_id, tool_name, Utc::now().to_rfc3339()], + ) + .context("[approval::store] insert_flow_trust")?; + Ok(()) + }) +} + +/// Whether `(flow_id, tool_name)` was previously granted "approve always for +/// this flow" trust. Consulted by [`super::gate::ApprovalGate::intercept_audited`] +/// before parking a `Workflow`-origin tool call. +pub fn is_flow_tool_trusted(config: &Config, flow_id: &str, tool_name: &str) -> Result { + with_connection(config, |conn| { + let exists: bool = conn + .query_row( + "SELECT EXISTS( + SELECT 1 FROM flow_tool_trust WHERE flow_id = ?1 AND tool_name = ?2 + )", + params![flow_id, tool_name], + |row| row.get(0), + ) + .context("[approval::store] is_flow_tool_trusted")?; + Ok(exists) + }) +} + fn expire_stale_with_now(conn: &Connection, now: DateTime) -> Result { let now_rfc3339 = now.to_rfc3339(); let deny = ApprovalDecision::Deny.as_str(); @@ -485,6 +568,23 @@ fn row_to_pending(row: &rusqlite::Row<'_>) -> rusqlite::Result let args_redacted = serde_json::from_str(&args_str).unwrap_or(serde_json::Value::Null); let created_str: String = row.get(5)?; let expires_opt: Option = row.get(6)?; + // Column 7 (`source_context`) is absent on rows written before this + // migration and on every plain chat-routed park — tolerate both a + // missing column read error and a NULL value as "no context" rather + // than failing the whole row decode (older SELECTs on a freshly + // migrated DB may race the column add on some SQLite builds). + let source_context_str: Option = row.get(7).unwrap_or(None); + let source_context = source_context_str.as_deref().and_then(|raw| { + serde_json::from_str::(raw) + .map_err(|err| { + tracing::warn!( + error = %err, + "[approval::store] failed to decode source_context JSON — treating as absent" + ); + err + }) + .ok() + }); // Note: column index 4 (`session_id`) is read on the SELECT but // intentionally not surfaced — see `PendingApproval` doc-comment. @@ -495,6 +595,7 @@ fn row_to_pending(row: &rusqlite::Row<'_>) -> rusqlite::Result args_redacted, created_at: parse_rfc3339(&created_str), expires_at: expires_opt.as_deref().map(parse_rfc3339), + source_context, }) } @@ -546,6 +647,7 @@ mod tests { args_redacted: json!({ "action": "execute", "tool_slug": "SLACK_SEND" }), created_at: Utc::now(), expires_at, + source_context: None, } } @@ -1160,4 +1262,112 @@ mod tests { "unexpected error: {err}" ); } + + // ── source_context / flow_tool_trust (flow-approval-surface, PR2) ────── + + fn flow_sample(request_id: &str, flow_id: &str, run_id: &str) -> PendingApproval { + PendingApproval::new( + request_id, + "composio", + "send slack message", + json!({ "action": "execute" }), + Some(Utc::now() + Duration::minutes(10)), + ) + .with_source_context(ApprovalSourceContext::Flow { + flow_id: flow_id.to_string(), + run_id: run_id.to_string(), + node_id: None, + }) + } + + #[test] + fn insert_pending_round_trips_flow_source_context() { + let (config, _dir) = test_config(); + let pending = flow_sample("flow-req-1", "flow-1", "run-1"); + insert_pending(&config, &pending, "sess-A").unwrap(); + + let rows = list_pending(&config).unwrap(); + assert_eq!(rows.len(), 1); + match &rows[0].source_context { + Some(ApprovalSourceContext::Flow { + flow_id, run_id, .. + }) => { + assert_eq!(flow_id, "flow-1"); + assert_eq!(run_id, "run-1"); + } + other => panic!("expected Flow source_context, got {other:?}"), + } + } + + #[test] + fn insert_pending_without_source_context_round_trips_none() { + let (config, _dir) = test_config(); + insert_pending(&config, &sample("chat-req-1", "sess-A"), "sess-A").unwrap(); + let rows = list_pending(&config).unwrap(); + assert_eq!(rows.len(), 1); + assert!( + rows[0].source_context.is_none(), + "chat-routed rows must not carry a source_context" + ); + } + + #[test] + fn decide_preserves_source_context_on_the_returned_row() { + let (config, _dir) = test_config(); + let pending = flow_sample("flow-req-2", "flow-2", "run-2"); + insert_pending(&config, &pending, "sess-A").unwrap(); + + let decided = decide( + &config, + "flow-req-2", + ApprovalDecision::ApproveAlwaysForFlow, + ) + .unwrap() + .expect("decided row"); + match decided.source_context { + Some(ApprovalSourceContext::Flow { + flow_id, run_id, .. + }) => { + assert_eq!(flow_id, "flow-2"); + assert_eq!(run_id, "run-2"); + } + other => panic!("expected Flow source_context on decided row, got {other:?}"), + } + } + + #[test] + fn list_pending_for_flow_run_filters_to_the_matching_flow_and_run() { + let (config, _dir) = test_config(); + insert_pending(&config, &flow_sample("a", "flow-1", "run-1"), "sess-A").unwrap(); + insert_pending(&config, &flow_sample("b", "flow-1", "run-2"), "sess-A").unwrap(); + insert_pending(&config, &flow_sample("c", "flow-2", "run-1"), "sess-A").unwrap(); + insert_pending(&config, &sample("d", "sess-A"), "sess-A").unwrap(); + + let rows = list_pending_for_flow_run(&config, "flow-1", "run-1").unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].request_id, "a"); + } + + #[test] + fn flow_tool_trust_round_trips() { + let (config, _dir) = test_config(); + assert!(!is_flow_tool_trusted(&config, "flow-1", "composio").unwrap()); + + insert_flow_trust(&config, "flow-1", "composio").unwrap(); + + assert!(is_flow_tool_trusted(&config, "flow-1", "composio").unwrap()); + // A different tool on the same flow, or the same tool on a different + // flow, must not be trusted by the one grant. + assert!(!is_flow_tool_trusted(&config, "flow-1", "pushover").unwrap()); + assert!(!is_flow_tool_trusted(&config, "flow-2", "composio").unwrap()); + } + + #[test] + fn insert_flow_trust_is_idempotent() { + let (config, _dir) = test_config(); + insert_flow_trust(&config, "flow-1", "composio").unwrap(); + // A second grant of the same pair must not error (INSERT OR IGNORE). + insert_flow_trust(&config, "flow-1", "composio").unwrap(); + assert!(is_flow_tool_trusted(&config, "flow-1", "composio").unwrap()); + } } diff --git a/src/openhuman/approval/types.rs b/src/openhuman/approval/types.rs index c5e8c46c32..94277d2d0a 100644 --- a/src/openhuman/approval/types.rs +++ b/src/openhuman/approval/types.rs @@ -29,6 +29,12 @@ pub struct PendingApproval { pub args_redacted: serde_json::Value, pub created_at: DateTime, pub expires_at: Option>, + /// Correlation context for a park raised from a non-chat source (e.g. a + /// `tinyflows` workflow run). `None` for the plain chat-routed path (and + /// for every other caller that never scopes a matching context) — kept + /// optional and additive so the chat path's wire shape never changes. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub source_context: Option, } impl PendingApproval { @@ -48,8 +54,46 @@ impl PendingApproval { args_redacted, created_at: Utc::now(), expires_at, + source_context: None, } } + + /// Attach an [`ApprovalSourceContext`] — used by + /// [`super::gate::ApprovalGate`] when parking a `Workflow`-origin tool + /// call so the row (and the `approval_list_pending` JSON the frontend + /// reads) carries the flow/run correlation. + pub fn with_source_context(mut self, ctx: ApprovalSourceContext) -> Self { + self.source_context = Some(ctx); + self + } +} + +/// Correlation context for a [`PendingApproval`] raised from a source other +/// than a live chat turn. Serialized internally-tagged on `kind` so the +/// frontend can discriminate on a single field: +/// `{ "kind": "flow", "flow_id": ..., "run_id": ..., "node_id": ... }`. +/// +/// Currently only the `Flow` source exists (a `tinyflows` workflow run, +/// issue B2/flow-approval-surface) — widen this enum if another correlated, +/// non-chat approval surface needs the same treatment. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ApprovalSourceContext { + Flow { + /// The `flows::Flow` id whose run parked this tool call. + flow_id: String, + /// The run's stable identifier — the tinyflows checkpointer thread + /// id (`flows::store::FlowRun.id` / `flows::ops::flows_run`'s + /// `thread_id`) — so a decision (or a per-flow "approve always" + /// trust grant) can be tied back to the exact run. + run_id: String, + /// The graph node that dispatched the call, when known. Not yet + /// populated by the gate (the tool-call dispatch path doesn't thread + /// a node id down to `ApprovalGate::intercept_audited` today) — + /// reserved for a follow-up that does. + #[serde(default, skip_serializing_if = "Option::is_none")] + node_id: Option, + }, } /// Durable audit row for an approval request after a decision. @@ -80,6 +124,18 @@ pub enum ApprovalDecision { /// so subsequent calls of the same tool skip the gate until the /// session ends or the core restarts. ApproveAlwaysForTool, + /// Run the call AND trust this exact `(flow_id, tool_name)` pair for the + /// flow the pending row's `source_context` names — see + /// [`ApprovalSourceContext::Flow`]. Unlike [`Self::ApproveAlwaysForTool`] + /// (a global, cross-flow allowlist), this grant is scoped to one + /// workflow: subsequent runs of *that* flow (including scheduled/ + /// triggered ones) auto-allow *that* tool without prompting, but a new + /// tool the flow starts calling still parks. Persisted via + /// `approval::store::insert_flow_trust`, not `autonomy.auto_approve`. + /// Only meaningful when `source_context` is `Some(Flow { .. })` — the + /// `approval_decide` RPC handler logs and no-ops the persistence step + /// otherwise (the decision itself still resolves the parked call). + ApproveAlwaysForFlow, /// Reject the call. The agent receives a structured error string. Deny, } @@ -89,6 +145,7 @@ impl ApprovalDecision { match self { Self::ApproveOnce => "approve_once", Self::ApproveAlwaysForTool => "approve_always_for_tool", + Self::ApproveAlwaysForFlow => "approve_always_for_flow", Self::Deny => "deny", } } @@ -98,13 +155,17 @@ impl ApprovalDecision { match s { "approve_once" => Some(Self::ApproveOnce), "approve_always_for_tool" => Some(Self::ApproveAlwaysForTool), + "approve_always_for_flow" => Some(Self::ApproveAlwaysForFlow), "deny" => Some(Self::Deny), _ => None, } } pub fn is_approve(self) -> bool { - matches!(self, Self::ApproveOnce | Self::ApproveAlwaysForTool) + matches!( + self, + Self::ApproveOnce | Self::ApproveAlwaysForTool | Self::ApproveAlwaysForFlow + ) } } @@ -166,6 +227,7 @@ mod tests { for d in [ ApprovalDecision::ApproveOnce, ApprovalDecision::ApproveAlwaysForTool, + ApprovalDecision::ApproveAlwaysForFlow, ApprovalDecision::Deny, ] { assert_eq!(ApprovalDecision::from_str(d.as_str()), Some(d)); @@ -181,9 +243,81 @@ mod tests { fn is_approve_true_for_approval_variants_only() { assert!(ApprovalDecision::ApproveOnce.is_approve()); assert!(ApprovalDecision::ApproveAlwaysForTool.is_approve()); + assert!(ApprovalDecision::ApproveAlwaysForFlow.is_approve()); assert!(!ApprovalDecision::Deny.is_approve()); } + #[test] + fn approve_always_for_flow_serializes_as_snake_case() { + let s = serde_json::to_string(&ApprovalDecision::ApproveAlwaysForFlow).unwrap(); + assert_eq!(s, "\"approve_always_for_flow\""); + } + + #[test] + fn source_context_flow_round_trips_as_internally_tagged_json() { + let ctx = ApprovalSourceContext::Flow { + flow_id: "flow-1".to_string(), + run_id: "run-1".to_string(), + node_id: None, + }; + let json = serde_json::to_value(&ctx).unwrap(); + assert_eq!(json["kind"], "flow"); + assert_eq!(json["flow_id"], "flow-1"); + assert_eq!(json["run_id"], "run-1"); + assert!( + json.get("node_id").is_none(), + "None node_id must be omitted, not null" + ); + + let back: ApprovalSourceContext = serde_json::from_value(json).unwrap(); + assert_eq!(back, ctx); + } + + #[test] + fn pending_approval_source_context_defaults_to_none_and_is_omitted_when_absent() { + let p = PendingApproval::new( + "req-1", + "composio", + "send email", + serde_json::json!({}), + None, + ); + assert!(p.source_context.is_none()); + let json = serde_json::to_value(&p).unwrap(); + assert!( + json.get("source_context").is_none(), + "absent source_context must not be serialized as null: {json}" + ); + } + + #[test] + fn with_source_context_attaches_flow_context() { + let p = PendingApproval::new( + "req-1", + "composio", + "send email", + serde_json::json!({}), + None, + ) + .with_source_context(ApprovalSourceContext::Flow { + flow_id: "flow-1".to_string(), + run_id: "run-1".to_string(), + node_id: Some("node-1".to_string()), + }); + match p.source_context { + Some(ApprovalSourceContext::Flow { + flow_id, + run_id, + node_id, + }) => { + assert_eq!(flow_id, "flow-1"); + assert_eq!(run_id, "run-1"); + assert_eq!(node_id.as_deref(), Some("node-1")); + } + other => panic!("expected Flow source_context, got {other:?}"), + } + } + #[test] fn approval_decision_serializes_as_snake_case() { let s = serde_json::to_string(&ApprovalDecision::ApproveAlwaysForTool).unwrap(); @@ -228,6 +362,7 @@ mod tests { args_redacted: serde_json::json!({ "tool_slug": "SLACK_SEND" }), created_at: Utc::now(), expires_at: None, + source_context: None, }; let dbg = format!("{p:?}"); assert!( diff --git a/src/openhuman/channels/host/adapters.rs b/src/openhuman/channels/host/adapters.rs index bcf1f88e54..8a017aad1c 100644 --- a/src/openhuman/channels/host/adapters.rs +++ b/src/openhuman/channels/host/adapters.rs @@ -185,6 +185,15 @@ impl ApprovalGate for CoreApprovalGate { Core::ApproveAlwaysForTool => { ApprovalDecision::Choice("approve_always_for_tool".to_string()) } + // `parse_approval_reply` never actually produces this variant + // (it only maps a free-text yes/no chat reply to + // ApproveOnce/Deny) — the "approve always for this workflow" + // decision is only ever picked from the dedicated + // `flow-gate-approval` notification action, not typed as a + // chat reply. Mapped here purely for match exhaustiveness. + Core::ApproveAlwaysForFlow => { + ApprovalDecision::Choice("approve_always_for_flow".to_string()) + } Core::Deny => ApprovalDecision::Deny, } }) diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index 6263454904..7c47e2af72 100644 --- a/src/openhuman/flows/ops.rs +++ b/src/openhuman/flows/ops.rs @@ -10,6 +10,7 @@ use serde_json::{json, Value}; use tinyflows::model::{NodeKind, TriggerKind, WorkflowGraph}; use crate::openhuman::agent::turn_origin::{with_origin, AgentTurnOrigin, TrustedAutomationSource}; +use crate::openhuman::approval::{FlowRunContext, APPROVAL_FLOW_RUN_CONTEXT}; use crate::openhuman::config::Config; use crate::openhuman::flows::bus; use crate::openhuman::flows::run_registry; @@ -1779,16 +1780,27 @@ pub async fn flows_run( thread_id.clone(), ), ); - let run = with_origin( - origin, - tinyflows::engine::run_with_checkpointer_journaled_observed( - &compiled, - input, - &caps, - checkpointer, - &thread_id, - journal.clone(), - &observer, + // Scope the flow/run correlation (issue flow-approval-surface, PR2) + // alongside the `Workflow` origin so a tool call the engine dispatches + // can, if it parks in the `ApprovalGate`, stamp its `PendingApproval` with + // `source_context = Flow { flow_id, run_id }` — the origin alone only + // carries `flow_id`. See `approval::gate::APPROVAL_FLOW_RUN_CONTEXT`. + let run = APPROVAL_FLOW_RUN_CONTEXT.scope( + FlowRunContext { + flow_id: flow_id.to_string(), + run_id: thread_id.clone(), + }, + with_origin( + origin, + tinyflows::engine::run_with_checkpointer_journaled_observed( + &compiled, + input, + &caps, + checkpointer, + &thread_id, + journal.clone(), + &observer, + ), ), ); let timed = tokio::time::timeout(std::time::Duration::from_secs(FLOW_RUN_TIMEOUT_SECS), run); @@ -1986,17 +1998,27 @@ pub async fn flows_resume( // `rejections` (issue G4 — deny semantics): a denied gate routes to its // `error` port (recovery branch) or, if it has none, fails the run. The // empty-rejections case is byte-for-byte the prior approve-only resume. - let run = with_origin( - origin, - tinyflows::engine::resume_with_checkpointer_journaled_observed( - &compiled, - &caps, - checkpointer, - thread_id, - approvals, - rejections, - journal.clone(), - &observer, + // + // Same flow/run correlation scope as `flows_run` (see its comment) — a + // resumed run can dispatch further tool calls that park, and those parks + // need `source_context` too. + let run = APPROVAL_FLOW_RUN_CONTEXT.scope( + FlowRunContext { + flow_id: flow_id.to_string(), + run_id: thread_id.to_string(), + }, + with_origin( + origin, + tinyflows::engine::resume_with_checkpointer_journaled_observed( + &compiled, + &caps, + checkpointer, + thread_id, + approvals, + rejections, + journal.clone(), + &observer, + ), ), ); From ba46b6d26e3cab46faa1bedd53f597c77f623dc0 Mon Sep 17 00:00:00 2001 From: "cyrus@tinyhumans.ai" Date: Wed, 8 Jul 2026 02:28:31 +0530 Subject: [PATCH 2/2] fix(approval): filter flow-approval test helper by flow_id to avoid cross-test flakiness find_flow_approval_requested drained the process-wide event bus for the first FlowApprovalRequested event, but other approval tests running concurrently (cargo test runs in parallel by default) also publish FlowApprovalRequested events with different flow_ids on the same singleton bus. Filter by expected_flow_id so the test only matches its own event instead of racing with unrelated ones. --- src/openhuman/approval/gate.rs | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/src/openhuman/approval/gate.rs b/src/openhuman/approval/gate.rs index 2aa5dbd28a..cf09480f2c 100644 --- a/src/openhuman/approval/gate.rs +++ b/src/openhuman/approval/gate.rs @@ -2152,12 +2152,16 @@ mod tests { let _ = handle.await.unwrap(); } - /// Drain `rx` (tolerating broadcast lag from unrelated concurrent - /// publishers on the process-wide bus) until a `FlowApprovalRequested` - /// arrives. + /// Drain `rx` until a `FlowApprovalRequested` for `expected_flow_id` + /// arrives. The event bus is process-wide and other tests in this file + /// (and elsewhere) publish on it concurrently — including other + /// `FlowApprovalRequested` events for *different* flow ids — so this must + /// filter by flow id, not just by variant, and tolerate both unrelated + /// events and broadcast lag rather than returning the first match. async fn find_flow_approval_requested( rx: &mut tokio::sync::broadcast::Receiver, - ) -> (String, String, String, String) { + expected_flow_id: &str, + ) -> (String, String, String) { loop { match rx.recv().await { Ok(crate::core::event_bus::DomainEvent::FlowApprovalRequested { @@ -2166,7 +2170,7 @@ mod tests { run_id, tool_name, .. - }) => return (request_id, flow_id, run_id, tool_name), + }) if flow_id == expected_flow_id => return (request_id, run_id, tool_name), Ok(_) => continue, Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, Err(tokio::sync::broadcast::error::RecvError::Closed) => {