feat(flows): Workflows B2 — triggers, resume, trust & run history

Cargo.toml

@@ -44,7 +44,7 @@ tinyplace = "1.0.1"
 # run on tinyagents). Powers the "Workflows" feature via the seam in
 # `src/openhuman/tinyflows/` + the `flows::` domain. Pulls tinyagents 1.3 transitively
 # (same version openhuman already uses — no conflict). Published on crates.io.
-tinyflows = "0.2"
+tinyflows = "0.3"
 # TinyAgents — Rust LLM orchestration framework (LangGraph/LangChain-style):
 # durable state graphs, agent-loop harness, model/tool registries, REPL +
 # `.rag` workflow language. openhuman's agent engine + orchestration run on this

src/core/event_bus/events.rs

@@ -426,6 +426,17 @@ pub enum DomainEvent {
         /// Optional job name for display/threading purposes.
         job_name: Option<String>,
     },
+    /// A `flow`-type cron job fired its schedule tick (issue B2,
+    /// `my_docs/ohxtf/b2-triggers-trust/01-triggers-and-trust.md` §1).
+    /// Published by `cron::scheduler` when a `JobType::Flow` job comes due;
+    /// carries only the flow id (the trigger payload for a `schedule` flow is
+    /// always empty — `flows::ops::flows_run` seeds `{}`). Consumed by
+    /// `flows::bus::FlowTriggerSubscriber`, which loads the flow, checks it is
+    /// still enabled with a `schedule` trigger, and spawns a run.
+    FlowScheduleTick {
+        /// Identifier of the `flows::Flow` to run.
+        flow_id: String,
+    },
 
     // ── Skills ──────────────────────────────────────────────────────────
     /// A skill was loaded into the runtime.
