diff --git a/Cargo.lock b/Cargo.lock index ad67f2e17..2bfc7fa61 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6693,9 +6693,9 @@ dependencies = [ [[package]] name = "tinyflows" -version = "0.2.0" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "11d5c7fbe95186c2a2b29cbfbcdde6c5241977261a55fc50bbdce203c1b568d2" +checksum = "17ea75c3f90edabea59dc15f5a43d290bc4175bc6f1e2e930da32a304a5844d9" dependencies = [ "async-trait", "futures-timer", diff --git a/Cargo.toml b/Cargo.toml index 252ecc694..c1545fdca 100644 --- a/Cargo.toml +++ b/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 diff --git a/app/src-tauri/Cargo.lock b/app/src-tauri/Cargo.lock index 445862461..83a08db47 100644 --- a/app/src-tauri/Cargo.lock +++ b/app/src-tauri/Cargo.lock @@ -9094,9 +9094,9 @@ dependencies = [ [[package]] name = "tinyflows" -version = "0.2.0" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "11d5c7fbe95186c2a2b29cbfbcdde6c5241977261a55fc50bbdce203c1b568d2" +checksum = "17ea75c3f90edabea59dc15f5a43d290bc4175bc6f1e2e930da32a304a5844d9" dependencies = [ "async-trait", "futures-timer", diff --git a/src/core/event_bus/events.rs b/src/core/event_bus/events.rs index 20d7b6827..fd88f8565 100644 --- a/src/core/event_bus/events.rs +++ b/src/core/event_bus/events.rs @@ -426,6 +426,17 @@ pub enum DomainEvent { /// Optional job name for display/threading purposes. job_name: Option, }, + /// 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", diff --git a/src/core/event_bus/events_tests.rs b/src/core/event_bus/events_tests.rs index 110dced61..fa1ebe909 100644 --- a/src/core/event_bus/events_tests.rs +++ b/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 { diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index ffc311b48..4c32d3388 100644 --- a/src/core/jsonrpc.rs +++ b/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( diff --git a/src/openhuman/agent/turn_origin.rs b/src/openhuman/agent/turn_origin.rs index 2d53dcd8d..01a4ef1b2 100644 --- a/src/openhuman/agent/turn_origin.rs +++ b/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! { diff --git a/src/openhuman/approval/gate.rs b/src/openhuman/approval/gate.rs index 13ad451e3..f6a6bd1d5 100644 --- a/src/openhuman/approval/gate.rs +++ b/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 diff --git a/src/openhuman/channels/runtime/startup.rs b/src/openhuman/channels/runtime/startup.rs index e6db8f63c..f337adc3f 100644 --- a/src/openhuman/channels/runtime/startup.rs +++ b/src/openhuman/channels/runtime/startup.rs @@ -703,11 +703,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). diff --git a/src/openhuman/cron/mod.rs b/src/openhuman/cron/mod.rs index 904e64d18..5602f2dd1 100644 --- a/src/openhuman/cron/mod.rs +++ b/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, diff --git a/src/openhuman/cron/scheduler.rs b/src/openhuman/cron/scheduler.rs index eb5878639..cecf1ff85 100644 --- a/src/openhuman/cron/scheduler.rs +++ b/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"; diff --git a/src/openhuman/cron/store.rs b/src/openhuman/cron/store.rs index 217efeea0..7f4e4ae13 100644 --- a/src/openhuman/cron/store.rs +++ b/src/openhuman/cron/store.rs @@ -139,6 +139,97 @@ pub fn add_agent_job_with_definition( get_job(config, &id) } +/// Registers the cron job that fires a `flows::Flow`'s `schedule` trigger +/// (issue B2). The flow's id is stored in `command` — a flow-schedule job has +/// no shell command / agent prompt of its own, it only needs to name which +/// flow to tick (see `JobType::Flow`'s doc). On fire the scheduler publishes +/// `DomainEvent::FlowScheduleTick { flow_id: command }` instead of running +/// anything; `flows::bus::FlowTriggerSubscriber` does the actual dispatch. +/// +/// Race-safe / idempotent: `bind_schedule_trigger` does check-then-act +/// (`find_flow_schedule_job` then this function), so two concurrent binds for +/// the same flow can both observe "no job yet". The `idx_cron_jobs_flow_command` +/// partial unique index (flow jobs only) turns the loser's `INSERT` into a +/// no-op via `ON CONFLICT ... DO NOTHING`, and that loser then looks up and +/// returns the winner's row instead of erroring — callers always get back +/// exactly one cron job for `flow_id`, never a duplicate and never a +/// constraint-violation error. +pub fn add_flow_schedule_job( + config: &Config, + flow_id: &str, + schedule: Schedule, +) -> Result { + let now = Utc::now(); + validate_schedule(&schedule, now)?; + let next_run = next_run_for_schedule(&schedule, now)?; + let id = Uuid::new_v4().to_string(); + let expression = schedule_cron_expression(&schedule).unwrap_or_default(); + let schedule_json = serde_json::to_string(&schedule)?; + let name = format!("flow:{flow_id}"); + + let inserted_rows = with_connection(config, |conn| { + let rows = conn + .execute( + "INSERT INTO cron_jobs ( + id, expression, command, schedule, job_type, prompt, name, session_target, model, + enabled, delivery, delete_after_run, created_at, next_run + ) VALUES (?1, ?2, ?3, ?4, 'flow', NULL, ?5, 'isolated', NULL, 1, ?6, 0, ?7, ?8) + ON CONFLICT (command) WHERE job_type = 'flow' DO NOTHING", + params![ + id, + expression, + flow_id, + schedule_json, + name, + serde_json::to_string(&DeliveryConfig::default())?, + now.to_rfc3339(), + next_run.to_rfc3339(), + ], + ) + .context("Failed to insert cron flow-schedule job")?; + Ok(rows) + })?; + + if inserted_rows > 0 { + get_job(config, &id) + } else { + // Lost the race — another caller already holds the flow-schedule job + // for this flow_id/command. Return its row rather than erroring so + // `add_flow_schedule_job` is safe to call twice concurrently. + tracing::debug!( + target: "cron", + %flow_id, + "[cron] add_flow_schedule_job: insert conflicted with an existing flow job — returning the existing binding" + ); + find_flow_schedule_job(config, flow_id)?.with_context(|| { + format!( + "add_flow_schedule_job: insert for flow '{flow_id}' conflicted but no existing \ + flow-schedule job was found" + ) + }) + } +} + +/// Finds the cron job (if any) registered for a flow's `schedule` trigger — +/// used by `flows::ops::flows_set_enabled` to make enable/disable idempotent +/// (re-use the existing binding rather than creating a duplicate) and to tear +/// it down on disable. +pub fn find_flow_schedule_job(config: &Config, flow_id: &str) -> Result> { + with_connection(config, |conn| { + let mut stmt = conn.prepare( + "SELECT id, expression, command, schedule, job_type, prompt, name, session_target, model, + enabled, delivery, delete_after_run, created_at, next_run, last_run, last_status, last_output, + agent_id + FROM cron_jobs WHERE job_type = 'flow' AND command = ?1 LIMIT 1", + )?; + let mut rows = stmt.query(params![flow_id])?; + match rows.next()? { + Some(row) => Ok(Some(map_cron_job_row(row)?)), + None => Ok(None), + } + }) +} + pub fn list_jobs(config: &Config) -> Result> { with_connection(config, |conn| { let mut stmt = conn.prepare( @@ -712,7 +803,18 @@ fn with_connection(config: &Config, f: impl FnOnce(&Connection) -> Result) ); CREATE INDEX IF NOT EXISTS idx_cron_runs_job_id ON cron_runs(job_id); CREATE INDEX IF NOT EXISTS idx_cron_runs_started_at ON cron_runs(started_at); - CREATE INDEX IF NOT EXISTS idx_cron_runs_job_started ON cron_runs(job_id, started_at);", + CREATE INDEX IF NOT EXISTS idx_cron_runs_job_started ON cron_runs(job_id, started_at); + + -- Guards against duplicate flow-schedule cron bindings under a + -- concurrent `bind_schedule_trigger` (issue B2 CodeRabbit finding): + -- `flows::ops::bind_schedule_trigger` does check-then-act + -- (`find_flow_schedule_job` then `add_flow_schedule_job`), so two + -- racing binds for the same flow could otherwise each observe 'no + -- job' and insert a duplicate. Scoped to `job_type = 'flow'` via a + -- partial index so it can never constrain shell/agent jobs, which + -- may legitimately share a `command`. + CREATE UNIQUE INDEX IF NOT EXISTS idx_cron_jobs_flow_command + ON cron_jobs(command) WHERE job_type = 'flow';", ) .context("Failed to initialize cron schema")?; diff --git a/src/openhuman/cron/store_tests.rs b/src/openhuman/cron/store_tests.rs index 3d42d13b9..87f67269a 100644 --- a/src/openhuman/cron/store_tests.rs +++ b/src/openhuman/cron/store_tests.rs @@ -474,6 +474,53 @@ fn dedup_named_jobs_keeps_earliest_when_history_tied() { assert!(get_job(&config, &job_b.id).is_err()); } +// ── add_flow_schedule_job race-safety (CodeRabbit finding A) ──────── + +#[test] +fn add_flow_schedule_job_twice_yields_a_single_row() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let schedule = Schedule::Cron { + expr: "0 9 * * *".into(), + tz: None, + active_hours: None, + }; + + let first = add_flow_schedule_job(&config, "flow-1", schedule.clone()).unwrap(); + let second = add_flow_schedule_job(&config, "flow-1", schedule).unwrap(); + + // Calling it twice for the same flow must not create a duplicate — the + // second call returns the same row the first one created. + assert_eq!(first.id, second.id); + + let flow_jobs: Vec<_> = list_jobs(&config) + .unwrap() + .into_iter() + .filter(|j| j.job_type == JobType::Flow && j.command == "flow-1") + .collect(); + assert_eq!( + flow_jobs.len(), + 1, + "exactly one job_type='flow' row should exist for flow-1" + ); +} + +#[test] +fn add_flow_schedule_job_unique_index_does_not_affect_shell_jobs() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + // Two shell jobs sharing the same command must both persist — the new + // partial unique index is scoped to job_type = 'flow' and must not + // constrain shell/agent jobs, which may legitimately share a command. + let shell_a = add_job(&config, "*/5 * * * *", "echo shared").unwrap(); + let shell_b = add_job(&config, "*/10 * * * *", "echo shared").unwrap(); + + assert!(get_job(&config, &shell_a.id).is_ok()); + assert!(get_job(&config, &shell_b.id).is_ok()); + assert_eq!(list_jobs(&config).unwrap().len(), 2); +} + #[test] fn dedup_named_jobs_ignores_unnamed_jobs() { let tmp = TempDir::new().unwrap(); diff --git a/src/openhuman/cron/tools/add.rs b/src/openhuman/cron/tools/add.rs index 4512ff42e..5f6e76fda 100644 --- a/src/openhuman/cron/tools/add.rs +++ b/src/openhuman/cron/tools/add.rs @@ -353,6 +353,17 @@ impl Tool for CronAddTool { delete_after_run, ) } + // `job_type` above is derived only from `Some("agent")`/`Some("shell")`/ + // the `prompt`-presence heuristic, so this arm is unreachable in + // practice — `JobType::Flow` rows are created internally by + // `flows::ops::flows_set_enabled` (via `cron::add_flow_schedule_job`), + // never through this agent-facing tool. Kept as an explicit error + // (not `unreachable!()`) so a future change to the heuristic above + // fails loudly with a clear message instead of panicking. + JobType::Flow => Err(anyhow::anyhow!( + "flow-type cron jobs are managed by the Workflows feature and cannot be \ + created via cron_add" + )), }; match result { diff --git a/src/openhuman/cron/types.rs b/src/openhuman/cron/types.rs index b88c728ab..db7d0652a 100644 --- a/src/openhuman/cron/types.rs +++ b/src/openhuman/cron/types.rs @@ -7,6 +7,14 @@ pub enum JobType { #[default] Shell, Agent, + /// A `flows::Flow` schedule trigger binding (issue B2). The job's + /// `command` column carries the bound flow's id (there is no shell + /// command / agent prompt to run); on fire the scheduler publishes + /// `DomainEvent::FlowScheduleTick { flow_id }` instead of running + /// anything itself — `flows::bus::FlowTriggerSubscriber` does the actual + /// dispatch. Created by `flows::ops::flows_set_enabled` (via + /// `cron::add_flow_schedule_job`), never via the `cron_add` agent tool. + Flow, } impl JobType { @@ -14,12 +22,15 @@ impl JobType { match self { Self::Shell => "shell", Self::Agent => "agent", + Self::Flow => "flow", } } pub(crate) fn parse(raw: &str) -> Self { if raw.eq_ignore_ascii_case("agent") { Self::Agent + } else if raw.eq_ignore_ascii_case("flow") { + Self::Flow } else { Self::Shell } @@ -268,9 +279,11 @@ mod tests { fn job_type_parse_and_as_str_roundtrip() { assert_eq!(JobType::parse("shell").as_str(), "shell"); assert_eq!(JobType::parse("agent").as_str(), "agent"); + assert_eq!(JobType::parse("flow").as_str(), "flow"); // Case-insensitive assert_eq!(JobType::parse("AGENT"), JobType::Agent); assert_eq!(JobType::parse("Agent"), JobType::Agent); + assert_eq!(JobType::parse("FLOW"), JobType::Flow); // Anything unknown falls back to Shell (the default) — guards // against unexpected legacy DB rows silently turning into Agent. assert_eq!(JobType::parse(""), JobType::Shell); diff --git a/src/openhuman/flows/bus.rs b/src/openhuman/flows/bus.rs index 1b0c9032a..a4d41cbc6 100644 --- a/src/openhuman/flows/bus.rs +++ b/src/openhuman/flows/bus.rs @@ -1,30 +1,199 @@ -//! Event bus handlers for the `flows::` domain. +//! Event bus handlers for the `flows::` domain (issue B2 — see +//! `my_docs/ohxtf/b2-triggers-trust/01-triggers-and-trust.md` §1). //! -//! B1 scope: this subscriber only **observes** and logs events that will later -//! drive automatic flow runs (B2). It does not yet map a matched event onto an -//! enabled [`crate::openhuman::flows::Flow`] or call `flows_run` — that bridge, -//! plus `TrustedAutomationSource::Workflow` for externally-triggered runs, is -//! B2+ (see `my_docs/ohxtf/b1-engine-seam-domain/07-execution-and-hitl.md`). +//! [`FlowTriggerSubscriber`] is the trigger → run bridge: it listens for the +//! normalized events a saved flow's trigger node can bind to +//! (`DomainEvent::FlowScheduleTick`, `ComposioTriggerReceived`, +//! `WebhookIncomingRequest`), matches them against enabled flows, and spawns +//! `flows::ops::flows_run` for each match. Matching helpers +//! ([`extract_trigger_kind`], [`extract_trigger_config`]) are also reused by +//! `flows::ops::flows_set_enabled` to bind/unbind a flow's automatic +//! dispatch on enable/disable. use crate::core::event_bus::{DomainEvent, EventHandler}; +use crate::openhuman::config::Config; +use crate::openhuman::flows::store; +use crate::openhuman::flows::Flow; use async_trait::async_trait; +use serde_json::Value; +use std::collections::HashSet; +use std::sync::{Arc, Mutex}; +use tinyflows::model::TriggerKind; -/// Listens for events that a saved flow's trigger node might match -/// (`cron`, `composio`, `system` domains) and logs them. -/// -/// A future revision resolves matched events to enabled flows and invokes -/// `flows::ops::flows_run` for each match. -pub struct FlowTriggerSubscriber; +/// Reads `trigger_kind` from a flow's trigger node config, deserializing into +/// `tinyflows::model::TriggerKind`. Returns `None` when the flow doesn't have +/// exactly one trigger node ([`tinyflows::model::WorkflowGraph::trigger`]) or +/// the `trigger_kind` discriminator is missing/invalid — callers treat that +/// as "no automatic binding", not an error (a `manual`-only or legacy graph +/// authored before B2 simply never fires itself). +pub(crate) fn extract_trigger_kind(flow: &Flow) -> Option { + let trigger = flow.graph.trigger()?; + serde_json::from_value(trigger.config.get("trigger_kind")?.clone()).ok() +} + +/// Returns the trigger node's full config value, for callers that need +/// kind-specific fields (`schedule` for `schedule`, `toolkit`/`trigger_slug` +/// for `app_event`, …). +pub(crate) fn extract_trigger_config(flow: &Flow) -> Option<&Value> { + Some(&flow.graph.trigger()?.config) +} + +/// True when `flow` is an enabled `app_event` flow bound to the given +/// Composio `toolkit`/`trigger_slug` (case-insensitive — Composio slugs are +/// conventionally upper-case but authoring surfaces may not normalize them). +fn matches_app_event(flow: &Flow, toolkit: &str, trigger_slug: &str) -> bool { + if !matches!(extract_trigger_kind(flow), Some(TriggerKind::AppEvent)) { + return false; + } + let Some(cfg) = extract_trigger_config(flow) else { + return false; + }; + let cfg_toolkit = cfg.get("toolkit").and_then(Value::as_str).unwrap_or(""); + let cfg_slug = cfg + .get("trigger_slug") + .and_then(Value::as_str) + .unwrap_or(""); + cfg_toolkit.eq_ignore_ascii_case(toolkit) && cfg_slug.eq_ignore_ascii_case(trigger_slug) +} + +/// Listens for normalized trigger events and starts runs for matching +/// enabled flows. See the module doc for the full contract. +pub struct FlowTriggerSubscriber { + config: Arc, + /// Process-local dedupe of trigger-driven dispatch, keyed by `flow_id` + /// (CodeRabbit finding B — overlapping runs for the same flow). A fast + /// cadence or trigger burst can otherwise fire `spawn_run` for the same + /// flow multiple times before the first run finishes, racing + /// `last_run_at`/`last_status` and doing duplicate work. This is + /// intentionally scoped to trigger-driven dispatch (this subscriber) — + /// the interactive `flows_run` RPC is NOT deduped, since a user + /// explicitly asking to run a flow again (e.g. while a scheduled run is + /// still in flight) is fine. + in_flight: Arc>>, +} impl FlowTriggerSubscriber { - pub fn new() -> Self { - Self + pub fn new(config: Arc) -> Self { + Self { + config, + in_flight: Arc::new(Mutex::new(HashSet::new())), + } + } + + /// Attempts to claim `flow_id` for a trigger-driven dispatch. Returns + /// `None` when a dispatch for the same flow is already in flight — the + /// caller should skip this tick. Returns `Some(guard)` on success; the + /// guard releases the claim on `Drop` (including on panic/early return), + /// so a run can never permanently wedge the flow out of future ticks. + fn try_acquire_dispatch(&self, flow_id: &str) -> Option { + let mut in_flight = self.in_flight.lock().unwrap_or_else(|e| e.into_inner()); + if !in_flight.insert(flow_id.to_string()) { + return None; + } + Some(InFlightGuard { + set: self.in_flight.clone(), + flow_id: flow_id.to_string(), + }) + } + + /// `DomainEvent::FlowScheduleTick` — a `flow`-type cron job fired. Loads + /// the one named flow, checks it is still enabled with a `schedule` + /// trigger (it may have been disabled/edited since the job was + /// registered), and dispatches it with an empty trigger payload. + async fn handle_schedule_tick(&self, flow_id: &str) { + let flow = match store::get_flow(&self.config, flow_id) { + Ok(Some(flow)) => flow, + Ok(None) => { + tracing::debug!(target: "flows", %flow_id, "[flows] schedule tick for unknown/removed flow — ignoring"); + return; + } + Err(e) => { + tracing::warn!(target: "flows", %flow_id, error = %e, "[flows] failed to load flow for schedule tick"); + return; + } + }; + if !flow.enabled { + tracing::debug!(target: "flows", %flow_id, "[flows] schedule tick for disabled flow — ignoring"); + return; + } + if !matches!(extract_trigger_kind(&flow), Some(TriggerKind::Schedule)) { + tracing::debug!(target: "flows", %flow_id, "[flows] schedule tick for flow whose trigger is no longer `schedule` — ignoring"); + return; + } + self.spawn_run(flow_id.to_string(), Value::Null); + } + + /// `DomainEvent::ComposioTriggerReceived` — scans every enabled flow for + /// an `app_event` trigger bound to this `toolkit`/`trigger_slug` and + /// dispatches each match with the event payload as the run input + /// (seeded into `run.trigger`, per the node-catalog contract). + async fn handle_app_event(&self, toolkit: &str, trigger_slug: &str, payload: &Value) { + let flows = match store::list_enabled_flows(&self.config) { + Ok(flows) => flows, + Err(e) => { + tracing::warn!(target: "flows", %toolkit, %trigger_slug, error = %e, "[flows] failed to list enabled flows for app_event dispatch"); + return; + } + }; + + let mut matched = 0usize; + for flow in flows { + if matches_app_event(&flow, toolkit, trigger_slug) { + matched += 1; + self.spawn_run(flow.id.clone(), payload.clone()); + } + } + tracing::debug!(target: "flows", %toolkit, %trigger_slug, matched, "[flows] app_event trigger matching complete"); + } + + /// Spawns a background `flows::ops::flows_run` for `flow_id`. Fire-and- + /// forget from the bus's perspective — `flows_run` itself records the + /// outcome onto the flow's summary fields and a `flow_runs` history row, + /// and surfaces a `CoreNotification` when the run pauses for approval. + /// + /// Skips the dispatch (see [`try_acquire_dispatch`]) if a trigger-driven + /// run for this `flow_id` is already in flight, so a fast schedule or a + /// burst of matching `app_event`s cannot run the same flow concurrently. + fn spawn_run(&self, flow_id: String, input: Value) { + let Some(guard) = self.try_acquire_dispatch(&flow_id) else { + tracing::debug!(target: "flows", %flow_id, "[flows] trigger: flow already running — skipping this tick"); + return; + }; + + let config = self.config.clone(); + tokio::spawn(async move { + // Held for the lifetime of the run; released on drop (including + // on panic) by `InFlightGuard`. + let _guard = guard; + tracing::info!(target: "flows", %flow_id, "[flows] trigger fired — starting run"); + match crate::openhuman::flows::ops::flows_run(&config, &flow_id, input).await { + Ok(_) => { + tracing::info!(target: "flows", %flow_id, "[flows] trigger-driven run finished") + } + Err(e) => { + tracing::warn!(target: "flows", %flow_id, error = %e, "[flows] trigger-driven run failed") + } + } + }); } } -impl Default for FlowTriggerSubscriber { - fn default() -> Self { - Self::new() +/// Drop guard releasing a [`FlowTriggerSubscriber::try_acquire_dispatch`] +/// claim. Removing the `flow_id` on `Drop` (rather than only on the happy +/// path) means a panicking or erroring `flows_run` still frees the flow up +/// for its next trigger tick. +struct InFlightGuard { + set: Arc>>, + flow_id: String, +} + +impl Drop for InFlightGuard { + fn drop(&mut self) { + // Recover from a poisoned lock (mirrors `try_acquire_dispatch`) so the + // flow_id is always removed — otherwise a poison would wedge this flow + // out of every future trigger dispatch, defeating the guard's purpose. + let mut set = self.set.lock().unwrap_or_else(|e| e.into_inner()); + set.remove(&self.flow_id); } } @@ -35,47 +204,248 @@ impl EventHandler for FlowTriggerSubscriber { } fn domains(&self) -> Option<&[&str]> { - Some(&["cron", "composio", "system"]) + Some(&["cron", "composio", "webhook", "system"]) } async fn handle(&self, event: &DomainEvent) { - // B1: log-only. Once trigger binding lands (B2), this maps the event - // onto enabled flows whose trigger node kind/config matches it, and - // runs each match through `flows::ops::flows_run`. - tracing::debug!( - target: "flows", - ?event, - "[flows] trigger subscriber observed event (B1: not yet dispatched to flows)" - ); + match event { + DomainEvent::FlowScheduleTick { flow_id } => self.handle_schedule_tick(flow_id).await, + DomainEvent::ComposioTriggerReceived { + toolkit, + trigger, + payload, + .. + } => self.handle_app_event(toolkit, trigger, payload).await, + DomainEvent::WebhookIncomingRequest { .. } => { + // Best-effort deviation (documented, not silently skipped — + // see `flows::ops::log_webhook_trigger_deferred` for the + // enable/disable-side note): a `webhook`-trigger flow needs a + // backend-provisioned tunnel + a UI surface for the resulting + // URL, neither of which exists yet. Never log the request's + // `raw_data` here — it is untrusted, possibly-sensitive + // inbound payload. + tracing::debug!( + target: "flows", + "[flows] observed WebhookIncomingRequest — webhook-trigger dispatch is not \ + implemented in B2 (pending backend tunnel provisioning + B3 UI); no flow \ + dispatched" + ); + } + other => { + // Anything else on our filtered domains (plain shell/agent + // `CronJobTriggered`, other Composio lifecycle events, + // system lifecycle, …) is not a flow trigger — ignore. Log + // only the variant name, never the event's Debug form: some + // sibling variants on these domains carry payloads we must + // not put in logs (e.g. `ComposioTriggerReceived::payload`). + tracing::trace!(target: "flows", variant = other.variant_name(), "[flows] ignoring unrelated event"); + } + } } } #[cfg(test)] mod tests { use super::*; + use crate::openhuman::flows::Flow; + use serde_json::json; + use tinyflows::model::{Node, NodeKind, WorkflowGraph}; + + fn test_config(tmp: &tempfile::TempDir) -> Arc { + let config = Config { + workspace_dir: tmp.path().join("workspace"), + action_dir: tmp.path().join("workspace"), + config_path: tmp.path().join("config.toml"), + ..Config::default() + }; + std::fs::create_dir_all(&config.workspace_dir).unwrap(); + Arc::new(config) + } + + fn trigger_node(config: Value) -> Node { + Node { + id: "t".to_string(), + kind: NodeKind::Trigger, + type_version: 1, + name: "Trigger".to_string(), + config, + ports: Vec::new(), + position: None, + } + } + + fn flow_with_trigger_config(id: &str, enabled: bool, trigger_config: Value) -> Flow { + Flow { + id: id.to_string(), + name: id.to_string(), + enabled, + graph: WorkflowGraph { + nodes: vec![trigger_node(trigger_config)], + ..Default::default() + }, + created_at: "2026-01-01T00:00:00Z".to_string(), + updated_at: "2026-01-01T00:00:00Z".to_string(), + last_run_at: None, + last_status: None, + require_approval: false, + } + } #[test] fn name_and_domains_are_stable() { - let sub = FlowTriggerSubscriber::new(); + let tmp = tempfile::TempDir::new().unwrap(); + let sub = FlowTriggerSubscriber::new(test_config(&tmp)); assert_eq!(sub.name(), "flows::trigger"); - assert_eq!(sub.domains(), Some(&["cron", "composio", "system"][..])); + assert_eq!( + sub.domains(), + Some(&["cron", "composio", "webhook", "system"][..]) + ); } #[tokio::test] async fn handle_does_not_panic_on_arbitrary_events() { - let sub = FlowTriggerSubscriber; + let tmp = tempfile::TempDir::new().unwrap(); + let sub = FlowTriggerSubscriber::new(test_config(&tmp)); sub.handle(&DomainEvent::CronJobTriggered { job_id: "j1".into(), job_name: "test".into(), job_type: "shell".into(), }) .await; + sub.handle(&DomainEvent::FlowScheduleTick { + flow_id: "missing-flow".into(), + }) + .await; + } + + #[test] + fn extract_trigger_kind_reads_schedule() { + let flow = flow_with_trigger_config( + "f1", + true, + json!({ "trigger_kind": "schedule", "schedule": "0 9 * * *" }), + ); + assert!(matches!( + extract_trigger_kind(&flow), + Some(TriggerKind::Schedule) + )); + } + + #[test] + fn extract_trigger_kind_none_for_missing_discriminator() { + let flow = flow_with_trigger_config("f1", true, json!({})); + assert!(extract_trigger_kind(&flow).is_none()); + } + + #[test] + fn extract_trigger_kind_none_for_invalid_discriminator() { + let flow = flow_with_trigger_config("f1", true, json!({ "trigger_kind": "not_a_kind" })); + assert!(extract_trigger_kind(&flow).is_none()); + } + + #[test] + fn matches_app_event_requires_toolkit_and_slug_match() { + let flow = flow_with_trigger_config( + "f1", + true, + json!({ "trigger_kind": "app_event", "toolkit": "gmail", "trigger_slug": "GMAIL_NEW_GMAIL_MESSAGE" }), + ); + assert!(matches_app_event(&flow, "gmail", "GMAIL_NEW_GMAIL_MESSAGE")); + // Case-insensitive. + assert!(matches_app_event(&flow, "Gmail", "gmail_new_gmail_message")); + // Wrong toolkit or slug does not match. + assert!(!matches_app_event( + &flow, + "slack", + "GMAIL_NEW_GMAIL_MESSAGE" + )); + assert!(!matches_app_event(&flow, "gmail", "SLACK_NEW_MESSAGE")); + } + + #[test] + fn matches_app_event_false_for_non_app_event_trigger() { + let flow = flow_with_trigger_config( + "f1", + true, + json!({ "trigger_kind": "schedule", "schedule": "0 9 * * *" }), + ); + assert!(!matches_app_event( + &flow, + "gmail", + "GMAIL_NEW_GMAIL_MESSAGE" + )); + } + + #[tokio::test] + async fn handle_app_event_ignores_disabled_flows() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = flow_with_trigger_config( + "disabled-flow", + false, + json!({ "trigger_kind": "app_event", "toolkit": "gmail", "trigger_slug": "GMAIL_NEW_GMAIL_MESSAGE" }), + ); + crate::openhuman::flows::store::upsert_flow(&config, &flow).unwrap(); + + // `list_enabled_flows` must not surface the disabled flow at all — + // proves the subscriber's dispatch source already excludes it, + // rather than asserting on a spawned background task's side effect. + let enabled = crate::openhuman::flows::store::list_enabled_flows(&config).unwrap(); + assert!(enabled.is_empty()); + } + + #[tokio::test] + async fn handle_schedule_tick_ignores_disabled_flow() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = flow_with_trigger_config( + "sched-flow", + false, + json!({ "trigger_kind": "schedule", "schedule": "0 9 * * *" }), + ); + crate::openhuman::flows::store::upsert_flow(&config, &flow).unwrap(); + + let sub = FlowTriggerSubscriber::new(config.clone()); + // Must not panic and must not spawn a run for a disabled flow — we + // can't directly observe "no run happened" without a full flows_run + // fixture, but this exercises the early-return path without error. + sub.handle(&DomainEvent::FlowScheduleTick { + flow_id: "sched-flow".into(), + }) + .await; + } + + // ── in-flight dedupe (CodeRabbit finding B) ───────────────────── + + #[test] + fn try_acquire_dispatch_skips_a_flow_already_in_flight() { + let tmp = tempfile::TempDir::new().unwrap(); + let sub = FlowTriggerSubscriber::new(test_config(&tmp)); + + let guard = sub + .try_acquire_dispatch("f1") + .expect("first claim for f1 should succeed"); + assert!( + sub.try_acquire_dispatch("f1").is_none(), + "a second claim for the same flow while the first is held must be skipped" + ); + + // A different flow is unaffected. + assert!(sub.try_acquire_dispatch("f2").is_some()); + + drop(guard); + assert!( + sub.try_acquire_dispatch("f1").is_some(), + "dropping the guard must release the claim so f1 can run again" + ); } #[test] fn default_constructs_the_same_as_new() { - let a = FlowTriggerSubscriber::new(); - let b = FlowTriggerSubscriber::default(); + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + let a = FlowTriggerSubscriber::new(config.clone()); + let b = FlowTriggerSubscriber::new(config); assert_eq!(a.name(), b.name()); } } diff --git a/src/openhuman/flows/mod.rs b/src/openhuman/flows/mod.rs index f1973add6..73fee8dd5 100644 --- a/src/openhuman/flows/mod.rs +++ b/src/openhuman/flows/mod.rs @@ -23,4 +23,4 @@ pub use schemas::{ // them to implement `tinyflows::caps::StateStore` without duplicating the // `flow_state` table's persistence logic. pub use store::{kv_get, kv_set}; -pub use types::Flow; +pub use types::{Flow, FlowRun, FlowRunStep}; diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index d3c945427..645be6007 100644 --- a/src/openhuman/flows/ops.rs +++ b/src/openhuman/flows/ops.rs @@ -1,20 +1,25 @@ //! Business logic for the `flows::` domain: validate-on-save CRUD plus the -//! end-to-end `flows_run` path. Delegated to from `schemas.rs`'s `handle_*` -//! RPC/CLI handlers, mirroring `src/openhuman/cron/ops.rs`. +//! end-to-end `flows_run` / `flows_resume` path. Delegated to from +//! `schemas.rs`'s `handle_*` RPC/CLI handlers, mirroring +//! `src/openhuman/cron/ops.rs`. use std::sync::Arc; +use chrono::Utc; use serde_json::{json, Value}; -use tinyflows::model::WorkflowGraph; +use tinyflows::model::{TriggerKind, WorkflowGraph}; +use crate::openhuman::agent::turn_origin::{with_origin, AgentTurnOrigin, TrustedAutomationSource}; use crate::openhuman::config::Config; +use crate::openhuman::flows::bus; use crate::openhuman::flows::store; -use crate::openhuman::flows::Flow; +use crate::openhuman::flows::types::FlowRunStep; +use crate::openhuman::flows::{Flow, FlowRun}; use crate::rpc::RpcOutcome; -/// Overall safety bound on a single `flows_run`. Individual capabilities have -/// their own timeouts (HTTP, sandbox), but a hung LLM/tool call must never let -/// the RPC block indefinitely — this caps the whole run. +/// Overall safety bound on a single `flows_run` / `flows_resume`. Individual +/// capabilities have their own timeouts (HTTP, sandbox), but a hung LLM/tool +/// call must never let the RPC block indefinitely — this caps the whole run. const FLOW_RUN_TIMEOUT_SECS: u64 = 600; /// Runs a raw graph JSON value through `tinyflows::migrate::migrate` (upgrade @@ -29,14 +34,30 @@ fn validate_and_migrate_graph(graph_json: Value) -> Result Result, String> { let graph = validate_and_migrate_graph(graph_json)?; - tracing::debug!(target: "flows", %name, node_count = graph.nodes.len(), "[flows] flows_create: persisting new flow"); - let flow = store::create_flow(config, name, graph).map_err(|e| e.to_string())?; + tracing::debug!(target: "flows", %name, node_count = graph.nodes.len(), require_approval, "[flows] flows_create: persisting new flow"); + let flow = + store::create_flow(config, name, graph, require_approval).map_err(|e| e.to_string())?; + + if flow.enabled { + tracing::debug!(target: "flows", flow_id = %flow.id, "[flows] flows_create: flow is enabled — binding automatic-dispatch trigger"); + bind_trigger(config, &flow); + } + Ok(RpcOutcome::single_log(flow, "flow created")) } @@ -54,30 +75,54 @@ pub async fn flows_list(config: &Config) -> Result>, String Ok(RpcOutcome::single_log(flows, "flows listed")) } -/// Updates a flow's name and/or graph. Re-validates the graph (whether newly -/// supplied or the existing one) before persisting, same as `flows_create`. +/// 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`. +/// +/// When the caller supplies a new `graph_json` and the flow is (still) +/// enabled, re-binds the automatic-dispatch trigger if the trigger +/// kind/config actually changed (e.g. a new schedule cron expression) — +/// otherwise the stale binding from the old graph would keep firing on the +/// old cadence, or a newly-added schedule would never get bound at all. +/// Skipped entirely for a name/`require_approval`-only update (no +/// `graph_json` supplied), since the trigger definitely didn't change. pub async fn flows_update( config: &Config, id: &str, name: Option, graph_json: Option, + require_approval: Option, ) -> Result, String> { let existing = store::get_flow(config, id) .map_err(|e| e.to_string())? .ok_or_else(|| format!("flow '{id}' not found"))?; - let new_name = name.unwrap_or(existing.name); + let new_name = name.unwrap_or_else(|| existing.name.clone()); + let new_require_approval = require_approval.unwrap_or(existing.require_approval); + let graph_changed = graph_json.is_some(); let graph = match graph_json { Some(raw) => validate_and_migrate_graph(raw)?, None => { tinyflows::validate::validate(&existing.graph).map_err(|e| e.to_string())?; - existing.graph + existing.graph.clone() } }; tracing::debug!(target: "flows", flow_id = %id, "[flows] flows_update: persisting changes"); - let updated = - store::update_flow_graph(config, id, new_name, graph).map_err(|e| e.to_string())?; + let updated = store::update_flow_graph(config, id, new_name, graph, new_require_approval) + .map_err(|e| e.to_string())?; + + if graph_changed && updated.enabled { + let trigger_unchanged = bus::extract_trigger_kind(&existing) + == bus::extract_trigger_kind(&updated) + && bus::extract_trigger_config(&existing) == bus::extract_trigger_config(&updated); + if !trigger_unchanged { + tracing::debug!(target: "flows", flow_id = %id, "[flows] flows_update: trigger changed on an enabled flow — rebinding automatic-dispatch trigger"); + unbind_trigger(config, &existing); + bind_trigger(config, &updated); + } + } + Ok(RpcOutcome::single_log( updated, format!("flow updated: {id}"), @@ -85,7 +130,25 @@ pub async fn flows_update( } /// Deletes a flow by id. +/// +/// Unbinds the flow's automatic-dispatch trigger (e.g. the schedule-trigger +/// cron job) *before* removing the flow definition. `flow_runs` cascades on +/// delete via a same-database `FOREIGN KEY ... ON DELETE CASCADE`, but a +/// bound cron job lives in the entirely separate `cron.db` — it does NOT +/// cascade — so skipping this would orphan the cron job, leaving it pointing +/// at a now-nonexistent `flow_id` forever. Best-effort: a lookup failure +/// (flow already gone, store error) is logged and does not block the delete +/// itself — `store::remove_flow` below still errors clearly if `id` doesn't +/// exist. pub async fn flows_delete(config: &Config, id: &str) -> Result, String> { + match store::get_flow(config, id) { + Ok(Some(flow)) => unbind_trigger(config, &flow), + Ok(None) => {} + Err(e) => { + tracing::warn!(target: "flows", flow_id = %id, error = %e, "[flows] flows_delete: failed to load flow before unbind — proceeding with delete anyway"); + } + } + store::remove_flow(config, id).map_err(|e| e.to_string())?; tracing::debug!(target: "flows", flow_id = %id, "[flows] flows_delete: removed"); Ok(RpcOutcome::new( @@ -94,31 +157,194 @@ pub async fn flows_delete(config: &Config, id: &str) -> Result )) } -/// Enables or disables a flow. B1 note: this does not yet gate anything at -/// run time (`flows_run` still runs a disabled flow on demand, mirroring -/// `cron::rpc::cron_run`'s "Run Now always works" behavior) — `enabled` will -/// gate automatic trigger-driven dispatch once `FlowTriggerSubscriber` -/// (`src/openhuman/flows/bus.rs`) is wired up to actually invoke flows (B2). +/// Enables or disables a flow. Enable/disable now (B2) binds/tears down the +/// flow's automatic trigger: +/// - `schedule` — registers/removes the backing `cron` job +/// (`cron::add_flow_schedule_job` / `cron::remove_job`) so +/// `flows::bus::FlowTriggerSubscriber` gets a `FlowScheduleTick` on the +/// configured cadence. +/// - `app_event` — no enable-time side effect needed: the subscriber matches +/// every `ComposioTriggerReceived` against `store::list_enabled_flows` at +/// dispatch time, so the `enabled` flag alone gates it. +/// - `webhook` — **not implemented** in B2 (best-effort deviation, see +/// `bind_trigger`'s webhook arm below and +/// `my_docs/ohxtf/b2-triggers-trust/01-triggers-and-trust.md` §1); logged, +/// not silently skipped. +/// - `manual` / anything else — no binding needed; `flows_run` always works. +/// +/// `flows_run` still runs a disabled flow on demand (mirrors +/// `cron::rpc::cron_run`'s "Run Now always works" behavior) — `enabled` only +/// gates *automatic* trigger-driven dispatch. pub async fn flows_set_enabled( config: &Config, id: &str, enabled: bool, ) -> Result, String> { let flow = store::set_enabled(config, id, enabled).map_err(|e| e.to_string())?; + + if enabled { + bind_trigger(config, &flow); + } else { + unbind_trigger(config, &flow); + } + Ok(RpcOutcome::single_log( flow, format!("flow {id} enabled={enabled}"), )) } -/// Runs a flow end-to-end: compile → build capabilities → durable -/// checkpointed run → record the outcome onto the flow's summary fields. +/// Registers the automatic-dispatch side effect for `flow`'s trigger kind, if +/// any. Best-effort: a binding failure is logged and does not fail the +/// `flows_set_enabled` call — the flow is still saved as enabled, it just +/// won't fire automatically until the underlying issue (invalid schedule, +/// cron store error, …) is fixed. +fn bind_trigger(config: &Config, flow: &Flow) { + match bus::extract_trigger_kind(flow) { + Some(TriggerKind::Schedule) => bind_schedule_trigger(config, flow), + Some(TriggerKind::Webhook) => log_webhook_trigger_deferred(flow, true), + _ => { + // `app_event` needs no enable-time binding (matched at dispatch + // time against `list_enabled_flows`); `manual`/`form`/others have + // no automatic-dispatch concept at all. + } + } +} + +/// Tears down the automatic-dispatch side effect for `flow`'s trigger kind, +/// mirroring [`bind_trigger`]. Best-effort, same rationale. +fn unbind_trigger(config: &Config, flow: &Flow) { + match bus::extract_trigger_kind(flow) { + Some(TriggerKind::Schedule) => unbind_schedule_trigger(config, &flow.id), + Some(TriggerKind::Webhook) => log_webhook_trigger_deferred(flow, false), + _ => {} + } +} + +/// Registers (or refreshes) the `cron` job backing a `schedule`-trigger +/// flow. Idempotent — re-uses an existing binding via +/// `cron::find_flow_schedule_job` rather than creating a duplicate, so this +/// is safe to call both from `flows_set_enabled` and from boot +/// reconciliation ([`reconcile_schedule_triggers_on_boot`]). +fn bind_schedule_trigger(config: &Config, flow: &Flow) { + let Some(trigger_config) = bus::extract_trigger_config(flow) else { + tracing::warn!(target: "flows", flow_id = %flow.id, "[flows] schedule trigger: flow has no single trigger node — cannot bind cron job"); + return; + }; + let Some(schedule_raw) = trigger_config.get("schedule").cloned() else { + tracing::warn!(target: "flows", flow_id = %flow.id, "[flows] schedule trigger config is missing `schedule` — cannot bind cron job"); + return; + }; + let schedule: crate::openhuman::cron::Schedule = match serde_json::from_value(schedule_raw) { + Ok(s) => s, + Err(e) => { + tracing::warn!(target: "flows", flow_id = %flow.id, error = %e, "[flows] invalid schedule trigger config — cannot bind cron job"); + return; + } + }; + + match crate::openhuman::cron::find_flow_schedule_job(config, &flow.id) { + Ok(Some(existing)) => { + let patch = crate::openhuman::cron::CronJobPatch { + enabled: Some(true), + schedule: Some(schedule), + ..Default::default() + }; + if let Err(e) = crate::openhuman::cron::update_job(config, &existing.id, patch) { + tracing::warn!(target: "flows", flow_id = %flow.id, cron_job_id = %existing.id, error = %e, "[flows] failed to refresh existing schedule-trigger cron job"); + } else { + tracing::debug!(target: "flows", flow_id = %flow.id, cron_job_id = %existing.id, "[flows] refreshed existing schedule-trigger cron job"); + } + } + Ok(None) => match crate::openhuman::cron::add_flow_schedule_job(config, &flow.id, schedule) + { + Ok(job) => { + tracing::info!(target: "flows", flow_id = %flow.id, cron_job_id = %job.id, "[flows] registered schedule-trigger cron job") + } + Err(e) => { + tracing::warn!(target: "flows", flow_id = %flow.id, error = %e, "[flows] failed to register schedule-trigger cron job") + } + }, + Err(e) => { + tracing::warn!(target: "flows", flow_id = %flow.id, error = %e, "[flows] failed to look up existing schedule-trigger cron job"); + } + } +} + +/// Removes the `cron` job backing a `schedule`-trigger flow, if one exists. +fn unbind_schedule_trigger(config: &Config, flow_id: &str) { + match crate::openhuman::cron::find_flow_schedule_job(config, flow_id) { + Ok(Some(job)) => { + if let Err(e) = crate::openhuman::cron::remove_job(config, &job.id) { + tracing::warn!(target: "flows", %flow_id, cron_job_id = %job.id, error = %e, "[flows] failed to remove schedule-trigger cron job"); + } else { + tracing::debug!(target: "flows", %flow_id, cron_job_id = %job.id, "[flows] removed schedule-trigger cron job"); + } + } + Ok(None) => {} + Err(e) => { + tracing::warn!(target: "flows", %flow_id, error = %e, "[flows] failed to look up schedule-trigger cron job for teardown"); + } + } +} + +/// Webhook trigger binding is a documented B2 stub (best-effort deviation): +/// registering a real inbound route requires provisioning a backend tunnel +/// (`webhooks::ops::create_tunnel`, a network call to the signed-in backend +/// account) plus a UI surface to show the resulting URL to the user — both +/// are B3 territory. Rather than silently doing nothing, this logs a clear, +/// actionable warning every time a `webhook`-trigger flow is enabled/disabled +/// so the gap is diagnosable. `flows::bus::FlowTriggerSubscriber` logs the +/// matching deferral on the inbound side (`WebhookIncomingRequest`). +fn log_webhook_trigger_deferred(flow: &Flow, enabled: bool) { + tracing::warn!( + target: "flows", + flow_id = %flow.id, + enabled, + "[flows] webhook trigger binding is not implemented in B2 (requires backend tunnel \ + provisioning + a UI surface for the resulting URL) — this flow will not fire \ + automatically from an inbound webhook until that lands" + ); +} + +/// Boot-time reconciliation: registers the `cron` job for every enabled, +/// `schedule`-trigger flow. Idempotent (delegates to [`bind_schedule_trigger`], +/// which re-uses an existing binding) — mirrors +/// `cron::seed::seed_proactive_agents_on_boot`'s "ensure jobs exist for +/// already-onboarded users upgrading from an older build" pattern, so a +/// flow enabled on a build that predates this cron binding (or whose binding +/// was lost some other way) gets its schedule re-registered on the next +/// boot without the user having to toggle it off and on. +pub async fn reconcile_schedule_triggers_on_boot(config: &Config) -> Result<(), String> { + let flows = store::list_enabled_flows(config).map_err(|e| e.to_string())?; + let mut reconciled = 0usize; + for flow in &flows { + if matches!(bus::extract_trigger_kind(flow), Some(TriggerKind::Schedule)) { + bind_schedule_trigger(config, flow); + reconciled += 1; + } + } + tracing::debug!(target: "flows", scanned = flows.len(), reconciled, "[flows] boot reconciliation of schedule-trigger cron jobs complete"); + Ok(()) +} + +/// Runs a saved flow end-to-end: compile → build capabilities → durable +/// checkpointed run → record the outcome onto the flow's summary fields and +/// into a `flow_runs` history row. /// /// Uses `tinyflows::engine::run_with_checkpointer` (not the simpler `run`) so /// a run that pauses at a human-in-the-loop approval gate is durably -/// checkpointed and can survive a process restart (resumed later via a -/// `flows_resume` RPC — B2+; see -/// `my_docs/ohxtf/b1-engine-seam-domain/07-execution-and-hitl.md`). +/// checkpointed and can survive a process restart (resumed later via +/// [`flows_resume`]; see +/// `my_docs/ohxtf/b1-engine-seam-domain/05-checkpointer-and-state.md`). +/// +/// The whole run is scoped under `AgentTurnOrigin::TrustedAutomation { +/// Workflow }` (issue B2) regardless of caller (an interactive RPC "Run" or +/// an automatic trigger dispatch from `flows::bus::FlowTriggerSubscriber`): +/// the trust argument is about the *flow* (a saved, validated graph whose +/// `tool_call`/`http_request` nodes are pre-declared), not about who started +/// the run — see `TrustedAutomationSource::Workflow`'s doc and +/// `my_docs/ohxtf/b2-triggers-trust/01-triggers-and-trust.md` §3. pub async fn flows_run( config: &Config, flow_id: &str, @@ -145,13 +371,16 @@ pub async fn flows_run( target: "flows", flow_id = %flow_id, thread_id = %thread_id, + require_approval = flow.require_approval, "[flows] flows_run: starting checkpointed run" ); + start_flow_run_row(config, &thread_id, flow_id); + // Record a failed attempt so `last_run_at`/`last_status` reflect reality // (a stop-policy engine/capability failure or a timeout) rather than // leaving the prior success/pending state on the flow. - let record_failed = || { + let record_failed = |error: &str| { if let Err(rec_err) = store::record_run(config, flow_id, "failed") { tracing::warn!( target: "flows", @@ -160,10 +389,14 @@ pub async fn flows_run( "[flows] flows_run: failed to record failed run" ); } + finish_flow_run_row(config, &thread_id, "failed", &[], &[], Some(error)); }; - let run = - tinyflows::engine::run_with_checkpointer(&compiled, input, &caps, checkpointer, &thread_id); + let origin = workflow_origin(flow_id, flow.require_approval); + let run = with_origin( + origin, + tinyflows::engine::run_with_checkpointer(&compiled, input, &caps, checkpointer, &thread_id), + ); let outcome = match tokio::time::timeout( std::time::Duration::from_secs(FLOW_RUN_TIMEOUT_SECS), run, @@ -172,14 +405,15 @@ pub async fn flows_run( { Ok(Ok(outcome)) => outcome, Ok(Err(e)) => { - record_failed(); + record_failed(&e.to_string()); tracing::warn!(target: "flows", flow_id = %flow_id, error = %e, "[flows] flows_run: run failed"); return Err(e.to_string()); } Err(_elapsed) => { - record_failed(); + let msg = format!("flow run timed out after {FLOW_RUN_TIMEOUT_SECS}s"); + record_failed(&msg); tracing::warn!(target: "flows", flow_id = %flow_id, timeout_secs = FLOW_RUN_TIMEOUT_SECS, "[flows] flows_run: run timed out"); - return Err(format!("flow run timed out after {FLOW_RUN_TIMEOUT_SECS}s")); + return Err(msg); } }; @@ -189,6 +423,15 @@ pub async fn flows_run( "pending_approval" }; store::record_run(config, flow_id, status).map_err(|e| e.to_string())?; + finish_flow_run_row( + config, + &thread_id, + status, + &reconstruct_steps(&outcome.output), + &outcome.pending_approvals, + None, + ); + notify_pending_approval(&flow, &thread_id, &outcome.pending_approvals); tracing::info!( target: "flows", @@ -208,6 +451,302 @@ pub async fn flows_run( )) } +/// Resumes a `flows_run` that paused at a human-in-the-loop approval gate, +/// continuing it from the durable checkpoint (`thread_id`) with +/// `approvals` newly granted. The UI approval card (B3) calls this once the +/// user decides. See `tinyflows::engine::resume_with_checkpointer`'s doc for +/// the resume mechanics. +/// +/// **Host-side approval guard (issue B2 finding #3):** tinyflows 0.2's +/// `resume_with_checkpointer` treats the resume call itself as approval of +/// whatever gate paused the run — its `approvals` argument is advisory only, +/// not enforced inside the crate (`flows_resume(..., approvals: [])` on a +/// paused run would otherwise still complete it). So before ever calling +/// into the engine, this loads the persisted `flow_runs` row for +/// `thread_id` (`flow_runs.id == thread_id`) and requires that `approvals` +/// names at least one of that row's *actually* pending node ids. A run +/// that isn't currently `pending_approval` (already completed, failed, or +/// unknown) is rejected outright — resuming an already-settled thread_id is +/// no longer treated as a harmless no-op, it's a clear error. +pub async fn flows_resume( + config: &Config, + flow_id: &str, + thread_id: &str, + approvals: Vec, +) -> Result, String> { + let flow = store::get_flow(config, flow_id) + .map_err(|e| e.to_string())? + .ok_or_else(|| format!("flow '{flow_id}' not found"))?; + + let run_record = store::get_flow_run(config, thread_id) + .map_err(|e| e.to_string())? + .ok_or_else(|| { + format!("no paused run to resume: no run recorded for thread '{thread_id}'") + })?; + if run_record.flow_id != flow_id { + return Err(format!( + "no paused run to resume: run '{thread_id}' belongs to flow '{}', not '{flow_id}'", + run_record.flow_id + )); + } + if run_record.status != "pending_approval" { + return Err(format!( + "no paused run to resume: run '{thread_id}' is not pending approval (status: {})", + run_record.status + )); + } + let matches_pending = approvals + .iter() + .any(|a| run_record.pending_approvals.contains(a)); + if !matches_pending { + tracing::warn!( + target: "flows", + flow_id = %flow_id, + %thread_id, + ?approvals, + pending = ?run_record.pending_approvals, + "[flows] flows_resume: rejected — caller approvals name none of the pending gates" + ); + return Err(format!( + "no pending approval matches: approvals {approvals:?} do not name any of the \ + currently pending gates {:?} for run '{thread_id}'", + run_record.pending_approvals + )); + } + + let compiled = tinyflows::compiler::compile(&flow.graph).map_err(|e| e.to_string())?; + let config_arc = Arc::new(config.clone()); + let caps = + crate::openhuman::tinyflows::build_capabilities(config_arc, format!("flow:{flow_id}")); + let checkpointer = + crate::openhuman::tinyflows::open_flow_checkpointer(config).map_err(|e| e.to_string())?; + + tracing::debug!( + target: "flows", + flow_id = %flow_id, + %thread_id, + approval_count = approvals.len(), + "[flows] flows_resume: resuming checkpointed run" + ); + + let origin = workflow_origin(flow_id, flow.require_approval); + let run = with_origin( + origin, + tinyflows::engine::resume_with_checkpointer( + &compiled, + &caps, + checkpointer, + thread_id, + approvals, + ), + ); + + let outcome = match tokio::time::timeout( + std::time::Duration::from_secs(FLOW_RUN_TIMEOUT_SECS), + run, + ) + .await + { + Ok(Ok(outcome)) => outcome, + Ok(Err(e)) => { + let _ = store::record_run(config, flow_id, "failed"); + finish_flow_run_row(config, thread_id, "failed", &[], &[], Some(&e.to_string())); + tracing::warn!(target: "flows", flow_id = %flow_id, %thread_id, error = %e, "[flows] flows_resume: run failed"); + return Err(e.to_string()); + } + Err(_elapsed) => { + let msg = format!("flow resume timed out after {FLOW_RUN_TIMEOUT_SECS}s"); + let _ = store::record_run(config, flow_id, "failed"); + finish_flow_run_row(config, thread_id, "failed", &[], &[], Some(&msg)); + tracing::warn!(target: "flows", flow_id = %flow_id, %thread_id, timeout_secs = FLOW_RUN_TIMEOUT_SECS, "[flows] flows_resume: run timed out"); + return Err(msg); + } + }; + + let status = if outcome.pending_approvals.is_empty() { + "completed" + } else { + "pending_approval" + }; + store::record_run(config, flow_id, status).map_err(|e| e.to_string())?; + finish_flow_run_row( + config, + thread_id, + status, + &reconstruct_steps(&outcome.output), + &outcome.pending_approvals, + None, + ); + notify_pending_approval(&flow, thread_id, &outcome.pending_approvals); + + tracing::info!( + target: "flows", + flow_id = %flow_id, + %thread_id, + status, + pending_approvals = outcome.pending_approvals.len(), + "[flows] flows_resume: finished" + ); + + Ok(RpcOutcome::single_log( + json!({ + "output": outcome.output, + "pending_approvals": outcome.pending_approvals, + "thread_id": thread_id, + }), + format!("flow resume {status}"), + )) +} + +/// Lists the most recent runs for a flow (newest first), for the B3 +/// run-history inspector. +pub async fn flows_list_runs( + config: &Config, + flow_id: &str, + limit: usize, +) -> Result>, String> { + let runs = store::list_flow_runs(config, flow_id, limit).map_err(|e| e.to_string())?; + Ok(RpcOutcome::single_log( + runs, + format!("flow runs listed: {flow_id}"), + )) +} + +/// Loads a single flow run record by id (== `thread_id`). +pub async fn flows_get_run(config: &Config, run_id: &str) -> Result, String> { + let run = store::get_flow_run(config, run_id) + .map_err(|e| e.to_string())? + .ok_or_else(|| format!("flow run '{run_id}' not found"))?; + Ok(RpcOutcome::single_log( + run, + format!("flow run loaded: {run_id}"), + )) +} + +/// Builds the `TrustedAutomation { Workflow }` origin scoped around every +/// `flows_run` / `flows_resume` invocation. See `flows_run`'s doc for why +/// this applies uniformly regardless of caller. +fn workflow_origin(flow_id: &str, require_approval: bool) -> AgentTurnOrigin { + AgentTurnOrigin::TrustedAutomation { + job_id: flow_id.to_string(), + source: TrustedAutomationSource::Workflow { require_approval }, + } +} + +/// Best-effort insert of the initial `"running"` `flow_runs` row. Logged, +/// never fails the run — run-history persistence is an observability aid, +/// not a correctness requirement of the run itself. +fn start_flow_run_row(config: &Config, thread_id: &str, flow_id: &str) { + let started_at = Utc::now().to_rfc3339(); + if let Err(e) = store::insert_flow_run(config, thread_id, flow_id, thread_id, &started_at) { + tracing::warn!(target: "flows", flow_id, thread_id, error = %e, "[flows] failed to persist flow run start"); + } +} + +/// Best-effort finalization of a `flow_runs` row. Logged, never fails the +/// run (see [`start_flow_run_row`]). +fn finish_flow_run_row( + config: &Config, + thread_id: &str, + status: &str, + steps: &[FlowRunStep], + pending_approvals: &[String], + error: Option<&str>, +) { + let finished_at = Utc::now().to_rfc3339(); + if let Err(e) = store::finish_flow_run( + config, + thread_id, + status, + &finished_at, + steps, + pending_approvals, + error, + ) { + tracing::warn!(target: "flows", thread_id, status, error = %e, "[flows] failed to persist flow run finish"); + } +} + +/// Reconstructs a lean per-node step list from a settled run's +/// `output["nodes"]` map. tinyflows 0.2's durable path installs a +/// `NoopObserver` (see `tinyflows/observability.rs`), so there is no live +/// step stream to persist — this is the B2 "good enough" substitute the +/// spec calls for; a richer per-step `RunObserver` is a tinyflows 0.3 item. +/// +/// // TODO(0.3): replace this reconstruction with a real `RunObserver` that +/// // streams `node_id`/`status`/`output`/`duration_ms` as each node +/// // finishes, once the durable run path supports installing one. +fn reconstruct_steps(output: &Value) -> Vec { + let Some(nodes) = output.get("nodes").and_then(Value::as_object) else { + return Vec::new(); + }; + nodes + .iter() + .map(|(node_id, slot)| FlowRunStep { + node_id: node_id.clone(), + output: slot.get("items").cloned().unwrap_or(Value::Null), + port: slot.get("port").and_then(Value::as_str).map(str::to_string), + }) + .collect() +} + +/// 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 paused run as a `CoreNotification` (category `Agents`) with an +/// "approve" action carrying `flow_id`/`thread_id`/`node_ids`, mirroring the +/// pattern `agent_meetings::calendar`'s auto-summarize "Ask" flow uses +/// (direct `publish_core_notification` call with an action payload, not the +/// generic `DomainEvent -> event_to_notification` bridge — this is a +/// flows-specific card with flow-specific action data, not a translation of +/// an existing broadcast event). No-op when nothing is pending. +fn notify_pending_approval(flow: &Flow, thread_id: &str, pending_approvals: &[String]) { + if pending_approvals.is_empty() { + return; + } + + use crate::openhuman::notifications::bus::publish_core_notification; + use crate::openhuman::notifications::types::{ + CoreNotificationAction, CoreNotificationCategory, CoreNotificationEvent, + }; + + let action_payload = json!({ + "flow_id": flow.id, + "thread_id": thread_id, + "node_ids": pending_approvals, + }); + + publish_core_notification(CoreNotificationEvent { + id: format!("flow-pending-approval:{}:{}", flow.id, thread_id), + category: CoreNotificationCategory::Agents, + title: "Workflow needs approval".to_string(), + body: format!( + "\"{}\" is waiting on {} approval{} before it can continue.", + flow.name, + pending_approvals.len(), + if pending_approvals.len() == 1 { + "" + } else { + "s" + } + ), + // No dedicated Workflows review route exists yet (B3 ships the UI); + // leave unset rather than link to a page that can't act on it. + deep_link: None, + timestamp_ms: now_ms(), + actions: Some(vec![CoreNotificationAction { + action_id: "approve".to_string(), + label: "Review".to_string(), + payload: Some(action_payload), + }]), + }); +} + #[cfg(test)] #[path = "ops_tests.rs"] mod tests; diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs index a21670d26..d782d4948 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/src/openhuman/flows/ops_tests.rs @@ -35,7 +35,7 @@ async fn flows_create_rejects_graph_without_trigger() { "edges": [] }); - let err = flows_create(&config, "bad".to_string(), graph_without_trigger) + let err = flows_create(&config, "bad".to_string(), graph_without_trigger, false) .await .expect_err("graph without a trigger must be rejected"); assert!( @@ -49,7 +49,7 @@ async fn flows_create_get_list_delete_roundtrip() { let tmp = TempDir::new().unwrap(); let config = test_config(&tmp); - let created = flows_create(&config, "demo".to_string(), trigger_only_graph()) + let created = flows_create(&config, "demo".to_string(), trigger_only_graph(), false) .await .unwrap(); let flow_id = created.value.id.clone(); @@ -70,7 +70,7 @@ async fn flows_create_get_list_delete_roundtrip() { async fn flows_set_enabled_toggles() { let tmp = TempDir::new().unwrap(); let config = test_config(&tmp); - let created = flows_create(&config, "demo".to_string(), trigger_only_graph()) + let created = flows_create(&config, "demo".to_string(), trigger_only_graph(), false) .await .unwrap(); assert!(created.value.enabled); @@ -90,7 +90,7 @@ async fn flows_set_enabled_toggles() { async fn flows_update_replaces_name_and_graph() { let tmp = TempDir::new().unwrap(); let config = test_config(&tmp); - let created = flows_create(&config, "demo".to_string(), trigger_only_graph()) + let created = flows_create(&config, "demo".to_string(), trigger_only_graph(), false) .await .unwrap(); @@ -102,6 +102,7 @@ async fn flows_update_replaces_name_and_graph() { &created.value.id, Some("renamed".to_string()), Some(new_graph), + None, ) .await .unwrap(); @@ -110,11 +111,32 @@ async fn flows_update_replaces_name_and_graph() { assert_eq!(updated.value.graph.name, "renamed-graph"); } +#[tokio::test] +async fn flows_update_can_set_require_approval() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "demo".to_string(), trigger_only_graph(), false) + .await + .unwrap(); + assert!(!created.value.require_approval); + + let updated = flows_update(&config, &created.value.id, None, None, Some(true)) + .await + .unwrap(); + assert!(updated.value.require_approval); + + // Omitting `require_approval` on a later update preserves the current value. + let unchanged = flows_update(&config, &created.value.id, None, None, None) + .await + .unwrap(); + assert!(unchanged.value.require_approval); +} + #[tokio::test] async fn flows_update_rejects_invalid_replacement_graph() { let tmp = TempDir::new().unwrap(); let config = test_config(&tmp); - let created = flows_create(&config, "demo".to_string(), trigger_only_graph()) + let created = flows_create(&config, "demo".to_string(), trigger_only_graph(), false) .await .unwrap(); @@ -124,7 +146,7 @@ async fn flows_update_rejects_invalid_replacement_graph() { "edges": [] }); - let err = flows_update(&config, &created.value.id, None, Some(invalid_graph)) + let err = flows_update(&config, &created.value.id, None, Some(invalid_graph), None) .await .expect_err("invalid replacement graph must be rejected"); assert!(err.contains("trigger")); @@ -134,7 +156,7 @@ async fn flows_update_rejects_invalid_replacement_graph() { async fn flows_run_completes_trigger_only_graph() { let tmp = TempDir::new().unwrap(); let config = test_config(&tmp); - let created = flows_create(&config, "demo".to_string(), trigger_only_graph()) + let created = flows_create(&config, "demo".to_string(), trigger_only_graph(), false) .await .unwrap(); @@ -171,7 +193,7 @@ async fn flows_run_reports_pending_approval_and_blocks_downstream() { ] }); - let created = flows_create(&config, "gated".to_string(), graph) + let created = flows_create(&config, "gated".to_string(), graph, false) .await .unwrap(); @@ -225,7 +247,7 @@ async fn flows_run_records_failed_status_when_a_node_errors() { "edges": [ { "from_node": "t", "to_node": "x" } ] }); - let created = flows_create(&config, "boom".to_string(), graph) + let created = flows_create(&config, "boom".to_string(), graph, false) .await .unwrap(); @@ -246,3 +268,508 @@ async fn flows_run_records_failed_status_when_a_node_errors() { "a failed run must stamp last_run_at" ); } + +// ── automatic-dispatch binding (issue B2 finding #1) ───────────────────── +// +// Live testing found that `flows_create` persisted a freshly-created, +// `enabled = true` schedule flow WITHOUT registering its cron job — only +// `flows_set_enabled` bound it. So a brand-new enabled schedule flow would +// silently never fire until an app restart (boot reconcile) or a manual +// disable→enable toggle. These tests exercise the fix directly against the +// real `cron` store (not a mock), the same way `bind_schedule_trigger` +// itself does. + +fn schedule_trigger_graph(cron_expr: &str) -> Value { + json!({ + "name": "scheduled", + "nodes": [ + { + "id": "t", + "kind": "trigger", + "name": "Trigger", + "config": { "trigger_kind": "schedule", "schedule": cron_expr } + } + ], + "edges": [] + }) +} + +#[tokio::test] +async fn flows_create_binds_schedule_cron_job_for_an_enabled_flow() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + let created = flows_create( + &config, + "scheduled".to_string(), + schedule_trigger_graph("0 9 * * *"), + false, + ) + .await + .unwrap(); + assert!(created.value.enabled, "flows_create defaults to enabled"); + + let job = crate::openhuman::cron::find_flow_schedule_job(&config, &created.value.id).unwrap(); + assert!( + job.is_some(), + "an enabled schedule flow must have its cron job bound immediately on create, not only \ + after a set_enabled toggle" + ); + assert_eq!(job.unwrap().expression, "0 9 * * *"); +} + +#[tokio::test] +async fn flows_delete_unbinds_schedule_cron_job() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create( + &config, + "scheduled".to_string(), + schedule_trigger_graph("0 9 * * *"), + false, + ) + .await + .unwrap(); + assert!( + crate::openhuman::cron::find_flow_schedule_job(&config, &created.value.id) + .unwrap() + .is_some(), + "precondition: cron job bound on create" + ); + + flows_delete(&config, &created.value.id).await.unwrap(); + + assert!( + crate::openhuman::cron::find_flow_schedule_job(&config, &created.value.id) + .unwrap() + .is_none(), + "deleting a flow must remove its schedule-trigger cron job — it lives in a separate \ + cron.db that flow_definitions' ON DELETE CASCADE cannot reach" + ); +} + +#[tokio::test] +async fn flows_update_rebinds_schedule_cron_job_when_trigger_schedule_changes() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create( + &config, + "scheduled".to_string(), + schedule_trigger_graph("0 9 * * *"), + false, + ) + .await + .unwrap(); + let old_job = crate::openhuman::cron::find_flow_schedule_job(&config, &created.value.id) + .unwrap() + .expect("cron job bound on create"); + assert_eq!(old_job.expression, "0 9 * * *"); + + flows_update( + &config, + &created.value.id, + None, + Some(schedule_trigger_graph("30 8 * * *")), + None, + ) + .await + .unwrap(); + + let new_job = crate::openhuman::cron::find_flow_schedule_job(&config, &created.value.id) + .unwrap() + .expect("cron job still bound after trigger schedule change"); + assert_eq!( + new_job.expression, "30 8 * * *", + "the bound cron job's schedule must reflect the new trigger config" + ); + + // No duplicate/orphaned job left behind for this flow. + let flow_jobs: Vec<_> = crate::openhuman::cron::list_jobs(&config) + .unwrap() + .into_iter() + .filter(|j| j.command == created.value.id) + .collect(); + assert_eq!(flow_jobs.len(), 1); +} + +#[tokio::test] +async fn flows_update_does_not_rebind_when_graph_is_not_supplied() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create( + &config, + "scheduled".to_string(), + schedule_trigger_graph("0 9 * * *"), + false, + ) + .await + .unwrap(); + let old_job = crate::openhuman::cron::find_flow_schedule_job(&config, &created.value.id) + .unwrap() + .expect("cron job bound on create"); + + // Name-only update: no graph_json supplied, so the trigger cannot have + // changed — the existing binding must be left untouched. + flows_update( + &config, + &created.value.id, + Some("renamed".to_string()), + None, + None, + ) + .await + .unwrap(); + + let job = crate::openhuman::cron::find_flow_schedule_job(&config, &created.value.id) + .unwrap() + .expect("cron job still bound"); + assert_eq!(job.id, old_job.id); + assert_eq!(job.expression, old_job.expression); +} + +// ── flows_resume (issue B2) ─────────────────────────────────────────────── + +fn approval_gated_graph() -> Value { + json!({ + "name": "approval-gated", + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Trigger" }, + { "id": "gate", "kind": "output_parser", "name": "Gate", "config": { "requires_approval": true } }, + { "id": "downstream", "kind": "output_parser", "name": "Downstream" } + ], + "edges": [ + { "from_node": "t", "to_node": "gate" }, + { "from_node": "gate", "to_node": "downstream" } + ] + }) +} + +#[tokio::test] +async fn flows_resume_continues_a_paused_run_to_completion() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "gated".to_string(), approval_gated_graph(), false) + .await + .unwrap(); + + let run = flows_run(&config, &created.value.id, json!({ "x": 1 })) + .await + .unwrap(); + let thread_id = run.value["thread_id"].as_str().unwrap().to_string(); + let pending: Vec = + serde_json::from_value(run.value["pending_approvals"].clone()).unwrap(); + assert_eq!(pending, vec!["gate".to_string()]); + + let resumed = flows_resume(&config, &created.value.id, &thread_id, pending) + .await + .unwrap(); + assert_eq!(resumed.value["pending_approvals"], json!([])); + assert!( + !resumed.value["output"]["nodes"]["downstream"]["items"].is_null(), + "downstream should run once the gate is approved via resume" + ); + + let reloaded = flows_get(&config, &created.value.id).await.unwrap(); + assert_eq!(reloaded.value.last_status.as_deref(), Some("completed")); + + // The run-history row must reflect the final completed status, not the + // intermediate pending_approval one it started at. + let run_row = flows_get_run(&config, &thread_id).await.unwrap(); + assert_eq!(run_row.value.status, "completed"); + assert!(run_row.value.pending_approvals.is_empty()); + assert!( + run_row + .value + .steps + .iter() + .any(|s| s.node_id == "downstream"), + "resume should reconstruct the downstream step that ran after approval" + ); +} + +#[tokio::test] +async fn flows_resume_missing_flow_errors() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let err = flows_resume(&config, "missing", "thread-1", vec![]) + .await + .expect_err("must error"); + assert!(err.contains("not found")); +} + +// ── flows_resume host-side approval guard (issue B2 finding #3) ────────── +// +// tinyflows 0.2's `resume_with_checkpointer` treats the resume call itself +// as approval of whatever gate paused the run — its `approvals` argument is +// advisory, not enforced by the crate. Live testing confirmed +// `flows_resume(..., approvals: [])` on a paused run still completed it. +// These tests exercise the host-side guard added in `flows::ops::flows_resume` +// that requires `approvals` to actually name a currently-pending gate, +// straight from the persisted `flow_runs` row, before ever calling into the +// engine. + +#[tokio::test] +async fn flows_resume_with_empty_approvals_is_rejected_and_does_not_complete_the_run() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "gated".to_string(), approval_gated_graph(), false) + .await + .unwrap(); + + let run = flows_run(&config, &created.value.id, json!({ "x": 1 })) + .await + .unwrap(); + let thread_id = run.value["thread_id"].as_str().unwrap().to_string(); + + let err = flows_resume(&config, &created.value.id, &thread_id, vec![]) + .await + .expect_err("an empty approvals list must not silently approve the pending gate"); + assert!( + err.contains("no pending approval matches"), + "expected a clear approval-mismatch error, got: {err}" + ); + + // The run must still be sitting at pending_approval, not completed. + let run_row = flows_get_run(&config, &thread_id).await.unwrap(); + assert_eq!(run_row.value.status, "pending_approval"); + assert_eq!(run_row.value.pending_approvals, vec!["gate".to_string()]); + + let reloaded = flows_get(&config, &created.value.id).await.unwrap(); + assert_eq!( + reloaded.value.last_status.as_deref(), + Some("pending_approval"), + "a rejected resume attempt must not overwrite the flow's last_status as completed" + ); +} + +#[tokio::test] +async fn flows_resume_with_mismatched_approvals_is_rejected() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "gated".to_string(), approval_gated_graph(), false) + .await + .unwrap(); + + let run = flows_run(&config, &created.value.id, json!({ "x": 1 })) + .await + .unwrap(); + let thread_id = run.value["thread_id"].as_str().unwrap().to_string(); + + // Names a node id that is not actually pending for this run. + let err = flows_resume( + &config, + &created.value.id, + &thread_id, + vec!["not-a-real-gate".to_string()], + ) + .await + .expect_err("approvals naming no actually-pending gate must be rejected"); + assert!(err.contains("no pending approval matches")); +} + +#[tokio::test] +async fn flows_resume_with_the_correct_gate_completes_and_runs_downstream() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "gated".to_string(), approval_gated_graph(), false) + .await + .unwrap(); + + let run = flows_run(&config, &created.value.id, json!({ "x": 1 })) + .await + .unwrap(); + let thread_id = run.value["thread_id"].as_str().unwrap().to_string(); + + let resumed = flows_resume( + &config, + &created.value.id, + &thread_id, + vec!["gate".to_string()], + ) + .await + .unwrap(); + assert_eq!(resumed.value["pending_approvals"], json!([])); + assert!( + !resumed.value["output"]["nodes"]["downstream"]["items"].is_null(), + "downstream should run once the correct gate is named in approvals" + ); + + let reloaded = flows_get(&config, &created.value.id).await.unwrap(); + assert_eq!(reloaded.value.last_status.as_deref(), Some("completed")); +} + +#[tokio::test] +async fn flows_resume_of_a_non_paused_run_errors_clearly() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "demo".to_string(), trigger_only_graph(), false) + .await + .unwrap(); + + // This run completes outright (no approval gate) — its recorded status + // is "completed", not "pending_approval". + let run = flows_run(&config, &created.value.id, json!({})) + .await + .unwrap(); + let thread_id = run.value["thread_id"].as_str().unwrap().to_string(); + + let err = flows_resume(&config, &created.value.id, &thread_id, vec![]) + .await + .expect_err("resuming an already-completed run must be a clear error, not a silent no-op"); + assert!( + err.contains("not pending approval") || err.contains("no paused run"), + "expected a clear non-paused-run error, got: {err}" + ); +} + +#[tokio::test] +async fn flows_resume_with_no_recorded_run_for_thread_id_errors_clearly() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "demo".to_string(), trigger_only_graph(), false) + .await + .unwrap(); + + let err = flows_resume( + &config, + &created.value.id, + "thread-that-was-never-started", + vec![], + ) + .await + .expect_err("must error when no run is recorded for this thread_id"); + assert!(err.contains("no paused run to resume")); +} + +// ── run history (flows_list_runs / flows_get_run) ──────────────────────── + +#[tokio::test] +async fn flows_run_persists_a_flow_run_row_queryable_via_list_and_get() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "demo".to_string(), trigger_only_graph(), false) + .await + .unwrap(); + + let run = flows_run(&config, &created.value.id, json!({ "hello": "world" })) + .await + .unwrap(); + let thread_id = run.value["thread_id"].as_str().unwrap().to_string(); + + let runs = flows_list_runs(&config, &created.value.id, 20) + .await + .unwrap(); + assert_eq!(runs.value.len(), 1); + assert_eq!(runs.value[0].id, thread_id); + assert_eq!(runs.value[0].status, "completed"); + + let single = flows_get_run(&config, &thread_id).await.unwrap(); + assert_eq!(single.value.flow_id, created.value.id); + assert_eq!(single.value.status, "completed"); + assert!( + single.value.steps.iter().any(|s| s.node_id == "t"), + "the trigger node's step should be reconstructed from output[\"nodes\"]" + ); +} + +#[tokio::test] +async fn flows_get_run_missing_run_errors() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let err = flows_get_run(&config, "missing-run") + .await + .expect_err("must error"); + assert!(err.contains("not found")); +} + +// ── pending-approval notification ──────────────────────────────────────── + +#[tokio::test] +async fn flows_run_emits_pending_approval_notification() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let mut rx = crate::openhuman::notifications::bus::subscribe_core_notifications(); + + let created = flows_create( + &config, + "gated-notify".to_string(), + approval_gated_graph(), + false, + ) + .await + .unwrap(); + + let run = flows_run(&config, &created.value.id, json!({})) + .await + .unwrap(); + let thread_id = run.value["thread_id"].as_str().unwrap().to_string(); + + // Filter for our notification specifically — the broadcast bus is + // process-global, so a concurrently-running test's notification could + // otherwise be received first. + let expected_prefix = format!("flow-pending-approval:{}:", created.value.id); + let mut found = None; + for _ in 0..20 { + match tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()).await { + Ok(Ok(n)) if n.id.starts_with(&expected_prefix) => { + found = Some(n); + break; + } + Ok(Ok(_unrelated)) => continue, + _ => break, + } + } + let notification = found.expect("expected a pending-approval notification for this flow"); + + assert_eq!( + notification.category, + crate::openhuman::notifications::types::CoreNotificationCategory::Agents + ); + let actions = notification + .actions + .expect("pending-approval notification must carry an action"); + let approve = actions + .iter() + .find(|a| a.action_id == "approve") + .expect("expected an 'approve' action"); + let payload = approve + .payload + .clone() + .expect("approve action must carry a payload"); + assert_eq!(payload["flow_id"], json!(created.value.id)); + assert_eq!(payload["thread_id"], json!(thread_id)); + assert_eq!(payload["node_ids"], json!(["gate"])); +} + +#[tokio::test] +async fn flows_run_does_not_notify_when_run_completes_without_pending_approvals() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let mut rx = crate::openhuman::notifications::bus::subscribe_core_notifications(); + + let created = flows_create(&config, "no-gate".to_string(), trigger_only_graph(), false) + .await + .unwrap(); + let created_id = created.value.id.clone(); + + flows_run(&config, &created.value.id, json!({})) + .await + .unwrap(); + + let expected_prefix = format!("flow-pending-approval:{created_id}:"); + let saw_notification = tokio::time::timeout(std::time::Duration::from_millis(300), async { + loop { + match rx.recv().await { + Ok(n) if n.id.starts_with(&expected_prefix) => return true, + Ok(_) => continue, + Err(_) => return false, + } + } + }) + .await + .unwrap_or(false); + assert!( + !saw_notification, + "a fully-completed run must not publish a pending-approval notification" + ); +} diff --git a/src/openhuman/flows/schemas.rs b/src/openhuman/flows/schemas.rs index f369e0a27..50c4212c6 100644 --- a/src/openhuman/flows/schemas.rs +++ b/src/openhuman/flows/schemas.rs @@ -32,6 +32,39 @@ fn flow_output() -> FieldSchema { } } +fn require_approval_input() -> FieldSchema { + FieldSchema { + name: "require_approval", + ty: TypeSchema::Option(Box::new(TypeSchema::Bool)), + comment: "Force a human-approval gate on every outbound tool/HTTP action this flow \ + takes, regardless of its saved-flow trust root. Defaults to `false`.", + required: false, + } +} + +fn run_output_fields() -> Vec { + vec![ + FieldSchema { + name: "output", + ty: TypeSchema::Json, + comment: "The run's final state (per-node items, trigger payload).", + required: true, + }, + FieldSchema { + name: "pending_approvals", + ty: TypeSchema::Array(Box::new(TypeSchema::String)), + comment: "Node ids paused awaiting human approval; empty once completed.", + required: true, + }, + FieldSchema { + name: "thread_id", + ty: TypeSchema::String, + comment: "Durable checkpoint thread id for this run (needed to resume).", + required: true, + }, + ] +} + pub fn all_controller_schemas() -> Vec { vec![ schemas("create"), @@ -41,6 +74,9 @@ pub fn all_controller_schemas() -> Vec { schemas("delete"), schemas("set_enabled"), schemas("run"), + schemas("resume"), + schemas("list_runs"), + schemas("get_run"), ] } @@ -74,6 +110,18 @@ pub fn all_registered_controllers() -> Vec { schema: schemas("run"), handler: handle_run, }, + RegisteredController { + schema: schemas("resume"), + handler: handle_resume, + }, + RegisteredController { + schema: schemas("list_runs"), + handler: handle_list_runs, + }, + RegisteredController { + schema: schemas("get_run"), + handler: handle_get_run, + }, ] } @@ -97,6 +145,7 @@ pub fn schemas(function: &str) -> ControllerSchema { "A tinyflows WorkflowGraph (nodes + edges); validated and migrated on save.", required: true, }, + require_approval_input(), ], outputs: vec![flow_output()], }, @@ -137,6 +186,7 @@ pub fn schemas(function: &str) -> ControllerSchema { comment: "Replacement WorkflowGraph, if changing it.", required: false, }, + require_approval_input(), ], outputs: vec![flow_output()], }, @@ -199,33 +249,81 @@ pub fn schemas(function: &str) -> ControllerSchema { outputs: vec![FieldSchema { name: "result", ty: TypeSchema::Object { - fields: vec![ - FieldSchema { - name: "output", - ty: TypeSchema::Json, - comment: "The run's final state (per-node items, trigger payload).", - required: true, - }, - FieldSchema { - name: "pending_approvals", - ty: TypeSchema::Array(Box::new(TypeSchema::String)), - comment: - "Node ids paused awaiting human approval; empty once completed.", - required: true, - }, - FieldSchema { - name: "thread_id", - ty: TypeSchema::String, - comment: - "Durable checkpoint thread id for this run (needed to resume).", - required: true, - }, - ], + fields: run_output_fields(), }, comment: "Run outcome payload.", required: true, }], }, + "resume" => ControllerSchema { + namespace: "flows", + function: "resume", + description: "Resume a flow run paused at a human-in-the-loop approval gate, \ + continuing from its durable checkpoint.", + inputs: vec![ + id_input("Identifier of the flow to resume."), + FieldSchema { + name: "thread_id", + ty: TypeSchema::String, + comment: + "The checkpoint thread id returned by `flows_run` / a prior `flows_resume`.", + required: true, + }, + FieldSchema { + name: "approvals", + ty: TypeSchema::Option(Box::new(TypeSchema::Array(Box::new( + TypeSchema::String, + )))), + comment: "Node ids being approved; defaults to an empty list.", + required: false, + }, + ], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Object { + fields: run_output_fields(), + }, + comment: "Resume outcome payload (same shape as `run`'s).", + required: true, + }], + }, + "list_runs" => ControllerSchema { + namespace: "flows", + function: "list_runs", + description: "List the most recent runs for a flow, newest first.", + inputs: vec![ + id_input("Identifier of the flow whose runs to list."), + FieldSchema { + name: "limit", + ty: TypeSchema::Option(Box::new(TypeSchema::U64)), + comment: "Maximum number of runs to return; defaults to 20.", + required: false, + }, + ], + outputs: vec![FieldSchema { + name: "runs", + ty: TypeSchema::Array(Box::new(TypeSchema::Ref("FlowRun"))), + comment: "Persisted run records for this flow, newest first.", + required: true, + }], + }, + "get_run" => ControllerSchema { + namespace: "flows", + function: "get_run", + description: "Load one persisted flow run record by its (checkpoint thread) id.", + inputs: vec![FieldSchema { + name: "run_id", + ty: TypeSchema::String, + comment: "Identifier of the run to load (== its checkpoint thread id).", + required: true, + }], + outputs: vec![FieldSchema { + name: "run", + ty: TypeSchema::Ref("FlowRun"), + comment: "The persisted run record.", + required: true, + }], + }, _other => ControllerSchema { namespace: "flows", function: "unknown", @@ -251,7 +349,11 @@ fn handle_create(params: Map) -> ControllerFuture { let config = config_rpc::load_config_with_timeout().await?; let name = read_required::(¶ms, "name")?; let graph = read_required::(¶ms, "graph")?; - to_json(ops::flows_create(&config, name, graph).await?) + let require_approval = params + .get("require_approval") + .and_then(Value::as_bool) + .unwrap_or(false); + to_json(ops::flows_create(&config, name, graph, require_approval).await?) }) } @@ -281,7 +383,8 @@ fn handle_update(params: Map) -> ControllerFuture { .transpose() .map_err(|e| format!("invalid 'name': {e}"))?; let graph = params.get("graph").filter(|v| !v.is_null()).cloned(); - to_json(ops::flows_update(&config, id.trim(), name, graph).await?) + let require_approval = params.get("require_approval").and_then(Value::as_bool); + to_json(ops::flows_update(&config, id.trim(), name, graph, require_approval).await?) }) } @@ -314,6 +417,44 @@ fn handle_run(params: Map) -> ControllerFuture { }) } +fn handle_resume(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let id = read_required::(¶ms, "id")?; + let thread_id = read_required::(¶ms, "thread_id")?; + let approvals: Vec = params + .get("approvals") + .filter(|v| !v.is_null()) + .cloned() + .map(serde_json::from_value) + .transpose() + .map_err(|e| format!("invalid 'approvals': {e}"))? + .unwrap_or_default(); + to_json(ops::flows_resume(&config, id.trim(), thread_id.trim(), approvals).await?) + }) +} + +fn handle_list_runs(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let id = read_required::(¶ms, "id")?; + let limit = params + .get("limit") + .and_then(Value::as_u64) + .and_then(|n| usize::try_from(n).ok()) + .unwrap_or(20); + to_json(ops::flows_list_runs(&config, id.trim(), limit).await?) + }) +} + +fn handle_get_run(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let run_id = read_required::(¶ms, "run_id")?; + to_json(ops::flows_get_run(&config, run_id.trim()).await?) + }) +} + fn read_required(params: &Map, key: &str) -> Result { let value = params .get(key) @@ -345,7 +486,10 @@ mod tests { "update", "delete", "set_enabled", - "run" + "run", + "resume", + "list_runs", + "get_run", ] ); } @@ -353,7 +497,7 @@ mod tests { #[test] fn all_registered_controllers_has_handler_per_schema() { let controllers = all_registered_controllers(); - assert_eq!(controllers.len(), 7); + assert_eq!(controllers.len(), 10); let names: Vec<_> = controllers.iter().map(|c| c.schema.function).collect(); assert_eq!( names, @@ -364,7 +508,10 @@ mod tests { "update", "delete", "set_enabled", - "run" + "run", + "resume", + "list_runs", + "get_run", ] ); } @@ -382,6 +529,17 @@ mod tests { assert_eq!(required, vec!["name", "graph"]); } + #[test] + fn schemas_create_require_approval_is_optional() { + let s = schemas("create"); + let field = s + .inputs + .iter() + .find(|f| f.name == "require_approval") + .unwrap(); + assert!(!field.required); + } + #[test] fn schemas_run_input_is_optional() { let s = schemas("run"); @@ -389,6 +547,39 @@ mod tests { assert!(!input.required); } + #[test] + fn schemas_resume_requires_id_and_thread_id_but_not_approvals() { + let s = schemas("resume"); + let required: Vec<_> = s + .inputs + .iter() + .filter(|f| f.required) + .map(|f| f.name) + .collect(); + assert_eq!(required, vec!["id", "thread_id"]); + let approvals = s.inputs.iter().find(|f| f.name == "approvals").unwrap(); + assert!(!approvals.required); + } + + #[test] + fn schemas_list_runs_limit_is_optional() { + let s = schemas("list_runs"); + let limit = s.inputs.iter().find(|f| f.name == "limit").unwrap(); + assert!(!limit.required); + } + + #[test] + fn schemas_get_run_requires_run_id() { + let s = schemas("get_run"); + let required: Vec<_> = s + .inputs + .iter() + .filter(|f| f.required) + .map(|f| f.name) + .collect(); + assert_eq!(required, vec!["run_id"]); + } + #[test] fn schemas_unknown_function_returns_placeholder() { let s = schemas("does-not-exist"); diff --git a/src/openhuman/flows/store.rs b/src/openhuman/flows/store.rs index d62365b2d..3e1a483d2 100644 --- a/src/openhuman/flows/store.rs +++ b/src/openhuman/flows/store.rs @@ -15,6 +15,7 @@ //! `checkpoints.db` (see `src/openhuman/tinyflows/mod.rs::open_flow_checkpointer`). use crate::openhuman::config::Config; +use crate::openhuman::flows::types::{FlowRun, FlowRunStep}; use crate::openhuman::flows::Flow; use anyhow::{Context, Result}; use chrono::Utc; @@ -59,30 +60,95 @@ fn with_connection(config: &Config, f: impl FnOnce(&Connection) -> Result) key TEXT NOT NULL, value TEXT NOT NULL, PRIMARY KEY (namespace, key) - );", + ); + + CREATE TABLE IF NOT EXISTS flow_runs ( + id TEXT PRIMARY KEY, + flow_id TEXT NOT NULL, + thread_id TEXT NOT NULL, + status TEXT NOT NULL, + started_at TEXT NOT NULL, + finished_at TEXT, + steps_json TEXT NOT NULL DEFAULT '[]', + pending_approvals_json TEXT NOT NULL DEFAULT '[]', + error TEXT, + FOREIGN KEY (flow_id) REFERENCES flow_definitions(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_flow_runs_flow_id ON flow_runs(flow_id); + CREATE INDEX IF NOT EXISTS idx_flow_runs_started_at ON flow_runs(started_at);", ) .context("Failed to initialize flows schema")?; + // `require_approval` (issue B2) — added post-hoc so a workspace created + // before this column existed still opens cleanly. Mirrors + // `cron::store`'s `add_column_if_missing` idiom. + add_column_if_missing( + &conn, + "flow_definitions", + "require_approval", + "INTEGER NOT NULL DEFAULT 0", + )?; + tracing::debug!(db = %db_path.display(), "[flows] store opened"); f(&conn) } +/// Adds `name` to `table` if it isn't already present, tolerating the race +/// where a concurrent process adds the same column between the `PRAGMA` +/// check and the `ALTER TABLE`. Mirrors `cron::store::add_column_if_missing` +/// (kept per-domain rather than shared — each store owns its own connection +/// helper and this is a handful of lines). +fn add_column_if_missing(conn: &Connection, table: &str, name: &str, sql_type: &str) -> Result<()> { + let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?; + let mut rows = stmt.query([])?; + while let Some(row) = rows.next()? { + let col_name: String = row.get(1)?; + if col_name == name { + return Ok(()); + } + } + drop(rows); + drop(stmt); + + match conn.execute( + &format!("ALTER TABLE {table} ADD COLUMN {name} {sql_type}"), + [], + ) { + Ok(_) => Ok(()), + Err(rusqlite::Error::SqliteFailure(err, Some(ref msg))) + if msg.contains("duplicate column name") => + { + tracing::debug!( + "[flows] column {table}.{name} already exists (concurrent migration): {err}" + ); + Ok(()) + } + Err(e) => Err(e).with_context(|| format!("Failed to add {table}.{name}")), + } +} + +/// Shared column list for every `flow_definitions` SELECT — keeps +/// [`map_flow_row`]'s positional `row.get(N)` calls in sync with the query. +const FLOW_DEFINITION_COLUMNS: &str = "id, name, graph_json, enabled, created_at, updated_at, \ + last_run_at, last_status, require_approval"; + /// Inserts or fully replaces a flow definition row. pub fn upsert_flow(config: &Config, flow: &Flow) -> Result<()> { let graph_json = serde_json::to_string(&flow.graph).context("Failed to serialize graph")?; with_connection(config, |conn| { conn.execute( "INSERT INTO flow_definitions - (id, name, graph_json, enabled, created_at, updated_at, last_run_at, last_status) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) + (id, name, graph_json, enabled, created_at, updated_at, last_run_at, last_status, require_approval) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) ON CONFLICT(id) DO UPDATE SET name = excluded.name, graph_json = excluded.graph_json, enabled = excluded.enabled, updated_at = excluded.updated_at, last_run_at = excluded.last_run_at, - last_status = excluded.last_status", + last_status = excluded.last_status, + require_approval = excluded.require_approval", params![ flow.id, flow.name, @@ -92,6 +158,7 @@ pub fn upsert_flow(config: &Config, flow: &Flow) -> Result<()> { flow.updated_at, flow.last_run_at, flow.last_status, + if flow.require_approval { 1 } else { 0 }, ], ) .context("Failed to upsert flow definition")?; @@ -106,6 +173,7 @@ pub fn create_flow( config: &Config, name: String, graph: tinyflows::model::WorkflowGraph, + require_approval: bool, ) -> Result { let now = Utc::now().to_rfc3339(); let flow = Flow { @@ -117,6 +185,7 @@ pub fn create_flow( updated_at: now, last_run_at: None, last_status: None, + require_approval, }; upsert_flow(config, &flow)?; Ok(flow) @@ -127,10 +196,9 @@ pub fn create_flow( /// under an older `schema_version` is upgraded on read. pub fn get_flow(config: &Config, id: &str) -> Result> { with_connection(config, |conn| { - let mut stmt = conn.prepare( - "SELECT id, name, graph_json, enabled, created_at, updated_at, last_run_at, last_status - FROM flow_definitions WHERE id = ?1", - )?; + let mut stmt = conn.prepare(&format!( + "SELECT {FLOW_DEFINITION_COLUMNS} FROM flow_definitions WHERE id = ?1" + ))?; let mut rows = stmt.query(params![id])?; match rows.next()? { Some(row) => Ok(Some(map_flow_row(row)?)), @@ -142,10 +210,31 @@ pub fn get_flow(config: &Config, id: &str) -> Result> { /// Lists all saved flows, migrating each graph on read (see [`get_flow`]). pub fn list_flows(config: &Config) -> Result> { with_connection(config, |conn| { - let mut stmt = conn.prepare( - "SELECT id, name, graph_json, enabled, created_at, updated_at, last_run_at, last_status - FROM flow_definitions ORDER BY created_at ASC", - )?; + let mut stmt = conn.prepare(&format!( + "SELECT {FLOW_DEFINITION_COLUMNS} FROM flow_definitions ORDER BY created_at ASC" + ))?; + let rows = stmt.query_map([], map_flow_row)?; + let mut flows = Vec::new(); + for row in rows { + flows.push(row?); + } + Ok(flows) + }) +} + +/// Lists only enabled flows, migrating each graph on read (see [`get_flow`]). +/// +/// Used by `flows::bus::FlowTriggerSubscriber` to match an inbound +/// `ComposioTriggerReceived` event against every enabled `app_event` flow — +/// scanning the (small) enabled set once per event is simpler and cheap +/// enough at expected flow counts; a dedicated toolkit/trigger_slug index is +/// a later optimization if this ever shows up as a bottleneck. +pub fn list_enabled_flows(config: &Config) -> Result> { + with_connection(config, |conn| { + let mut stmt = conn.prepare(&format!( + "SELECT {FLOW_DEFINITION_COLUMNS} FROM flow_definitions WHERE enabled = 1 \ + ORDER BY created_at ASC" + ))?; let rows = stmt.query_map([], map_flow_row)?; let mut flows = Vec::new(); for row in rows { @@ -185,13 +274,14 @@ pub fn set_enabled(config: &Config, id: &str, enabled: bool) -> Result { get_flow(config, id)?.ok_or_else(|| anyhow::anyhow!("flow '{id}' not found after update")) } -/// Replaces a flow's name/graph (re-validated by the caller before this is -/// invoked) in place, bumping `updated_at`. +/// Replaces a flow's name/graph/`require_approval` (re-validated by the +/// caller before this is invoked) in place, bumping `updated_at`. pub fn update_flow_graph( config: &Config, id: &str, name: String, graph: tinyflows::model::WorkflowGraph, + require_approval: bool, ) -> Result { let graph_json = serde_json::to_string(&graph).context("Failed to serialize graph")?; let now = Utc::now().to_rfc3339(); @@ -200,8 +290,15 @@ pub fn update_flow_graph( // `enabled` / `last_run_at` / `last_status` from a read-modify-write. let changed = with_connection(config, |conn| { conn.execute( - "UPDATE flow_definitions SET name = ?1, graph_json = ?2, updated_at = ?3 WHERE id = ?4", - params![name, graph_json, now, id], + "UPDATE flow_definitions SET name = ?1, graph_json = ?2, updated_at = ?3, \ + require_approval = ?4 WHERE id = ?5", + params![ + name, + graph_json, + now, + if require_approval { 1 } else { 0 }, + id + ], ) .context("Failed to update flow") })?; @@ -246,6 +343,7 @@ fn map_flow_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { updated_at: row.get(5)?, last_run_at: row.get(6)?, last_status: row.get(7)?, + require_approval: row.get::<_, i64>(8)? != 0, }) } @@ -296,6 +394,113 @@ pub fn kv_set( }) } +/// Shared column list for every `flow_runs` SELECT — keeps +/// [`map_flow_run_row`]'s positional `row.get(N)` calls in sync. +const FLOW_RUN_COLUMNS: &str = "id, flow_id, thread_id, status, started_at, finished_at, \ + steps_json, pending_approvals_json, error"; + +/// Inserts the initial `"running"` row for a new `flows_run` / `flows_resume` +/// invocation. `id` and `thread_id` are the same value in practice (the +/// tinyflows checkpointer thread id doubles as the run's stable identifier), +/// kept as two columns because they answer two different questions (row +/// identity vs. the checkpointer key `flows_resume` needs). +pub fn insert_flow_run( + config: &Config, + id: &str, + flow_id: &str, + thread_id: &str, + started_at: &str, +) -> Result<()> { + with_connection(config, |conn| { + conn.execute( + "INSERT INTO flow_runs (id, flow_id, thread_id, status, started_at) + VALUES (?1, ?2, ?3, 'running', ?4)", + params![id, flow_id, thread_id, started_at], + ) + .context("Failed to insert flow run")?; + Ok(()) + }) +} + +/// Finalizes a flow run row: settles its terminal `status`, `finished_at`, +/// reconstructed `steps`, `pending_approvals`, and (on failure) `error`. +/// Called once a `flows_run` / `flows_resume` invocation settles — including +/// the timeout / capability-error paths, so a row never gets stuck at +/// `"running"` when the process is still up. +pub fn finish_flow_run( + config: &Config, + id: &str, + status: &str, + finished_at: &str, + steps: &[FlowRunStep], + pending_approvals: &[String], + error: Option<&str>, +) -> Result<()> { + let steps_json = serde_json::to_string(steps).context("Failed to serialize flow run steps")?; + let pending_json = serde_json::to_string(pending_approvals) + .context("Failed to serialize flow run pending approvals")?; + with_connection(config, |conn| { + conn.execute( + "UPDATE flow_runs SET status = ?1, finished_at = ?2, steps_json = ?3, \ + pending_approvals_json = ?4, error = ?5 WHERE id = ?6", + params![status, finished_at, steps_json, pending_json, error, id], + ) + .context("Failed to finish flow run")?; + Ok(()) + }) +} + +/// Loads one flow run by id (== thread_id). +pub fn get_flow_run(config: &Config, id: &str) -> Result> { + with_connection(config, |conn| { + let mut stmt = conn.prepare(&format!( + "SELECT {FLOW_RUN_COLUMNS} FROM flow_runs WHERE id = ?1" + ))?; + let mut rows = stmt.query(params![id])?; + match rows.next()? { + Some(row) => Ok(Some(map_flow_run_row(row)?)), + None => Ok(None), + } + }) +} + +/// Lists the most recent runs for a flow, newest first. +pub fn list_flow_runs(config: &Config, flow_id: &str, limit: usize) -> Result> { + with_connection(config, |conn| { + let lim = i64::try_from(limit.max(1)).context("Run history limit overflow")?; + let mut stmt = conn.prepare(&format!( + "SELECT {FLOW_RUN_COLUMNS} FROM flow_runs WHERE flow_id = ?1 \ + ORDER BY started_at DESC, id DESC LIMIT ?2" + ))?; + let rows = stmt.query_map(params![flow_id, lim], map_flow_run_row)?; + let mut runs = Vec::new(); + for row in rows { + runs.push(row?); + } + Ok(runs) + }) +} + +fn map_flow_run_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let steps_raw: String = row.get(6)?; + let steps: Vec = serde_json::from_str(&steps_raw).map_err(sql_conversion_error)?; + let pending_raw: String = row.get(7)?; + let pending_approvals: Vec = + serde_json::from_str(&pending_raw).map_err(sql_conversion_error)?; + + Ok(FlowRun { + id: row.get(0)?, + flow_id: row.get(1)?, + thread_id: row.get(2)?, + status: row.get(3)?, + started_at: row.get(4)?, + finished_at: row.get(5)?, + steps, + pending_approvals, + error: row.get(8)?, + }) +} + #[cfg(test)] #[path = "store_tests.rs"] mod tests; diff --git a/src/openhuman/flows/store_tests.rs b/src/openhuman/flows/store_tests.rs index 249a5f5ea..7bbab20e8 100644 --- a/src/openhuman/flows/store_tests.rs +++ b/src/openhuman/flows/store_tests.rs @@ -34,7 +34,7 @@ fn create_get_list_delete_roundtrip() { let tmp = TempDir::new().unwrap(); let config = test_config(&tmp); - let flow = create_flow(&config, "demo".to_string(), trigger_graph()).unwrap(); + let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false).unwrap(); assert_eq!(flow.name, "demo"); assert!(flow.enabled); @@ -70,7 +70,7 @@ fn remove_flow_errors_when_not_found() { fn set_enabled_toggles_and_persists() { let tmp = TempDir::new().unwrap(); let config = test_config(&tmp); - let flow = create_flow(&config, "demo".to_string(), trigger_graph()).unwrap(); + let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false).unwrap(); assert!(flow.enabled); let disabled = set_enabled(&config, &flow.id, false).unwrap(); @@ -87,11 +87,12 @@ fn set_enabled_toggles_and_persists() { fn update_flow_graph_bumps_updated_at_and_preserves_created_at() { let tmp = TempDir::new().unwrap(); let config = test_config(&tmp); - let flow = create_flow(&config, "demo".to_string(), trigger_graph()).unwrap(); + let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false).unwrap(); let mut new_graph = trigger_graph(); new_graph.name = "renamed-graph".to_string(); - let updated = update_flow_graph(&config, &flow.id, "renamed".to_string(), new_graph).unwrap(); + let updated = + update_flow_graph(&config, &flow.id, "renamed".to_string(), new_graph, false).unwrap(); assert_eq!(updated.name, "renamed"); assert_eq!(updated.created_at, flow.created_at); @@ -102,7 +103,7 @@ fn update_flow_graph_bumps_updated_at_and_preserves_created_at() { fn record_run_sets_last_run_fields() { let tmp = TempDir::new().unwrap(); let config = test_config(&tmp); - let flow = create_flow(&config, "demo".to_string(), trigger_graph()).unwrap(); + let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false).unwrap(); assert!(flow.last_run_at.is_none()); record_run(&config, &flow.id, "completed").unwrap(); @@ -167,3 +168,230 @@ fn kv_get_set_round_trips_and_is_namespace_scoped() { Some(serde_json::json!(2)) ); } + +// ── require_approval ───────────────────────────────────────────────────── + +#[test] +fn create_flow_persists_require_approval() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + let flow = create_flow(&config, "demo".to_string(), trigger_graph(), true).unwrap(); + assert!(flow.require_approval); + + let reloaded = get_flow(&config, &flow.id).unwrap().unwrap(); + assert!(reloaded.require_approval); +} + +#[test] +fn update_flow_graph_can_change_require_approval() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false).unwrap(); + assert!(!flow.require_approval); + + let updated = + update_flow_graph(&config, &flow.id, flow.name.clone(), trigger_graph(), true).unwrap(); + assert!(updated.require_approval); + + let reloaded = get_flow(&config, &flow.id).unwrap().unwrap(); + assert!(reloaded.require_approval); +} + +#[test] +fn legacy_flow_definitions_row_without_require_approval_column_defaults_false() { + // A row inserted before the `require_approval` column existed (the + // `add_column_if_missing` ALTER runs on every `with_connection` call, so + // this simulates a workspace opened once on an older build). + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + let legacy_graph_json = serde_json::to_string(&trigger_graph()).unwrap(); + with_connection(&config, |conn| { + conn.execute( + "INSERT INTO flow_definitions + (id, name, graph_json, enabled, created_at, updated_at, last_run_at, last_status) + VALUES ('legacy-2', 'legacy', ?1, 1, '2020-01-01T00:00:00Z', '2020-01-01T00:00:00Z', NULL, NULL)", + rusqlite::params![legacy_graph_json], + )?; + Ok(()) + }) + .unwrap(); + + let loaded = get_flow(&config, "legacy-2").unwrap().expect("row present"); + assert!(!loaded.require_approval); +} + +// ── list_enabled_flows ──────────────────────────────────────────────────── + +#[test] +fn list_enabled_flows_excludes_disabled() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + let enabled_flow = create_flow(&config, "enabled".to_string(), trigger_graph(), false).unwrap(); + let disabled_flow = + create_flow(&config, "disabled".to_string(), trigger_graph(), false).unwrap(); + set_enabled(&config, &disabled_flow.id, false).unwrap(); + + let enabled = list_enabled_flows(&config).unwrap(); + assert_eq!(enabled.len(), 1); + assert_eq!(enabled[0].id, enabled_flow.id); +} + +// ── flow_runs CRUD ──────────────────────────────────────────────────────── + +#[test] +fn flow_run_insert_finish_get_round_trip() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false).unwrap(); + + let thread_id = format!("flow:{}:run-1", flow.id); + insert_flow_run( + &config, + &thread_id, + &flow.id, + &thread_id, + "2026-01-01T00:00:00Z", + ) + .unwrap(); + + let running = get_flow_run(&config, &thread_id) + .unwrap() + .expect("row present"); + assert_eq!(running.status, "running"); + assert!(running.finished_at.is_none()); + assert!(running.steps.is_empty()); + + let steps = vec![FlowRunStep { + node_id: "t".to_string(), + output: serde_json::json!([{"json": {"x": 1}}]), + port: None, + }]; + finish_flow_run( + &config, + &thread_id, + "completed", + "2026-01-01T00:00:01Z", + &steps, + &[], + None, + ) + .unwrap(); + + let finished = get_flow_run(&config, &thread_id) + .unwrap() + .expect("row present"); + assert_eq!(finished.status, "completed"); + assert_eq!( + finished.finished_at.as_deref(), + Some("2026-01-01T00:00:01Z") + ); + assert_eq!(finished.steps.len(), 1); + assert_eq!(finished.steps[0].node_id, "t"); + assert!(finished.pending_approvals.is_empty()); + assert!(finished.error.is_none()); +} + +#[test] +fn finish_flow_run_records_error_on_failure() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false).unwrap(); + let thread_id = format!("flow:{}:run-2", flow.id); + insert_flow_run( + &config, + &thread_id, + &flow.id, + &thread_id, + "2026-01-01T00:00:00Z", + ) + .unwrap(); + + finish_flow_run( + &config, + &thread_id, + "failed", + "2026-01-01T00:00:01Z", + &[], + &[], + Some("boom"), + ) + .unwrap(); + + let finished = get_flow_run(&config, &thread_id).unwrap().unwrap(); + assert_eq!(finished.status, "failed"); + assert_eq!(finished.error.as_deref(), Some("boom")); +} + +#[test] +fn get_flow_run_returns_none_for_unknown_id() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + assert!(get_flow_run(&config, "missing").unwrap().is_none()); +} + +#[test] +fn list_flow_runs_orders_newest_first_and_is_scoped_to_flow() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow_a = create_flow(&config, "a".to_string(), trigger_graph(), false).unwrap(); + let flow_b = create_flow(&config, "b".to_string(), trigger_graph(), false).unwrap(); + + insert_flow_run( + &config, + "run-a1", + &flow_a.id, + "run-a1", + "2026-01-01T00:00:00Z", + ) + .unwrap(); + insert_flow_run( + &config, + "run-a2", + &flow_a.id, + "run-a2", + "2026-01-02T00:00:00Z", + ) + .unwrap(); + insert_flow_run( + &config, + "run-b1", + &flow_b.id, + "run-b1", + "2026-01-01T00:00:00Z", + ) + .unwrap(); + + let runs_a = list_flow_runs(&config, &flow_a.id, 10).unwrap(); + assert_eq!(runs_a.len(), 2); + assert_eq!(runs_a[0].id, "run-a2", "newest run must come first"); + assert_eq!(runs_a[1].id, "run-a1"); + + let runs_b = list_flow_runs(&config, &flow_b.id, 10).unwrap(); + assert_eq!(runs_b.len(), 1); + assert_eq!(runs_b[0].id, "run-b1"); +} + +#[test] +fn list_flow_runs_respects_limit() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false).unwrap(); + + for i in 0..3 { + let id = format!("run-{i}"); + insert_flow_run( + &config, + &id, + &flow.id, + &id, + &format!("2026-01-0{}T00:00:00Z", i + 1), + ) + .unwrap(); + } + + let limited = list_flow_runs(&config, &flow.id, 2).unwrap(); + assert_eq!(limited.len(), 2); +} diff --git a/src/openhuman/flows/types.rs b/src/openhuman/flows/types.rs index c3b13e67e..9251dcd2e 100644 --- a/src/openhuman/flows/types.rs +++ b/src/openhuman/flows/types.rs @@ -28,6 +28,68 @@ pub struct Flow { pub last_run_at: Option, /// Outcome of the most recent run: `"completed"` | `"pending_approval"` | `"failed"`. pub last_status: Option, + /// "Require approval for outbound actions" (issue B2). When `true`, the + /// approval gate does NOT auto-allow this flow's `TrustedAutomation + /// { Workflow }` trust root — every external_effect tool/HTTP call the + /// flow makes still parks for a real decision, regardless of how the run + /// was triggered. See `src/openhuman/approval/gate.rs` and + /// `src/openhuman/agent/turn_origin.rs::TrustedAutomationSource::Workflow`. + #[serde(default)] + pub require_approval: bool, +} + +/// One reconstructed step of a persisted [`FlowRun`] (issue B2, run-history +/// inspector). tinyflows 0.2's durable path installs a `NoopObserver` (see +/// `src/openhuman/tinyflows/observability.rs`), so there is no live per-step +/// stream to persist — instead, `flows::ops` reconstructs a lean step list +/// straight from `RunOutcome.output["nodes"]` after the run settles: each +/// entry is a node id plus its emitted items and the output port it took, if +/// any. There is no per-step timing or input/attempt data in 0.2. +/// +/// // TODO(0.3): a richer `RunObserver` that streams per-step +/// // node_id/status/output/duration_ms live (see +/// // `tinyflows::observability::ExecutionStep`) would let this type carry +/// // real timing/attempt data instead of being reconstructed post-hoc. +#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)] +pub struct FlowRunStep { + /// The node's id within the flow's graph. + pub node_id: String, + /// The node's emitted items for this run (`output["nodes"][id]["items"]`). + pub output: serde_json::Value, + /// The output port the node routed on, if it picked one (branching / + /// switch nodes) — `output["nodes"][id]["port"]`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub port: Option, +} + +/// 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)] +pub struct FlowRun { + /// Stable identifier for this run — the same value as `thread_id` (the + /// tinyflows checkpointer key), so a run row can be found either way. + pub id: String, + /// The flow this run belongs to. + pub flow_id: String, + /// The tinyflows checkpointer thread id (needed to `flows_resume`). + pub thread_id: String, + /// `"running"` | `"completed"` | `"pending_approval"` | `"failed"`. + pub status: String, + /// RFC3339 timestamp when the run started. + pub started_at: String, + /// RFC3339 timestamp when the run last settled (completed/paused/failed). + /// `None` while a run row is still `"running"`. + pub finished_at: Option, + /// Reconstructed per-node steps (see [`FlowRunStep`]). + #[serde(default)] + pub steps: Vec, + /// Node ids paused awaiting human approval when `status == + /// "pending_approval"`; empty otherwise. + #[serde(default)] + pub pending_approvals: Vec, + /// Error message when `status == "failed"`. + #[serde(default)] + pub error: Option, } #[cfg(test)] @@ -61,11 +123,66 @@ mod tests { updated_at: "2026-01-01T00:00:00Z".to_string(), last_run_at: None, last_status: None, + require_approval: false, }; let json = serde_json::to_string(&flow).expect("serialize"); let back: Flow = serde_json::from_str(&json).expect("deserialize"); assert_eq!(back.id, flow.id); assert_eq!(back.graph, flow.graph); assert!(back.last_run_at.is_none()); + assert!(!back.require_approval); + } + + #[test] + fn flow_require_approval_defaults_false_when_omitted_from_json() { + // Legacy/serialized JSON authored before the field existed must still + // deserialize (SQLite rows are migrated via `add_column_if_missing`, + // but any bare JSON fixture should also default safely). + let json = serde_json::json!({ + "id": "flow_1", + "name": "demo", + "enabled": true, + "graph": sample_graph(), + "created_at": "2026-01-01T00:00:00Z", + "updated_at": "2026-01-01T00:00:00Z", + }); + let flow: Flow = serde_json::from_value(json).expect("deserialize"); + assert!(!flow.require_approval); + } + + #[test] + fn flow_run_round_trips_through_json() { + let run = FlowRun { + id: "flow:flow_1:run-uuid".to_string(), + flow_id: "flow_1".to_string(), + thread_id: "flow:flow_1:run-uuid".to_string(), + status: "completed".to_string(), + started_at: "2026-01-01T00:00:00Z".to_string(), + finished_at: Some("2026-01-01T00:00:01Z".to_string()), + steps: vec![FlowRunStep { + node_id: "t".to_string(), + output: serde_json::json!([{"json": {"hello": "world"}}]), + port: None, + }], + pending_approvals: Vec::new(), + error: None, + }; + let json = serde_json::to_string(&run).expect("serialize"); + let back: FlowRun = serde_json::from_str(&json).expect("deserialize"); + assert_eq!(back.id, run.id); + assert_eq!(back.steps.len(), 1); + assert_eq!(back.steps[0].node_id, "t"); + assert!(back.steps[0].port.is_none()); + } + + #[test] + fn flow_run_step_omits_port_when_none() { + let step = FlowRunStep { + node_id: "n".to_string(), + output: serde_json::Value::Null, + port: None, + }; + let v = serde_json::to_value(&step).unwrap(); + assert!(v.get("port").is_none()); } } diff --git a/src/openhuman/tinyflows/caps.rs b/src/openhuman/tinyflows/caps.rs index 2655df474..4781802da 100644 --- a/src/openhuman/tinyflows/caps.rs +++ b/src/openhuman/tinyflows/caps.rs @@ -141,14 +141,130 @@ impl LlmProvider for OpenHumanLlm { } } +/// Parses a `"composio::"` `connection_ref` (see the +/// node catalog, `my_docs/ohxtf/commons/12-node-catalog-0.2.md`) and returns +/// the trailing connection id segment. Values that don't match this shape +/// return `None` — the caller logs and falls back to the ambient session +/// account (only Direct mode can actually forward the id today; see +/// [`OpenHumanTools::invoke`]'s doc for the Backend-mode gap this leaves +/// open). +pub(crate) fn composio_connection_id(conn: &str) -> Option<&str> { + let rest = conn.strip_prefix("composio:")?; + let id = rest.rsplit(':').next()?; + (!id.is_empty()).then_some(id) +} + +/// Parses a `"http_cred:"` `connection_ref` for [`OpenHumanHttp`]. No +/// host-side HTTP credential store exists yet — this only extracts the name +/// so the adapter can log a clear, actionable warning instead of silently +/// ignoring the reference. See [`OpenHumanHttp::request`]'s doc. +pub(crate) fn http_cred_name(conn: &str) -> Option<&str> { + let name = conn.strip_prefix("http_cred:")?.trim(); + (!name.is_empty()).then_some(name) +} + +/// Strict, deny-by-default curation check for flow `tool_call` nodes (issue +/// B2 finding #2). +/// +/// This is intentionally **stricter** than +/// `memory_sync::composio::providers::is_action_visible_with_pref` — the +/// helper the normal agent tool-call loop uses. That helper is permissive by +/// design for a toolkit it doesn't recognize: it falls back to the +/// `classify_unknown` heuristic and lets the slug through (scope-gated), and +/// treats a prefix-less slug as unconditionally visible. That's safe in the +/// agent loop because the model only ever sees slugs the *backend itself* +/// returned from live tool discovery (`composio_list_tools`) — there is no +/// path for the model to invent a slug that reaches this check. A flow's +/// `tool_call.slug`, by contrast, is a free-form string the flow *author* +/// typed when building the graph; it never round-trips through Composio +/// discovery before `invoke` is called. So here a slug is allowed **only** +/// if it resolves to a real, known toolkit AND is present in that toolkit's +/// curated catalog: +/// - `toolkit_from_slug` fails to extract anything (empty/blank slug) → reject. +/// - the extracted toolkit has no registered provider curated list AND no +/// static `catalog_for_toolkit` entry (i.e. it isn't one of OpenHuman's +/// known/curated toolkits at all — including a made-up prefix like +/// `madeupkit`, or a prefix-less slug like `noop` which `toolkit_from_slug` +/// degrades to treating as its own single-segment "toolkit") → reject. +/// - the toolkit has a catalog but `slug` isn't one of its entries → reject. +/// - otherwise, apply the same per-user read/write/admin scope preference +/// the agent loop uses (`UserScopePref::allows`). +/// +/// // TODO(0.3): this hard-rejects any *real* Composio toolkit that simply +/// // isn't in the static `catalog_for_toolkit` map yet (there is no +/// // host-side, offline way to ask "is this actually a valid Composio +/// // toolkit/action" beyond the curated catalogs OpenHuman ships). That's +/// // an accepted trade-off for a genuine allowlist rather than a residual +/// // gap to silently work around — extending `catalog_for_toolkit` (or, if +/// // a live catalog lookup becomes available, consulting it here) is how a +/// // newly-supported toolkit gets flow tool-call support. +async fn is_curated_flow_tool(slug: &str) -> bool { + use crate::openhuman::memory_sync::composio::providers::{ + catalog_for_toolkit, find_curated, get_provider, load_user_scope_or_default, + toolkit_from_slug, + }; + + let Some(toolkit) = toolkit_from_slug(slug) else { + return false; + }; + let catalog = get_provider(&toolkit) + .and_then(|p| p.curated_tools()) + .or_else(|| catalog_for_toolkit(&toolkit)); + let Some(catalog) = catalog else { + return false; + }; + let Some(curated) = find_curated(catalog, slug) else { + return false; + }; + let pref = load_user_scope_or_default(&toolkit).await; + pref.allows(curated.scope) +} + /// [`ToolInvoker`] adapter over Composio (`src/openhuman/composio/client.rs`). /// -/// **B1 deviation (tracked, see `my_docs/ohxtf/commons/11-gotchas-and-decisions.md`):** -/// `connection_ref` is logged but not forwarded — `execute_tool` (backend mode) -/// takes no connection id and resolves the ambient signed-in account; direct -/// mode uses `config.composio.entity_id`. Fine for a single-account desktop -/// user; must be resolved before multi-account or B2 trigger runs. There is -/// also no curated-tool-set / scope filter yet — `invoke` will call any slug. +/// **B2 (closes two B1 deviations, see +/// `my_docs/ohxtf/b2-triggers-trust/01-triggers-and-trust.md` §4-5):** +/// - **Curation + scope (hard allowlist)**: every call is checked against +/// [`is_curated_flow_tool`] — a deny-by-default gate that only allows a +/// slug resolving to a *known, curated* toolkit action, unlike the general +/// agent tool-call path's more permissive +/// `memory_sync::composio::providers::is_action_visible_with_pref` (see +/// [`is_curated_flow_tool`]'s doc for why the two differ). A non-curated / +/// unrecognized / out-of-scope slug is rejected with +/// `EngineError::Capability("tool not permitted: ")` before any +/// Composio call. **As of tinyflows 0.3 this is load-bearing, not merely +/// defense-in-depth**: integration-node config (including `slug`) is now +/// `=`-expression evaluated against upstream/trigger data before `invoke`, +/// so a trigger payload *can* influence which tool a `=`-derived slug +/// resolves to. The curation gate runs on the **resolved** slug (verified: +/// a `=item.tool`-derived unknown slug is rejected here before Composio), +/// constraining any data-derived tool to the user's curated, in-scope, +/// connected set — and it still closes the case where an author hand-types +/// an arbitrary/typo'd slug. +/// - **connection_ref**: `conn` (`"composio::"`) is +/// now parsed and forwarded to `direct_execute` (Composio Direct mode). +/// Backend mode's `execute_tool` still has no per-call account-scoping +/// path — that's a backend API gap, not something this seam can close +/// alone — so a `connection_ref` under Backend mode logs a warning and +/// falls back to the ambient signed-in account (documented stub; see +/// `composio_connection_id`). +/// - **Trust gate**: invocation is also routed through the OpenHuman +/// `ApprovalGate` (mirrors `tinyagents/middleware.rs::ApprovalSecurityMiddleware`) +/// before dispatch, closing the Codex P1 finding that flow tool nodes +/// bypassed the Network/tool approval gate entirely. `ops::flows_run` / +/// `flows_resume` scope a `TrustedAutomation { Workflow }` origin around +/// the whole run, so the gate either auto-allows (pre-declared trust root) +/// or — when the flow's `require_approval` is set — parks for a real +/// decision. No gate installed (unit tests, some hosts) means no gating, +/// same as the existing agent tool-loop middleware. +/// +/// // SECURITY NOTE (tinyflows 0.3, now the pinned version): integration nodes +/// // `=`-resolve config from upstream/trigger data, so a trigger-driven flow +/// // whose `slug`/`url` is `=`-derived lets untrusted trigger data pick *which* +/// // curated + in-scope + connected tool/endpoint runs (blast radius bounded by +/// // the curation + scope + connection checks above and the approval gate). +/// // For such flows authors should set `require_approval`. FOLLOW-UP: auto-force +/// // approval when a trigger-driven run's tool/http config contains `=`-exprs. pub struct OpenHumanTools { pub config: Arc, } @@ -156,41 +272,110 @@ pub struct OpenHumanTools { #[async_trait] impl ToolInvoker for OpenHumanTools { async fn invoke(&self, slug: &str, args: Value, conn: Option<&str>) -> Result { - if let Some(c) = conn { - tracing::debug!( + // Curation + scope gate — hard allowlist (see [`is_curated_flow_tool`]'s + // doc for why this differs from the general agent tool-call path). + // Runs before anything else — a rejected slug never reaches the + // composio client at all. + if !is_curated_flow_tool(slug).await { + tracing::warn!( target: "flows", %slug, - conn = %c, - "[flows] tool conn (backend resolves ambient account; not forwarded in B1)" + "[flows] tool_call: rejected — not a recognized curated toolkit action, or out \ + of the user's configured scope" ); + return Err(EngineError::Capability(format!( + "tool not permitted: {slug}" + ))); + } + + // Approval gate (see the struct doc). Mirrors + // `tinyagents/middleware.rs::ApprovalSecurityMiddleware::wrap_tool`'s + // shape exactly: compute summary/redacted args only when a gate is + // installed, deny short-circuits before any composio call, allow + // records an audit id to close out after the call resolves. + let mut audit_id: Option = None; + if let Some(gate) = crate::openhuman::approval::ApprovalGate::try_global() { + let summary = crate::openhuman::approval::summarize_action(slug, &args); + let redacted = crate::openhuman::approval::redact_args(&args); + let (outcome, request_id) = gate.intercept_audited(slug, &summary, redacted).await; + match outcome { + crate::openhuman::approval::GateOutcome::Deny { reason } => { + return Err(EngineError::Capability(reason)); + } + crate::openhuman::approval::GateOutcome::Allow => audit_id = request_id, + } } let kind = create_composio_client(&self.config) .map_err(|e| EngineError::Capability(e.to_string()))?; let args_opt = if args.is_null() { None } else { Some(args) }; + let connection_id = conn.and_then(composio_connection_id); - tracing::debug!(target: "flows", %slug, mode = kind.mode(), "[flows] tool_call: invoking composio tool"); + tracing::debug!( + target: "flows", + %slug, + mode = kind.mode(), + has_connection_ref = connection_id.is_some(), + "[flows] tool_call: invoking composio tool" + ); let response = match kind { - ComposioClientKind::Backend(client) => client - .execute_tool(slug, args_opt) - .await - .map_err(|e| EngineError::Capability(e.to_string()))?, - ComposioClientKind::Direct(tool) => { - direct_execute(&tool, slug, args_opt, &self.config.composio.entity_id, None) + ComposioClientKind::Backend(client) => { + if connection_id.is_some() { + tracing::warn!( + target: "flows", + %slug, + "[flows] tool_call: connection_ref set but backend mode has no per-call \ + account-scoping path yet — using the ambient session account \ + (documented stub, see caps.rs's OpenHumanTools doc)" + ); + } + client + .execute_tool(slug, args_opt) .await - .map_err(|e| EngineError::Capability(e.to_string()))? + .map_err(|e| EngineError::Capability(e.to_string())) } + ComposioClientKind::Direct(tool) => direct_execute( + &tool, + slug, + args_opt, + &self.config.composio.entity_id, + connection_id, + ) + .await + .map_err(|e| EngineError::Capability(e.to_string())), }; - serde_json::to_value(response).map_err(|e| EngineError::Capability(e.to_string())) + if let Some(id) = audit_id { + if let Some(gate) = crate::openhuman::approval::ApprovalGate::try_global() { + let exec = if response.is_ok() { + crate::openhuman::approval::ExecutionOutcome::Success + } else { + crate::openhuman::approval::ExecutionOutcome::Failure + }; + gate.record_execution( + &id, + exec, + response.as_ref().err().map(ToString::to_string).as_deref(), + ); + } + } + + serde_json::to_value(response?).map_err(|e| EngineError::Capability(e.to_string())) } } /// [`HttpClient`] adapter over `HttpRequestTool` /// (`src/openhuman/tools/impl/network/http_request.rs`). Allowlist + DNS-rebind -/// guard + Network gating live inside `execute`, so this adapter gets them for -/// free. +/// guard live inside `execute`, so this adapter gets them for free. +/// +/// **B2:** also routes through the OpenHuman `ApprovalGate` before dispatch +/// (same rationale/shape as [`OpenHumanTools::invoke`] — closes the Codex P1 +/// finding that flow HTTP nodes bypassed the Network approval gate). A +/// `"http_cred:"` `connection_ref` is parsed but there is no HTTP +/// credential store to resolve it against yet (documented stub, see +/// `http_cred_name`) — the request proceeds without injecting stored +/// credentials. pub struct OpenHumanHttp { pub security: Arc, pub http_config: HttpRequestConfig, @@ -199,8 +384,31 @@ pub struct OpenHumanHttp { #[async_trait] impl HttpClient for OpenHumanHttp { async fn request(&self, request: Value, conn: Option<&str>) -> Result { - if let Some(c) = conn { - tracing::debug!(target: "flows", conn = %c, "[flows] http conn (not resolved in B1)"); + const TOOL_NAME: &str = "flows_http_request"; + + let mut audit_id: Option = None; + if let Some(gate) = crate::openhuman::approval::ApprovalGate::try_global() { + let summary = crate::openhuman::approval::summarize_action(TOOL_NAME, &request); + let redacted = crate::openhuman::approval::redact_args(&request); + let (outcome, request_id) = gate.intercept_audited(TOOL_NAME, &summary, redacted).await; + match outcome { + crate::openhuman::approval::GateOutcome::Deny { reason } => { + return Err(EngineError::Capability(reason)); + } + crate::openhuman::approval::GateOutcome::Allow => audit_id = request_id, + } + } + + if let Some(name) = conn.and_then(http_cred_name) { + tracing::warn!( + target: "flows", + cred = %name, + "[flows] http_request: connection_ref names an http_cred secret, but no HTTP \ + credential store exists yet — proceeding WITHOUT injecting stored credentials \ + (documented stub, see caps.rs's OpenHumanHttp doc)" + ); + } else if let Some(c) = conn { + tracing::debug!(target: "flows", conn = %c, "[flows] http conn: unrecognized connection_ref prefix (expected `http_cred:`) — ignoring"); } let tool = HttpRequestTool::new( @@ -221,20 +429,36 @@ impl HttpClient for OpenHumanHttp { // config is the request descriptor; `HttpRequestTool::execute` reads // only those keys and ignores the rest (e.g. `connection_ref`, // `on_error`), so passing the whole config through is safe. - let result = tool - .execute(request) - .await - .map_err(|e| EngineError::Capability(e.to_string()))?; + let result = tool.execute(request).await; - // `HttpRequestTool::execute` always returns `Ok`, using `is_error` to - // signal a failed request (non-2xx, DNS/allowlist rejection, timeout, - // …) — surface that as a capability error so the engine's - // `on_error`/`retry` policy can act on it. - if result.is_error { - return Err(EngineError::Capability(result.text())); + let outcome: Result = match result { + Ok(result) if result.is_error => { + // `HttpRequestTool::execute` always returns `Ok`, using + // `is_error` to signal a failed request (non-2xx, DNS/allowlist + // rejection, timeout, …) — surface that as a capability error + // so the engine's `on_error`/`retry` policy can act on it. + Err(EngineError::Capability(result.text())) + } + Ok(result) => Ok(json!({ "text": result.text() })), + Err(e) => Err(EngineError::Capability(e.to_string())), + }; + + if let Some(id) = audit_id { + if let Some(gate) = crate::openhuman::approval::ApprovalGate::try_global() { + let exec = if outcome.is_ok() { + crate::openhuman::approval::ExecutionOutcome::Success + } else { + crate::openhuman::approval::ExecutionOutcome::Failure + }; + gate.record_execution( + &id, + exec, + outcome.as_ref().err().map(ToString::to_string).as_deref(), + ); + } } - Ok(json!({ "text": result.text() })) + outcome } } diff --git a/src/openhuman/tinyflows/tests.rs b/src/openhuman/tinyflows/tests.rs index 4c4d2a5b2..c57f47061 100644 --- a/src/openhuman/tinyflows/tests.rs +++ b/src/openhuman/tinyflows/tests.rs @@ -23,14 +23,14 @@ use std::sync::Arc; use serde_json::json; use tempfile::TempDir; -use tinyflows::caps::{CodeLanguage, CodeRunner, HttpClient, StateStore}; +use tinyflows::caps::{CodeLanguage, CodeRunner, HttpClient, StateStore, ToolInvoker}; use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; use crate::openhuman::config::Config; use crate::openhuman::security::SecurityPolicy; use super::build_capabilities; -use super::caps::{FlowStateStore, OpenHumanCode, OpenHumanHttp}; +use super::caps::{FlowStateStore, OpenHumanCode, OpenHumanHttp, OpenHumanTools}; fn test_config(tmp: &TempDir) -> Arc { let config = Config { @@ -212,3 +212,137 @@ async fn code_adapter_javascript_passthrough_round_trips_json() { .expect("javascript passthrough should succeed when node is present"); assert_eq!(result, input); } + +// ── Tool curation / scope + connection_ref (issue B2) ───────────────────── +// +// No `ApprovalGate` is installed in this test binary (see the module doc on +// `flows::bus`'s tests and the trust-model tests in `approval::gate` for the +// gate-level behavior) — these tests exercise the *curation* gate, which is +// independent of the approval gate and runs first, so they stay deterministic +// without any global state. + +fn tools_adapter(config: Arc) -> OpenHumanTools { + OpenHumanTools { config } +} + +#[tokio::test] +async fn tools_invoke_rejects_a_non_curated_slug_for_a_known_toolkit() { + let tmp = TempDir::new().unwrap(); + let tools = tools_adapter(test_config(&tmp)); + + // "gmail" has a curated catalog; this action is not in it, so curation + // must reject regardless of the user's read/write/admin scope prefs. + let err = tools + .invoke("GMAIL_NOT_A_REAL_CURATED_ACTION", json!({}), None) + .await + .expect_err("a non-curated action for a curated toolkit must be rejected"); + let msg = err.to_string(); + assert!( + msg.contains("tool not permitted"), + "expected a curation rejection message, got: {msg}" + ); + assert!(msg.contains("GMAIL_NOT_A_REAL_CURATED_ACTION")); +} + +#[tokio::test] +async fn tools_invoke_rejects_an_unrecognized_toolkit_slug() { + // Issue B2 finding #2 (deny-by-default): a made-up toolkit prefix that + // isn't in any curated catalog must be rejected — not passed through on + // a permissive "unknown toolkit" heuristic. Live testing confirmed this + // used to reach Composio (and only failed there for lack of a signed-in + // session), which is not a hard allowlist. + let tmp = TempDir::new().unwrap(); + let tools = tools_adapter(test_config(&tmp)); + + let err = tools + .invoke("madeupkit_dostuff", json!({}), None) + .await + .expect_err("an unrecognized toolkit slug must be rejected by curation"); + let msg = err.to_string(); + assert!( + msg.contains("tool not permitted"), + "expected a curation rejection message, got: {msg}" + ); + assert!(msg.contains("madeupkit_dostuff")); +} + +#[tokio::test] +async fn tools_invoke_rejects_a_prefix_less_slug() { + // "noop" has no curated catalog (`catalog_for_toolkit` returns `None` + // for the single-segment "toolkit" `toolkit_from_slug` degrades it to), + // so the hard allowlist in `is_curated_flow_tool` rejects it outright — + // unlike the general agent tool-call path's `is_action_visible_with_pref`, + // which falls back to the permissive `classify_unknown` heuristic and + // would let this slug through. + let tmp = TempDir::new().unwrap(); + let tools = tools_adapter(test_config(&tmp)); + + let err = tools + .invoke("noop", json!({}), None) + .await + .expect_err("a prefix-less/unrecognized slug must be rejected by curation"); + assert!( + err.to_string().contains("tool not permitted"), + "expected a curation rejection message, got: {err}" + ); +} + +#[tokio::test] +async fn tools_invoke_does_not_reject_a_known_curated_slug_at_the_curation_gate() { + // A real curated action for a known toolkit must clear the curation + // gate — it may still fail further downstream (no composio client + // configured in this test environment), but that failure must NOT be + // the "tool not permitted" curation-rejection message. + let tmp = TempDir::new().unwrap(); + let tools = tools_adapter(test_config(&tmp)); + + let err = tools + .invoke("GMAIL_SEND_EMAIL", json!({}), None) + .await + .expect_err("no composio client is configured in the test environment"); + assert!( + !err.to_string().contains("tool not permitted"), + "a known curated slug must not be rejected by curation, got: {err}" + ); +} + +#[test] +fn composio_connection_id_parses_toolkit_prefixed_ref() { + assert_eq!( + super::caps::composio_connection_id("composio:slack:acct_123"), + Some("acct_123") + ); + // Trailing segment only — works even without a toolkit segment present. + assert_eq!( + super::caps::composio_connection_id("composio::acct_1"), + Some("acct_1") + ); +} + +#[test] +fn composio_connection_id_returns_none_for_non_composio_ref_or_empty_id() { + assert_eq!( + super::caps::composio_connection_id("http_cred:my-secret"), + None + ); + assert_eq!(super::caps::composio_connection_id("composio:"), None); + assert_eq!(super::caps::composio_connection_id("composio:slack:"), None); +} + +#[test] +fn http_cred_name_parses_and_trims() { + assert_eq!( + super::caps::http_cred_name("http_cred:my-secret"), + Some("my-secret") + ); + assert_eq!( + super::caps::http_cred_name("http_cred: spaced "), + Some("spaced") + ); +} + +#[test] +fn http_cred_name_returns_none_for_non_http_cred_ref_or_empty_name() { + assert_eq!(super::caps::http_cred_name("composio:slack:acct_1"), None); + assert_eq!(super::caps::http_cred_name("http_cred:"), None); +}