feat(flows): tinyflows integration Phase 2 — credentials & connections (Composio accounts, encrypted http_cred, flows_list_connections, tier gating) (#4561)

src/openhuman/credentials/http_creds.rs

@@ -0,0 +1,519 @@
+//! Named HTTP credentials for `http_request` flow nodes.
+//!
+//! A flow's `http_request` node can carry a `connection_ref` of the shape
+//! `"http_cred:<name>"`. This module is the host-side store those names resolve
+//! against: each record is an **injection template** (bearer token, HTTP basic
+//! user:pass, or a raw custom header) whose secret material is encrypted at
+//! rest with the same [`SecretStore`](crate::openhuman::keyring::SecretStore)
+//! (ChaCha20-Poly1305) the auth-profile store uses.
+//!
+//! **Security contract:** the secret value NEVER leaves this module except as
+//! the header it is injected into, server-side, inside
+//! `tinyflows::caps::OpenHumanHttp::request`. It is never returned to the UI,
+//! handed to the flow engine/graph, or logged. List/summary shapes carry only
+//! the name + scheme + non-secret template fields ([`HttpCredentialSummary`]).
+
+use std::collections::BTreeMap;
+use std::fs;
+use std::path::{Path, PathBuf};
+
+use anyhow::{Context, Result};
+use base64::engine::Engine as _;
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+use crate::openhuman::config::Config;
+use crate::openhuman::keyring::SecretStore;
+
+const STORE_FILENAME: &str = "http-credentials.json";
+const CURRENT_SCHEMA_VERSION: u32 = 1;
+
+/// How a credential is presented on the outbound request.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum HttpCredentialScheme {
+    /// `Authorization: Bearer <secret>`.
+    Bearer,
+    /// `Authorization: Basic base64(<username>:<secret>)`.
+    Basic,
+    /// A raw custom header: `<header_name>: <secret>` (e.g. `X-API-Key`).
+    Header,
+}
+
+impl HttpCredentialScheme {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            HttpCredentialScheme::Bearer => "bearer",
+            HttpCredentialScheme::Basic => "basic",
+            HttpCredentialScheme::Header => "header",
+        }
+    }
+}
+
+/// A resolved HTTP credential, secret in the clear in memory. Produced only by
+/// [`HttpCredentialsStore::get`] and consumed only by the server-side injector.
+#[derive(Debug, Clone)]
+pub struct HttpCredential {
+    pub name: String,
+    pub scheme: HttpCredentialScheme,
+    /// Header name for the [`HttpCredentialScheme::Header`] scheme (e.g.
+    /// `X-API-Key`). Ignored for bearer/basic.
+    pub header_name: Option<String>,
+    /// Username for the [`HttpCredentialScheme::Basic`] scheme. Ignored
+    /// otherwise. Not itself a secret, but stored alongside the secret.
+    pub username: Option<String>,
+    /// The secret material: bearer token, basic password, or raw header value.
+    pub secret: String,
+    pub created_at: DateTime<Utc>,
+    pub updated_at: DateTime<Utc>,
+}
+
+impl HttpCredential {
+    pub fn bearer(name: impl Into<String>, token: impl Into<String>) -> Self {
+        let now = Utc::now();
+        Self {
+            name: name.into(),
+            scheme: HttpCredentialScheme::Bearer,
+            header_name: None,
+            username: None,
+            secret: token.into(),
+            created_at: now,
+            updated_at: now,
+        }
+    }
+
+    pub fn basic(
+        name: impl Into<String>,
+        username: impl Into<String>,
+        password: impl Into<String>,
+    ) -> Self {
+        let now = Utc::now();
+        Self {
+            name: name.into(),
+            scheme: HttpCredentialScheme::Basic,
+            header_name: None,
+            username: Some(username.into()),
+            secret: password.into(),
+            created_at: now,
+            updated_at: now,
+        }
+    }
+
+    pub fn header(
+        name: impl Into<String>,
+        header_name: impl Into<String>,
+        value: impl Into<String>,
+    ) -> Self {
+        let now = Utc::now();
+        Self {
+            name: name.into(),
+            scheme: HttpCredentialScheme::Header,
+            header_name: Some(header_name.into()),
+            username: None,
+            secret: value.into(),
+            created_at: now,
+            updated_at: now,
+        }
+    }
+
+    /// The `(header_name, header_value)` pair to inject onto the outbound
+    /// request. **The returned value contains the secret** — callers must merge
+    /// it into the request server-side and must never log or echo it.
+    pub fn to_header(&self) -> Result<(String, String)> {
+        match self.scheme {
+            HttpCredentialScheme::Bearer => {
+                anyhow::ensure!(
+                    !self.secret.trim().is_empty(),
+                    "http_cred '{}': bearer token is empty",
+                    self.name
+                );
+                Ok((
+                    "Authorization".to_string(),
+                    format!("Bearer {}", self.secret),
+                ))
+            }
+            HttpCredentialScheme::Basic => {
+                let username = self.username.as_deref().unwrap_or_default();
+                let encoded = base64::engine::general_purpose::STANDARD
+                    .encode(format!("{username}:{}", self.secret));
+                Ok(("Authorization".to_string(), format!("Basic {encoded}")))
+            }
+            HttpCredentialScheme::Header => {
+                let header_name = self
+                    .header_name
+                    .as_deref()
+                    .map(str::trim)
+                    .filter(|h| !h.is_empty())
+                    .with_context(|| {
+                        format!(
+                            "http_cred '{}': header scheme requires a non-empty header_name",
+                            self.name
+                        )
+                    })?;
+                anyhow::ensure!(
+                    !self.secret.trim().is_empty(),
+                    "http_cred '{}': header value is empty",
+                    self.name
+                );
+                Ok((header_name.to_string(), self.secret.clone()))
+            }
+        }
+    }
+}
+
+/// Secret-free description of a stored credential — safe to return to the UI /
+/// list surfaces (e.g. a future `flows_list_connections`).
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct HttpCredentialSummary {
+    pub name: String,
+    pub scheme: String,
+    pub header_name: Option<String>,
+    pub username: Option<String>,
+    pub updated_at: String,
+}
+
+/// On-disk record. `secret` is stored as `enc2:<hex>` ciphertext (or plaintext
+/// when `secrets.encrypt = false`, matching the auth-profile store's behavior).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct PersistedHttpCredential {
+    scheme: String,
+    #[serde(default)]
+    header_name: Option<String>,
+    #[serde(default)]
+    username: Option<String>,
+    /// Encrypted secret material.
+    secret: String,
+    created_at: String,
+    updated_at: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct PersistedHttpCredentials {
+    schema_version: u32,
+    updated_at: String,
+    credentials: BTreeMap<String, PersistedHttpCredential>,
+}
+
+impl Default for PersistedHttpCredentials {
+    fn default() -> Self {
+        Self {
+            schema_version: CURRENT_SCHEMA_VERSION,
+            updated_at: Utc::now().to_rfc3339(),
+            credentials: BTreeMap::new(),
+        }
+    }
+}
+
+/// Encrypted-at-rest store of named HTTP credentials.
+#[derive(Debug, Clone)]
+pub struct HttpCredentialsStore {
+    path: PathBuf,
+    secret_store: SecretStore,
+}
+
+impl HttpCredentialsStore {
+    pub fn from_config(config: &Config) -> Self {
+        let state_dir = super::state_dir_from_config(config);
+        Self::new(&state_dir, config.secrets.encrypt)
+    }
+
+    pub fn new(state_dir: &Path, encrypt_secrets: bool) -> Self {
+        Self {
+            path: state_dir.join(STORE_FILENAME),
+            secret_store: SecretStore::new(state_dir, encrypt_secrets),
+        }
+    }
+
+    /// Normalize a credential name into the stable storage key. Names are
+    /// case-insensitive and trimmed so `http_cred:Stripe ` and `stripe` resolve
+    /// to the same record.
+    fn normalize_name(name: &str) -> String {
+        name.trim().to_ascii_lowercase()
+    }
+
+    /// List all stored credentials as secret-free summaries.
+    pub fn list(&self) -> Result<Vec<HttpCredentialSummary>> {
+        let persisted = self.read_persisted()?;
+        Ok(persisted
+            .credentials
+            .into_iter()
+            .map(|(name, rec)| HttpCredentialSummary {
+                name,
+                scheme: rec.scheme,
+                header_name: rec.header_name,
+                username: rec.username,
+                updated_at: rec.updated_at,
+            })
+            .collect())
+    }
+
+    /// Resolve a credential name to its secret-bearing record, decrypting the
+    /// secret. Returns `Ok(None)` when no such credential exists.
+    pub fn get(&self, name: &str) -> Result<Option<HttpCredential>> {
+        let key = Self::normalize_name(name);
+        let persisted = self.read_persisted()?;
+        let Some(rec) = persisted.credentials.get(&key) else {
+            log::debug!(target: "credentials", "[credentials] http_cred get miss name={key}");
+            return Ok(None);
+        };
+
+        let scheme = parse_scheme(&rec.scheme).with_context(|| {
+            format!("http_cred '{key}' has unrecognized scheme {:?}", rec.scheme)
+        })?;
+        let secret = self
+            .secret_store
+            .decrypt(&rec.secret)
+            .with_context(|| format!("failed to decrypt http_cred '{key}' secret"))?;
+
+        log::debug!(
+            target: "credentials",
+            "[credentials] http_cred get hit name={key} scheme={}",
+            scheme.as_str()
+        );
+        Ok(Some(HttpCredential {
+            name: key,
+            scheme,
+            header_name: rec.header_name.clone(),
+            username: rec.username.clone(),
+            secret,
+            created_at: parse_dt(&rec.created_at),
+            updated_at: parse_dt(&rec.updated_at),
+        }))
+    }
+
+    /// Insert or replace a credential, encrypting its secret at rest.
+    pub fn upsert(&self, cred: &HttpCredential) -> Result<()> {
+        let key = Self::normalize_name(&cred.name);
+        anyhow::ensure!(!key.is_empty(), "http_cred name cannot be empty");
+
+        let mut persisted = self.read_persisted()?;
+        let encrypted = self
+            .secret_store
+            .encrypt(&cred.secret)
+            .context("failed to encrypt http_cred secret")?;
+
+        let created_at = persisted
+            .credentials
+            .get(&key)
+            .map(|r| r.created_at.clone())
+            .unwrap_or_else(|| cred.created_at.to_rfc3339());
+
+        persisted.credentials.insert(
+            key.clone(),
+            PersistedHttpCredential {
+                scheme: cred.scheme.as_str().to_string(),
+                header_name: cred.header_name.clone(),
+                username: cred.username.clone(),
+                secret: encrypted,
+                created_at,
+                updated_at: Utc::now().to_rfc3339(),
+            },
+        );
+        persisted.updated_at = Utc::now().to_rfc3339();
+        self.write_persisted(&persisted)?;
+        log::info!(
+            target: "credentials",
+            "[credentials] http_cred upserted name={key} scheme={} (secret redacted)",
+            cred.scheme.as_str()
+        );
+        Ok(())
+    }
+
+    /// Remove a credential by name. Returns whether a record was removed.
+    pub fn remove(&self, name: &str) -> Result<bool> {
+        let key = Self::normalize_name(name);
+        let mut persisted = self.read_persisted()?;
+        let removed = persisted.credentials.remove(&key).is_some();
+        if removed {
+            persisted.updated_at = Utc::now().to_rfc3339();
+            self.write_persisted(&persisted)?;
+            log::info!(target: "credentials", "[credentials] http_cred removed name={key}");
+        }
+        Ok(removed)
+    }
+
+    fn read_persisted(&self) -> Result<PersistedHttpCredentials> {
+        if !self.path.exists() {
+            return Ok(PersistedHttpCredentials::default());
+        }
+        let bytes = fs::read(&self.path).with_context(|| {
+            format!(
+                "failed to read http-credentials store at {}",
+                self.path.display()
+            )
+        })?;
+        if bytes.is_empty() {
+            return Ok(PersistedHttpCredentials::default());
+        }
+        serde_json::from_slice(&bytes).with_context(|| {
+            format!(
+                "http-credentials store at {} is not valid JSON",
+                self.path.display()
+            )
+        })
+    }
+
+    fn write_persisted(&self, persisted: &PersistedHttpCredentials) -> Result<()> {
+        if let Some(parent) = self.path.parent() {
+            fs::create_dir_all(parent).with_context(|| {
+                format!(
+                    "failed to create http-credentials dir at {}",
+                    parent.display()
+                )
+            })?;
+        }
+        let json = serde_json::to_vec_pretty(persisted)
+            .context("failed to serialize http-credentials store")?;
+        // Atomic publish: write to a unique tmp then rename over the store so a
+        // concurrent reader never observes a torn file.
+        let tmp_name = format!(
+            "{STORE_FILENAME}.tmp.{}.{}",
+            std::process::id(),
+            Utc::now().timestamp_nanos_opt().unwrap_or_default()
+        );
+        let tmp_path = self.path.with_file_name(tmp_name);
+        fs::write(&tmp_path, &json)
+            .with_context(|| format!("failed to write {}", tmp_path.display()))?;
+        if let Err(e) = fs::rename(&tmp_path, &self.path) {
+            let _ = fs::remove_file(&tmp_path);
+            return Err(e).with_context(|| {
+                format!(
+                    "failed to replace http-credentials store at {}",
+                    self.path.display()
+                )
+            });
+        }
+        Ok(())
+    }
+}
+
+fn parse_scheme(raw: &str) -> Option<HttpCredentialScheme> {
+    match raw.trim().to_ascii_lowercase().as_str() {
+        "bearer" => Some(HttpCredentialScheme::Bearer),
+        "basic" => Some(HttpCredentialScheme::Basic),
+        "header" => Some(HttpCredentialScheme::Header),
+        _ => None,
+    }
+}
+
+fn parse_dt(raw: &str) -> DateTime<Utc> {
+    DateTime::parse_from_rfc3339(raw)
+        .map(|dt| dt.with_timezone(&Utc))
+        .unwrap_or_else(|_| Utc::now())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn temp_store() -> (tempfile::TempDir, HttpCredentialsStore) {
+        let dir = tempfile::tempdir().expect("tempdir");
+        // encrypt=true exercises the ChaCha20-Poly1305 at-rest path.
+        let store = HttpCredentialsStore::new(dir.path(), true);
+        (dir, store)
+    }
+
+    #[test]
+    fn bearer_to_header_is_authorization_bearer() {
+        let cred = HttpCredential::bearer("stripe", "sk_live_abc123");
+        let (name, value) = cred.to_header().unwrap();
+        assert_eq!(name, "Authorization");
+        assert_eq!(value, "Bearer sk_live_abc123");
+    }
+
+    #[test]
+    fn basic_to_header_is_base64_user_pass() {
+        let cred = HttpCredential::basic("acme", "alice", "hunter2");
+        let (name, value) = cred.to_header().unwrap();
+        assert_eq!(name, "Authorization");
+        // base64("alice:hunter2")
+        let expected = base64::engine::general_purpose::STANDARD.encode("alice:hunter2");
+        assert_eq!(value, format!("Basic {expected}"));
+    }
+
+    #[test]
+    fn header_scheme_uses_custom_header_name() {
+        let cred = HttpCredential::header("apikey", "X-API-Key", "topsecret");
+        let (name, value) = cred.to_header().unwrap();
+        assert_eq!(name, "X-API-Key");
+        assert_eq!(value, "topsecret");
+    }
+
+    #[test]
+    fn header_scheme_without_header_name_errors() {
+        let mut cred = HttpCredential::header("apikey", "X-API-Key", "topsecret");
+        cred.header_name = None;
+        assert!(cred.to_header().is_err());
+    }
+
+    #[test]
+    fn roundtrip_encrypts_secret_at_rest() {
+        let (dir, store) = temp_store();
+        let secret = "sk_live_super_secret_value";
+        store
+            .upsert(&HttpCredential::bearer("stripe", secret))
+            .unwrap();
+
+        // The on-disk file must NOT contain the plaintext secret.
+        let raw = std::fs::read_to_string(dir.path().join(STORE_FILENAME)).unwrap();
+        assert!(
+            !raw.contains(secret),
+            "plaintext secret leaked into on-disk store: {raw}"
+        );
+        assert!(raw.contains("enc2:"), "secret was not encrypted: {raw}");
+
+        // But get() decrypts it back.
+        let got = store.get("stripe").unwrap().expect("credential present");
+        assert_eq!(got.secret, secret);
+        assert_eq!(got.scheme, HttpCredentialScheme::Bearer);
+    }
+
+    #[test]
+    fn name_resolution_is_case_insensitive_and_trimmed() {
+        let (_dir, store) = temp_store();
+        store
+            .upsert(&HttpCredential::bearer("Stripe", "tok"))
+            .unwrap();
+        assert!(store.get("  STRIPE ").unwrap().is_some());
+        assert!(store.get("stripe").unwrap().is_some());
+    }
+
+    #[test]
+    fn list_never_exposes_secrets() {
+        let (_dir, store) = temp_store();
+        store
+            .upsert(&HttpCredential::header("apikey", "X-API-Key", "topsecret"))
+            .unwrap();
+        let summaries = store.list().unwrap();
+        assert_eq!(summaries.len(), 1);
+        let s = &summaries[0];
+        assert_eq!(s.name, "apikey");
+        assert_eq!(s.scheme, "header");
+        assert_eq!(s.header_name.as_deref(), Some("X-API-Key"));
+        // The summary type has no secret field at all — assert via serialization
+        // that "topsecret" never appears.
+        let json = serde_json::to_string(&summaries).unwrap();
+        assert!(
+            !json.contains("topsecret"),
+            "secret leaked into summary: {json}"
+        );
+    }
+
+    #[test]
+    fn get_unknown_name_returns_none() {
+        let (_dir, store) = temp_store();
+        assert!(store.get("does-not-exist").unwrap().is_none());
+    }
+
+    #[test]
+    fn remove_deletes_record() {
+        let (_dir, store) = temp_store();
+        store
+            .upsert(&HttpCredential::bearer("stripe", "tok"))
+            .unwrap();
+        assert!(store.remove("stripe").unwrap());
+        assert!(store.get("stripe").unwrap().is_none());
+        assert!(!store.remove("stripe").unwrap());
+    }
+}

