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

This commit is contained in:
Cyrus Gray
2026-07-03 23:45:50 +05:30
committed by GitHub
parent d735c43a4a
commit 68e59959fb
25 changed files with 3088 additions and 174 deletions
Generated
+2 -2
View File
@@ -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",
+1 -1
View File
@@ -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
+2 -2
View File
@@ -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",
+14 -1
View File
@@ -426,6 +426,17 @@ pub enum DomainEvent {
/// Optional job name for display/threading purposes.
job_name: Option<String>,
},
/// A `flow`-type cron job fired its schedule tick (issue B2,
/// `my_docs/ohxtf/b2-triggers-trust/01-triggers-and-trust.md` §1).
/// Published by `cron::scheduler` when a `JobType::Flow` job comes due;
/// carries only the flow id (the trigger payload for a `schedule` flow is
/// always empty — `flows::ops::flows_run` seeds `{}`). Consumed by
/// `flows::bus::FlowTriggerSubscriber`, which loads the flow, checks it is
/// still enabled with a `schedule` trigger, and spawns a run.
FlowScheduleTick {
/// Identifier of the `flows::Flow` to run.
flow_id: String,
},
// ── Skills ──────────────────────────────────────────────────────────
/// A skill was loaded into the runtime.
@@ -1281,7 +1292,8 @@ impl DomainEvent {
Self::CronJobTriggered { .. }
| Self::CronJobCompleted { .. }
| Self::CronDeliveryRequested { .. }
| Self::ProactiveMessageRequested { .. } => "cron",
| Self::ProactiveMessageRequested { .. }
| Self::FlowScheduleTick { .. } => "cron",
Self::WorkflowLoaded { .. }
| Self::WorkflowStopped { .. }
@@ -1439,6 +1451,7 @@ impl DomainEvent {
Self::CronJobCompleted { .. } => "CronJobCompleted",
Self::CronDeliveryRequested { .. } => "CronDeliveryRequested",
Self::ProactiveMessageRequested { .. } => "ProactiveMessageRequested",
Self::FlowScheduleTick { .. } => "FlowScheduleTick",
Self::WorkflowLoaded { .. } => "WorkflowLoaded",
Self::WorkflowStopped { .. } => "WorkflowStopped",
Self::WorkflowStartFailed { .. } => "WorkflowStartFailed",
+6
View File
@@ -203,6 +203,12 @@ fn all_variants_have_correct_domain() {
},
"cron",
),
(
DomainEvent::FlowScheduleTick {
flow_id: "flow-1".into(),
},
"cron",
),
// Workflow
(
DomainEvent::WorkflowLoaded {
+25
View File
@@ -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(
+19
View File
@@ -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! {
+118 -3
View File
@@ -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
+5 -5
View File
@@ -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).
+4 -3
View File
@@ -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,
+27
View File
@@ -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";
+103 -1
View File
@@ -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<CronJob> {
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<Option<CronJob>> {
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<Vec<CronJob>> {
with_connection(config, |conn| {
let mut stmt = conn.prepare(
@@ -712,7 +803,18 @@ fn with_connection<T>(config: &Config, f: impl FnOnce(&Connection) -> Result<T>)
);
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")?;
+47
View File
@@ -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();
+11
View File
@@ -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 {
+13
View File
@@ -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);
+401 -31
View File
@@ -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<TriggerKind> {
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<Config>,
/// 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<Mutex<HashSet<String>>>,
}
impl FlowTriggerSubscriber {
pub fn new() -> Self {
Self
pub fn new(config: Arc<Config>) -> 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<InFlightGuard> {
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<Mutex<HashSet<String>>>,
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<Config> {
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());
}
}
+1 -1
View File
@@ -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};
+570 -31
View File
@@ -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<WorkflowGraph, String
}
/// Creates a new flow from a name and a raw graph JSON value.
///
/// `store::create_flow` defaults new flows to `enabled = true` — this binds
/// the flow's automatic-dispatch side effect (e.g. registers the
/// schedule-trigger cron job) immediately, reusing the same [`bind_trigger`]
/// helper `flows_set_enabled` uses. Without this, a freshly-created enabled
/// schedule flow would silently never fire until an app restart (boot
/// reconcile) or a manual disable→enable toggle. Best-effort, same as
/// `flows_set_enabled`: a binding failure is logged, not fatal to create.
pub async fn flows_create(
config: &Config,
name: String,
graph_json: Value,
require_approval: bool,
) -> Result<RpcOutcome<Flow>, 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<RpcOutcome<Vec<Flow>>, 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<String>,
graph_json: Option<Value>,
require_approval: Option<bool>,
) -> Result<RpcOutcome<Flow>, 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<RpcOutcome<Value>, 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<RpcOutcome<Value>
))
}
/// 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<RpcOutcome<Flow>, 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<String>,
) -> Result<RpcOutcome<Value>, 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<RpcOutcome<Vec<FlowRun>>, 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<RpcOutcome<FlowRun>, 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<FlowRunStep> {
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;
+536 -9
View File
@@ -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<String> =
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"
);
}
+218 -27
View File
@@ -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<FieldSchema> {
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<ControllerSchema> {
vec![
schemas("create"),
@@ -41,6 +74,9 @@ pub fn all_controller_schemas() -> Vec<ControllerSchema> {
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<RegisteredController> {
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<String, Value>) -> ControllerFuture {
let config = config_rpc::load_config_with_timeout().await?;
let name = read_required::<String>(&params, "name")?;
let graph = read_required::<Value>(&params, "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<String, Value>) -> 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<String, Value>) -> ControllerFuture {
})
}
fn handle_resume(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let config = config_rpc::load_config_with_timeout().await?;
let id = read_required::<String>(&params, "id")?;
let thread_id = read_required::<String>(&params, "thread_id")?;
let approvals: Vec<String> = 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<String, Value>) -> ControllerFuture {
Box::pin(async move {
let config = config_rpc::load_config_with_timeout().await?;
let id = read_required::<String>(&params, "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<String, Value>) -> ControllerFuture {
Box::pin(async move {
let config = config_rpc::load_config_with_timeout().await?;
let run_id = read_required::<String>(&params, "run_id")?;
to_json(ops::flows_get_run(&config, run_id.trim()).await?)
})
}
fn read_required<T: DeserializeOwned>(params: &Map<String, Value>, key: &str) -> Result<T, String> {
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");
+221 -16
View File
@@ -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<T>(config: &Config, f: impl FnOnce(&Connection) -> Result<T>)
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<Flow> {
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<Option<Flow>> {
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<Option<Flow>> {
/// Lists all saved flows, migrating each graph on read (see [`get_flow`]).
pub fn list_flows(config: &Config) -> Result<Vec<Flow>> {
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<Vec<Flow>> {
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<Flow> {
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<Flow> {
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<Flow> {
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<Option<FlowRun>> {
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<Vec<FlowRun>> {
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<FlowRun> {
let steps_raw: String = row.get(6)?;
let steps: Vec<FlowRunStep> = serde_json::from_str(&steps_raw).map_err(sql_conversion_error)?;
let pending_raw: String = row.get(7)?;
let pending_approvals: Vec<String> =
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;
+233 -5
View File
@@ -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);
}
+117
View File
@@ -28,6 +28,68 @@ pub struct Flow {
pub last_run_at: Option<String>,
/// Outcome of the most recent run: `"completed"` | `"pending_approval"` | `"failed"`.
pub last_status: Option<String>,
/// "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<String>,
}
/// A persisted record of one `flows_run` / `flows_resume` invocation, for the
/// B3 run-history inspector. Written by `flows::store` from `flows::ops`.
#[derive(Debug, Clone, Serialize, Deserialize)]
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<String>,
/// Reconstructed per-node steps (see [`FlowRunStep`]).
#[serde(default)]
pub steps: Vec<FlowRunStep>,
/// Node ids paused awaiting human approval when `status ==
/// "pending_approval"`; empty otherwise.
#[serde(default)]
pub pending_approvals: Vec<String>,
/// Error message when `status == "failed"`.
#[serde(default)]
pub error: Option<String>,
}
#[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());
}
}
+258 -34
View File
@@ -141,14 +141,130 @@ impl LlmProvider for OpenHumanLlm {
}
}
/// Parses a `"composio:<toolkit>:<connection_id>"` `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:<name>"` `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: <slug>")` 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:<toolkit>:<connection_id>"`) 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<Config>,
}
@@ -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<Value> {
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<String> = 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:<name>"` `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<SecurityPolicy>,
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<Value> {
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<String> = 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:<name>`) — 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<Value> = 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
}
}
+136 -2
View File
@@ -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<Config> {
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<Config>) -> 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);
}