loopflow 0.12.8

Run steps and flows with coding agents
Documentation
use std::os::unix::process::CommandExt;
use std::process::Command;

use anyhow::{anyhow, Context};

use crate::durable::{AgentInvocationId, BoundaryState, InvocationSurface};
use crate::engine::wave_home::HomeRoute;
use crate::lf::InvocationCommand;
use crate::store::{open_store, storage_config_from_env, Store};

pub fn run(command: &InvocationCommand) -> anyhow::Result<()> {
    let runtime = tokio::runtime::Runtime::new()?;
    runtime.block_on(run_async(command))
}

async fn run_async(command: &InvocationCommand) -> anyhow::Result<()> {
    let store = open_shared_store().await?;
    match command {
        InvocationCommand::List { active, json } => {
            let surfaces = store.invocation_surfaces(*active).await?;
            if *json {
                println!("{}", serde_json::to_string_pretty(&surfaces)?);
            } else if surfaces.is_empty() {
                println!("No AgentInvocations.");
            } else {
                for surface in surfaces {
                    print_surface(&surface, false)?;
                }
            }
        }
        InvocationCommand::Status {
            invocation_id,
            json,
        }
        | InvocationCommand::Attach {
            invocation_id,
            json,
        } => {
            let surface = load_surface(&store, invocation_id).await?;
            print_surface(&surface, *json)?;
        }
        InvocationCommand::Handback {
            invocation_id,
            outcome,
            json,
        } => {
            let invocation_id = parse_invocation_id(invocation_id)?;
            let outcome = parse_outcome(outcome)?;
            let surface = store.handback_invocation(&invocation_id, outcome).await?;
            print_surface(&surface, *json)?;
        }
        InvocationCommand::Present { invocation_id } => {
            let surface = load_surface(&store, invocation_id).await?;
            present(&surface)?;
        }
    }
    Ok(())
}

async fn open_shared_store() -> anyhow::Result<Store> {
    let config = storage_config_from_env().context("resolve the shared Loopflow store")?;
    open_store(&config)
        .await
        .context("open the shared Loopflow store")
}

async fn load_surface(store: &Store, invocation_id: &str) -> anyhow::Result<InvocationSurface> {
    let invocation_id = parse_invocation_id(invocation_id)?;
    store
        .invocation_surface(&invocation_id)
        .await?
        .ok_or_else(|| anyhow!("AgentInvocation {invocation_id} not found"))
}

fn parse_invocation_id(value: &str) -> anyhow::Result<AgentInvocationId> {
    AgentInvocationId::parse(value).map_err(Into::into)
}

fn parse_outcome(value: &str) -> anyhow::Result<BoundaryState> {
    match value {
        "succeeded" => Ok(BoundaryState::Succeeded),
        "failed" => Ok(BoundaryState::Failed),
        "interrupted" => Ok(BoundaryState::Interrupted),
        "unknown" => Ok(BoundaryState::Unknown),
        value => Err(anyhow!(
            "invalid AgentInvocation handback outcome {value:?}"
        )),
    }
}

fn print_surface(surface: &InvocationSurface, json: bool) -> anyhow::Result<()> {
    if json {
        println!("{}", serde_json::to_string_pretty(surface)?);
    } else {
        println!(
            "{}  {:?}\n  work: {}:{}\n  provider: {}\n  home: {}\n  cwd: {}",
            surface.invocation.id,
            surface.run.state,
            surface.work.kind(),
            surface.work.id(),
            surface.invocation.route.provider,
            surface.home_route,
            surface
                .run
                .cwd
                .as_ref()
                .map_or_else(|| "-".to_string(), |cwd| cwd.display().to_string()),
        );
    }
    Ok(())
}

fn present(surface: &InvocationSurface) -> anyhow::Result<()> {
    let argv = surface
        .attach_argv
        .as_ref()
        .ok_or_else(|| anyhow!("Invocation {} has no attach route", surface.invocation.id))?;
    let (program, args) = argv.split_first().ok_or_else(|| {
        anyhow!(
            "Invocation {} has an empty attach route",
            surface.invocation.id
        )
    })?;
    let home = HomeRoute::parse(&surface.home_route)
        .ok_or_else(|| anyhow!("invalid Home route {:?}", surface.home_route))?;
    let mut command = if let Some(destination) = home.ssh_destination() {
        let mut command = Command::new("ssh");
        if let Some(port) = home.ssh_port() {
            command.args(["-p", &port.to_string()]);
        }
        command.arg(destination).arg("--").arg(program).args(args);
        command
    } else {
        let mut command = Command::new(program);
        command.args(args);
        if let Some(cwd) = &surface.run.cwd {
            command.current_dir(cwd);
        }
        command
    };
    let error = command.exec();
    Err(anyhow!(
        "failed to exec AgentInvocation attach route: {error}"
    ))
}