src/openhuman/credentials/mod.rs

@@ -3,6 +3,7 @@
 pub mod bus;
 pub mod cli;
 mod core;
+pub mod http_creds;
 pub mod ops;
 pub mod profiles;
 pub mod responses;
@@ -16,6 +17,9 @@ pub use crate::api::rest::{
     BackendOAuthClient, ConnectResponse, IntegrationSummary, IntegrationTokensHandoff,
 };
 pub use core::*;
+pub use http_creds::{
+    HttpCredential, HttpCredentialScheme, HttpCredentialSummary, HttpCredentialsStore,
+};
 pub use ops as rpc;
 pub use ops::*;
 // Direct-mode (BYO Composio API key) credential helpers.

src/openhuman/flows/mod.rs

@@ -29,4 +29,4 @@ pub use schemas::{
 // lives in the sibling `tinyflows` domain and persists each finished step onto
 // the `flow_runs` row through this function as the run executes.
 pub use store::{kv_get, kv_set, upsert_flow_run_step};
-pub use types::{Flow, FlowRun, FlowRunStep, FlowRunTrigger, FlowValidation};
+pub use types::{Flow, FlowConnection, FlowRun, FlowRunStep, FlowRunTrigger, FlowValidation};

src/openhuman/flows/ops.rs

@@ -14,7 +14,7 @@ use crate::openhuman::config::Config;
 use crate::openhuman::flows::bus;
 use crate::openhuman::flows::run_registry;
 use crate::openhuman::flows::store;
-use crate::openhuman::flows::types::{FlowRunStep, FlowRunTrigger};
+use crate::openhuman::flows::types::{FlowConnection, FlowRunStep, FlowRunTrigger};
 use crate::openhuman::flows::{Flow, FlowRun};
 use crate::rpc::RpcOutcome;
 
@@ -33,6 +33,45 @@ const FLOW_RUN_TIMEOUT_SECS: u64 = 600;
 /// this is a dedicated flows-side TTL, not a reuse of the approval store's.
 const FLOW_PARKED_TTL_SECS: i64 = 600;
 
+// ─────────────────────────────────────────────────────────────────────────────
+// Phase 2 — autonomy-tier gating of acting flow nodes
+// ─────────────────────────────────────────────────────────────────────────────
+//
+// A `flows_run` / `flows_resume` executes under a `TrustedAutomation { Workflow }`
+// origin (see `workflow_origin` below), but the *acting power* of a run is still
+// bounded by the user's `[autonomy]` tier — the same `SecurityPolicy`
+// (`src/openhuman/security/`) the agent tool-loop honors, built via
+// `SecurityPolicy::from_config(&config.autonomy, …)` inside
+// `tinyflows::caps::build_capabilities`.
+//
+// Before an acting node dispatches, its capability adapter
+// (`src/openhuman/tinyflows/caps.rs::enforce_node_tier_gate`) maps the node to a
+// `CommandClass` and consults `SecurityPolicy::gate_decision`. `Block` refuses
+// outright (`[policy-blocked]` error, no dispatch); `Prompt`/`Allow` fall through
+// to the process-global `ApprovalGate`, which performs the human round-trip for
+// `Prompt` exactly as the agent tool-loop does. Node → class → per-tier decision:
+//
+//   Flow node        CommandClass   read-only     supervised    full
+//   ────────────     ────────────   ──────────    ──────────    ──────────
+//   http_request     Network        BLOCK         Prompt        Prompt
+//   code             Write          BLOCK         Prompt        Allow
+//   tool_call        (curation +    (curated +    Prompt        Prompt/Allow¹
+//                     ApprovalGate)   scope gate)
+//   agent (llm)      — (no acting side effect; not tier-gated, only the
+//                        inference/privacy chokepoint applies)
+//   state (kv)       — (host-internal flow KV; not an outbound act)
+//
+//   ¹ tool_call routes through the deny-by-default curation/scope gate plus the
+//     ApprovalGate rather than `gate_decision`; a Network-class Composio action
+//     still prompts under supervised/full and the curation gate is the hard
+//     allowlist. See `caps.rs::OpenHumanTools`.
+//
+// `Network` is never `Allow` in any tier (always `Prompt` when not blocked), so
+// even a full-tier http_request node prompts unless a pre-declared trust root /
+// `auto_approve` short-circuits the ApprovalGate — matching `curl`/`shell`.
+// `Write` (code) is `Allow` under full, so trusted automations run sandboxed
+// code unattended; read-only blocks both outright.
+
 /// Runs a raw graph JSON value through `tinyflows::migrate::migrate` (upgrade
 /// an older-schema definition to current), deserializes it, and rejects a
 /// structurally invalid graph via `tinyflows::validate::validate` — so a bad
@@ -209,6 +248,181 @@ pub async fn flows_list(config: &Config) -> Result<RpcOutcome<Vec<Flow>>, String
     Ok(RpcOutcome::single_log(flows, "flows listed"))
 }
 
