feat(flows): backend — surface flow-run approvals + per-flow trust (correlation, notify, socket) (#4673)

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
@@ -1412,7 +1438,8 @@ impl DomainEvent {
             Self::ApprovalRequested { .. }
             | Self::ApprovalDecided { .. }
             | Self::ApprovalGateOverrideIgnored { .. }
-            | Self::ApprovalGateDisabled { .. } => "approval",
+            | Self::ApprovalGateDisabled { .. }
+            | Self::FlowApprovalRequested { .. } => "approval",
 
             Self::PlanReviewRequested { .. } | Self::PlanReviewDecided { .. } => "plan_review",
 
@@ -1549,6 +1576,7 @@ impl DomainEvent {
             Self::SessionExpired { .. } => "SessionExpired",
             Self::ApprovalRequested { .. } => "ApprovalRequested",
             Self::ApprovalDecided { .. } => "ApprovalDecided",
+            Self::FlowApprovalRequested { .. } => "FlowApprovalRequested",
             Self::PlanReviewRequested { .. } => "PlanReviewRequested",
             Self::PlanReviewDecided { .. } => "PlanReviewDecided",
             Self::ApprovalGateOverrideIgnored { .. } => "ApprovalGateOverrideIgnored",

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);
+                }
                 _ => {}
             }
         }

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<Vec<PendingApproval>> {
+        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<bool> {
+        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,315 @@ 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` 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<crate::core::event_bus::DomainEvent>,
+        expected_flow_id: &str,
+    ) -> (String, String, String) {
+        loop {
+            match rx.recv().await {
+                Ok(crate::core::event_bus::DomainEvent::FlowApprovalRequested {
+                    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) => {
+                    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")
+                }
+            }
+        }
+    }
 }

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

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(),

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<String, Value>) -> ControllerFuture {
         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}'"
+                "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)

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<Vec<PendingApproval>> {
         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<usize> {
     })
 }
 
+/// 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<Vec<PendingApproval>> {
+    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<bool> {
+    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<Utc>) -> Result<usize> {
     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<PendingApproval>
     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<String> = 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<String> = row.get(7).unwrap_or(None);
+    let source_context = source_context_str.as_deref().and_then(|raw| {
+        serde_json::from_str::<ApprovalSourceContext>(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<PendingApproval>
         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());
+    }
 }

src/openhuman/approval/types.rs

@@ -29,6 +29,12 @@ pub struct PendingApproval {
     pub args_redacted: serde_json::Value,
     pub created_at: DateTime<Utc>,
     pub expires_at: Option<DateTime<Utc>>,
+    /// 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<ApprovalSourceContext>,
 }
 
 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<String>,
+    },
 }
 
 /// 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!(

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

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,
+            ),
         ),
     );