@@ -1281,7 +1292,8 @@ impl DomainEvent {
             Self::CronJobTriggered { .. }
             | Self::CronJobCompleted { .. }
             | Self::CronDeliveryRequested { .. }
-            | Self::ProactiveMessageRequested { .. } => "cron",
+            | Self::ProactiveMessageRequested { .. }
+            | Self::FlowScheduleTick { .. } => "cron",
 
             Self::WorkflowLoaded { .. }
             | Self::WorkflowStopped { .. }
@@ -1439,6 +1451,7 @@ impl DomainEvent {
             Self::CronJobCompleted { .. } => "CronJobCompleted",
             Self::CronDeliveryRequested { .. } => "CronDeliveryRequested",
             Self::ProactiveMessageRequested { .. } => "ProactiveMessageRequested",
+            Self::FlowScheduleTick { .. } => "FlowScheduleTick",
             Self::WorkflowLoaded { .. } => "WorkflowLoaded",
             Self::WorkflowStopped { .. } => "WorkflowStopped",
             Self::WorkflowStartFailed { .. } => "WorkflowStartFailed",

src/core/event_bus/events_tests.rs

@@ -203,6 +203,12 @@ fn all_variants_have_correct_domain() {
             },
             "cron",
         ),
+        (
+            DomainEvent::FlowScheduleTick {
+                flow_id: "flow-1".into(),
+            },
+            "cron",
+        ),
         // Workflow
         (
             DomainEvent::WorkflowLoaded {

src/core/jsonrpc.rs

@@ -2115,6 +2115,17 @@ async fn run_server_inner(
                 {
                     log::warn!("[cron] boot seed of proactive agent jobs failed: {e}");
                 }
+                // Re-register the cron job for every enabled, schedule-trigger
+                // flow (issue B2) — idempotent, so a flow whose binding
+                // predates this feature (or was otherwise lost) gets its
+                // schedule re-registered without the user re-toggling it.
+                if let Err(e) =
+                    crate::openhuman::flows::ops::reconcile_schedule_triggers_on_boot(&config).await
+                {
+                    log::warn!(
+                        "[flows] boot reconciliation of schedule-trigger cron jobs failed: {e}"
+                    );
+                }
                 if let Err(e) = crate::openhuman::cron::scheduler::run(config).await {
                     log::error!("[cron] scheduler loop ended with error: {e}");
                 }
@@ -2222,6 +2233,20 @@ fn register_domain_subscribers(
             log::warn!("[event_bus] failed to register channel subscriber — bus not initialized");
         }
 
+        // Flows trigger dispatch (issue B2): maps FlowScheduleTick /
+        // ComposioTriggerReceived / WebhookIncomingRequest onto enabled flows and
+        // runs `flows::ops::flows_run`. Registered here (unconditional core boot,
+        // Once-guarded) rather than under channel startup, so schedule/app-event
+        // workflows still dispatch when no realtime channel is configured or
+        // `OPENHUMAN_DISABLE_CHANNEL_LISTENERS` short-circuits `start_channels`.
+        if let Some(handle) = crate::core::event_bus::subscribe_global(Arc::new(
+            crate::openhuman::flows::bus::FlowTriggerSubscriber::new(Arc::new(config.clone())),
+        )) {
+            std::mem::forget(handle);
+        } else {
+            log::warn!("[event_bus] failed to register flows trigger subscriber — bus not initialized");
+        }
+
         crate::openhuman::health::bus::register_health_subscriber();
         crate::openhuman::notifications::register_notification_bridge_subscriber(config.clone());
         crate::openhuman::memory_conversations::register_conversation_persistence_subscriber(

src/openhuman/agent/turn_origin.rs

@@ -78,6 +78,25 @@ pub enum TrustedAutomationSource {
     /// Autonomous continuation of a thread goal: the heartbeat injected a turn
     /// to keep working an idle `active` goal the user explicitly created.
     GoalContinuation,
+    /// A saved, enabled `flows::Flow` (tinyflows workflow) executing via
+    /// `flows::ops::flows_run` / `flows_resume` (issue B2, see
+    /// `my_docs/ohxtf/b2-triggers-trust/01-triggers-and-trust.md` §3). The
+    /// flow's `tool_call`/`http_request` nodes were pre-declared (their
+    /// `slug`/`url` are static graph config, never `=`-expression evaluated
+    /// in tinyflows 0.2 — see `my_docs/ohxtf/commons/12-node-catalog-0.2.md`)
+    /// and validated when the flow was saved, so the *action* carries a trust
+    /// root the same way a user-authored cron job's prompt does. The runtime
+    /// trigger payload (webhook body, Composio event, …) stays untrusted —
+    /// nothing in it can introduce a *new* action, only feed the pre-declared
+    /// one's arguments.
+    Workflow {
+        /// Mirrors `Flow::require_approval`: when `true` the gate does NOT
+        /// auto-allow this trust root — every external_effect call still
+        /// parks for a real decision (same shape as `GoalContinuation`),
+        /// letting a user force human review on a specific flow's outbound
+        /// actions regardless of the trust root above.
+        require_approval: bool,
+    },
 }
 
 tokio::task_local! {

src/openhuman/approval/gate.rs

@@ -320,12 +320,21 @@ impl ApprovalGate {
         // 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
-        // origin and fall through to the parking flow below.
-        let is_goal_continuation = matches!(
+        // origin and fall through to the parking flow below. A workflow run
+        // whose flow has `require_approval` set gets the same treatment — the
+        // user explicitly asked for every outbound action on that flow to be
+        // gated, and a global tool allowlist must not silently override that
+        // per-flow choice.
+        let bypass_auto_approve_shortcut = matches!(
             &origin,
             AgentTurnOrigin::TrustedAutomation {
                 source: TrustedAutomationSource::GoalContinuation,
                 ..
+            } | AgentTurnOrigin::TrustedAutomation {
+                source: TrustedAutomationSource::Workflow {
+                    require_approval: true
+                },
+                ..
             }
         );
 
@@ -335,7 +344,7 @@ impl ApprovalGate {
         // live policy) takes effect on the very next tool call; fall back to the
         // gate's boot-time config when no live policy is installed (e.g. a CLI
         // invocation that never started a session runtime, or a unit test).
-        if !is_goal_continuation && self.tool_is_auto_approved(tool_name) {
+        if !bypass_auto_approve_shortcut && self.tool_is_auto_approved(tool_name) {
             tracing::debug!(
                 tool = tool_name,
                 "[approval::gate] auto_approve allowlist hit, skipping prompt"
@@ -446,6 +455,44 @@ impl ApprovalGate {
                 // irreversible external action. Read/compute tools (not gated
                 // here) still make progress on the goal.
             }
+            AgentTurnOrigin::TrustedAutomation {
+                source:
+                    TrustedAutomationSource::Workflow {
+                        require_approval: false,
+                    },
+                job_id,
+            } => {
+                tracing::debug!(
+                    tool = tool_name,
+                    flow_id = %job_id,
+                    "[approval::gate] trusted workflow automation — pre-declared action, \
+                     allowing without prompt"
+                );
+                return (GateOutcome::Allow, None);
+            }
+            AgentTurnOrigin::TrustedAutomation {
+                source:
+                    TrustedAutomationSource::Workflow {
+                        require_approval: true,
+                    },
+                job_id,
+            } => {
+                tracing::info!(
+                    tool = tool_name,
+                    flow_id = %job_id,
+                    "[approval::gate] workflow run has require_approval enabled — parking for \
+                     HITL review instead of auto-allowing the trust root"
+                );
+                // Fall through to the parking flow (same shape as
+                // GoalContinuation): persists a `pending_approvals` audit row
+                // and publishes `ApprovalRequested`. There is no chat thread to
+                // route the prompt to for a background/triggered flow run yet
+                // (B3 will add a dedicated review surface) — a caller can still
+                // decide it via `approval_decide` (e.g. a generic pending-
+                // approvals list) before the TTL elapses; absent a decision this
+                // TTL-denies, the conservative fail-closed default for a
+                // user-forced HITL gate.
+            }
             AgentTurnOrigin::Cli => {
                 tracing::debug!(
                     tool = tool_name,
@@ -1351,6 +1398,74 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn intercept_with_workflow_origin_trust_root_allows_without_prompt() {
+        // A saved+enabled flow's pre-declared tool/HTTP action (trust root,
+        // `require_approval: false`) is allowed without a prompt.
+        let (gate, _dir) = test_gate();
+        let origin = AgentTurnOrigin::TrustedAutomation {
+            job_id: "flow-1".into(),
+            source: TrustedAutomationSource::Workflow {
+                require_approval: false,
+            },
+        };
+        let outcome = turn_origin::with_origin(
+            origin,
+            gate.intercept("composio", "post to slack", serde_json::json!({})),
+        )
+        .await;
+        assert!(matches!(outcome, GateOutcome::Allow));
+        assert!(
+            gate.list_pending().unwrap().is_empty(),
+            "a trusted workflow action must not persist a pending row"
+        );
+    }
+
+    #[tokio::test]
+    async fn intercept_with_workflow_require_approval_persists_and_ttl_denies() {
+        // A per-flow `require_approval: true` toggle forces every external
+        // action through the HITL gate even though the origin carries a
+        // trust root — same conservative park-and-audit shape as
+        // `GoalContinuation` / `ExternalChannel`, since there is no flow
+        // review surface to route the prompt to yet (B3).
+        let (gate, _dir) = test_gate(); // 2s TTL
+        let gate = Arc::new(gate);
+        let origin = AgentTurnOrigin::TrustedAutomation {
+            job_id: "flow-2".into(),
+            source: TrustedAutomationSource::Workflow {
+                require_approval: true,
+            },
+        };
+
+        let g = gate.clone();
+        let handle = tokio::spawn(async move {
+            turn_origin::with_origin(
+                origin,
+                g.intercept("composio", "post to slack", serde_json::json!({})),
+            )
+            .await
+        });
+
+        let mut tries = 0;
+        loop {
+            if !gate.list_pending().unwrap().is_empty() {
+                break;
+            }
+            tries += 1;
+            assert!(
+                tries < 50,
+                "audit row never appeared for require_approval workflow origin"
+            );
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+
+        let outcome = handle.await.unwrap();
+        match outcome {
+            GateOutcome::Deny { reason } => assert!(reason.contains("timed out")),
+            other => panic!("expected deny, got {other:?}"),
+        }
+    }
+
     #[tokio::test]
     async fn intercept_with_trusted_subconscious_origin_allows_without_prompt() {
         // Subconscious ticks on internal-only memory are trusted automation

src/openhuman/channels/runtime/startup.rs

@@ -702,11 +702,11 @@ pub async fn start_channels(mut config: Config) -> Result<()> {
     let _cron_delivery_handle = bus.subscribe(Arc::new(
         crate::openhuman::cron::bus::CronDeliverySubscriber::new(Arc::clone(&channels_by_name)),
     ));
-    // Register the flows trigger subscriber (B1: observes matched events and
-    // logs them; B2 maps them onto enabled flows and calls `flows_run`).
-    let _flows_trigger_handle = bus.subscribe(Arc::new(
-        crate::openhuman::flows::bus::FlowTriggerSubscriber::new(),
-    ));
+    // NOTE: the flows `FlowTriggerSubscriber` is registered in
+    // `jsonrpc.rs::register_domain_subscribers` (unconditional core boot), NOT
+    // here — `start_channels` is skipped when no channel is configured or
+    // `OPENHUMAN_DISABLE_CHANNEL_LISTENERS` is set, which would otherwise leave
+    // schedule/app-event workflows undispatched (issue B2 review).
     // Register the proactive message subscriber so morning briefings,
     // welcome messages, and other proactive agent output gets routed to
     // the user's active channel (+ always to web).

src/openhuman/cron/mod.rs

@@ -21,9 +21,10 @@ pub use schemas::{
 };
 #[allow(unused_imports)]
 pub use store::{
-    add_agent_job, add_agent_job_with_definition, add_job, add_shell_job, clear_all_jobs,
-    dedup_named_jobs, delete_queued_runs, due_jobs, get_job, list_jobs, list_runs, record_last_run,
-    record_run, remove_job, reschedule_after_run, update_job,
+    add_agent_job, add_agent_job_with_definition, add_flow_schedule_job, add_job, add_shell_job,
+    clear_all_jobs, dedup_named_jobs, delete_queued_runs, due_jobs, find_flow_schedule_job,
+    get_job, list_jobs, list_runs, record_last_run, record_run, remove_job, reschedule_after_run,
+    update_job,
 };
 pub use types::{
     ActiveHours, CronJob, CronJobPatch, CronRun, DeliveryConfig, JobType, Schedule, SessionTarget,

src/openhuman/cron/scheduler.rs

@@ -453,6 +453,10 @@ async fn execute_job_with_retry(
                 (success, output, None)
             }
             JobType::Agent => run_agent_job(config, job).await,
+            JobType::Flow => {
+                let (success, output) = run_flow_schedule_job(job);
+                (success, output, None)
+            }
         };
         last_output = output;
         if agent_error.is_some() {
@@ -890,6 +894,29 @@ async fn run_agent_job(config: &Config, job: &CronJob) -> (bool, String, Option<
     }
 }
 
+/// Fires a `JobType::Flow` job: publishes `DomainEvent::FlowScheduleTick` for
+/// the bound flow id (stored in `job.command`, see `JobType::Flow`'s doc) and
+/// returns immediately. This job type does no work itself — dispatching the
+/// actual `flows::ops::flows_run` happens asynchronously in
+/// `flows::bus::FlowTriggerSubscriber`, which is the sole consumer of this
+/// event (kept out of the cron domain so cron stays flow-agnostic).
+fn run_flow_schedule_job(job: &CronJob) -> (bool, String) {
+    let flow_id = job.command.clone();
+    tracing::info!(
+        target: "flows",
+        job_id = %job.id,
+        %flow_id,
+        "[cron] flow schedule tick — publishing FlowScheduleTick"
+    );
+    publish_global(DomainEvent::FlowScheduleTick {
+        flow_id: flow_id.clone(),
+    });
+    (
+        true,
+        format!("flow schedule tick emitted for flow {flow_id}"),
+    )
+}
+
 /// Placeholder recorded in run history when an agent job succeeds but returns
 /// no text. Never delivered to chat — used only for the run-history record.
 const EMPTY_AGENT_OUTPUT: &str = "agent job executed";