+/// Lists the connection sources a flow node's `connection_ref` can attach to:
+/// Composio connected accounts (`kind = "composio"`) and stored HTTP
+/// credentials (`kind = "http"`). This is the picker source for the Workflows
+/// UI (and the agent's flow-authoring surface) — it returns ids + display
+/// labels + kind ONLY, never any secret material.
+///
+/// The two sources are aggregated independently and are individually
+/// fault-tolerant: a transient Composio backend/network failure (or an
+/// unconfigured Direct-mode key) yields zero Composio entries but still returns
+/// the HTTP credential half, and vice-versa. A failure in one source never
+/// fails the whole picker.
+pub async fn flows_list_connections(
+    config: &Config,
+) -> Result<RpcOutcome<Vec<FlowConnection>>, String> {
+    tracing::debug!(
+        "[flows] rpc flows_list_connections: aggregating composio + http_cred picker sources"
+    );
+    let mut logs = Vec::new();
+
+    // 1. Composio connected accounts. Direct mode without a configured key
+    //    already short-circuits to an empty list (a valid setup state, not an
+    //    error); a backend outage returns Err — tolerate it so the picker still
+    //    surfaces HTTP credentials.
+    let composio_conns =
+        match crate::openhuman::composio::ops::composio_list_connections(config).await {
+            Ok(outcome) => {
+                tracing::debug!(
+                    count = outcome.value.connections.len(),
+                    "[flows] flows_list_connections: composio source returned connections"
+                );
+                outcome.value.connections
+            }
+            Err(e) => {
+                tracing::warn!(
+                    error = %e,
+                    "[flows] flows_list_connections: composio source unavailable — \
+                     returning http_cred entries only"
+                );
+                logs.push(format!(
+                    "flows_list_connections: composio source unavailable ({e})"
+                ));
+                Vec::new()
+            }
+        };
+
+    // 2. Named HTTP credentials — secret-free summaries (the store never hands
+    //    out secret material here; injection happens server-side in
+    //    `tinyflows::caps::OpenHumanHttp`).
+    let http_creds =
+        match crate::openhuman::credentials::HttpCredentialsStore::from_config(config).list() {
+            Ok(list) => {
+                tracing::debug!(
+                    count = list.len(),
+                    "[flows] flows_list_connections: http_cred store returned summaries"
+                );
+                list
+            }
+            Err(e) => {
+                tracing::warn!(
+                    error = %e,
+                    "[flows] flows_list_connections: http_cred store read failed — \
+                     returning composio entries only"
+                );
+                logs.push(format!(
+                    "flows_list_connections: http_cred store unavailable ({e})"
+                ));
+                Vec::new()
+            }
+        };
+
+    let connections = build_flow_connections(composio_conns, http_creds);
+    tracing::debug!(
+        total = connections.len(),
+        "[flows] flows_list_connections: aggregated picker sources"
+    );
+    logs.push(format!(
+        "flows_list_connections: {} connection(s)",
+        connections.len()
+    ));
+    Ok(RpcOutcome::new(connections, logs))
+}
+
+/// Fold Composio connected accounts + named HTTP credentials into the flat,
+/// secret-free [`FlowConnection`] picker list. Only ACTIVE Composio connections
+/// are surfaced — a pending/expired OAuth account cannot execute a tool, so it
+/// would be a dead pick. Pure (no I/O) so the aggregation shape is
+/// unit-testable without a live backend.
+fn build_flow_connections(
+    composio: Vec<crate::openhuman::composio::ComposioConnection>,
+    http: Vec<crate::openhuman::credentials::HttpCredentialSummary>,
+) -> Vec<FlowConnection> {
+    let mut out = Vec::with_capacity(composio.len() + http.len());
+    for conn in composio {
+        if !conn.is_active() {
+            tracing::debug!(
+                toolkit = %conn.toolkit,
+                connection_id = %conn.id,
+                status = %conn.status,
+                "[flows] flows_list_connections: skipping non-active composio connection"
+            );
+            continue;
+        }
+        let toolkit = conn.normalized_toolkit();
+        out.push(FlowConnection {
+            // Exactly the shape `tinyflows::caps::composio_connection_id` parses.
+            connection_ref: format!("composio:{}:{}", toolkit, conn.id),
+            kind: "composio".to_string(),
+            display: composio_connection_display(&toolkit, &conn),
+            toolkit: Some(toolkit),
+            scheme: None,
+        });
+    }
+    for cred in http {
+        out.push(FlowConnection {
+            // Exactly the shape `tinyflows::caps::http_cred_name` parses.
+            connection_ref: format!("http_cred:{}", cred.name),
+            kind: "http".to_string(),
+            display: http_credential_display(&cred),
+            toolkit: None,
+            scheme: Some(cred.scheme),
+        });
+    }
+    out
+}
+
+/// Human-readable picker label for a Composio connected account, e.g.
+/// `"Gmail · user@example.com"`. Prefers email, then workspace/team, then
+/// handle; falls back to the title-cased toolkit alone when no identity is
+/// cached. The identity fields are display metadata (already surfaced by
+/// `composio_list_connections`), never secret material.
+fn composio_connection_display(
+    toolkit: &str,
+    conn: &crate::openhuman::composio::ComposioConnection,
+) -> String {
+    let title = title_case_toolkit(toolkit);
+    let identity = conn
+        .account_email
+        .as_deref()
+        .or(conn.workspace.as_deref())
+        .or(conn.username.as_deref())
+        .map(str::trim)
+        .filter(|s| !s.is_empty());
+    match identity {
+        Some(id) => format!("{title} · {id}"),
+        None => title,
+    }
+}
+
+/// Human-readable picker label for a named HTTP credential, e.g.
+/// `"stripe (bearer)"`. Only the (non-secret) name + scheme — never the value.
+fn http_credential_display(cred: &crate::openhuman::credentials::HttpCredentialSummary) -> String {
+    format!("{} ({})", cred.name, cred.scheme)
+}
+
+/// Title-case a toolkit slug for display: `"gmail"` → `"Gmail"`,
+/// `"google_calendar"` → `"Google Calendar"`. Best-effort cosmetic only.
+fn title_case_toolkit(toolkit: &str) -> String {
+    let trimmed = toolkit.trim();
+    if trimmed.is_empty() {
+        return String::new();
+    }
+    trimmed
+        .split(|c| c == '_' || c == '-' || c == ' ')
+        .filter(|w| !w.is_empty())
+        .map(|word| {
+            let mut chars = word.chars();
+            match chars.next() {
+                Some(first) => first.to_uppercase().collect::<String>() + chars.as_str(),
+                None => String::new(),
+            }
+        })
+        .collect::<Vec<_>>()
+        .join(" ")
+}
+
 /// Updates a flow's name, graph, and/or `require_approval` toggle.
 /// Re-validates the graph (whether newly supplied or the existing one)
 /// before persisting, same as `flows_create`.

