feat(flows): Workflows B2 — triggers, resume, trust & run history#4447
Conversation
… reconcile
- event_bus: add DomainEvent::FlowScheduleTick { flow_id } (domain "cron",
wired into domain()/variant_name() + a table-driven test).
- cron: minimal additive JobType::Flow — a flow-type cron job stores the flow
id in its command column and its scheduler branch emits FlowScheduleTick when
it fires (no cron schema change). add_flow_schedule_job/find_flow_schedule_job
for idempotent (un)binding.
- jsonrpc: boot hook calls flows::ops::reconcile_schedule_triggers_on_boot next
to the existing cron seed hook, re-registering cron jobs for enabled schedule
flows.
- app/src-tauri/Cargo.lock: sync tinyagents/tinyflows transitive deps that
landed in the root lockfile in B1 but were never synced into the Tauri shell
lockfile (pre-existing B1 debt).
- bus: FlowTriggerSubscriber (now holding Arc<Config>) dispatches real runs — matches FlowScheduleTick (schedule), ComposioTriggerReceived (toolkit+ trigger_slug), WebhookIncomingRequest (documented deferral); spawns flows_run per match. extract_trigger_kind/config helpers. - ops: flows_set_enabled binds/unbinds cron schedule jobs; boot reconciliation; flows_resume (resume_with_checkpointer); pending-approval CoreNotification (Agents category, approve action); run-history rows written from run/resume; require_approval threaded through create/update. - store: flow_runs table + FlowRun CRUD + list_enabled_flows; require_approval column (add_column_if_missing). - schemas: flows_resume / flows_list_runs / flows_get_run controllers (+ tests). - startup: pass Arc<Config> to FlowTriggerSubscriber. - Run-history steps reconstructed from outcome.output (lean; richer observer is a tinyflows 0.3 follow-up).
- turn_origin: add TrustedAutomationSource::Workflow { require_approval }.
- approval/gate: Workflow arm mirrors Cron (require_approval=false → Allow, no
chat routing) / GoalContinuation (require_approval=true → park-and-audit,
fail-closed with no chat surface → TTL-deny unless resolved); added to the
auto_approve allowlist-bypass so a global grant can't override an explicit
per-flow require_approval. No non-flow path changed.
- caps: OpenHumanTools/OpenHumanHttp now route through ApprovalGate
(mirroring ApprovalSecurityMiddleware) — closes the deferred B1 Codex P1
(flow HTTP/tool nodes bypassing the Network approval gate). Curation:
invoke() gates on is_action_visible_with_pref + per-user scope before
Composio; rejects non-curated slug. connection_ref: composio:<tk>:<id>
forwarded to direct_execute (Direct mode); backend-mode account scoping +
http_cred:<name> store are documented TODO(0.3) stubs with warnings.
- flows_run/flows_resume scope the Workflow origin around every run via
with_origin (trust is about the saved flow, not the caller).
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughThis PR adds workflow-triggered flow execution, per-flow approval requirements, durable run history, RPC support for run operations, and deny-by-default gating for tool and HTTP execution. ChangesFlow scheduling, approval, and execution
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Scheduler
participant EventBus
participant FlowTriggerSubscriber
participant flows_ops as "flows::ops"
participant ApprovalGate
participant Store as "flow_runs store"
Scheduler->>EventBus: publish FlowScheduleTick(flow_id)
EventBus->>FlowTriggerSubscriber: handle(FlowScheduleTick)
FlowTriggerSubscriber->>flows_ops: flows_run(flow_id)
flows_ops->>Store: insert_flow_run
flows_ops->>ApprovalGate: intercept_audited(TrustedAutomation::Workflow)
alt require_approval false
ApprovalGate-->>flows_ops: Allow
flows_ops->>Store: finish_flow_run
else require_approval true
ApprovalGate-->>flows_ops: pending approval
flows_ops->>Store: finish_flow_run
flows_ops->>flows_ops: notify_pending_approval
end
sequenceDiagram
participant Caller
participant flows_ops as "flows::ops::flows_resume"
participant Store as "flow_runs store"
participant Engine as "tinyflows::engine"
Caller->>flows_ops: flows_resume(flow_id, thread_id, approvals)
flows_ops->>Store: get_flow_run(thread_id)
Store-->>flows_ops: FlowRun
flows_ops->>flows_ops: verify approvals
flows_ops->>Engine: resume_with_checkpointer
Engine-->>flows_ops: result
flows_ops->>Store: finish_flow_run
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: fddb14bdf4
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| let _flows_trigger_handle = bus.subscribe(Arc::new( | ||
| crate::openhuman::flows::bus::FlowTriggerSubscriber::new(), | ||
| crate::openhuman::flows::bus::FlowTriggerSubscriber::new(Arc::new(config.clone())), | ||
| )); |
There was a problem hiding this comment.
When the desktop has no realtime channel integration configured, or OPENHUMAN_DISABLE_CHANNEL_LISTENERS is set, run_server_inner returns before calling start_channels, so this is the only FlowTriggerSubscriber registration and it never happens. Cron still publishes FlowScheduleTick, but with no subscriber the schedule/app-event workflows are never dispatched in that default no-channel context; register this subscriber during core/event-bus startup instead of under channel startup.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in 67d1631 — good catch. Moved the FlowTriggerSubscriber registration out of start_channels into jsonrpc.rs::register_domain_subscribers (unconditional core boot, Once-guarded, subscribe_global + mem::forget — the same pattern the webhook/channel subscribers use), and dropped it from start_channels. So schedule/app-event workflows now dispatch even with no realtime channel / OPENHUMAN_DISABLE_CHANNEL_LISTENERS, and there's no double subscriber (which would have double-dispatched every trigger).
| let flow = | ||
| store::create_flow(config, name, graph, require_approval).map_err(|e| e.to_string())?; |
There was a problem hiding this comment.
New flows are persisted as enabled, but this create path returns without calling the new trigger-binding code. Because boot reconciliation has already run in a live process, creating a schedule-trigger workflow via flows_create leaves it without a backing cron job, so it will not emit FlowScheduleTick until the app restarts or the user toggles enabled off/on.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Already addressed in 36a0446 (pushed after the commit you reviewed) — flows_create now calls bind_trigger when the new flow is enabled, so a schedule flow gets its cron job immediately (verified live: 0→1 flow cron jobs right after create). flows_delete also unbinds now.
| let updated = store::update_flow_graph(config, id, new_name, graph, new_require_approval) | ||
| .map_err(|e| e.to_string())?; |
There was a problem hiding this comment.
When an enabled flow's trigger graph changes here, the backing cron job created by bind_schedule_trigger is not reconciled. Editing a schedule from one cadence to another keeps the old cron schedule until restart/toggle, and switching away from schedule leaves a stale cron job firing ticks that are ignored; update should unbind/rebind based on the old and new trigger for enabled flows.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Already addressed in 36a0446 — flows_update now diffs the old vs new trigger (extract_trigger_kind/extract_trigger_config) for an enabled flow and unbinds+rebinds when the schedule changes (and unbinds when switching away from schedule), so an edited cadence takes effect without a restart/toggle and no stale cron job is left firing ignored ticks.
…resume guard Found by live end-to-end testing of the B2 surface: 1. Enabled schedule flows weren't scheduled until a set_enabled toggle/restart: flows_create now binds the trigger when the new flow is enabled; flows_delete unbinds (the cron job lives in a separate cron.db — no FK cascade); flows_update re-binds when an enabled flow's trigger schedule changes. 2. Tool curation was permissive for unknown toolkits (is_action_visible_with_pref is lenient by design for the agent loop, unsafe for author-written flow slugs): new is_curated_flow_tool() is a hard allowlist — toolkit must resolve to a real catalog AND the slug must be in it AND pass user scope; else 'tool not permitted'. madeupkit_dostuff / prefix-less noop now rejected; GMAIL_SEND_EMAIL still passes. 3. flows_resume ignored the approvals list (tinyflows 0.2 treats resume itself as approval): host-side guard now requires a recorded pending_approval run whose pending gates intersect the caller's approvals; empty/mismatched → rejected; resuming a non-paused run → clear error. flows tests 67 pass, tinyflows 14 pass (1 pre-existing #[ignore]); cargo check + fmt clean.
Follow-up: live end-to-end testing found + fixed 3 gaps (
|
coderabbitai
Bot
added
feature
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/openhuman/cron/store.rs`:
- Around line 142-204: The flow schedule binding path uses a race-prone
check-then-insert flow between find_flow_schedule_job and add_flow_schedule_job,
so concurrent calls can create duplicate cron_jobs for the same flow. Add a
database uniqueness constraint for the flow job identity (job_type plus command,
or an equivalent partial unique index for job_type = 'flow') and update
add_flow_schedule_job to handle the resulting insert conflict gracefully. Keep
find_flow_schedule_job as the lookup used by flows::ops::bind_schedule_trigger,
but make the storage layer enforce one cron job per flow.
In `@src/openhuman/flows/bus.rs`:
- Around line 69-136: `handle_schedule_tick`, `handle_app_event`, and
`spawn_run` currently dispatch `flows::ops::flows_run` without any in-flight
guard, so the same `flow_id` can run concurrently and race on summary state. Add
a per-flow active-run check or dedupe/queue mechanism before calling `spawn_run`
(and/or inside `flows_run`) so only one run per flow can execute at a time,
using the existing `flow_id` and `flows_run` symbols to anchor the change.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 256e8f95-3984-4881-95a1-5773ac42bb54
⛔ Files ignored due to path filters (1)
app/src-tauri/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (21)
src/core/event_bus/events.rssrc/core/event_bus/events_tests.rssrc/core/jsonrpc.rssrc/openhuman/agent/turn_origin.rssrc/openhuman/approval/gate.rssrc/openhuman/channels/runtime/startup.rssrc/openhuman/cron/mod.rssrc/openhuman/cron/scheduler.rssrc/openhuman/cron/store.rssrc/openhuman/cron/tools/add.rssrc/openhuman/cron/types.rssrc/openhuman/flows/bus.rssrc/openhuman/flows/mod.rssrc/openhuman/flows/ops.rssrc/openhuman/flows/ops_tests.rssrc/openhuman/flows/schemas.rssrc/openhuman/flows/store.rssrc/openhuman/flows/store_tests.rssrc/openhuman/flows/types.rssrc/openhuman/tinyflows/caps.rssrc/openhuman/tinyflows/tests.rs
… (Codex P1) FlowTriggerSubscriber was registered only inside start_channels, which run_server_inner skips when no realtime channel is configured or OPENHUMAN_DISABLE_CHANNEL_LISTENERS is set — leaving cron's FlowScheduleTick (and Composio app-events) with no subscriber, so scheduled/app-event workflows never dispatched in the default no-channel desktop context. Move registration into jsonrpc.rs::register_domain_subscribers (unconditional core boot, Once-guarded, subscribe_global + mem::forget — same pattern as the webhook/channel subscribers), and drop it from start_channels to avoid a double subscriber (which would double-dispatch every trigger).
There was a problem hiding this comment.
🧹 Nitpick comments (1)
src/core/jsonrpc.rs (1)
2236-2249: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUpdate the aggregate registration log to include "flows".
The success-path summary log Extends the register_domain_subscribers Once initialization to register a new global event-bus subscriber, flows::bus::FlowTriggerSubscriber, wired with Arc::new(config.clone()) at Line 2247 correctly warns on failure, but the aggregate
log::info!at Lines 2389-2391 that lists all subscribers registered in this block ("webhook, channel, health, conversation, composio, restart, proactive, agent, session_expired, mcp_client") was not updated to mention "flows". Since this is the one line operators scan to confirm boot wiring succeeded, its omission makes it harder to confirm the new flows-trigger dispatch path came up correctly.✏️ Proposed fix
log::info!( - "[event_bus] domain subscribers registered (webhook, channel, health, conversation, composio, restart, proactive, agent, session_expired, mcp_client)" + "[event_bus] domain subscribers registered (webhook, channel, health, conversation, composio, restart, proactive, agent, session_expired, mcp_client, flows)" );Also applies to: 2389-2391
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/core/jsonrpc.rs` around lines 2236 - 2249, The aggregate boot summary in register_domain_subscribers is missing the new flows subscriber, so update the success-path log to include “flows” alongside the existing subscriber list. Make the change in the same initialization block that registers FlowTriggerSubscriber with subscribe_global so the operator-facing log matches the actual set of registered subscribers.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@src/core/jsonrpc.rs`:
- Around line 2236-2249: The aggregate boot summary in
register_domain_subscribers is missing the new flows subscriber, so update the
success-path log to include “flows” alongside the existing subscriber list. Make
the change in the same initialization block that registers FlowTriggerSubscriber
with subscribe_global so the operator-facing log matches the actual set of
registered subscribers.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: f372554f-4ef3-49aa-92be-ce3e7919fc8f
📒 Files selected for processing (2)
src/core/jsonrpc.rssrc/openhuman/channels/runtime/startup.rs
…overlapping runs - cron/store: partial UNIQUE index idx_cron_jobs_flow_command ON cron_jobs(command) WHERE job_type='flow' (never touches shell/agent jobs). add_flow_schedule_job is now race-safe: INSERT ... ON CONFLICT (command) WHERE job_type='flow' DO NOTHING, returning the existing row when the insert loses the race — so concurrent binds yield exactly one flow cron job, never a constraint error. - flows/bus: FlowTriggerSubscriber gains an in-flight HashSet + try_acquire_dispatch with a Drop-guard, so a trigger burst / fast cadence can't run the same flow_id concurrently (guard released even on panic/error). Scoped to trigger dispatch; interactive flows_run RPC is intentionally not deduped. flows tests 68 pass; cron pass (bar the known pre-existing scheduler flake).
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/openhuman/flows/bus.rs`:
- Around line 190-196: The InFlightGuard::drop cleanup currently skips removal
when the mutex is poisoned, which can leave the flow_id stuck in in_flight and
block future dispatches. Update the Drop implementation to recover from
poisoning the same way try_acquire_dispatch does, by taking the inner set even
on a poisoned lock and always removing self.flow_id. Keep the fix localized to
InFlightGuard and its set lock handling so the guard reliably clears state
during drop.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 5fd50486-4ec2-4428-acde-8e8c3de9381a
📒 Files selected for processing (3)
src/openhuman/cron/store.rssrc/openhuman/cron/store_tests.rssrc/openhuman/flows/bus.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- src/openhuman/cron/store.rs
tinyflows 0.3 resolves `=` expressions in integration-node config (agent /
tool_call / http_request) against a { item, items, run } scope — so a node can
now use an upstream node's / the trigger's output in its params (the north-star
'AI draft → Slack' data handoff). The capability traits are UNCHANGED (the crate
resolves `=` before calling them), so the seam adapters need no changes.
Verified live: url `=item.target` resolves from the run input (SSRF guard blocked
the resolved host); a `=item.tool`-derived unknown slug is still rejected by the
hard curation gate ('tool not permitted'). This is exactly why B2's hard curation
+ call-time approval gate are load-bearing now: 0.3 makes `slug`/`url` data-
derivable, so untrusted trigger data can influence *which* curated+connected tool
runs — bounded by curation/scope/connection + the gate; authors should set
require_approval for trigger-driven `=`-tool flows (comment + follow-up noted).
flows 68 tests + tinyflows 14 pass against 0.3; cargo check + fmt clean.
…ers-trust # Conflicts: # app/src-tauri/Cargo.lock
…bit) Drop used `if let Ok(mut set) = self.set.lock()`, silently no-oping on a poisoned lock — which would leave the flow_id in the in-flight set forever, permanently skipping that flow from future trigger dispatches. Mirror try_acquire_dispatch's `unwrap_or_else(|e| e.into_inner())` poison recovery.
1 participant
Summary
B2 of the Workflows feature — makes saved workflows fire themselves on events, resumable, and able to
act safely. Builds on B1 (#4433). Delivered as 3 reviewed commits, each independently verified.
RPC surface grows to 10:
openhuman.flows_{create,get,list,update,delete,set_enabled,run,resume,list_runs,get_run}.What's in it
1. Triggers → run bridge (
FlowScheduleTickcron event +flows/bus.rs)FlowTriggerSubscriberdispatches real runs onFlowScheduleTick(schedule),ComposioTriggerReceived(matched by
toolkit+trigger_slug), and — deferred —WebhookIncomingRequest.flows_set_enabledbinds/unbinds a cron schedule job (idempotent); boot reconciliation re-registersenabled schedule flows. Cron gains a minimal additive
JobType::Flow(stores the flow id incommand,emits
FlowScheduleTick) — no cron schema change, existing shell/agent jobs untouched.2.
flows_resume+ approval notificationopenhuman.flows_resume { id, thread_id, approvals }→resume_with_checkpointer.CoreNotification(Agents category, "approve" action carrying{flow_id, thread_id, node_ids}).3. Trust model + tool curation + connection_ref (security-sensitive)
TrustedAutomationSource::Workflow { require_approval }+ approval-gate arms.require_approval=falsetrusts the saved flow's pre-declared actions (mirrors Cron);
require_approval=trueis fail-closed(parks + TTL-denies; a global "always allow" grant cannot override it). The
Workfloworigin is scopedaround every run — so the gate sees it at tool/HTTP call time (task-local, propagates through the engine's
join_allfan-out; a missing origin falls through toUnknown→ deny).OpenHumanTools/OpenHumanHttpnow route through theApprovalGate (mirroring
ApprovalSecurityMiddleware) — flow HTTP/tool nodes no longer bypass the Networkapproval path.
invokegates onis_action_visible_with_pref+ per-user scope before Composio; anon-curated/out-of-scope slug is rejected.
connection_ref:composio:<toolkit>:<id>is forwarded todirect_execute(Direct mode).4. Run history —
flow_runstable +flows_list_runs/flows_get_run; steps reconstructed fromoutcome.output(lean — the 0.2 durable path usesNoopObserver; a richer observer is a tinyflows 0.3 item).Verification
Each of the 3 commits was reviewed by an independent verification pass (architectobot):
Suite:
GGML_NATIVE=OFF cargo checkclean ·cargo fmt --checkclean ·cargo test --libacrossflows/tinyflows/approval/cron/turn_origin/event_bus = 428 passed, 0 failed (1 pre-existing
#[ignore]).openhuman-core schema | grep flows_lists all 10 controllers.Deliberate stubs / deferrals (documented, fail-safe)
webhooks::create_tunnel(a backend network call)FlowTriggerSubscriberlogs the deferral, never dispatches.connection_ref: Composio backendexecute_toolhas no per-call account-scoping API →logs a warning, falls back to the ambient session account (backend gap, not closable in the seam).
http_cred:<name>: no host HTTP credential store exists → parsed, warned, proceeds without injectedcreds (fail-safe).
slug is static config in 0.2; tighten alongside the tinyflows 0.3
=data-binding release).Non-goals (later phases)
UI (approval card, canvas, run inspector) = B3–B5. The tinyflows
=data-binding (crate 0.3) — being doneseparately — is what finally lets a node use an upstream node's output; this PR notes where it plugs in.
Summary by CodeRabbit
require_approval) and durable flow run history with resume, list runs, and get a run.