7.4 Platform and the task-kind registry

// app/src/platform.rs
use std::{collections::BTreeMap, path::Path, sync::Arc};
use bo_core::ports::{DocumentParser, IntegrationConnector, LlmProvider};
use graph_flow::{FlowRunner, SessionStorage};
use loco_rs::app::AppContext;
use crate::{definitions::Definitions, engine, kinds::KindRegistry, settings::Settings};

pub struct CompiledWorkflow { pub def: crate::definitions::WorkflowDef, pub runner: FlowRunner }

pub struct Platform {
    pub defs: Definitions,
    pub llms: BTreeMap<String, Arc<dyn LlmProvider>>,           // keyed by models.yaml key
    pub parsers: BTreeMap<String, Arc<dyn DocumentParser>>,
    pub connectors: BTreeMap<String, Arc<dyn IntegrationConnector>>,
    pub sessions: Arc<dyn SessionStorage>,
    pub workflows: BTreeMap<String, CompiledWorkflow>,
}

impl Platform {
    pub async fn load(ctx: &AppContext, s: &Settings) -> anyhow::Result<Self> {
        let defs = Definitions::load(Path::new(&s.definitions_dir))?;
        let llms = engine::adapters::build_llms(&defs, s)?;           // TODO(impl): VllmProvider per model
        let parsers = engine::adapters::build_parsers(s)?;            // TODO(impl): DoclingClient per parser
        let connectors = engine::adapters::build_connectors(&defs)?;  // TODO(impl): adapter by `kind`
        let sessions: Arc<dyn SessionStorage> = Arc::new(engine::session_storage::SeaOrmSessionStorage::new(ctx.db.clone()));

        let mut p = Self { defs, llms, parsers, connectors, sessions, workflows: BTreeMap::new() };
        let kinds = KindRegistry::builtin();
        for (id, def) in p.defs.workflows.clone() {
            if !def.enabled { continue; }
            let graph = engine::compiler::compile(&def, &kinds, &p)?;
            let runner = FlowRunner::new(Arc::new(graph), p.sessions.clone());   // source-verified signature
            p.workflows.insert(id, CompiledWorkflow { def, runner });
        }
        Ok(p)
    }
}

// app/src/kinds/mod.rs — Rust behaviours that YAML composes
use std::{collections::BTreeMap, sync::Arc};
use graph_flow::Task;
use crate::{definitions::NodeDef, platform::Platform};

pub mod connector_invoke; pub mod document_parse; pub mod flow_end;
pub mod human_approval; pub mod llm_extract; pub mod records_match; pub mod rules_validate;

/// A factory that turns one YAML node into a graph-flow task. It must validate `node.with` and fail loudly.
pub type KindFactory = fn(node: &NodeDef, platform: &Platform) -> anyhow::Result<Arc<dyn Task>>;