src/openhuman/flows/ops_tests.rs

@@ -1357,3 +1357,146 @@ async fn flows_set_enabled_schedule_flow_has_no_warning() {
         enabled.logs
     );
 }
+
+// ── flows_list_connections (picker source) ──────────────────────────────
+
+use crate::openhuman::composio::ComposioConnection;
+use crate::openhuman::credentials::{HttpCredential, HttpCredentialSummary, HttpCredentialsStore};
+
+fn composio_conn(id: &str, toolkit: &str, status: &str, email: Option<&str>) -> ComposioConnection {
+    ComposioConnection {
+        id: id.to_string(),
+        toolkit: toolkit.to_string(),
+        status: status.to_string(),
+        created_at: None,
+        account_email: email.map(str::to_string),
+        workspace: None,
+        username: None,
+    }
+}
+
+fn http_summary(name: &str, scheme: &str) -> HttpCredentialSummary {
+    HttpCredentialSummary {
+        name: name.to_string(),
+        scheme: scheme.to_string(),
+        header_name: None,
+        username: None,
+        updated_at: "2026-01-01T00:00:00Z".to_string(),
+    }
+}
+
+#[test]
+fn build_flow_connections_emits_parseable_refs_for_both_kinds() {
+    let composio = vec![composio_conn(
+        "ca_abc",
+        "Gmail",
+        "ACTIVE",
+        Some("user@example.com"),
+    )];
+    let http = vec![http_summary("stripe", "bearer")];
+
+    let out = build_flow_connections(composio, http);
+    assert_eq!(out.len(), 2);
+
+    let gmail = &out[0];
+    assert_eq!(gmail.kind, "composio");
+    // Toolkit is normalized (lowercased) and the ref round-trips through the
+    // exact parser the caps seam uses on execution.
+    assert_eq!(gmail.connection_ref, "composio:gmail:ca_abc");
+    assert_eq!(
+        crate::openhuman::tinyflows::caps::composio_connection_id(&gmail.connection_ref),
+        Some("ca_abc")
+    );
+    assert_eq!(gmail.toolkit.as_deref(), Some("gmail"));
+    assert_eq!(gmail.display, "Gmail · user@example.com");
+    assert!(gmail.scheme.is_none());
+
+    let stripe = &out[1];
+    assert_eq!(stripe.kind, "http");
+    assert_eq!(stripe.connection_ref, "http_cred:stripe");
+    assert_eq!(
+        crate::openhuman::tinyflows::caps::http_cred_name(&stripe.connection_ref),
+        Some("stripe")
+    );
+    assert_eq!(stripe.scheme.as_deref(), Some("bearer"));
+    assert_eq!(stripe.display, "stripe (bearer)");
+    assert!(stripe.toolkit.is_none());
+}
+
+#[test]
+fn build_flow_connections_skips_non_active_composio_accounts() {
+    let composio = vec![
+        composio_conn("ca_ok", "notion", "ACTIVE", None),
+        composio_conn("ca_pending", "slack", "PENDING", None),
+    ];
+    let out = build_flow_connections(composio, Vec::new());
+    assert_eq!(out.len(), 1, "only the ACTIVE connection is surfaced");
+    assert_eq!(out[0].connection_ref, "composio:notion:ca_ok");
+    // No cached identity → title-cased toolkit alone.
+    assert_eq!(out[0].display, "Notion");
+}
+
+#[test]
+fn build_flow_connections_never_carries_secret_fields() {
+    let out = build_flow_connections(
+        vec![composio_conn("ca_abc", "gmail", "ACTIVE", Some("u@x.io"))],
+        vec![http_summary("stripe", "header")],
+    );
+    let json = serde_json::to_string(&out).unwrap();
+    // The serialized picker payload must expose only ref/kind/display/toolkit/
+    // scheme — no secret-bearing key names at all.
+    for banned in [
+        "secret", "token", "password", "\"key\"", "apiKey", "api_key",
+    ] {
+        assert!(
+            !json
+                .to_ascii_lowercase()
+                .contains(&banned.to_ascii_lowercase()),
+            "serialized FlowConnection leaked a secret-bearing field ({banned}): {json}"
+        );
+    }
+}
+
+#[test]
+fn title_case_toolkit_handles_underscores_and_dashes() {
+    assert_eq!(title_case_toolkit("gmail"), "Gmail");
+    assert_eq!(title_case_toolkit("google_calendar"), "Google Calendar");
+    assert_eq!(title_case_toolkit("google-sheets"), "Google Sheets");
+    assert_eq!(title_case_toolkit(""), "");
+}
+
+#[tokio::test]
+async fn flows_list_connections_aggregates_http_creds_and_tolerates_composio() {
+    let tmp = TempDir::new().unwrap();
+    let mut config = test_config(&tmp);
+    // Force Direct mode with no key so the composio source short-circuits to an
+    // empty list offline (no network) — proving the aggregation still returns
+    // the HTTP-credential half.
+    config.composio.mode = crate::openhuman::config::schema::COMPOSIO_MODE_DIRECT.to_string();
+    // Secrets in the clear at rest for the test (mirrors the E2E config).
+    config.secrets.encrypt = false;
+
+    // Seed one HTTP credential through the same store the op reads.
+    let store = HttpCredentialsStore::from_config(&config);
+    store
+        .upsert(&HttpCredential::bearer("stripe", "sk_live_seed_secret"))
+        .unwrap();
+
+    let outcome = flows_list_connections(&config).await.unwrap();
+    let refs: Vec<_> = outcome
+        .value
+        .iter()
+        .map(|c| c.connection_ref.as_str())
+        .collect();
+    assert!(
+        refs.contains(&"http_cred:stripe"),
+        "http_cred must be surfaced: {refs:?}"
+    );
+
+    // The secret must never appear anywhere in the RPC payload.
+    let json = serde_json::to_string(&outcome.value).unwrap();
+    assert!(
+        !json.contains("sk_live_seed_secret"),
+        "secret leaked into flows_list_connections payload: {json}"
+    );
+}

