7.7 Controllers, workers, tasks

// app/src/controllers/agents.rs — the web app builds its agent list from this (no hard-coded agents)
use axum::{extract::State, routing::get};
use loco_rs::{app::AppContext, controller::{format, Routes}, Result};
use axum::response::Response;

async fn list(State(ctx): State<AppContext>) -> Result<Response> {
    let p = crate::app::platform(&ctx)?;
    let agents: Vec<_> = p.workflows.values().map(|w| serde_json::json!({
        "id": w.def.id, "title": w.def.title, "description": w.def.description,
        "version": w.def.version, "ui": w.def.ui, "state_whitelist": w.def.state_whitelist,
    })).collect();
    format::json(agents)                                   // source-verified helper
}
pub fn routes() -> Routes { Routes::new().prefix("api/agents").add("/", get(list)) }

// app/src/controllers/agui.rs — AG-UI endpoint as a Loco controller
use std::convert::Infallible;
use axum::{extract::{Path, State}, response::sse::{KeepAlive, Sse}, routing::post, Json};
use bo_agui::{to_sse, supports_interrupt_outcome, Event, Interrupt, RunAgentInput, RunOutcome};
use futures::Stream;
use graph_flow::{ExecutionStatus, Session, SessionStorage};
use loco_rs::{app::AppContext, controller::Routes, Result};
use tokio::sync::mpsc;

async fn run(
    State(ctx): State<AppContext>,
    // auth: loco_rs::controller::extractor::auth::JWT,     // VERIFY path; gives claims → user → tenant + roles
    Path(workflow): Path<String>,
    Json(input): Json<RunAgentInput>,
) -> Result<Sse<impl Stream<Item = std::result::Result<axum::response::sse::Event, Infallible>>>> {
    let p = crate::app::platform(&ctx)?;
    let settings: crate::settings::Settings = ctx.config.settings()?;
    let (tx, mut rx) = mpsc::channel::<Event>(64);

    tokio::spawn(async move {
        let (tid, rid) = (input.thread_id.clone(), input.run_id.clone());
        let _ = tx.send(Event::RunStarted { thread_id: tid.clone(), run_id: rid.clone(),
            protocol_version: Some(settings.agui.protocol_version.clone()) }).await;
        let Some(wf) = p.workflows.get(&workflow) else {
            let _ = tx.send(Event::RunError { message: format!("unknown workflow `{workflow}`"), code: Some("not_found".into()) }).await;
            return;
        };

        // Start or resume. Session id == AG-UI threadId.
        let prep: std::result::Result<(), String> = async {
            match p.sessions.get(&tid).await.map_err(|e| e.to_string())? {
                None => {
                    let s = Session::new_from_task(tid.clone(), &wf.def.entry);     // source-verified
                    s.context.set("session_id", tid.clone()).map_err(|e| e.to_string())?;
                    if let Some(o) = input.forwarded_props.as_object() {
                        for (k, v) in o { s.context.set(k.clone(), v.clone()).map_err(|e| e.to_string())?; }
                    }
                    // TODO(impl): set tenant + actor from JWT; insert workflow_runs row; RBAC `runs.start` from rbac.yaml
                    p.sessions.save(s).await.map_err(|e| e.to_string())
                }
                Some(s) => {
                    let waiting = s.context.get::<bool>("awaiting_approval").unwrap_or(false);
                    match (waiting, input.resume.first()) {
                        (true, None) => Err("thread has a pending interrupt; resume required".into()),
                        (true, Some(r)) => {
                            // TODO(impl): RBAC `approvals.resolve`, roles + four_eyes from `approval_request`,
                            //             idempotent upsert into approvals, audit row.
                            let decision = if r.status == "cancelled" { serde_json::json!({ "approved": false }) }
                                           else { r.payload.clone().unwrap_or_default() };
                            s.context.set("approval", decision).map_err(|e| e.to_string())?;
                            p.sessions.save(s).await.map_err(|e| e.to_string())
                        }
                        (false, _) => Ok(()),
                    }
                }
            }
        }.await;
        if let Err(message) = prep { let _ = tx.send(Event::RunError { message, code: None }).await; return; }

        // Drive until pause or completion, emitting state after each step.
        loop {
            let res = match wf.runner.run(&tid).await {                          // source-verified
                Ok(r) => r,
                Err(e) => { let _ = tx.send(Event::RunError { message: e.to_string(), code: None }).await; return; }
            };
            let snapshot = project_state(&*p.sessions, &tid, &wf.def.state_whitelist).await;
            let _ = tx.send(Event::StateSnapshot { snapshot: snapshot.clone() }).await;
            if let Some(text) = &res.response { emit_text(&tx, text).await; }
            match res.status {
                ExecutionStatus::Paused { .. } => continue,
                ExecutionStatus::Completed => {
                    let _ = tx.send(Event::RunFinished { thread_id: tid, run_id: rid, result: None, outcome: Some(RunOutcome::Success) }).await;
                    return;
                }
                ExecutionStatus::WaitingForInput => {
                    let _ = tx.send(Event::MessagesSnapshot { messages: vec![] }).await;   // TODO(impl): history
                    if supports_interrupt_outcome(&input) {
                        let i = Interrupt { id: format!("{tid}:approval"), reason: "confirmation".into(),
                            message: res.response.clone(), response_schema: None, expires_at: None,
                            metadata: Some(serde_json::json!({ "workflow": workflow, "ui": wf.def.ui })) };
                        let _ = tx.send(Event::RunFinished { thread_id: tid, run_id: rid, result: None,
                            outcome: Some(RunOutcome::Interrupt { interrupts: vec![i] }) }).await;
                    } else {
                        let _ = tx.send(Event::Custom { name: settings.agui.legacy_interrupt_event.clone(),
                            value: serde_json::json!({ "id": format!("{tid}:approval"), "state": snapshot }) }).await;
                        let _ = tx.send(Event::RunFinished { thread_id: tid, run_id: rid, result: None, outcome: None }).await;
                    }
                    return;
                }
            }
        }
    });

    let stream = async_stream::stream! { while let Some(ev) = rx.recv().await { yield Ok(to_sse(&ev)); } };
    Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
}