pub struct KindRegistry(BTreeMap<&'static str, KindFactory>);
impl KindRegistry {
    /// Adding a behaviour = one module + one line here. Nothing else changes.
    pub fn builtin() -> Self {
        let mut m: BTreeMap<&'static str, KindFactory> = BTreeMap::new();
        m.insert("document.parse", document_parse::build);
        m.insert("llm.extract", llm_extract::build);
        m.insert("rules.validate", rules_validate::build);
        m.insert("records.match", records_match::build);
        m.insert("human.approval", human_approval::build);
        m.insert("connector.invoke", connector_invoke::build);
        m.insert("flow.end", flow_end::build);
        // TODO(impl): llm.generate, llm.classify, vector.search, sql.query, mcp.tool, notify.email, flow.fanout
        Self(m)
    }
    pub fn get(&self, kind: &str) -> Option<&KindFactory> { self.0.get(kind) }
    pub fn names(&self) -> Vec<&'static str> { self.0.keys().copied().collect() }
}

// app/src/kinds/llm_extract.rs — example kind, fully parameterised from YAML
use std::sync::Arc;
use async_trait::async_trait;
use bo_core::ports::LlmProvider;
use graph_flow::{Context, GraphError, NextAction, Task, TaskResult};
use serde::Deserialize;
use crate::{definitions::{NodeDef, PromptDef}, platform::Platform};

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct With { model: String, prompt: String, schema: String, input: String, output: String }

struct LlmExtract { id: String, llm: Arc<dyn LlmProvider>, prompt: PromptDef, schema: serde_json::Value, input: String, output: String }

pub fn build(node: &NodeDef, p: &Platform) -> anyhow::Result<Arc<dyn Task>> {
    let w: With = serde_json::from_value(node.with.clone())?;
    Ok(Arc::new(LlmExtract {
        id: node.id.clone(),
        llm: p.llms.get(&w.model).cloned().ok_or_else(|| anyhow::anyhow!("unknown model `{}`", w.model))?,
        prompt: p.defs.prompts.get(&w.prompt).cloned().ok_or_else(|| anyhow::anyhow!("unknown prompt `{}`", w.prompt))?,
        schema: p.defs.schemas.get(&w.schema).cloned().ok_or_else(|| anyhow::anyhow!("unknown schema `{}`", w.schema))?,
        input: w.input, output: w.output,
    }))
}

#[async_trait]
impl Task for LlmExtract {
    fn id(&self) -> &str { &self.id }
    async fn run(&self, ctx: Context) -> graph_flow::Result<TaskResult> {
        let fail = |m: String| GraphError::TaskExecutionFailed(m);
        let text: String = ctx.get(&self.input).ok_or_else(|| fail(format!("missing `{}`", self.input)))?;
        let max = self.prompt.guards["max_input_chars"].as_u64().unwrap_or(200_000) as usize;
        let tag = self.prompt.guards["fence_tag"].as_str().unwrap_or("document");
        // Untrusted document text: neutralise the fence tag, then substitute into the user template.
        let safe: String = text.replace(&format!("</{tag}>"), "").chars().take(max).collect();
        let user = self.prompt.user.replace("{{ input }}", &safe);
        let json = self.llm.extract_json(&self.prompt.system, &user, &self.schema).await.map_err(|e| fail(e.to_string()))?;
        // TODO(impl): jsonschema validation of `json` against self.schema; retry policy from NodeDef.retry
        ctx.set(self.output.clone(), json)?;                                   // source-verified: Context::set is sync
        Ok(TaskResult::new(Some(format!("{}: extracted", self.id)), NextAction::ContinueAndExecute))
    }
}

// app/src/kinds/human_approval.rs
use std::sync::Arc;
use async_trait::async_trait;
use graph_flow::{Context, NextAction, Task, TaskResult};
use serde::Deserialize;
use crate::{definitions::{Condition, NodeDef}, platform::Platform};

#[derive(Deserialize, Clone)]
#[serde(deny_unknown_fields)]
pub struct With {
    #[serde(default = "confirmation")] pub reason: String,
    pub roles: Vec<String>,
    #[serde(default)] pub four_eyes: bool,
    #[serde(default)] pub expires_in_hours: Option<u64>,
    #[serde(default)] pub auto_approve: Option<AutoApprove>,
    #[serde(default)] pub always_require_when: Option<Condition>,
}
#[derive(Deserialize, Clone)]
#[serde(deny_unknown_fields)]
pub struct AutoApprove { pub enabled: bool, pub when: Condition }
fn confirmation() -> String { "confirmation".into() }

struct HumanApproval { id: String, with: With }
pub fn build(node: &NodeDef, _p: &Platform) -> anyhow::Result<Arc<dyn Task>> {
    Ok(Arc::new(HumanApproval { id: node.id.clone(), with: serde_json::from_value(node.with.clone())? }))
}

#[async_trait]
impl Task for HumanApproval {
    fn id(&self) -> &str { &self.id }
    async fn run(&self, ctx: Context) -> graph_flow::Result<TaskResult> {
        if ctx.get::<serde_json::Value>("approval").is_some() {
            ctx.set("awaiting_approval", false)?;
            return Ok(TaskResult::new(Some("Decision recorded".into()), NextAction::ContinueAndExecute));
        }
        let forced = self.with.always_require_when.as_ref().is_some_and(|c| c.eval(&ctx));
        let auto = self.with.auto_approve.as_ref().is_some_and(|a| a.enabled && a.when.eval(&ctx));
        if auto && !forced {
            ctx.set("approval", serde_json::json!({ "approved": true, "by": "policy:auto_approve" }))?;
            return Ok(TaskResult::new(Some("Auto-approved by policy".into()), NextAction::ContinueAndExecute));
        }
        // The controller reads these keys to build the AG-UI interrupt and the approvals row.
        ctx.set("awaiting_approval", true)?;
        ctx.set("approval_request", serde_json::json!({
            "node": self.id, "reason": self.with.reason, "roles": self.with.roles,
            "four_eyes": self.with.four_eyes, "expires_in_hours": self.with.expires_in_hours }))?;
        Ok(TaskResult::new(Some("Awaiting human approval".into()), NextAction::WaitForInput))
    }
}