src/openhuman/flows/schemas.rs

@@ -65,12 +65,55 @@ fn run_output_fields() -> Vec<FieldSchema> {
     ]
 }
 
+/// Field schema for one `FlowConnection` element of `flows_list_connections`'s
+/// output. Kept in one place so the schema mirrors
+/// `flows::types::FlowConnection` exactly — and documents that no secret field
+/// exists on the wire.
+fn flow_connection_fields() -> Vec<FieldSchema> {
+    vec![
+        FieldSchema {
+            name: "connection_ref",
+            ty: TypeSchema::String,
+            comment: "Ready-to-use `connection_ref` to stamp onto a node: \
+                      `composio:<toolkit>:<connection_id>` or `http_cred:<name>`.",
+            required: true,
+        },
+        FieldSchema {
+            name: "kind",
+            ty: TypeSchema::String,
+            comment: "Source kind: `composio` | `http`.",
+            required: true,
+        },
+        FieldSchema {
+            name: "display",
+            ty: TypeSchema::String,
+            comment: "Human-readable picker label (e.g. `Gmail · user@example.com`). \
+                      Never secret material.",
+            required: true,
+        },
+        FieldSchema {
+            name: "toolkit",
+            ty: TypeSchema::Option(Box::new(TypeSchema::String)),
+            comment: "Composio toolkit slug (kind `composio` only).",
+            required: false,
+        },
+        FieldSchema {
+            name: "scheme",
+            ty: TypeSchema::Option(Box::new(TypeSchema::String)),
+            comment: "HTTP credential injection scheme (kind `http` only): \
+                      `bearer` | `basic` | `header`.",
+            required: false,
+        },
+    ]
+}
+
 pub fn all_controller_schemas() -> Vec<ControllerSchema> {
     vec![
         schemas("create"),
         schemas("validate"),
         schemas("get"),
         schemas("list"),
+        schemas("list_connections"),
         schemas("update"),
         schemas("delete"),
         schemas("set_enabled"),
@@ -100,6 +143,10 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
             schema: schemas("list"),
             handler: handle_list,
         },
+        RegisteredController {
+            schema: schemas("list_connections"),
+            handler: handle_list_connections,
+        },
         RegisteredController {
             schema: schemas("update"),
             handler: handle_update,
@@ -212,6 +259,25 @@ pub fn schemas(function: &str) -> ControllerSchema {
                 required: true,
             }],
         },
