From ab9b39c2e06ddfc6fe09e4275a20c4c4012bebaf Mon Sep 17 00:00:00 2001 From: oxoxDev <164490987+oxoxDev@users.noreply.github.com> Date: Fri, 12 Jun 2026 01:25:28 +0530 Subject: [PATCH] feat(workflow-runs): live workflow execution engine (#3375 PR2) (#3600) --- src/openhuman/agent_orchestration/ops.rs | 25 + .../workflow_runs/engine.rs | 855 ++++++++++++++++++ .../workflow_runs/engine_tests.rs | 518 +++++++++++ .../agent_orchestration/workflow_runs/mod.rs | 18 +- .../workflow_runs/schemas.rs | 157 +++- tests/json_rpc_e2e.rs | 152 ++++ 6 files changed, 1715 insertions(+), 10 deletions(-) create mode 100644 src/openhuman/agent_orchestration/workflow_runs/engine.rs create mode 100644 src/openhuman/agent_orchestration/workflow_runs/engine_tests.rs diff --git a/src/openhuman/agent_orchestration/ops.rs b/src/openhuman/agent_orchestration/ops.rs index dca45b3c2..5ce97ba9d 100644 --- a/src/openhuman/agent_orchestration/ops.rs +++ b/src/openhuman/agent_orchestration/ops.rs @@ -252,6 +252,31 @@ impl AgentOrchestrationSession { Ok(snapshot) } + /// Abort every in-flight child task and mark non-terminal children + /// [`AgentStatus::Cancelled`]. + /// + /// Used by the workflow engine on stop/interrupt to drain a session's + /// running children without going through per-child `close_agent` calls. + /// Idempotent — children already terminal are left untouched. Wakes waiters + /// so any pending `wait_agents` resolves immediately. + pub async fn abort_all(&self) { + let mut state = self.state.lock().await; + let task_ids: Vec = state.tasks.keys().cloned().collect(); + for id in task_ids { + if let Some(task) = state.tasks.remove(&id) { + task.abort(); + } + } + for record in state.agents.values_mut() { + if !record.snapshot.status.is_terminal() { + record.snapshot.status = AgentStatus::Cancelled; + record.snapshot.updated_at = now(); + } + } + drop(state); + self.notify.notify_waiters(); + } + /// Spawn a linked follow-up child from an existing child record. /// /// `request.orchestration_id` identifies the previous child and diff --git a/src/openhuman/agent_orchestration/workflow_runs/engine.rs b/src/openhuman/agent_orchestration/workflow_runs/engine.rs new file mode 100644 index 000000000..20acba586 --- /dev/null +++ b/src/openhuman/agent_orchestration/workflow_runs/engine.rs @@ -0,0 +1,855 @@ +//! Live execution engine for durable workflow runs (#3375 PR2). +//! +//! PR1 shipped the declarative [`WorkflowDefinition`] model, the durable +//! [`WorkflowRun`] ledger, and the read controllers. This module makes runs +//! actually *execute*: [`start_workflow_run`] resolves a definition, creates a +//! `Running` ledger row, and spawns a non-blocking engine task that walks the +//! phase DAG in dependency order, fanning out each phase's agents through the +//! programmatic [`AgentOrchestrationSession`] with bounded concurrency, then +//! persisting phase outputs after every phase. [`stop_workflow_run`] flips a +//! cancellation signal the loop checks between phases (→ `Interrupted`); +//! [`resume_workflow_run`] reloads a run and continues from the first +//! incomplete phase. +//! +//! ## Root parent context (the one real unknown) +//! +//! Child agents are spawned via [`AgentOrchestrationSession::spawn_agent`], +//! which reads the *parent execution context* from a task-local +//! ([`current_parent`]). The engine runs from a controller-spawned background +//! task — there is **no** agent turn on the stack, so the task-local is unset +//! and a naive spawn would fail with `NoParentContext`. +//! +//! The fix mirrors the production blueprint in +//! [`crate::openhuman::agent::triage::escalation`]: build a *root* +//! [`ParentExecutionContext`] from a real [`Agent`] (`Agent::from_config`) and +//! run the whole phase loop inside [`with_parent_context`]. Every +//! `spawn_agent` call nested in that scope then resolves `current_parent()` to +//! the root, inheriting a real provider, tool registry, memory, and model — the +//! same construction path `agent_chat` uses. + +use std::collections::{HashMap, HashSet}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, OnceLock}; + +use anyhow::{anyhow, Context, Result}; +use serde_json::{json, Value}; + +use crate::openhuman::agent::harness::fork_context::{with_parent_context, ParentExecutionContext}; +use crate::openhuman::agent::Agent; +use crate::openhuman::config::Config; +use crate::openhuman::session_db::run_ledger::{ + get_workflow_run, upsert_workflow_run, WorkflowRun, WorkflowRunStatus, WorkflowRunUpsert, +}; + +use super::ops::definition_by_id; +use super::types::{WorkflowDefinition, WorkflowPhase}; + +const LOG_TARGET: &str = "workflow_run_engine"; + +/// Per-phase status stored inside the run's `phase_states` JSON column. +const PHASE_PENDING: &str = "pending"; +const PHASE_RUNNING: &str = "running"; +const PHASE_COMPLETED: &str = "completed"; +const PHASE_FAILED: &str = "failed"; + +// ─────────────────────────────────────────────────────────────────────────── +// Cancellation registry +// ─────────────────────────────────────────────────────────────────────────── + +/// Process-wide map of `run_id -> cancellation flag`. `stop_workflow_run` +/// flips the flag; the engine loop checks it between phases and aborts in-flight +/// child tasks via the orchestration session before marking the run +/// `Interrupted`. +fn cancel_registry() -> &'static Mutex>> { + static REGISTRY: OnceLock>>> = OnceLock::new(); + REGISTRY.get_or_init(|| Mutex::new(HashMap::new())) +} + +/// Register (or reuse) a cancellation flag for `run_id`. +fn register_cancel_flag(run_id: &str) -> Arc { + let mut map = cancel_registry().lock().expect("cancel registry poisoned"); + map.entry(run_id.to_string()) + .or_insert_with(|| Arc::new(AtomicBool::new(false))) + .clone() +} + +/// Look up an existing cancellation flag for `run_id`, if one is registered. +fn lookup_cancel_flag(run_id: &str) -> Option> { + cancel_registry() + .lock() + .expect("cancel registry poisoned") + .get(run_id) + .cloned() +} + +/// Drop a run's cancellation flag once the engine loop is done with it. +fn clear_cancel_flag(run_id: &str) { + cancel_registry() + .lock() + .expect("cancel registry poisoned") + .remove(run_id); +} + +// ─────────────────────────────────────────────────────────────────────────── +// Public entry points +// ─────────────────────────────────────────────────────────────────────────── + +/// Start a new workflow run and return immediately. +/// +/// Resolves `definition_id` to a builtin [`WorkflowDefinition`], creates a +/// `Running` ledger row with `phase_states` initialised to one `pending` entry +/// per phase, persists it, then `tokio::spawn`s the engine loop. The returned +/// [`WorkflowRun`] is the freshly-created row (status `Running`); callers poll +/// `workflow_run_get` to observe progress. +pub async fn start_workflow_run( + config: &Config, + definition_id: &str, + input: Value, + parent_thread_id: Option, +) -> Result { + log::debug!( + target: LOG_TARGET, + "[workflow_run_engine] start.entry definition={definition_id} parent_thread={parent_thread_id:?}" + ); + let definition = definition_by_id(definition_id) + .ok_or_else(|| anyhow!("unknown workflow definition: {definition_id}"))?; + + let run_id = format!("wfrun-{}", uuid::Uuid::new_v4()); + let phase_states = init_phase_states(&definition); + + let run = upsert_workflow_run( + config, + WorkflowRunUpsert { + id: run_id.clone(), + definition_id: definition.id.clone(), + parent_thread_id, + input: input.clone(), + phase_states, + child_run_ids: Vec::new(), + status: WorkflowRunStatus::Running, + summary: None, + started_at: None, + completed_at: None, + }, + ) + .context("persist initial workflow run")?; + + register_cancel_flag(&run_id); + + // Spawn the engine loop. Clone what the task needs (the engine reloads + // config inside the task so it can build a real Agent without holding a + // borrow across the spawn boundary). + let task_run_id = run_id.clone(); + tokio::spawn(async move { + match Config::load_or_init().await { + Ok(task_config) => { + run_engine_loop(&task_config, &task_run_id, definition).await; + } + Err(err) => { + log::error!( + target: LOG_TARGET, + "[workflow_run_engine] start.config_load_failed run={task_run_id} err={err}" + ); + } + } + }); + + log::debug!( + target: LOG_TARGET, + "[workflow_run_engine] start.spawned run={run_id} phases={}", + run.phase_states.as_object().map(|m| m.len()).unwrap_or(0) + ); + Ok(run) +} + +/// Signal a running workflow to stop after its current phase. +/// +/// Flips the run's cancellation flag (checked by the loop between phases) and +/// eagerly marks the persisted row `Interrupted` so a poller sees the intent +/// immediately even while the in-flight phase drains. Idempotent: stopping a +/// terminal or unknown run is a no-op that returns the current row. +pub async fn stop_workflow_run(config: &Config, id: &str) -> Result> { + log::debug!(target: LOG_TARGET, "[workflow_run_engine] stop.entry run={id}"); + let Some(run) = get_workflow_run(config, id)? else { + log::debug!(target: LOG_TARGET, "[workflow_run_engine] stop.unknown run={id}"); + return Ok(None); + }; + + if matches!( + run.status, + WorkflowRunStatus::Completed | WorkflowRunStatus::Failed | WorkflowRunStatus::Cancelled + ) { + log::debug!( + target: LOG_TARGET, + "[workflow_run_engine] stop.already_terminal run={id} status={}", + run.status.as_str() + ); + return Ok(Some(run)); + } + + if let Some(flag) = lookup_cancel_flag(id) { + flag.store(true, Ordering::SeqCst); + } else { + // No live loop (e.g. process restart) — register a flag anyway so a + // future resume observes the stop intent. + register_cancel_flag(id).store(true, Ordering::SeqCst); + } + + let updated = upsert_workflow_run( + config, + WorkflowRunUpsert { + id: run.id.clone(), + definition_id: run.definition_id.clone(), + parent_thread_id: run.parent_thread_id.clone(), + input: run.input.clone(), + phase_states: run.phase_states.clone(), + child_run_ids: run.child_run_ids.clone(), + status: WorkflowRunStatus::Interrupted, + summary: run.summary.clone(), + started_at: Some(run.started_at), + completed_at: None, + }, + ) + .context("persist workflow run interrupt")?; + + log::debug!(target: LOG_TARGET, "[workflow_run_engine] stop.marked_interrupted run={id}"); + Ok(Some(updated)) +} + +/// Resume an interrupted (or otherwise incomplete) workflow run. +/// +/// Reloads the run, clears any stale cancellation flag, flips the row back to +/// `Running`, and spawns a fresh engine loop. Phases already `completed` in +/// `phase_states` are skipped; the loop continues from the first incomplete +/// phase whose dependencies are satisfied. Returns the run row (now `Running`), +/// or an error if the run is unknown / already terminal-complete / its +/// definition no longer exists. +pub async fn resume_workflow_run(config: &Config, id: &str) -> Result { + log::debug!(target: LOG_TARGET, "[workflow_run_engine] resume.entry run={id}"); + let run = get_workflow_run(config, id)?.ok_or_else(|| anyhow!("unknown workflow run: {id}"))?; + + if matches!(run.status, WorkflowRunStatus::Completed) { + return Err(anyhow!("workflow run {id} is already completed")); + } + + let definition = definition_by_id(&run.definition_id) + .ok_or_else(|| anyhow!("definition {} no longer exists", run.definition_id))?; + + // Clear any prior cancellation intent and re-register a fresh flag. + clear_cancel_flag(id); + register_cancel_flag(id); + + let resumed = upsert_workflow_run( + config, + WorkflowRunUpsert { + id: run.id.clone(), + definition_id: run.definition_id.clone(), + parent_thread_id: run.parent_thread_id.clone(), + input: run.input.clone(), + phase_states: run.phase_states.clone(), + child_run_ids: run.child_run_ids.clone(), + status: WorkflowRunStatus::Running, + summary: run.summary.clone(), + started_at: Some(run.started_at), + completed_at: None, + }, + ) + .context("persist workflow run resume")?; + + let task_run_id = id.to_string(); + tokio::spawn(async move { + match Config::load_or_init().await { + Ok(task_config) => { + run_engine_loop(&task_config, &task_run_id, definition).await; + } + Err(err) => { + log::error!( + target: LOG_TARGET, + "[workflow_run_engine] resume.config_load_failed run={task_run_id} err={err}" + ); + } + } + }); + + log::debug!(target: LOG_TARGET, "[workflow_run_engine] resume.spawned run={id}"); + Ok(resumed) +} + +// ─────────────────────────────────────────────────────────────────────────── +// Engine loop +// ─────────────────────────────────────────────────────────────────────────── + +/// Build the root parent context + drive the phase DAG to completion. +/// +/// Separated from [`start_workflow_run`] so it can run on the spawned task with +/// an owned [`Config`]. Errors are recorded on the run row (status `Failed`) +/// rather than propagated — there is no caller to receive them. +async fn run_engine_loop(config: &Config, run_id: &str, definition: WorkflowDefinition) { + let cancel = lookup_cancel_flag(run_id).unwrap_or_else(|| register_cancel_flag(run_id)); + + let outcome = match build_root_parent(config).await { + Ok(parent) => { + with_parent_context(parent, async { + drive_phases(config, run_id, &definition, &cancel).await + }) + .await + } + Err(err) => Err(err), + }; + + if let Err(err) = outcome { + log::error!( + target: LOG_TARGET, + "[workflow_run_engine] loop.failed run={run_id} err={err}" + ); + // Best-effort terminal failure write, preserving partial phase state. + if let Ok(Some(run)) = get_workflow_run(config, run_id) { + if !matches!( + run.status, + WorkflowRunStatus::Completed + | WorkflowRunStatus::Failed + | WorkflowRunStatus::Cancelled + | WorkflowRunStatus::Interrupted + ) { + let _ = persist( + config, + &run, + run.phase_states.clone(), + run.child_run_ids.clone(), + WorkflowRunStatus::Failed, + Some(format!("engine error: {err}")), + true, + ); + } + } + } + + clear_cancel_flag(run_id); +} + +/// Build a root [`ParentExecutionContext`] from a config-built [`Agent`]. +/// +/// Mirrors `triage::escalation::dispatch_target_agent` — the proven path for +/// running sub-agents without an enclosing agent turn. +async fn build_root_parent(config: &Config) -> Result { + // The engine fans out builtin sub-agents (planner / researcher / critic / + // summarizer), so the agent-definition registry MUST be initialised before + // any spawn. The full runtime boot (`bootstrap_core_runtime`) does this, but + // the engine can also be reached from contexts that only built the HTTP + // router (e.g. JSON-RPC e2e harness) — so init defensively here. `OnceLock` + // makes this idempotent: a no-op when the registry is already loaded. + if crate::openhuman::agent::harness::AgentDefinitionRegistry::global().is_none() { + if let Err(err) = crate::openhuman::agent::harness::AgentDefinitionRegistry::init_global( + &config.workspace_dir, + ) { + log::warn!( + target: LOG_TARGET, + "[workflow_run_engine] root_parent.registry_init_failed err={err}" + ); + } + } + + let mut agent = Agent::from_config(config) + .context("build Agent from config for workflow engine root parent")?; + + let integrations = crate::openhuman::composio::fetch_connected_integrations(config).await; + agent.set_connected_integrations(integrations); + + Ok(ParentExecutionContext { + agent_definition_id: "workflow_engine".to_string(), + allowed_subagent_ids: HashSet::new(), + provider: agent.provider_arc(), + all_tools: agent.tools_arc(), + all_tool_specs: agent.tool_specs_arc(), + model_name: agent.model_name().to_string(), + temperature: agent.temperature(), + workspace_dir: agent.workspace_dir().to_path_buf(), + memory: agent.memory_arc(), + agent_config: agent.agent_config().clone(), + workflows: Arc::new(agent.workflows().to_vec()), + memory_context: Arc::new(None), + session_id: format!("workflow-{}", uuid::Uuid::new_v4()), + channel: "workflow".to_string(), + connected_integrations: agent.connected_integrations().to_vec(), + tool_call_format: crate::openhuman::context::prompt::ToolCallFormat::PFormat, + session_key: agent.session_key().to_string(), + session_parent_prefix: agent.session_parent_prefix().map(str::to_string), + on_progress: None, + run_queue: None, + }) +} + +/// Topologically walk the phase DAG, executing each phase whose dependencies +/// are all `completed`. Persists after every phase. Returns `Ok(())` once every +/// phase is completed, the run is interrupted, or a phase fails (the terminal +/// status is written to the row before returning `Ok`). Returns `Err` only for +/// engine-internal failures (e.g. ledger write errors). +async fn drive_phases( + config: &Config, + run_id: &str, + definition: &WorkflowDefinition, + cancel: &Arc, +) -> Result<()> { + use crate::openhuman::agent_orchestration::{ + AgentOrchestrationSession, AgentStatus, SpawnAgentRequest, WaitAgentOptions, + }; + + let session = AgentOrchestrationSession::new(format!("workflow-engine-{run_id}")); + let mut total_spawned: u32 = 0; + + // Optional per-run model override. When present in the run input + // (`{"modelOverride": "..."}`), every child is forced onto the parent + // (root) provider with this model instead of its declarative workload + // provider. Production runs omit it (agents use their configured provider); + // deterministic mock-backend tests set it so children resolve to the + // injected mock provider. + let model_override = get_workflow_run(config, run_id)? + .and_then(|r| { + r.input + .get("modelOverride") + .and_then(Value::as_str) + .map(str::to_string) + }) + .filter(|m| !m.trim().is_empty()); + + loop { + // Reload the run each iteration so we read the latest phase_states (and + // so a resume picks up persisted progress). + let run = get_workflow_run(config, run_id)? + .ok_or_else(|| anyhow!("workflow run {run_id} vanished mid-loop"))?; + let mut phase_states = run.phase_states.clone(); + let mut child_run_ids = run.child_run_ids.clone(); + + // Cancellation check between phases. + if cancel.load(Ordering::SeqCst) { + log::debug!( + target: LOG_TARGET, + "[workflow_run_engine] loop.cancelled run={run_id}" + ); + session.abort_all().await; + persist( + config, + &run, + phase_states, + child_run_ids, + WorkflowRunStatus::Interrupted, + None, + false, + )?; + return Ok(()); + } + + // Find the next runnable phase: pending, with all deps completed. + let Some(phase) = next_runnable_phase(definition, &phase_states) else { + // No runnable phase left. Either everything is done, or we're + // blocked (which a validated DAG shouldn't be). + if all_phases_completed(definition, &phase_states) { + let summary = synthesize_summary(definition, &phase_states); + log::debug!( + target: LOG_TARGET, + "[workflow_run_engine] loop.completed run={run_id} summary_chars={}", + summary.as_deref().map(str::len).unwrap_or(0) + ); + persist( + config, + &run, + phase_states, + child_run_ids, + WorkflowRunStatus::Completed, + summary, + true, + )?; + } else { + log::warn!( + target: LOG_TARGET, + "[workflow_run_engine] loop.stuck run={run_id} no_runnable_phase" + ); + persist( + config, + &run, + phase_states, + child_run_ids, + WorkflowRunStatus::Failed, + Some("no runnable phase (dependency deadlock)".to_string()), + true, + )?; + } + return Ok(()); + }; + + log::debug!( + target: LOG_TARGET, + "[workflow_run_engine] phase.start run={run_id} phase={} agents={} spawned_so_far={}", + phase.name, + phase.agent_ids.len(), + total_spawned + ); + set_phase_status(&mut phase_states, &phase.name, PHASE_RUNNING, None); + persist( + config, + &run, + phase_states.clone(), + child_run_ids.clone(), + WorkflowRunStatus::Running, + None, + false, + )?; + + // Thread prior phases' outputs into this phase's prompt context. + let upstream_context = upstream_outputs(definition, phase, &phase_states); + + // Spawn the phase's agents in concurrency-bounded batches, never + // exceeding the run-wide `max_children` cap. + let mut phase_outputs: Vec = Vec::new(); + let mut phase_failed: Option = None; + let mut index_in_phase = 0usize; + + let concurrency = definition.default_concurrency.max(1); + let mut pending = phase.agent_ids.to_vec(); + + 'phase: while !pending.is_empty() { + // Cancellation can also land mid-phase between batches. + if cancel.load(Ordering::SeqCst) { + session.abort_all().await; + persist( + config, + &run, + phase_states, + child_run_ids, + WorkflowRunStatus::Interrupted, + None, + false, + )?; + return Ok(()); + } + + let budget_left = definition.max_children.saturating_sub(total_spawned); + if budget_left == 0 { + phase_failed = Some(format!( + "max_children cap ({}) reached before phase '{}' completed", + definition.max_children, phase.name + )); + break 'phase; + } + let batch_size = (concurrency as usize) + .min(budget_left as usize) + .min(pending.len()) + .max(1); + let batch: Vec = pending.drain(..batch_size).collect(); + + let mut batch_ids: Vec<(String, String)> = Vec::new(); // (orchestration_id, agent_id) + for agent_id in &batch { + let prompt = phase_prompt(&run.input, phase, index_in_phase, &upstream_context); + index_in_phase += 1; + match session + .spawn_agent(SpawnAgentRequest { + agent_id: agent_id.clone(), + prompt, + model: model_override.clone(), + ..Default::default() + }) + .await + { + Ok(resp) => { + total_spawned += 1; + child_run_ids.push(resp.orchestration_id.clone()); + batch_ids.push((resp.orchestration_id, agent_id.clone())); + } + Err(err) => { + phase_failed = Some(format!("spawn failed for agent '{agent_id}': {err}")); + break 'phase; + } + } + } + + // Wait for this batch to reach terminal status. + let wait = session + .wait_agents(WaitAgentOptions { + orchestration_ids: batch_ids.iter().map(|(id, _)| id.clone()).collect(), + timeout_ms: None, + }) + .await + .map_err(|e| anyhow!("wait_agents failed: {e}"))?; + + for snapshot in &wait.agents { + match snapshot.status { + AgentStatus::Completed => { + phase_outputs.push(json!({ + "orchestrationId": snapshot.orchestration_id, + "agentId": snapshot.agent_id, + "output": snapshot.result_summary.clone().unwrap_or_default(), + })); + } + AgentStatus::Failed | AgentStatus::Cancelled | AgentStatus::Closed => { + phase_failed = Some(format!( + "child '{}' (agent '{}') ended {}: {}", + snapshot.orchestration_id, + snapshot.agent_id, + serde_json::to_value(snapshot.status) + .ok() + .and_then(|v| v.as_str().map(str::to_string)) + .unwrap_or_else(|| "non-completed".to_string()), + snapshot.error.clone().unwrap_or_default() + )); + break 'phase; + } + AgentStatus::Pending | AgentStatus::Running | AgentStatus::Waiting => { + // wait_agents with no timeout only returns on terminal, + // so this is defensive. + phase_failed = Some(format!( + "child '{}' returned non-terminal status", + snapshot.orchestration_id + )); + break 'phase; + } + } + } + } + + if let Some(reason) = phase_failed { + log::warn!( + target: LOG_TARGET, + "[workflow_run_engine] phase.failed run={run_id} phase={} reason={reason}", + phase.name + ); + set_phase_status( + &mut phase_states, + &phase.name, + PHASE_FAILED, + Some(json!([])), + ); + set_phase_reason(&mut phase_states, &phase.name, &reason); + persist( + config, + &run, + phase_states, + child_run_ids, + WorkflowRunStatus::Failed, + Some(reason), + true, + )?; + return Ok(()); + } + + log::debug!( + target: LOG_TARGET, + "[workflow_run_engine] phase.done run={run_id} phase={} outputs={}", + phase.name, + phase_outputs.len() + ); + set_phase_status( + &mut phase_states, + &phase.name, + PHASE_COMPLETED, + Some(Value::Array(phase_outputs)), + ); + persist( + config, + &run, + phase_states, + child_run_ids, + WorkflowRunStatus::Running, + None, + false, + )?; + } +} + +// ─────────────────────────────────────────────────────────────────────────── +// Phase-state helpers +// ─────────────────────────────────────────────────────────────────────────── + +/// Initialise `phase_states` to one `pending` entry per phase, preserving +/// declaration order via an object keyed by phase name. +fn init_phase_states(definition: &WorkflowDefinition) -> Value { + let mut map = serde_json::Map::new(); + for phase in &definition.phases { + map.insert( + phase.name.clone(), + json!({ "status": PHASE_PENDING, "outputs": [] }), + ); + } + Value::Object(map) +} + +fn phase_status<'a>(phase_states: &'a Value, name: &str) -> Option<&'a str> { + phase_states + .get(name) + .and_then(|p| p.get("status")) + .and_then(Value::as_str) +} + +fn set_phase_status(phase_states: &mut Value, name: &str, status: &str, outputs: Option) { + if let Some(obj) = phase_states.as_object_mut() { + let entry = obj + .entry(name.to_string()) + .or_insert_with(|| json!({ "status": PHASE_PENDING, "outputs": [] })); + if let Some(entry_obj) = entry.as_object_mut() { + entry_obj.insert("status".to_string(), json!(status)); + if let Some(out) = outputs { + entry_obj.insert("outputs".to_string(), out); + } + } + } +} + +fn set_phase_reason(phase_states: &mut Value, name: &str, reason: &str) { + if let Some(obj) = phase_states.as_object_mut() { + if let Some(entry) = obj.get_mut(name).and_then(Value::as_object_mut) { + entry.insert("reason".to_string(), json!(reason)); + } + } +} + +/// The first phase that is `pending` (or missing) and whose every dependency is +/// `completed`. Definition order breaks ties so the walk is deterministic. +fn next_runnable_phase<'a>( + definition: &'a WorkflowDefinition, + phase_states: &Value, +) -> Option<&'a WorkflowPhase> { + definition.phases.iter().find(|phase| { + let status = phase_status(phase_states, &phase.name).unwrap_or(PHASE_PENDING); + if status == PHASE_COMPLETED || status == PHASE_RUNNING { + return false; + } + phase + .depends_on + .iter() + .all(|dep| phase_status(phase_states, dep) == Some(PHASE_COMPLETED)) + }) +} + +fn all_phases_completed(definition: &WorkflowDefinition, phase_states: &Value) -> bool { + definition + .phases + .iter() + .all(|phase| phase_status(phase_states, &phase.name) == Some(PHASE_COMPLETED)) +} + +/// Collect the outputs of every completed phase this phase depends on, so they +/// can be threaded into the downstream prompt. +fn upstream_outputs( + _definition: &WorkflowDefinition, + phase: &WorkflowPhase, + phase_states: &Value, +) -> Vec { + let mut out = Vec::new(); + for dep in &phase.depends_on { + if let Some(outputs) = phase_states + .get(dep) + .and_then(|p| p.get("outputs")) + .and_then(Value::as_array) + { + for item in outputs { + if let Some(text) = item.get("output").and_then(Value::as_str) { + if !text.trim().is_empty() { + out.push(json!({ "phase": dep, "output": text })); + } + } + } + } + } + out +} + +/// Build the prompt for one child in a phase: the run input + the phase's +/// description + upstream outputs threaded in as context. +fn phase_prompt( + input: &Value, + phase: &WorkflowPhase, + index_in_phase: usize, + upstream: &[Value], +) -> String { + let question = input + .get("question") + .or_else(|| input.get("input")) + .and_then(Value::as_str) + .map(str::to_string) + .unwrap_or_else(|| input.to_string()); + + let mut prompt = format!( + "Workflow phase: {}\n{}\n\nInput:\n{}\n", + phase.name, phase.description, question + ); + if phase.agent_ids.len() > 1 { + prompt.push_str(&format!( + "\n(You are worker #{} in this phase.)\n", + index_in_phase + 1 + )); + } + if !upstream.is_empty() { + prompt.push_str("\nContext from prior phases:\n"); + for item in upstream { + if let (Some(p), Some(o)) = ( + item.get("phase").and_then(Value::as_str), + item.get("output").and_then(Value::as_str), + ) { + prompt.push_str(&format!("- [{p}] {o}\n")); + } + } + } + prompt +} + +/// The synthesize phase's combined output becomes the run summary. Falls back +/// to the last completed phase's output if no phase is literally named +/// `synthesize`. +fn synthesize_summary(definition: &WorkflowDefinition, phase_states: &Value) -> Option { + let pick = |name: &str| -> Option { + let outputs = phase_states + .get(name) + .and_then(|p| p.get("outputs")) + .and_then(Value::as_array)?; + let joined = outputs + .iter() + .filter_map(|o| o.get("output").and_then(Value::as_str)) + .filter(|s| !s.trim().is_empty()) + .collect::>() + .join("\n"); + (!joined.trim().is_empty()).then_some(joined) + }; + + if let Some(summary) = pick("synthesize") { + return Some(summary); + } + // Fall back to the last phase in declaration order with non-empty output. + definition + .phases + .iter() + .rev() + .find_map(|phase| pick(&phase.name)) +} + +/// Persist a run-state update. `terminal` controls whether `completed_at` is +/// stamped. +#[allow(clippy::too_many_arguments)] +fn persist( + config: &Config, + run: &WorkflowRun, + phase_states: Value, + child_run_ids: Vec, + status: WorkflowRunStatus, + summary: Option, + terminal: bool, +) -> Result { + upsert_workflow_run( + config, + WorkflowRunUpsert { + id: run.id.clone(), + definition_id: run.definition_id.clone(), + parent_thread_id: run.parent_thread_id.clone(), + input: run.input.clone(), + phase_states, + child_run_ids, + status, + summary, + started_at: Some(run.started_at), + completed_at: terminal.then(chrono::Utc::now), + }, + ) + .context("persist workflow run state") +} + +#[cfg(test)] +#[path = "engine_tests.rs"] +mod engine_tests; diff --git a/src/openhuman/agent_orchestration/workflow_runs/engine_tests.rs b/src/openhuman/agent_orchestration/workflow_runs/engine_tests.rs new file mode 100644 index 000000000..95724d2d6 --- /dev/null +++ b/src/openhuman/agent_orchestration/workflow_runs/engine_tests.rs @@ -0,0 +1,518 @@ +//! Engine unit tests (#3375 PR2). +//! +//! These exercise the phase scheduler ([`super::drive_phases`]) directly with a +//! mock `Provider` so child agents resolve deterministically and never touch the +//! network. The full [`super::start_workflow_run`] entry point (which builds a +//! real `Agent` from config) is covered by the JSON-RPC e2e test over the live +//! core stack with the mock backend. +//! +//! Test strategy: install a mock [`ParentExecutionContext`] via +//! [`with_parent_context`] (mirroring `agent_orchestration::ops_tests`), seed a +//! `Running` workflow-run row in a tempdir-backed ledger, then drive a custom +//! definition and assert on the persisted `phase_states` + run status. + +use std::collections::HashSet; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::Arc; + +use async_trait::async_trait; +use parking_lot::Mutex; +use serde_json::{json, Value}; +use tokio::time::{sleep, Duration}; + +use super::*; +use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry; +use crate::openhuman::agent::harness::fork_context::{with_parent_context, ParentExecutionContext}; +use crate::openhuman::config::{AgentConfig, Config}; +use crate::openhuman::context::prompt::ToolCallFormat; +use crate::openhuman::inference::provider::traits::ProviderCapabilities; +use crate::openhuman::inference::provider::{ChatRequest, ChatResponse, Provider}; +use crate::openhuman::memory::{Memory, MemoryCategory, MemoryEntry, NamespaceSummary, RecallOpts}; +use crate::openhuman::session_db::run_ledger::{ + get_workflow_run, upsert_workflow_run, WorkflowRunUpsert, +}; +use crate::openhuman::tools::{Tool, ToolSpec}; + +use super::super::types::{WorkflowDefinition, WorkflowPhase, WorkflowSafetyTier}; + +// ── Mocks (mirrors agent_orchestration::ops_tests) ────────────────────────── + +#[derive(Default)] +struct NoopMemory; + +#[async_trait] +impl Memory for NoopMemory { + async fn store( + &self, + _namespace: &str, + _key: &str, + _content: &str, + _category: MemoryCategory, + _session_id: Option<&str>, + ) -> anyhow::Result<()> { + Ok(()) + } + async fn recall( + &self, + _query: &str, + _limit: usize, + _opts: RecallOpts<'_>, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + async fn get(&self, _namespace: &str, _key: &str) -> anyhow::Result> { + Ok(None) + } + async fn list( + &self, + _namespace: Option<&str>, + _category: Option<&MemoryCategory>, + _session_id: Option<&str>, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + async fn forget(&self, _namespace: &str, _key: &str) -> anyhow::Result { + Ok(false) + } + async fn namespace_summaries(&self) -> anyhow::Result> { + Ok(Vec::new()) + } + async fn count(&self) -> anyhow::Result { + Ok(0) + } + async fn health_check(&self) -> bool { + true + } + fn name(&self) -> &str { + "noop" + } +} + +fn text_response(text: impl Into) -> ChatResponse { + ChatResponse { + text: Some(text.into()), + tool_calls: Vec::new(), + usage: None, + reasoning_content: None, + } +} + +/// Mock provider that records peak concurrency and answers each child with a +/// short deterministic completion. Sleeps briefly so overlapping spawns are +/// observable for the concurrency-cap assertions. +#[derive(Clone, Default)] +struct PeakProvider { + calls: Arc, + active: Arc, + max_active: Arc, + prompts: Arc>>, + fail_on: Arc>>, +} + +impl PeakProvider { + fn calls(&self) -> usize { + self.calls.load(Ordering::SeqCst) + } + fn max_active(&self) -> usize { + self.max_active.load(Ordering::SeqCst) + } + fn fail_when_prompt_contains(&self, needle: &str) { + *self.fail_on.lock() = Some(needle.to_string()); + } + fn record_peak(&self, current: usize) { + let mut observed = self.max_active.load(Ordering::SeqCst); + while current > observed { + match self.max_active.compare_exchange( + observed, + current, + Ordering::SeqCst, + Ordering::SeqCst, + ) { + Ok(_) => break, + Err(next) => observed = next, + } + } + } +} + +#[async_trait] +impl Provider for PeakProvider { + fn capabilities(&self) -> ProviderCapabilities { + ProviderCapabilities { + native_tool_calling: true, + vision: false, + } + } + async fn chat_with_system( + &self, + _system_prompt: Option<&str>, + _message: &str, + _model: &str, + _temperature: f64, + ) -> anyhow::Result { + Ok("ok".to_string()) + } + async fn chat( + &self, + request: ChatRequest<'_>, + _model: &str, + _temperature: f64, + ) -> anyhow::Result { + self.calls.fetch_add(1, Ordering::SeqCst); + let current = self.active.fetch_add(1, Ordering::SeqCst) + 1; + self.record_peak(current); + sleep(Duration::from_millis(40)).await; + let flattened = request + .messages + .iter() + .map(|m| m.content.as_str()) + .collect::>() + .join("\n"); + self.prompts.lock().push(flattened.clone()); + self.active.fetch_sub(1, Ordering::SeqCst); + + if let Some(needle) = self.fail_on.lock().as_ref() { + if flattened.contains(needle.as_str()) { + return Err(anyhow::anyhow!("mock provider forced failure")); + } + } + Ok(text_response("PHASE_OUTPUT_OK")) + } +} + +fn mock_parent(provider: Arc) -> ParentExecutionContext { + ParentExecutionContext { + agent_definition_id: "workflow_engine".to_string(), + allowed_subagent_ids: HashSet::new(), + provider, + all_tools: Arc::new(Vec::>::new()), + all_tool_specs: Arc::new(Vec::::new()), + model_name: "test-model".to_string(), + temperature: 0.2, + workspace_dir: std::env::temp_dir(), + memory: Arc::new(NoopMemory), + agent_config: AgentConfig::default(), + workflows: Arc::new(Vec::new()), + memory_context: Arc::new(None), + session_id: "workflow-engine-test".to_string(), + channel: "test".to_string(), + connected_integrations: Vec::new(), + tool_call_format: ToolCallFormat::PFormat, + session_key: "0_workflow_engine".to_string(), + session_parent_prefix: None, + on_progress: None, + run_queue: None, + } +} + +fn test_config() -> (tempfile::TempDir, Config) { + let dir = tempfile::tempdir().expect("tempdir"); + let config = Config { + workspace_dir: dir.path().to_path_buf(), + ..Config::default() + }; + (dir, config) +} + +/// Seed a `Running` workflow run row for `definition`, returning its id. +fn seed_run(config: &Config, definition: &WorkflowDefinition, input: Value) -> String { + let id = format!("wfrun-test-{}", uuid::Uuid::new_v4()); + upsert_workflow_run( + config, + WorkflowRunUpsert { + id: id.clone(), + definition_id: definition.id.clone(), + parent_thread_id: None, + input, + phase_states: init_phase_states(definition), + child_run_ids: Vec::new(), + status: WorkflowRunStatus::Running, + summary: None, + started_at: None, + completed_at: None, + }, + ) + .expect("seed run"); + id +} + +/// A linear A→B→C definition using `code_executor` (a builtin leaf agent), with +/// `concurrency` workers in phase B. +fn linear_def(concurrency: u32, max_children: u32, parallel_in_b: usize) -> WorkflowDefinition { + WorkflowDefinition { + id: "test_linear".to_string(), + name: "test linear".to_string(), + description: "A then parallel B then C".to_string(), + phases: vec![ + WorkflowPhase { + name: "a".to_string(), + description: "phase a".to_string(), + agent_ids: vec!["code_executor".to_string()], + depends_on: vec![], + }, + WorkflowPhase { + name: "b".to_string(), + description: "phase b PARALLEL".to_string(), + agent_ids: vec!["code_executor".to_string(); parallel_in_b], + depends_on: vec!["a".to_string()], + }, + WorkflowPhase { + name: "c".to_string(), + description: "phase c synthesize".to_string(), + agent_ids: vec!["code_executor".to_string()], + depends_on: vec!["b".to_string()], + }, + ], + default_concurrency: concurrency, + max_children, + safety_tier: WorkflowSafetyTier::ReadOnly, + } +} + +fn status_of(config: &Config, id: &str) -> WorkflowRunStatus { + get_workflow_run(config, id).unwrap().unwrap().status +} + +fn phase_states(config: &Config, id: &str) -> Value { + get_workflow_run(config, id).unwrap().unwrap().phase_states +} + +// ── Tests ─────────────────────────────────────────────────────────────────── + +#[tokio::test] +async fn unit_phases_execute_in_dependency_order() { + AgentDefinitionRegistry::init_global_builtins().unwrap(); + let (_dir, config) = test_config(); + let provider = PeakProvider::default(); + let def = linear_def(2, 8, 2); + let id = seed_run( + &config, + &def, + json!({ "question": "what is X?", "modelOverride": "test-model" }), + ); + let cancel = Arc::new(AtomicBool::new(false)); + + with_parent_context(mock_parent(Arc::new(provider.clone())), async { + drive_phases(&config, &id, &def, &cancel).await + }) + .await + .expect("drive_phases ok"); + + assert_eq!(status_of(&config, &id), WorkflowRunStatus::Completed); + let states = phase_states(&config, &id); + for phase in ["a", "b", "c"] { + assert_eq!( + states + .get(phase) + .and_then(|p| p.get("status")) + .and_then(Value::as_str), + Some("completed"), + "phase {phase} should be completed: {states}" + ); + } + // 1 (a) + 2 (b) + 1 (c) = 4 children executed. + assert_eq!(provider.calls(), 4, "all four children should run"); + + // Summary comes from the last phase's output (no phase literally named + // 'synthesize', so the fallback picks phase c). + let summary = get_workflow_run(&config, &id).unwrap().unwrap().summary; + assert!( + summary + .as_deref() + .unwrap_or_default() + .contains("PHASE_OUTPUT_OK"), + "summary should carry final phase output: {summary:?}" + ); +} + +#[tokio::test] +async fn unit_concurrency_cap_is_respected() { + AgentDefinitionRegistry::init_global_builtins().unwrap(); + let (_dir, config) = test_config(); + let provider = PeakProvider::default(); + // 4 parallel workers in phase b, but concurrency capped at 2. + let def = linear_def(2, 16, 4); + let id = seed_run( + &config, + &def, + json!({ "question": "q", "modelOverride": "test-model" }), + ); + let cancel = Arc::new(AtomicBool::new(false)); + + with_parent_context(mock_parent(Arc::new(provider.clone())), async { + drive_phases(&config, &id, &def, &cancel).await + }) + .await + .expect("drive_phases ok"); + + assert_eq!(status_of(&config, &id), WorkflowRunStatus::Completed); + assert!( + provider.max_active() <= 2, + "concurrency cap of 2 exceeded: peak={}", + provider.max_active() + ); +} + +#[tokio::test] +async fn unit_max_children_hard_cap_fails_run() { + AgentDefinitionRegistry::init_global_builtins().unwrap(); + let (_dir, config) = test_config(); + let provider = PeakProvider::default(); + // a(1) + b(4) = 5 needed, but max_children = 3 → run fails in phase b. + let def = linear_def(2, 3, 4); + let id = seed_run( + &config, + &def, + json!({ "question": "q", "modelOverride": "test-model" }), + ); + let cancel = Arc::new(AtomicBool::new(false)); + + with_parent_context(mock_parent(Arc::new(provider.clone())), async { + drive_phases(&config, &id, &def, &cancel).await + }) + .await + .expect("drive_phases returns Ok with terminal Failed state"); + + assert_eq!(status_of(&config, &id), WorkflowRunStatus::Failed); + let summary = get_workflow_run(&config, &id).unwrap().unwrap().summary; + assert!( + summary + .as_deref() + .unwrap_or_default() + .contains("max_children"), + "failure reason should mention max_children: {summary:?}" + ); + // Phase a completed before the cap bit; partial state preserved. + let states = phase_states(&config, &id); + assert_eq!( + states + .get("a") + .and_then(|p| p.get("status")) + .and_then(Value::as_str), + Some("completed") + ); +} + +#[tokio::test] +async fn unit_failed_child_marks_run_failed_with_partial_state() { + AgentDefinitionRegistry::init_global_builtins().unwrap(); + let (_dir, config) = test_config(); + let provider = PeakProvider::default(); + // Force the phase-b child to fail (its prompt mentions "PARALLEL"). + provider.fail_when_prompt_contains("phase b PARALLEL"); + let def = linear_def(2, 8, 1); + let id = seed_run( + &config, + &def, + json!({ "question": "q", "modelOverride": "test-model" }), + ); + let cancel = Arc::new(AtomicBool::new(false)); + + with_parent_context(mock_parent(Arc::new(provider.clone())), async { + drive_phases(&config, &id, &def, &cancel).await + }) + .await + .expect("drive_phases returns Ok with terminal Failed state"); + + assert_eq!(status_of(&config, &id), WorkflowRunStatus::Failed); + let states = phase_states(&config, &id); + assert_eq!( + states + .get("a") + .and_then(|p| p.get("status")) + .and_then(Value::as_str), + Some("completed"), + "phase a should have completed before b failed" + ); + assert_eq!( + states + .get("b") + .and_then(|p| p.get("status")) + .and_then(Value::as_str), + Some("failed") + ); + assert_eq!( + states + .get("c") + .and_then(|p| p.get("status")) + .and_then(Value::as_str), + Some("pending"), + "phase c never ran" + ); +} + +#[tokio::test] +async fn unit_stop_mid_run_marks_interrupted() { + AgentDefinitionRegistry::init_global_builtins().unwrap(); + let (_dir, config) = test_config(); + let provider = PeakProvider::default(); + let def = linear_def(2, 8, 1); + let id = seed_run( + &config, + &def, + json!({ "question": "q", "modelOverride": "test-model" }), + ); + // Pre-flip the cancel flag so the loop interrupts before the first phase. + let cancel = Arc::new(AtomicBool::new(true)); + + with_parent_context(mock_parent(Arc::new(provider.clone())), async { + drive_phases(&config, &id, &def, &cancel).await + }) + .await + .expect("drive_phases ok"); + + assert_eq!(status_of(&config, &id), WorkflowRunStatus::Interrupted); + // No children ran — cancellation was checked before the first phase. + assert_eq!(provider.calls(), 0); +} + +#[tokio::test] +async fn unit_resume_skips_completed_phases() { + AgentDefinitionRegistry::init_global_builtins().unwrap(); + let (_dir, config) = test_config(); + let provider = PeakProvider::default(); + let def = linear_def(2, 8, 1); + let id = seed_run( + &config, + &def, + json!({ "question": "q", "modelOverride": "test-model" }), + ); + + // Mark phase 'a' already completed (simulating a prior partial run). + let run = get_workflow_run(&config, &id).unwrap().unwrap(); + let mut states = run.phase_states.clone(); + states["a"]["status"] = json!("completed"); + states["a"]["outputs"] = + json!([{ "orchestrationId": "x", "agentId": "code_executor", "output": "A_DONE" }]); + upsert_workflow_run( + &config, + WorkflowRunUpsert { + id: id.clone(), + definition_id: run.definition_id.clone(), + parent_thread_id: None, + input: run.input.clone(), + phase_states: states, + child_run_ids: Vec::new(), + status: WorkflowRunStatus::Running, + summary: None, + started_at: Some(run.started_at), + completed_at: None, + }, + ) + .unwrap(); + + let cancel = Arc::new(AtomicBool::new(false)); + with_parent_context(mock_parent(Arc::new(provider.clone())), async { + drive_phases(&config, &id, &def, &cancel).await + }) + .await + .expect("drive_phases ok"); + + assert_eq!(status_of(&config, &id), WorkflowRunStatus::Completed); + // Only phases b and c run on resume (a was already completed): 2 calls. + assert_eq!( + provider.calls(), + 2, + "resume must skip the completed phase a" + ); +} diff --git a/src/openhuman/agent_orchestration/workflow_runs/mod.rs b/src/openhuman/agent_orchestration/workflow_runs/mod.rs index 5f1e88969..470315484 100644 --- a/src/openhuman/agent_orchestration/workflow_runs/mod.rs +++ b/src/openhuman/agent_orchestration/workflow_runs/mod.rs @@ -6,20 +6,26 @@ //! table) rather than the main chat context, so runs can be listed, inspected, //! and — once the engine lands — stopped and resumed. //! -//! PR1 scope (this module today): the declarative definition model, the builtin -//! read-only "parallel research with cross-checking" workflow, structural + -//! agent validation, and the read controllers (`workflow_run_list_definitions`, -//! `workflow_run_list`, `workflow_run_get`). The live execution engine -//! (start/stop/resume, phase scheduling over `spawn_parallel_agents`, -//! concurrency caps, approval gate) is a follow-up PR. +//! PR1 scope: the declarative definition model, the builtin read-only +//! "parallel research with cross-checking" workflow, structural + agent +//! validation, and the read controllers (`workflow_run_list_definitions`, +//! `workflow_run_list`, `workflow_run_get`). +//! +//! PR2 scope (`engine.rs`): the live execution engine — `start`/`stop`/`resume` +//! controllers, phase scheduling that walks the dependency DAG and fans out each +//! phase's agents through the programmatic `AgentOrchestrationSession` with +//! bounded concurrency and a run-wide `max_children` cap, persisting phase +//! outputs to the run ledger after every phase. //! //! Namespace note: this is distinct from the existing `workflows` domain, which //! handles SKILL.md / WORKFLOW.md bundle discovery. +mod engine; mod ops; mod schemas; pub mod types; +pub use engine::{resume_workflow_run, start_workflow_run, stop_workflow_run}; pub use ops::{ builtin_definitions, definition_by_id, get_run, list_definitions, list_runs, validate_definition, validate_structure, PARALLEL_RESEARCH_ID, diff --git a/src/openhuman/agent_orchestration/workflow_runs/schemas.rs b/src/openhuman/agent_orchestration/workflow_runs/schemas.rs index 0d8248353..2b064de2c 100644 --- a/src/openhuman/agent_orchestration/workflow_runs/schemas.rs +++ b/src/openhuman/agent_orchestration/workflow_runs/schemas.rs @@ -1,9 +1,9 @@ //! Controller schemas + JSON-RPC dispatchers for durable workflow runs. //! -//! Read-only surface (PR1): list builtin definitions, list durable runs, get a -//! run by id. Namespace `workflow_run` — distinct from the existing -//! `workflows` domain (SKILL.md/WORKFLOW.md discovery). Start / stop / resume -//! controllers land with the execution engine in a follow-up PR. +//! Read surface (PR1): list builtin definitions, list durable runs, get a run +//! by id. Execution surface (PR2): start / stop / resume, delegating to the +//! [`super::engine`] phase scheduler. Namespace `workflow_run` — distinct from +//! the existing `workflows` domain (SKILL.md/WORKFLOW.md discovery). use serde_json::{Map, Value}; @@ -19,6 +19,9 @@ pub fn all_controller_schemas() -> Vec { schema_for("workflow_run_list_definitions"), schema_for("workflow_run_list"), schema_for("workflow_run_get"), + schema_for("workflow_run_start"), + schema_for("workflow_run_stop"), + schema_for("workflow_run_resume"), ] } @@ -37,6 +40,18 @@ pub fn all_registered_controllers() -> Vec { schema: schema_for("workflow_run_get"), handler: handle_get, }, + RegisteredController { + schema: schema_for("workflow_run_start"), + handler: handle_start, + }, + RegisteredController { + schema: schema_for("workflow_run_stop"), + handler: handle_stop, + }, + RegisteredController { + schema: schema_for("workflow_run_resume"), + handler: handle_resume, + }, ] } @@ -75,6 +90,44 @@ fn schema_for(function: &str) -> ControllerSchema { inputs: vec![required_str("id", "Workflow run id.")], outputs: vec![json_output("workflowRun", "WorkflowRun payload or null.")], }, + "workflow_run_start" => ControllerSchema { + namespace: "workflow_run", + function: "start", + description: "Start a workflow run from a definition id; returns the created \ + Running run. Execution proceeds asynchronously — poll \ + workflow_run_get for progress.", + inputs: vec![ + required_str("definitionId", "Workflow definition id to run."), + optional_json("input", "Run input (e.g. { \"question\": \"...\" })."), + optional_str("parentThreadId", "Originating thread id, for lineage."), + ], + outputs: vec![json_output( + "workflowRun", + "The created WorkflowRun (status running).", + )], + }, + "workflow_run_stop" => ControllerSchema { + namespace: "workflow_run", + function: "stop", + description: "Stop a running workflow after its current phase; marks the run \ + interrupted and aborts in-flight child agents.", + inputs: vec![required_str("id", "Workflow run id.")], + outputs: vec![json_output( + "workflowRun", + "Updated WorkflowRun payload or null.", + )], + }, + "workflow_run_resume" => ControllerSchema { + namespace: "workflow_run", + function: "resume", + description: "Resume an interrupted workflow run, skipping phases already \ + completed and continuing from the first incomplete phase.", + inputs: vec![required_str("id", "Workflow run id.")], + outputs: vec![json_output( + "workflowRun", + "The resumed WorkflowRun (status running).", + )], + }, other => unreachable!("unknown workflow_run schema: {other}"), } } @@ -139,6 +192,85 @@ fn handle_get(params: Map) -> ControllerFuture { }) } +fn handle_start(params: Map) -> ControllerFuture { + Box::pin(async move { + let cid = new_correlation_id(); + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] start.entry"); + let config = config_rpc::load_config_with_timeout().await.inspect_err(|err| { + log::warn!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] start.config_failed err={err}"); + })?; + let definition_id = params + .get("definitionId") + .and_then(|v| v.as_str()) + .ok_or_else(|| "missing required param: definitionId".to_string())? + .to_string(); + let input = params.get("input").cloned().unwrap_or(Value::Null); + let parent_thread_id = params + .get("parentThreadId") + .and_then(|v| v.as_str()) + .map(str::to_string); + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] start.definition={definition_id}"); + let run = super::engine::start_workflow_run( + &config, + &definition_id, + input, + parent_thread_id, + ) + .await + .map_err(|e| { + let s = e.to_string(); + log::warn!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] start.error err={s}"); + s + })?; + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] start.success id={}", run.id); + to_json(serde_json::json!({ "workflowRun": run })) + }) +} + +fn handle_stop(params: Map) -> ControllerFuture { + Box::pin(async move { + let cid = new_correlation_id(); + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] stop.entry"); + let config = config_rpc::load_config_with_timeout().await.inspect_err(|err| { + log::warn!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] stop.config_failed err={err}"); + })?; + let id = params + .get("id") + .and_then(|v| v.as_str()) + .ok_or_else(|| "missing required param: id".to_string())?; + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] stop.id={id}"); + let run = super::engine::stop_workflow_run(&config, id).await.map_err(|e| { + let s = e.to_string(); + log::warn!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] stop.error id={id} err={s}"); + s + })?; + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] stop.success id={id} found={}", run.is_some()); + to_json(serde_json::json!({ "workflowRun": run })) + }) +} + +fn handle_resume(params: Map) -> ControllerFuture { + Box::pin(async move { + let cid = new_correlation_id(); + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] resume.entry"); + let config = config_rpc::load_config_with_timeout().await.inspect_err(|err| { + log::warn!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] resume.config_failed err={err}"); + })?; + let id = params + .get("id") + .and_then(|v| v.as_str()) + .ok_or_else(|| "missing required param: id".to_string())?; + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] resume.id={id}"); + let run = super::engine::resume_workflow_run(&config, id).await.map_err(|e| { + let s = e.to_string(); + log::warn!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] resume.error id={id} err={s}"); + s + })?; + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] resume.success id={}", run.id); + to_json(serde_json::json!({ "workflowRun": run })) + }) +} + fn to_json(value: T) -> Result { RpcOutcome::new(value, vec![]).into_cli_compatible_json() } @@ -174,6 +306,15 @@ fn optional_u64(name: &'static str, comment: &'static str) -> FieldSchema { } } +fn optional_json(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Option(Box::new(TypeSchema::Json)), + comment, + required: false, + } +} + fn json_output(name: &'static str, comment: &'static str) -> FieldSchema { FieldSchema { name, @@ -192,7 +333,15 @@ mod tests { let schemas = all_controller_schemas(); let registered = all_registered_controllers(); assert_eq!(schemas.len(), registered.len()); + assert_eq!( + schemas.len(), + 6, + "expected 3 read + 3 execution controllers" + ); assert!(schemas.iter().all(|s| s.namespace == "workflow_run")); assert_eq!(schema_for("workflow_run_get").function, "get"); + assert_eq!(schema_for("workflow_run_start").function, "start"); + assert_eq!(schema_for("workflow_run_stop").function, "stop"); + assert_eq!(schema_for("workflow_run_resume").function, "resume"); } } diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index 83a1ea2de..a33011ddb 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -11936,3 +11936,155 @@ async fn json_rpc_memory_sources_list_keeps_multiple_active_connections_per_tool rpc_join.abort(); composio_join.abort(); } + +// ───────────────────────────────────────────────────────────────────────────── +// Workflow run execution engine (#3375 PR2) +// +// Starts the builtin read-only "parallel research with cross-checking" workflow +// over the live JSON-RPC core with the mock backend, then polls workflow_run_get +// until the run reaches `completed` with a non-empty `summary`. Child agents are +// pinned to the mock provider via `modelOverride` so every phase resolves against +// the in-process mock `/openai/v1/chat/completions` (no live network). +// +// Self-contained: boots its own mock + core stack and tears them down. Appended +// at the END of the file per the multi-agent coordination protocol (#3370 epic). +// ───────────────────────────────────────────────────────────────────────────── + +#[test] +fn json_rpc_workflow_run_engine_executes_builtin_to_completion() { + run_json_rpc_e2e_on_agent_stack( + "json_rpc_workflow_run_engine_executes_builtin_to_completion", + json_rpc_workflow_run_engine_executes_builtin_to_completion_inner, + ); +} + +async fn json_rpc_workflow_run_engine_executes_builtin_to_completion_inner() { + let _env_lock = json_rpc_e2e_env_lock(); + let tmp = tempdir().expect("tempdir"); + let home = tmp.path(); + let openhuman_home = home.join(".openhuman"); + + let _home_guard = EnvVarGuard::set_to_path("HOME", home); + let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); + let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); + let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); + + let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; + let mock_origin = format!("http://{}", mock_addr); + + write_min_config(&openhuman_home, &mock_origin); + let user_scoped_dir = openhuman_home.join("users").join("e2e-user"); + write_min_config(&user_scoped_dir, &mock_origin); + + let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; + let rpc_base = format!("http://{}", rpc_addr); + + tokio::time::sleep(Duration::from_millis(100)).await; + + // Authenticate so the child agents' provider has a backend session. + let store = post_json_rpc( + &rpc_base, + 37_500_1, + "openhuman.auth_store_session", + json!({ "token": "e2e-test-jwt", "user_id": "e2e-user" }), + ) + .await; + assert_no_jsonrpc_error(&store, "store_session"); + + // Sanity: the builtin definition is listed. + let defs = post_json_rpc( + &rpc_base, + 37_500_2, + "openhuman.workflow_run_list_definitions", + json!({}), + ) + .await; + let defs_result = assert_no_jsonrpc_error(&defs, "list_definitions"); + let defs_body = peel_logs_envelope(defs_result.get("result").unwrap_or(defs_result)); + let definition_id = defs_body + .get("definitions") + .and_then(Value::as_array) + .and_then(|arr| arr.first()) + .and_then(|d| d.get("id")) + .and_then(Value::as_str) + .expect("at least one builtin workflow definition") + .to_string(); + assert_eq!(definition_id, "parallel_research_cross_check"); + + // Start the workflow. Children inherit the parent (root) provider via the + // `modelOverride` pin → every phase hits the mock chat-completions route. + let start = post_json_rpc( + &rpc_base, + 37_500_3, + "openhuman.workflow_run_start", + json!({ + "definitionId": definition_id, + "input": { "question": "What is the capital of France?", "modelOverride": "e2e-mock-model" }, + }), + ) + .await; + let start_result = assert_no_jsonrpc_error(&start, "workflow_run_start"); + let start_body = peel_logs_envelope(start_result.get("result").unwrap_or(start_result)); + let run = start_body + .get("workflowRun") + .expect("workflowRun in start response"); + let run_id = run + .get("id") + .and_then(Value::as_str) + .expect("run id") + .to_string(); + assert_eq!( + run.get("status").and_then(Value::as_str), + Some("running"), + "fresh run should be running: {run}" + ); + + // Poll workflow_run_get until terminal (or timeout). The four-phase builtin + // fans out 5 children total against the mock — completes quickly. + let mut final_status = String::new(); + let mut final_summary = String::new(); + for attempt in 0..120 { + tokio::time::sleep(Duration::from_millis(250)).await; + let got = post_json_rpc( + &rpc_base, + 37_500_100 + attempt, + "openhuman.workflow_run_get", + json!({ "id": run_id }), + ) + .await; + let got_result = assert_no_jsonrpc_error(&got, "workflow_run_get"); + let got_body = peel_logs_envelope(got_result.get("result").unwrap_or(got_result)); + let wf = got_body + .get("workflowRun") + .expect("workflowRun payload present"); + let status = wf + .get("status") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + if matches!( + status.as_str(), + "completed" | "failed" | "cancelled" | "interrupted" + ) { + final_status = status; + final_summary = wf + .get("summary") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + break; + } + } + + assert_eq!( + final_status, "completed", + "workflow run must reach completed; got status={final_status:?} summary={final_summary:?}" + ); + assert!( + !final_summary.trim().is_empty(), + "completed workflow run must carry a non-empty summary" + ); + + rpc_join.abort(); + mock_join.abort(); +}