/// Only keys listed in the workflow's `state_whitelist` ever reach the browser.
async fn project_state(storage: &dyn SessionStorage, sid: &str, whitelist: &[String]) -> serde_json::Value {
    let mut out = serde_json::Map::new();
    if let Ok(Some(s)) = storage.get(sid).await {
        for k in whitelist { if let Some(v) = s.context.get::<serde_json::Value>(k) { out.insert(k.clone(), v); } }
    }
    serde_json::Value::Object(out)
}
async fn emit_text(tx: &mpsc::Sender<Event>, text: &str) {
    let id = uuid::Uuid::now_v7().to_string();
    let _ = tx.send(Event::TextMessageStart { message_id: id.clone(), role: "assistant".into() }).await;
    let _ = tx.send(Event::TextMessageContent { message_id: id.clone(), delta: text.into() }).await;
    let _ = tx.send(Event::TextMessageEnd { message_id: id }).await;
}

pub fn routes() -> Routes { Routes::new().prefix("agui").add("/{workflow}", post(run)) }

// app/src/workers/ingest.rs — background document ingestion on Loco's Postgres queue
use async_trait::async_trait;
use loco_rs::{app::AppContext, bgworker::BackgroundWorker, Result};
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
pub struct IngestArgs { pub tenant_id: uuid::Uuid, pub document_id: uuid::Uuid, pub parser: String, pub start_workflow: Option<String> }

pub struct IngestWorker { ctx: AppContext }

#[async_trait]
impl BackgroundWorker<IngestArgs> for IngestWorker {
    fn build(ctx: &AppContext) -> Self { Self { ctx: ctx.clone() } }
    fn queue() -> Option<String> { Some("ingest".into()) }
    fn max_retries() -> u32 { 3 }                                   // source-verified trait methods
    async fn perform(&self, args: IngestArgs) -> Result<()> {
        let p = crate::app::platform(&self.ctx)?;
        let _parser = p.parsers.get(&args.parser).ok_or_else(|| loco_rs::Error::Message(format!("unknown parser `{}`", args.parser)))?;
        // TODO(impl): load bytes from ctx.storage, parse, store parsed_md/json, chunk + embed + upsert vectors,
        //             then optionally create a workflow session and enqueue DriveWorkflowWorker.
        Ok(())
    }
}
// Enqueue from a controller:  IngestWorker::perform_later(&ctx, args).await?;

// app/src/tasks/config_validate.rs — `bo-app task config_validate`
use async_trait::async_trait;
use loco_rs::{app::AppContext, task::{Task, TaskInfo, Vars}, Result};

pub struct ConfigValidate;
#[async_trait]
impl Task for ConfigValidate {
    fn task(&self) -> TaskInfo {
        TaskInfo { name: "config_validate".into(), detail: "Load and validate every YAML definition, then exit".into() }
    }
    async fn run(&self, ctx: &AppContext, _vars: &Vars) -> Result<()> {
        // after_context already ran: reaching this line means everything loaded and compiled.
        let p = crate::app::platform(ctx)?;
        println!("OK: {} workflows, {} connectors, {} models", p.workflows.len(), p.connectors.len(), p.defs.models.models.len());
        Ok(())
    }
}
// tasks/schema_export.rs: write schemars::schema_for!(WorkflowDef) etc. to schemas/*.json — TODO(impl)