+        "list_connections" => ControllerSchema {
+            namespace: "flows",
+            function: "list_connections",
+            description: "List the connection sources a flow node's `connection_ref` can attach \
+                          to: Composio connected accounts (kind `composio`) and stored HTTP \
+                          credentials (kind `http`). Returns ids + display labels + kind ONLY — \
+                          never any secret material (OAuth/bearer tokens, passwords, and API \
+                          keys stay server-side and are injected only at execution time).",
+            inputs: vec![],
+            outputs: vec![FieldSchema {
+                name: "connections",
+                ty: TypeSchema::Array(Box::new(TypeSchema::Object {
+                    fields: flow_connection_fields(),
+                })),
+                comment: "Resolvable connections for the flows picker (composio + http), \
+                          secret-free.",
+                required: true,
+            }],
+        },
         "update" => ControllerSchema {
             namespace: "flows",
             function: "update",
@@ -476,6 +542,13 @@ fn handle_list(_params: Map<String, Value>) -> ControllerFuture {
     })
 }
 
+fn handle_list_connections(_params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let config = config_rpc::load_config_with_timeout().await?;
+        to_json(ops::flows_list_connections(&config).await?)
+    })
+}
+
 fn handle_update(params: Map<String, Value>) -> ControllerFuture {
     Box::pin(async move {
         let config = config_rpc::load_config_with_timeout().await?;
@@ -614,6 +687,7 @@ mod tests {
                 "validate",
                 "get",
                 "list",
+                "list_connections",
                 "update",
                 "delete",
                 "set_enabled",
@@ -629,7 +703,7 @@ mod tests {
     #[test]
     fn all_registered_controllers_has_handler_per_schema() {
         let controllers = all_registered_controllers();
-        assert_eq!(controllers.len(), 12);
+        assert_eq!(controllers.len(), 13);
         let names: Vec<_> = controllers.iter().map(|c| c.schema.function).collect();
         assert_eq!(
             names,
@@ -638,6 +712,7 @@ mod tests {
                 "validate",
                 "get",
                 "list",
+                "list_connections",
                 "update",
                 "delete",
                 "set_enabled",
@@ -650,6 +725,41 @@ mod tests {
         );
     }
 
+    #[test]
+    fn schemas_list_connections_has_no_inputs_and_secret_free_outputs() {
+        let s = schemas("list_connections");
+        assert_eq!(s.namespace, "flows");
+        assert!(s.inputs.is_empty());
+        // The only output is the `connections` array.
+        assert_eq!(s.outputs.len(), 1);
+        assert_eq!(s.outputs[0].name, "connections");
+        // No field on a FlowConnection element may resemble secret material.
+        if let TypeSchema::Array(inner) = &s.outputs[0].ty {
+            if let TypeSchema::Object { fields } = inner.as_ref() {
+                let names: Vec<_> = fields.iter().map(|f| f.name).collect();
+                assert_eq!(
+                    names,
+                    vec!["connection_ref", "kind", "display", "toolkit", "scheme"]
+                );
+                for f in fields {
+                    let n = f.name.to_ascii_lowercase();
+                    assert!(
+                        !n.contains("secret")
+                            && !n.contains("token")
+                            && !n.contains("password")
+                            && !n.contains("key"),
+                        "flow_connection field '{}' looks secret-bearing",
+                        f.name
+                    );
+                }
+            } else {
+                panic!("connections element type is not an Object");
+            }
+        } else {
+            panic!("connections output is not an Array");
+        }
+    }
+
     #[test]
     fn schemas_create_requires_name_and_graph() {
         let s = schemas("create");

src/openhuman/flows/types.rs

@@ -121,6 +121,42 @@ pub struct FlowRunStep {
     pub duration_ms: Option<u64>,
 }
 
+/// A resolvable connection the flows UI / agent picker can attach to a node's
+/// `connection_ref`. Aggregated by `openhuman.flows_list_connections` from two
+/// host-side sources:
+///
+/// - **Composio connected accounts** (`kind = "composio"`) — each active OAuth
+///   integration instance, emitted as a ready-to-use
+///   `"composio:<toolkit>:<connection_id>"` ref (the exact shape
+///   `tinyflows::caps::composio_connection_id` parses back on execution).
+/// - **Named HTTP credentials** (`kind = "http"`) — each stored injection
+///   template, emitted as `"http_cred:<name>"` (the shape
+///   `tinyflows::caps::http_cred_name` parses).
+///
+/// **Security contract:** carries only non-secret identity — the
+/// `connection_ref` string plus a display label (and toolkit/scheme hints).
+/// It NEVER carries secret material (OAuth tokens, bearer tokens, passwords,
+/// API keys). Those stay server-side and are injected only inside the
+/// `tinyflows::caps` adapters at execution time.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct FlowConnection {
+    /// The ready-to-use `connection_ref` value to stamp onto a node:
+    /// `"composio:<toolkit>:<connection_id>"` or `"http_cred:<name>"`.
+    pub connection_ref: String,
+    /// Source kind: `"composio"` | `"http"`.
+    pub kind: String,
+    /// Human-readable label for the picker, e.g. `"Gmail · user@example.com"`
+    /// or `"stripe (bearer)"`. Never contains secret material.
+    pub display: String,
+    /// Composio toolkit slug (`kind = "composio"` only), e.g. `"gmail"`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub toolkit: Option<String>,
+    /// HTTP credential injection scheme (`kind = "http"` only):
+    /// `"bearer"` | `"basic"` | `"header"`. Not a secret.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub scheme: Option<String>,
+}
+
 /// A persisted record of one `flows_run` / `flows_resume` invocation, for the
 /// B3 run-history inspector. Written by `flows::store` from `flows::ops`.
 #[derive(Debug, Clone, Serialize, Deserialize)]

src/openhuman/tinyflows/tests.rs

@@ -90,6 +90,9 @@ fn http_adapter(allowed_domains: Vec<String>) -> OpenHumanHttp {
             allowed_domains,
             ..Default::default()
         },
+        http_creds: Arc::new(
+            crate::openhuman::credentials::HttpCredentialsStore::from_config(&config),
+        ),
     }
 }
 
@@ -203,7 +206,12 @@ async fn engine_run_drives_trigger_to_http_request_through_the_real_seam() {
 async fn code_adapter_javascript_passthrough_round_trips_json() {
     let tmp = TempDir::new().unwrap();
     let config = test_config(&tmp);
-    let runner = OpenHumanCode { config };
+    let security = Arc::new(SecurityPolicy::from_config(
+        &config.autonomy,
+        &config.workspace_dir,
+        &config.action_dir,
+    ));
+    let runner = OpenHumanCode { config, security };
 
     let input = json!([{ "json": { "n": 7 } }]);
     let result = runner

tests/json_rpc_e2e.rs

@@ -12730,6 +12730,68 @@ async fn json_rpc_flows_validate_reports_warnings_and_errors() {
     rpc_join.abort();
 }
 
+/// `openhuman.flows_list_connections` (PHASE 2): the connection picker source.
+/// Aggregates Composio connected accounts + stored HTTP credentials into a flat
+/// list of `connection_ref` + display + kind — and NEVER any secret material.
+///
+/// We seed one named HTTP credential (a bearer token) through the same
+/// host-side store the RPC reads, then assert the RPC surfaces it as
+/// `http_cred:<name>` with `kind = "http"` and that the token value never
+/// appears anywhere in the RPC payload. The Composio half is exercised for
+/// fault-tolerance: the mock upstream has no connected-accounts route, so the
+/// Composio source fails and is tolerated (the RPC still returns the HTTP half
+/// rather than erroring).
+#[tokio::test]
+async fn json_rpc_flows_list_connections_aggregates_secret_free() {
+    let _env_lock = json_rpc_e2e_env_lock();
+    let (rpc_base, _tmp, api_join, rpc_join, _guards) = boot_flows_rpc_env().await;
+
+    // Seed an HTTP credential through the same encrypted-at-rest store the op
+    // reads (config resolves under the guarded HOME set by boot_flows_rpc_env).
+    let seed_config = openhuman_core::openhuman::config::load_config_with_timeout()
+        .await
+        .expect("load config to seed http_cred");
+    const SECRET: &str = "sk_live_flows_list_connections_seed";
+    openhuman_core::openhuman::credentials::HttpCredentialsStore::from_config(&seed_config)
+        .upsert(&openhuman_core::openhuman::credentials::HttpCredential::bearer("stripe", SECRET))
+        .expect("seed http_cred");
+
+    let resp = post_json_rpc(
+        &rpc_base,
+        9330,
+        "openhuman.flows_list_connections",
+        json!({}),
+    )
+    .await;
+    let raw = assert_no_jsonrpc_error(&resp, "flows_list_connections");
+
+    // The seeded secret must never appear anywhere in the RPC response.
+    let raw_str = raw.to_string();
+    assert!(
+        !raw_str.contains(SECRET),
+        "secret leaked into flows_list_connections payload: {raw_str}"
+    );
+
+    let connections = peel_logs_envelope(raw)
+        .as_array()
+        .expect("connections is an array")
+        .clone();
+
+    let stripe = connections
+        .iter()
+        .find(|c| c.get("connection_ref").and_then(Value::as_str) == Some("http_cred:stripe"))
+        .expect("seeded http_cred surfaced in picker");
+    assert_eq!(stripe.get("kind").and_then(Value::as_str), Some("http"));
+    assert_eq!(stripe.get("scheme").and_then(Value::as_str), Some("bearer"));
+    assert!(
+        stripe.get("display").and_then(Value::as_str).is_some(),
+        "http_cred entry must carry a display label"
+    );
+
+    api_join.abort();
+    rpc_join.abort();
+}
+
 /// Task 4 / #3090: when a web-chat request is sent with
 /// `speak_reply: true`, `run_chat_task` should drive the agent's final text
 /// through `voice::reply_speech::synthesize_reply` after the turn completes.