diff --git a/src/core/all.rs b/src/core/all.rs index 1d8399c27..d2d1e0f25 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -296,6 +296,9 @@ fn build_registered_controllers() -> Vec { // Background agent command center — read-only grouped view over the run ledger controllers .extend(crate::openhuman::agent_orchestration::all_command_center_registered_controllers()); + // Durable dynamic workflow runs — definitions + read surface over the run ledger + controllers + .extend(crate::openhuman::agent_orchestration::all_workflow_run_registered_controllers()); controllers } @@ -424,6 +427,8 @@ fn build_declared_controller_schemas() -> Vec { schemas.extend(crate::openhuman::session_db::all_session_db_controller_schemas()); // Background agent command center schemas.extend(crate::openhuman::agent_orchestration::all_command_center_controller_schemas()); + // Durable dynamic workflow runs + schemas.extend(crate::openhuman::agent_orchestration::all_workflow_run_controller_schemas()); schemas } @@ -510,6 +515,9 @@ pub fn namespace_description(namespace: &str) -> Option<&'static str> { "agent_work" => Some( "Background agent command center — recent agent runs grouped by status (needs-input, working, completed, failed, stopped).", ), + "workflow_run" => Some( + "Durable dynamic workflow runs — declarative multi-agent definitions and the read surface over persisted runs.", + ), "billing" => Some("Subscription plan, payment links, and credit top-up via the backend."), "team" => Some("Team member management, invites, and role changes via the backend."), "tool_registry" => Some( diff --git a/src/openhuman/agent_orchestration/mod.rs b/src/openhuman/agent_orchestration/mod.rs index 191d32a65..58c6fea84 100644 --- a/src/openhuman/agent_orchestration/mod.rs +++ b/src/openhuman/agent_orchestration/mod.rs @@ -9,6 +9,7 @@ pub mod command_center; mod ops; pub mod tools; pub mod types; +pub mod workflow_runs; #[cfg(test)] mod ops_tests; @@ -22,3 +23,6 @@ pub use types::{ FollowUpRequest, MessageAgentRequest, ResumeAgentRequest, SpawnAgentRequest, SpawnAgentResponse, WaitAgentOptions, WaitAgentResponse, }; +pub use workflow_runs::{ + all_workflow_run_controller_schemas, all_workflow_run_registered_controllers, +}; diff --git a/src/openhuman/agent_orchestration/workflow_runs/mod.rs b/src/openhuman/agent_orchestration/workflow_runs/mod.rs new file mode 100644 index 000000000..5f1e88969 --- /dev/null +++ b/src/openhuman/agent_orchestration/workflow_runs/mod.rs @@ -0,0 +1,34 @@ +//! Durable dynamic workflow runs (issue #3375). +//! +//! A first-class, repeatable multi-agent orchestration model: a declarative +//! [`WorkflowDefinition`] (phase graph) coordinates many child agents, and each +//! run's durable state lives in `session_db::run_ledger` (the `workflow_runs` +//! 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. +//! +//! Namespace note: this is distinct from the existing `workflows` domain, which +//! handles SKILL.md / WORKFLOW.md bundle discovery. + +mod ops; +mod schemas; +pub mod types; + +pub use ops::{ + builtin_definitions, definition_by_id, get_run, list_definitions, list_runs, + validate_definition, validate_structure, PARALLEL_RESEARCH_ID, +}; +pub use schemas::{ + all_controller_schemas as all_workflow_run_controller_schemas, + all_registered_controllers as all_workflow_run_registered_controllers, +}; +pub use types::{ + DefinitionError, WorkflowDefinition, WorkflowDefinitionListResponse, WorkflowPhase, + WorkflowSafetyTier, +}; diff --git a/src/openhuman/agent_orchestration/workflow_runs/ops.rs b/src/openhuman/agent_orchestration/workflow_runs/ops.rs new file mode 100644 index 000000000..017a4e2b9 --- /dev/null +++ b/src/openhuman/agent_orchestration/workflow_runs/ops.rs @@ -0,0 +1,423 @@ +//! Workflow-definition catalog + read surface over durable workflow runs. +//! +//! PR1 scope: expose the builtin [`WorkflowDefinition`]s, validate them +//! (structure + agent existence), and read durable [`WorkflowRun`]s from +//! `session_db::run_ledger`. No execution engine yet — starting / stopping / +//! resuming runs lands in a follow-up PR. + +use std::collections::{HashMap, HashSet, VecDeque}; + +use anyhow::Result; + +use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry; +use crate::openhuman::config::Config; +use crate::openhuman::session_db::run_ledger::{ + get_workflow_run, list_workflow_runs, WorkflowRun, WorkflowRunListRequest, + WorkflowRunListResponse, +}; + +use super::types::{ + DefinitionError, WorkflowDefinition, WorkflowDefinitionListResponse, WorkflowPhase, + WorkflowSafetyTier, +}; + +/// Id of the first shipped (read-only) workflow. +pub const PARALLEL_RESEARCH_ID: &str = "parallel_research_cross_check"; + +/// All builtin workflow definitions. +/// +/// First (and only) shipped workflow is a read-only "parallel research with +/// cross-checking" pipeline: decompose the question, fan out researchers, +/// have a critic cross-check the claims, then synthesize a cited report. +pub fn builtin_definitions() -> Vec { + vec![WorkflowDefinition { + id: PARALLEL_RESEARCH_ID.to_string(), + name: "Parallel research with cross-checking".to_string(), + description: "Decompose a question into angles, research them in parallel, cross-check \ + the claims with a critic, then synthesize a cited report. Read-only." + .to_string(), + phases: vec![ + WorkflowPhase { + name: "decompose".to_string(), + description: "Break the question into independent research angles.".to_string(), + agent_ids: vec!["planner".to_string()], + depends_on: vec![], + }, + WorkflowPhase { + name: "research".to_string(), + description: "Research each angle in parallel.".to_string(), + agent_ids: vec!["researcher".to_string(), "researcher".to_string()], + depends_on: vec!["decompose".to_string()], + }, + WorkflowPhase { + name: "cross_check".to_string(), + description: "Adversarially cross-check the gathered claims.".to_string(), + agent_ids: vec!["critic".to_string()], + depends_on: vec!["research".to_string()], + }, + WorkflowPhase { + name: "synthesize".to_string(), + description: "Synthesize a single cited report.".to_string(), + agent_ids: vec!["summarizer".to_string()], + depends_on: vec!["cross_check".to_string()], + }, + ], + default_concurrency: 2, + max_children: 8, + safety_tier: WorkflowSafetyTier::ReadOnly, + }] +} + +/// Look up one builtin definition by id. +pub fn definition_by_id(id: &str) -> Option { + builtin_definitions().into_iter().find(|d| d.id == id) +} + +/// List available workflow definitions (builtins for now). +pub fn list_definitions() -> WorkflowDefinitionListResponse { + let definitions = builtin_definitions(); + WorkflowDefinitionListResponse { + count: definitions.len(), + definitions, + } +} + +/// Validate a definition's structure (registry-independent). +/// +/// Checks: at least one phase; unique phase names; non-empty phases; +/// `depends_on` references existing phases; no dependency cycles. +pub fn validate_structure(def: &WorkflowDefinition) -> Vec { + log::debug!( + target: "workflow_run", + "[workflow_run] validate_structure.entry id={} phases={}", + def.id, + def.phases.len() + ); + let mut errors = Vec::new(); + if def.phases.is_empty() { + errors.push(DefinitionError::NoPhases); + log::debug!(target: "workflow_run", "[workflow_run] validate_structure.exit id={} errors=1 reason=no_phases", def.id); + return errors; + } + + let mut seen: HashSet<&str> = HashSet::new(); + for phase in &def.phases { + if !seen.insert(phase.name.as_str()) { + errors.push(DefinitionError::DuplicatePhase { + name: phase.name.clone(), + }); + } + if phase.agent_ids.is_empty() { + errors.push(DefinitionError::EmptyPhase { + phase: phase.name.clone(), + }); + } + } + + let names: HashSet<&str> = def.phases.iter().map(|p| p.name.as_str()).collect(); + for phase in &def.phases { + for dep in &phase.depends_on { + if !names.contains(dep.as_str()) { + errors.push(DefinitionError::UnknownDependency { + phase: phase.name.clone(), + depends_on: dep.clone(), + }); + } + } + } + + if has_cycle(def) { + errors.push(DefinitionError::CyclicDependency); + } + + if def.default_concurrency == 0 || def.max_children == 0 { + errors.push(DefinitionError::InvalidConcurrency { + default_concurrency: def.default_concurrency, + max_children: def.max_children, + }); + } + + log::debug!( + target: "workflow_run", + "[workflow_run] validate_structure.exit id={} errors={}", + def.id, + errors.len() + ); + errors +} + +/// Validate that every agent referenced by a definition is resolvable through +/// the provided lookup. Kept generic so it is testable without the global +/// registry. +pub fn validate_agents(def: &WorkflowDefinition, is_known: F) -> Vec +where + F: Fn(&str) -> bool, +{ + log::debug!( + target: "workflow_run", + "[workflow_run] validate_agents.entry id={} phases={}", + def.id, + def.phases.len() + ); + let mut errors = Vec::new(); + for phase in &def.phases { + for agent_id in &phase.agent_ids { + if !is_known(agent_id) { + errors.push(DefinitionError::UnknownAgent { + phase: phase.name.clone(), + agent_id: agent_id.clone(), + }); + } + } + } + log::debug!( + target: "workflow_run", + "[workflow_run] validate_agents.exit id={} unknown={}", + def.id, + errors.len() + ); + errors +} + +/// Full validation against the live agent registry. +/// +/// Always runs the structural checks. Agent-existence checks run only when the +/// registry is initialized, so callers in a registry-less context (e.g. early +/// boot, some tests) are not given false `UnknownAgent` errors. +pub fn validate_definition(def: &WorkflowDefinition) -> Vec { + log::debug!(target: "workflow_run", "[workflow_run] validate_definition.entry id={}", def.id); + let mut errors = validate_structure(def); + match AgentDefinitionRegistry::global() { + Some(registry) => { + errors.extend(validate_agents(def, |id| registry.get(id).is_some())); + } + None => { + log::debug!( + target: "workflow_run", + "[workflow_run][registry] validate_definition.skip_agents id={} reason=registry_uninitialized", + def.id + ); + } + } + log::debug!( + target: "workflow_run", + "[workflow_run] validate_definition.exit id={} errors={}", + def.id, + errors.len() + ); + errors +} + +/// Kahn's-algorithm cycle check over the phase dependency graph. Edges that +/// point at unknown phases are ignored here (reported separately). +fn has_cycle(def: &WorkflowDefinition) -> bool { + let names: HashSet<&str> = def.phases.iter().map(|p| p.name.as_str()).collect(); + let mut indegree: HashMap<&str, usize> = + def.phases.iter().map(|p| (p.name.as_str(), 0)).collect(); + let mut adjacency: HashMap<&str, Vec<&str>> = HashMap::new(); + for phase in &def.phases { + for dep in &phase.depends_on { + let dep = dep.as_str(); + if names.contains(dep) { + // edge dep -> phase + adjacency.entry(dep).or_default().push(phase.name.as_str()); + *indegree.entry(phase.name.as_str()).or_insert(0) += 1; + } + } + } + + let mut queue: VecDeque<&str> = indegree + .iter() + .filter(|(_, &d)| d == 0) + .map(|(&n, _)| n) + .collect(); + let mut visited = 0usize; + while let Some(node) = queue.pop_front() { + visited += 1; + if let Some(children) = adjacency.get(node) { + for &child in children { + let entry = indegree.get_mut(child).expect("child in indegree"); + *entry -= 1; + if *entry == 0 { + queue.push_back(child); + } + } + } + } + // Compare against the unique-node count, not `def.phases.len()`: the graph + // is keyed by unique phase names, so duplicate names (reported separately as + // `DuplicatePhase`) would otherwise trip a false `CyclicDependency`. + visited != indegree.len() +} + +/// List durable workflow runs (delegates to the run ledger). +pub fn list_runs( + config: &Config, + request: &WorkflowRunListRequest, +) -> Result { + log::debug!( + target: "workflow_run", + "[workflow_run] list_runs.entry definition={:?} status={:?}", + request.definition_id, + request.status + ); + list_workflow_runs(config, request) +} + +/// Get one durable workflow run by id (delegates to the run ledger). +pub fn get_run(config: &Config, id: &str) -> Result> { + log::debug!(target: "workflow_run", "[workflow_run] get_run.entry id={id}"); + get_workflow_run(config, id) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn good_def() -> WorkflowDefinition { + definition_by_id(PARALLEL_RESEARCH_ID).expect("builtin present") + } + + #[test] + fn builtin_is_structurally_valid() { + assert!(validate_structure(&good_def()).is_empty()); + } + + #[test] + fn builtin_agents_pass_when_all_known() { + // Treat the four referenced agents as registered. + let known = ["planner", "researcher", "critic", "summarizer"]; + let errors = validate_agents(&good_def(), |id| known.contains(&id)); + assert!(errors.is_empty(), "unexpected: {errors:?}"); + } + + #[test] + fn unknown_agent_is_reported() { + let errors = validate_agents(&good_def(), |id| id == "researcher"); + // planner, critic, summarizer are unknown -> 3 errors. + assert_eq!(errors.len(), 3); + assert!(errors.iter().any( + |e| matches!(e, DefinitionError::UnknownAgent { agent_id, .. } if agent_id == "planner") + )); + } + + #[test] + fn no_phases_is_rejected() { + let mut def = good_def(); + def.phases.clear(); + assert_eq!(validate_structure(&def), vec![DefinitionError::NoPhases]); + } + + #[test] + fn duplicate_and_empty_phase_are_reported() { + let def = WorkflowDefinition { + phases: vec![ + WorkflowPhase { + name: "a".into(), + description: String::new(), + agent_ids: vec!["researcher".into()], + depends_on: vec![], + }, + WorkflowPhase { + name: "a".into(), + description: String::new(), + agent_ids: vec![], + depends_on: vec![], + }, + ], + ..good_def() + }; + let errors = validate_structure(&def); + assert!(errors.contains(&DefinitionError::DuplicatePhase { name: "a".into() })); + assert!(errors.contains(&DefinitionError::EmptyPhase { phase: "a".into() })); + } + + #[test] + fn unknown_dependency_is_reported() { + let def = WorkflowDefinition { + phases: vec![WorkflowPhase { + name: "only".into(), + description: String::new(), + agent_ids: vec!["researcher".into()], + depends_on: vec!["ghost".into()], + }], + ..good_def() + }; + let errors = validate_structure(&def); + assert!(errors.contains(&DefinitionError::UnknownDependency { + phase: "only".into(), + depends_on: "ghost".into(), + })); + } + + #[test] + fn cycle_is_detected() { + let def = WorkflowDefinition { + phases: vec![ + WorkflowPhase { + name: "a".into(), + description: String::new(), + agent_ids: vec!["researcher".into()], + depends_on: vec!["b".into()], + }, + WorkflowPhase { + name: "b".into(), + description: String::new(), + agent_ids: vec!["researcher".into()], + depends_on: vec!["a".into()], + }, + ], + ..good_def() + }; + assert!(validate_structure(&def).contains(&DefinitionError::CyclicDependency)); + } + + #[test] + fn duplicate_phase_names_do_not_report_false_cycle() { + let def = WorkflowDefinition { + phases: vec![ + WorkflowPhase { + name: "a".into(), + description: String::new(), + agent_ids: vec!["researcher".into()], + depends_on: vec![], + }, + WorkflowPhase { + name: "a".into(), + description: String::new(), + agent_ids: vec!["researcher".into()], + depends_on: vec![], + }, + ], + ..good_def() + }; + let errors = validate_structure(&def); + assert!(errors.contains(&DefinitionError::DuplicatePhase { name: "a".into() })); + assert!( + !errors.contains(&DefinitionError::CyclicDependency), + "duplicate names must not trip a false cycle: {errors:?}" + ); + } + + #[test] + fn zero_concurrency_is_rejected() { + let def = WorkflowDefinition { + default_concurrency: 0, + max_children: 0, + ..good_def() + }; + assert!( + validate_structure(&def).contains(&DefinitionError::InvalidConcurrency { + default_concurrency: 0, + max_children: 0, + }) + ); + } + + #[test] + fn list_definitions_returns_builtins() { + let resp = list_definitions(); + assert_eq!(resp.count, 1); + assert_eq!(resp.definitions[0].id, PARALLEL_RESEARCH_ID); + } +} diff --git a/src/openhuman/agent_orchestration/workflow_runs/schemas.rs b/src/openhuman/agent_orchestration/workflow_runs/schemas.rs new file mode 100644 index 000000000..0d8248353 --- /dev/null +++ b/src/openhuman/agent_orchestration/workflow_runs/schemas.rs @@ -0,0 +1,198 @@ +//! 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. + +use serde_json::{Map, Value}; + +use crate::core::all::{ControllerFuture, RegisteredController}; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; +use crate::openhuman::config::rpc as config_rpc; +use crate::openhuman::session_db::run_ledger::WorkflowRunListRequest; +use crate::rpc::RpcOutcome; + +/// Controller schemas exposed by the workflow-runs module. +pub fn all_controller_schemas() -> Vec { + vec![ + schema_for("workflow_run_list_definitions"), + schema_for("workflow_run_list"), + schema_for("workflow_run_get"), + ] +} + +/// Registered controllers (schema + handler) for workflow runs. +pub fn all_registered_controllers() -> Vec { + vec![ + RegisteredController { + schema: schema_for("workflow_run_list_definitions"), + handler: handle_list_definitions, + }, + RegisteredController { + schema: schema_for("workflow_run_list"), + handler: handle_list, + }, + RegisteredController { + schema: schema_for("workflow_run_get"), + handler: handle_get, + }, + ] +} + +fn schema_for(function: &str) -> ControllerSchema { + match function { + "workflow_run_list_definitions" => ControllerSchema { + namespace: "workflow_run", + function: "list_definitions", + description: "List available declarative workflow definitions (builtins).", + inputs: vec![], + outputs: vec![json_output( + "result", + "WorkflowDefinitionListResponse with definitions and count.", + )], + }, + "workflow_run_list" => ControllerSchema { + namespace: "workflow_run", + function: "list", + description: "List durable workflow runs with optional filters and pagination.", + inputs: vec![ + optional_str("definitionId", "Filter by workflow definition id."), + optional_str("status", "Filter by run status."), + optional_str("parentThreadId", "Filter by parent thread id."), + optional_u64("limit", "Max runs to return (default 50, max 500)."), + optional_u64("offset", "Pagination offset."), + ], + outputs: vec![json_output( + "result", + "WorkflowRunListResponse with runs array and count.", + )], + }, + "workflow_run_get" => ControllerSchema { + namespace: "workflow_run", + function: "get", + description: "Get a durable workflow run by id.", + inputs: vec![required_str("id", "Workflow run id.")], + outputs: vec![json_output("workflowRun", "WorkflowRun payload or null.")], + }, + other => unreachable!("unknown workflow_run schema: {other}"), + } +} + +fn handle_list_definitions(_params: Map) -> ControllerFuture { + Box::pin(async move { + let cid = new_correlation_id(); + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] list_definitions.entry"); + let response = super::ops::list_definitions(); + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] list_definitions.success count={}", response.count); + to_json(response) + }) +} + +fn handle_list(params: Map) -> ControllerFuture { + Box::pin(async move { + let cid = new_correlation_id(); + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] list.entry"); + let config = config_rpc::load_config_with_timeout().await.inspect_err(|err| { + log::warn!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] list.config_failed err={err}"); + })?; + let request: WorkflowRunListRequest = if params.is_empty() { + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] list.branch=default_request"); + WorkflowRunListRequest::default() + } else { + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] list.branch=parsed_params"); + serde_json::from_value(Value::Object(params)).map_err(|e| { + let s = format!("invalid workflow run list params: {e}"); + log::warn!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] list.bad_params err={s}"); + s + })? + }; + let response = super::ops::list_runs(&config, &request).map_err(|e| { + let s = e.to_string(); + log::warn!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] list.error err={s}"); + s + })?; + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] list.success count={}", response.count); + to_json(response) + }) +} + +fn handle_get(params: Map) -> ControllerFuture { + Box::pin(async move { + let cid = new_correlation_id(); + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] get.entry"); + let config = config_rpc::load_config_with_timeout().await.inspect_err(|err| { + log::warn!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] get.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}] get.id={id}"); + let run = super::ops::get_run(&config, id).map_err(|e| { + let s = e.to_string(); + log::warn!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] get.error id={id} err={s}"); + s + })?; + log::debug!(target: "workflow_run_rpc", "[workflow_run_rpc][{cid}] get.success id={id} found={}", run.is_some()); + to_json(serde_json::json!({ "workflowRun": run })) + }) +} + +fn to_json(value: T) -> Result { + RpcOutcome::new(value, vec![]).into_cli_compatible_json() +} + +fn new_correlation_id() -> String { + uuid::Uuid::new_v4().simple().to_string()[..8].to_string() +} + +fn required_str(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::String, + comment, + required: true, + } +} + +fn optional_str(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Option(Box::new(TypeSchema::String)), + comment, + required: false, + } +} + +fn optional_u64(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Option(Box::new(TypeSchema::U64)), + comment, + required: false, + } +} + +fn json_output(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Json, + comment, + required: true, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn registered_controllers_match_schemas() { + let schemas = all_controller_schemas(); + let registered = all_registered_controllers(); + assert_eq!(schemas.len(), registered.len()); + assert!(schemas.iter().all(|s| s.namespace == "workflow_run")); + assert_eq!(schema_for("workflow_run_get").function, "get"); + } +} diff --git a/src/openhuman/agent_orchestration/workflow_runs/types.rs b/src/openhuman/agent_orchestration/workflow_runs/types.rs new file mode 100644 index 000000000..60c128ce7 --- /dev/null +++ b/src/openhuman/agent_orchestration/workflow_runs/types.rs @@ -0,0 +1,110 @@ +//! Declarative workflow-definition types for durable multi-agent runs (#3375). +//! +//! A [`WorkflowDefinition`] is a static, declarative phase graph — NOT an +//! arbitrary script. Each phase names the agents it fans out to and the phases +//! it depends on, so the runtime (added in a follow-up PR) can schedule phases +//! in dependency order with bounded concurrency. Keeping definitions +//! declarative avoids the security surface of arbitrary in-process script +//! execution. +//! +//! This PR ships the definition model + the read surface (list definitions, +//! list/get durable runs from `session_db::run_ledger`). The live execution +//! engine is deferred to a follow-up. + +use serde::Serialize; + +/// Safety tier of a workflow — governs what its child agents may do. +/// +/// Only [`WorkflowSafetyTier::ReadOnly`] workflows ship first; edit-capable +/// tiers gate on explicit user approval once the engine lands. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum WorkflowSafetyTier { + /// Child agents may only read / research — no file writes, no external + /// side effects beyond fetching. + ReadOnly, + /// Child agents may take standard non-destructive actions. + Standard, + /// Child agents may edit files (requires worktree isolation + approval). + EditCapable, +} + +impl WorkflowSafetyTier { + /// Stable wire string. + pub fn as_str(self) -> &'static str { + match self { + WorkflowSafetyTier::ReadOnly => "read_only", + WorkflowSafetyTier::Standard => "standard", + WorkflowSafetyTier::EditCapable => "edit_capable", + } + } +} + +/// One phase of a workflow: a set of agents fanned out in parallel once every +/// phase it `depends_on` has completed. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct WorkflowPhase { + /// Phase name, unique within a definition; referenced by `depends_on`. + pub name: String, + /// Human-readable purpose of the phase. + pub description: String, + /// Agent definition ids spawned (in parallel) during this phase. + pub agent_ids: Vec, + /// Names of phases that must complete before this one starts. + pub depends_on: Vec, +} + +/// A declarative, repeatable workflow definition. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct WorkflowDefinition { + /// Stable definition id (e.g. `parallel_research_cross_check`). + pub id: String, + /// Display name. + pub name: String, + /// What the workflow does and when to use it. + pub description: String, + /// Ordered list of phases (dependency edges encoded via `depends_on`). + pub phases: Vec, + /// Default max agents run concurrently within a phase. + pub default_concurrency: u32, + /// Hard cap on total child agents across the whole run. + pub max_children: u32, + /// Safety tier (drives approval + isolation policy at run time). + pub safety_tier: WorkflowSafetyTier, +} + +/// Response wrapper for the list-definitions controller. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct WorkflowDefinitionListResponse { + /// Available workflow definitions (builtins for now). + pub definitions: Vec, + /// Total count. + pub count: usize, +} + +/// A validation problem found in a [`WorkflowDefinition`]. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub enum DefinitionError { + /// A phase referenced an agent id not present in the registry. + UnknownAgent { phase: String, agent_id: String }, + /// A `depends_on` named a phase that does not exist. + UnknownDependency { phase: String, depends_on: String }, + /// Two phases share the same name. + DuplicatePhase { name: String }, + /// A phase has no agents. + EmptyPhase { phase: String }, + /// The dependency graph contains a cycle. + CyclicDependency, + /// The definition declares no phases. + NoPhases, + /// `default_concurrency` or `max_children` is zero — an executor would + /// deadlock or trivially fail to launch any child. + InvalidConcurrency { + default_concurrency: u32, + max_children: u32, + }, +} diff --git a/src/openhuman/session_db/run_ledger/mod.rs b/src/openhuman/session_db/run_ledger/mod.rs index 65fffc0e4..4c970c75e 100644 --- a/src/openhuman/session_db/run_ledger/mod.rs +++ b/src/openhuman/session_db/run_ledger/mod.rs @@ -10,11 +10,12 @@ pub mod store; pub mod types; pub use ops::{ - append_run_event, get_agent_run, list_agent_runs, list_recent_run_events, upsert_agent_run, - upsert_run_telemetry, upsert_workflow_run, + append_run_event, get_agent_run, get_workflow_run, list_agent_runs, list_recent_run_events, + list_workflow_runs, upsert_agent_run, upsert_run_telemetry, upsert_workflow_run, }; pub use types::{ AgentRun, AgentRunKind, AgentRunListRequest, AgentRunListResponse, AgentRunStatus, AgentRunUpsert, RunEvent, RunEventAppend, RunEventListRequest, RunEventListResponse, - RunTelemetry, RunTelemetryUpsert, WorkflowRun, WorkflowRunStatus, WorkflowRunUpsert, + RunTelemetry, RunTelemetryUpsert, WorkflowRun, WorkflowRunListRequest, WorkflowRunListResponse, + WorkflowRunStatus, WorkflowRunUpsert, }; diff --git a/src/openhuman/session_db/run_ledger/ops.rs b/src/openhuman/session_db/run_ledger/ops.rs index 407ad20b3..a28e48278 100644 --- a/src/openhuman/session_db/run_ledger/ops.rs +++ b/src/openhuman/session_db/run_ledger/ops.rs @@ -9,7 +9,7 @@ use super::store::init_run_ledger_schema; use super::types::{ AgentRun, AgentRunListRequest, AgentRunListResponse, AgentRunStatus, AgentRunUpsert, RunEvent, RunEventAppend, RunEventListRequest, RunEventListResponse, RunTelemetry, RunTelemetryUpsert, - WorkflowRun, WorkflowRunUpsert, + WorkflowRun, WorkflowRunListRequest, WorkflowRunListResponse, WorkflowRunUpsert, }; const LOG_PREFIX: &str = "[session_db:run_ledger]"; @@ -322,7 +322,8 @@ pub fn list_recent_run_events( }) } -fn get_workflow_run(config: &Config, id: &str) -> Result> { +pub fn get_workflow_run(config: &Config, id: &str) -> Result> { + log::debug!("{LOG_PREFIX} get_workflow_run.entry id={id}"); crate::openhuman::session_db::store::with_connection(config, |conn| { init_run_ledger_schema(conn)?; let mut stmt = conn.prepare( @@ -330,9 +331,100 @@ fn get_workflow_run(config: &Config, id: &str) -> Result> { child_run_ids_json, status, summary, started_at, updated_at, completed_at FROM workflow_runs WHERE id = ?1", )?; - Ok(stmt + let run = stmt .query_row(params![id], map_workflow_run_row) - .optional()?) + .optional()?; + log::debug!( + "{LOG_PREFIX} get_workflow_run.exit id={id} found={}", + run.is_some() + ); + Ok(run) + }) +} + +/// List durable workflow runs, most-recently-updated first, with optional +/// filters (definition id, status, parent thread) and pagination. Mirrors +/// [`list_agent_runs`] for the workflow_runs table. +pub fn list_workflow_runs( + config: &Config, + request: &WorkflowRunListRequest, +) -> Result { + log::debug!( + "{LOG_PREFIX} list_workflow_runs.entry definition={:?} status={:?} parent_thread={:?} limit={:?} offset={:?}", + request.definition_id, + request.status, + request.parent_thread_id, + request.limit, + request.offset + ); + crate::openhuman::session_db::store::with_connection(config, |conn| { + init_run_ledger_schema(conn)?; + let mut where_clauses = Vec::new(); + let mut values: Vec> = Vec::new(); + + if let Some(definition) = request + .definition_id + .as_deref() + .filter(|s| !s.trim().is_empty()) + { + values.push(Box::new(definition.to_string())); + where_clauses.push(format!("definition_id = ?{}", values.len())); + } + if let Some(status) = request.status.as_deref().filter(|s| !s.trim().is_empty()) { + values.push(Box::new(status.to_string())); + where_clauses.push(format!("status = ?{}", values.len())); + } + if let Some(thread) = request + .parent_thread_id + .as_deref() + .filter(|s| !s.trim().is_empty()) + { + values.push(Box::new(thread.to_string())); + where_clauses.push(format!("parent_thread_id = ?{}", values.len())); + } + + let where_sql = if where_clauses.is_empty() { + String::new() + } else { + format!("WHERE {}", where_clauses.join(" AND ")) + }; + let count_sql = format!("SELECT COUNT(*) FROM workflow_runs {where_sql}"); + let params_ref: Vec<&dyn rusqlite::types::ToSql> = + values.iter().map(|v| v.as_ref()).collect(); + let count = conn.query_row(&count_sql, params_ref.as_slice(), |row| { + row.get::<_, i64>(0) + })? as usize; + + let limit = request.limit.unwrap_or(50).min(500) as i64; + // `offset` is `u64`; convert checked so a value > i64::MAX surfaces a + // clear error instead of wrapping negative and corrupting pagination. + let offset = i64::try_from(request.offset.unwrap_or(0)) + .context("workflow run list offset exceeds i64::MAX")?; + values.push(Box::new(limit)); + let limit_idx = values.len(); + values.push(Box::new(offset)); + let offset_idx = values.len(); + + let query_sql = format!( + "SELECT id, definition_id, parent_thread_id, input_json, phase_states_json, + child_run_ids_json, status, summary, started_at, updated_at, completed_at + FROM workflow_runs {where_sql} + ORDER BY updated_at DESC + LIMIT ?{limit_idx} OFFSET ?{offset_idx}" + ); + let params_ref: Vec<&dyn rusqlite::types::ToSql> = + values.iter().map(|v| v.as_ref()).collect(); + let mut stmt = conn.prepare(&query_sql)?; + let rows = stmt.query_map(params_ref.as_slice(), map_workflow_run_row)?; + let mut runs = Vec::new(); + for row in rows { + runs.push(row?); + } + log::debug!( + "{LOG_PREFIX} list_workflow_runs.exit count={count} returned={}", + runs.len() + ); + Ok(WorkflowRunListResponse { runs, count }) }) } diff --git a/src/openhuman/session_db/run_ledger/types.rs b/src/openhuman/session_db/run_ledger/types.rs index 5066fca23..686d3e2cb 100644 --- a/src/openhuman/session_db/run_ledger/types.rs +++ b/src/openhuman/session_db/run_ledger/types.rs @@ -263,6 +263,30 @@ pub struct AgentRunListResponse { pub count: usize, } +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct WorkflowRunListRequest { + #[serde(default)] + pub definition_id: Option, + #[serde(default)] + pub status: Option, + #[serde(default)] + pub parent_thread_id: Option, + /// `u64` to match the `TypeSchema::U64` the controller advertises (the RPC + /// scalar-coercion layer only handles `U64`). Capped at 500 in `list_workflow_runs`. + #[serde(default)] + pub limit: Option, + #[serde(default)] + pub offset: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct WorkflowRunListResponse { + pub runs: Vec, + pub count: usize, +} + #[derive(Debug, Clone, Default, Deserialize)] #[serde(rename_all = "camelCase")] pub struct RunEventListRequest { diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index 40d440a7b..600611e4f 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -3090,6 +3090,111 @@ async fn json_rpc_agent_work_list_groups_runs_by_bucket() { rpc_join.abort(); } +#[tokio::test] +async fn json_rpc_workflow_run_definitions_and_runs_roundtrip() { + 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_url_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); + let _api_url_guard = EnvVarGuard::unset("OPENHUMAN_API_URL"); + + let (api_addr, api_join) = serve_on_ephemeral(mock_upstream_router()).await; + let api_origin = format!("http://{api_addr}"); + write_min_config(openhuman_home.as_path(), &api_origin); + + let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; + let rpc_base = format!("http://{rpc_addr}"); + + let config = openhuman_core::openhuman::config::Config::load_or_init() + .await + .expect("load config"); + + // Builtin definitions are available with no seeding. + let defs = post_json_rpc( + &rpc_base, + 9141, + "openhuman.workflow_run_list_definitions", + json!({}), + ) + .await; + let defs_outer = assert_no_jsonrpc_error(&defs, "workflow_run_list_definitions"); + assert_eq!( + defs_outer.get("count").and_then(serde_json::Value::as_u64), + Some(1) + ); + assert_eq!( + defs_outer + .get("definitions") + .and_then(|d| d.get(0)) + .and_then(|d| d.get("id")) + .and_then(serde_json::Value::as_str), + Some("parallel_research_cross_check") + ); + + // Seed a durable workflow run, then list + get it. + openhuman_core::openhuman::session_db::run_ledger::upsert_workflow_run( + &config, + openhuman_core::openhuman::session_db::run_ledger::WorkflowRunUpsert { + id: "wf-run-1".to_string(), + definition_id: "parallel_research_cross_check".to_string(), + parent_thread_id: Some("thread-wf-1".to_string()), + input: json!({ "question": "test" }), + phase_states: json!({ "decompose": "completed" }), + child_run_ids: vec!["child-1".to_string()], + status: openhuman_core::openhuman::session_db::run_ledger::WorkflowRunStatus::Running, + summary: None, + started_at: None, + completed_at: None, + }, + ) + .expect("seed workflow run"); + + let list = post_json_rpc( + &rpc_base, + 9142, + "openhuman.workflow_run_list", + json!({ "definitionId": "parallel_research_cross_check" }), + ) + .await; + let list_outer = assert_no_jsonrpc_error(&list, "workflow_run_list"); + assert_eq!( + list_outer.get("count").and_then(serde_json::Value::as_u64), + Some(1) + ); + assert_eq!( + list_outer + .get("runs") + .and_then(|r| r.get(0)) + .and_then(|r| r.get("status")) + .and_then(serde_json::Value::as_str), + Some("running") + ); + + let get = post_json_rpc( + &rpc_base, + 9143, + "openhuman.workflow_run_get", + json!({ "id": "wf-run-1" }), + ) + .await; + let get_outer = assert_no_jsonrpc_error(&get, "workflow_run_get"); + assert_eq!( + get_outer + .get("workflowRun") + .and_then(|r| r.get("definitionId")) + .and_then(serde_json::Value::as_str), + Some("parallel_research_cross_check") + ); + + api_join.abort(); + rpc_join.abort(); +} + #[tokio::test] async fn json_rpc_task_board_brief_roundtrips_across_todos_and_threads_rpc() { let _env_lock = json_rpc_e2e_env_lock();