feat(agent_orchestration): durable workflow-run model + read surface (#3375) (#3500)

src/core/all.rs

@@ -296,6 +296,9 @@ fn build_registered_controllers() -> Vec<RegisteredController> {
     // 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<ControllerSchema> {
     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(

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,
+};

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,
+};

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<WorkflowDefinition> {
+    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<WorkflowDefinition> {
+    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<DefinitionError> {
+    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<F>(def: &WorkflowDefinition, is_known: F) -> Vec<DefinitionError>
+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<DefinitionError> {
+    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<WorkflowRunListResponse> {
+    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<Option<WorkflowRun>> {
+    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);
+    }
+}

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<ControllerSchema> {
+    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<RegisteredController> {
+    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<String, Value>) -> 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<String, Value>) -> 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<String, Value>) -> 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<T: serde::Serialize>(value: T) -> Result<Value, String> {
+    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");
+    }
+}

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<String>,
+    /// Names of phases that must complete before this one starts.
+    pub depends_on: Vec<String>,
+}
+
+/// 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<WorkflowPhase>,
+    /// 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<WorkflowDefinition>,
+    /// 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,
+    },
+}

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,
 };

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<Option<WorkflowRun>> {
+pub fn get_workflow_run(config: &Config, id: &str) -> Result<Option<WorkflowRun>> {
+    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<Option<WorkflowRun>> {
                     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<WorkflowRunListResponse> {
+    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<Box<dyn rusqlite::types::ToSql>> = 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 })
     })
 }
 

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<String>,
+    #[serde(default)]
+    pub status: Option<String>,
+    #[serde(default)]
+    pub parent_thread_id: Option<String>,
+    /// `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<u64>,
+    #[serde(default)]
+    pub offset: Option<u64>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WorkflowRunListResponse {
+    pub runs: Vec<WorkflowRun>,
+    pub count: usize,
+}
+
 #[derive(Debug, Clone, Default, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct RunEventListRequest {

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();