admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-27 21:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(workflow-runs): live workflow execution engine (#3375 PR2) (#3600)

Browse Source
This commit is contained in:
oxoxDev
2026-06-11 12:55:28 -07:00
committed by GitHub
parent 29ce30b73e
commit ab9b39c2e0
6 changed files with 1715 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openhuman/agent_orchestration/ops.rs
+25
View File
@@ -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<String> = 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
src/openhuman/agent_orchestration/workflow_runs/engine.rs
+855
View File
@@ -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<HashMap<String, Arc<AtomicBool>>> {
static REGISTRY: OnceLock<Mutex<HashMap<String, Arc<AtomicBool>>>> = 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<AtomicBool> {
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<Arc<AtomicBool>> {
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<String>,
) -> Result<WorkflowRun> {
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<Option<WorkflowRun>> {
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<WorkflowRun> {
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<ParentExecutionContext> {
// 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<AtomicBool>,
) -> 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<Value> = Vec::new();
let mut phase_failed: Option<String> = 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<String> = 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<Value>) {
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<Value> {
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<String> {
let pick = |name: &str| -> Option<String> {
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::<Vec<_>>()
.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<String>,
status: WorkflowRunStatus,
summary: Option<String>,
terminal: bool,
) -> Result<WorkflowRun> {
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;
src/openhuman/agent_orchestration/workflow_runs/engine_tests.rs
+518
View File
@@ -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<Vec<MemoryEntry>> {
Ok(Vec::new())
}
async fn get(&self, _namespace: &str, _key: &str) -> anyhow::Result<Option<MemoryEntry>> {
Ok(None)
}
async fn list(
&self,
_namespace: Option<&str>,
_category: Option<&MemoryCategory>,
_session_id: Option<&str>,
) -> anyhow::Result<Vec<MemoryEntry>> {
Ok(Vec::new())
}
async fn forget(&self, _namespace: &str, _key: &str) -> anyhow::Result<bool> {
Ok(false)
}
async fn namespace_summaries(&self) -> anyhow::Result<Vec<NamespaceSummary>> {
Ok(Vec::new())
}
async fn count(&self) -> anyhow::Result<usize> {
Ok(0)
}
async fn health_check(&self) -> bool {
true
}
fn name(&self) -> &str {
"noop"
}
}
fn text_response(text: impl Into<String>) -> 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<AtomicUsize>,
active: Arc<AtomicUsize>,
max_active: Arc<AtomicUsize>,
prompts: Arc<Mutex<Vec<String>>>,
fail_on: Arc<Mutex<Option<String>>>,
}
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<String> {
Ok("ok".to_string())
}
async fn chat(
&self,
request: ChatRequest<'_>,
_model: &str,
_temperature: f64,
) -> anyhow::Result<ChatResponse> {
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::<Vec<_>>()
.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<dyn Provider>) -> ParentExecutionContext {
ParentExecutionContext {
agent_definition_id: "workflow_engine".to_string(),
allowed_subagent_ids: HashSet::new(),
provider,
all_tools: Arc::new(Vec::<Box<dyn Tool>>::new()),
all_tool_specs: Arc::new(Vec::<ToolSpec>::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"
);
}
src/openhuman/agent_orchestration/workflow_runs/mod.rs
+12 -6
View File
@@ -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,
src/openhuman/agent_orchestration/workflow_runs/schemas.rs
+153 -4
View File
@@ -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<ControllerSchema> {
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<RegisteredController> {
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<String, Value>) -> ControllerFuture {
})
}
fn handle_start(params: Map<String, Value>) -> 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<String, Value>) -> 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<String, Value>) -> 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<T: serde::Serialize>(value: T) -> Result<Value, String> {
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");
}
}
tests/json_rpc_e2e.rs
+152
View File
@@ -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();
}