objectiveai_cli/command/tools/
run.rs

1//! `tools run` — resolve a tool by `(owner, name, version)`, build its
2//! command from the current platform's exec vector + the caller's
3//! args, run it with the tool's version folder's `cli/` subdir as the
4//! working directory (per `resolve_tool`), and yield each
5//! stdout/stderr line as a [`ResponseItem`] as it arrives. A non-zero
6//! exit code surfaces as a final `Err(Error::ToolExit(code))`.
7
8use std::pin::Pin;
9use std::process::Stdio;
10
11use futures::Stream;
12use objectiveai_sdk::cli::command::tools::run::{Request, ResponseItem};
13use objectiveai_sdk::cli::{Error as CliError, ErrorType};
14use tokio::process::Command;
15
16use crate::child_io::{PipeEvent, spawn_pipe_reader};
17use crate::context::Context;
18use crate::error::Error;
19
20type ItemStream = Pin<Box<dyn Stream<Item = Result<ResponseItem, Error>> + Send>>;
21
22pub async fn execute(ctx: &Context, request: Request) -> Result<ItemStream, Error> {
23    let coord = format!("{}/{}/{}", request.owner, request.name, request.version);
24    let (exec, cwd) = ctx
25        .filesystem
26        .resolve_tool(&request.owner, &request.name, &request.version)
27        .await
28        .ok_or_else(|| Error::ToolNotFound(coord.clone()))?;
29
30    // The command is the tool's exec vector merged with the caller's
31    // args, verbatim — neither array's strings are inspected or
32    // mutated. The first element is the program; the rest are its
33    // arguments. CWD is the tool's `cli/` subdir (from `resolve_tool`);
34    // `objectiveai.json` lives in the parent version folder.
35    let mut argv = exec;
36    argv.extend(request.args);
37    let mut argv = argv.into_iter();
38    let program = argv.next().ok_or_else(|| {
39        Error::ToolNotFound(format!("{coord} (empty exec)"))
40    })?;
41
42    let program = crate::spawn::resolve_program(program, &cwd);
43
44    // Per-tool scratch space inside the (transient) state tree —
45    // tools that persist files write here, never into their own
46    // (possibly committed) version folder.
47    let state_dir = ctx
48        .filesystem
49        .state_dir()
50        .join("tools")
51        .join(&request.owner)
52        .join(&request.name)
53        .join(&request.version);
54    tokio::fs::create_dir_all(&state_dir)
55        .await
56        .map_err(Error::ToolSpawn)?;
57
58    // Per-tool database compartment: an owned schema plus readonly
59    // access to the base objectiveai tables, handed to the child as
60    // a role-scoped connection URL. Provisioning is idempotent; a
61    // failure fails the run loudly rather than spawning a child with
62    // a silently missing database.
63    let postgres_url = crate::db::compartment::ensure(
64        ctx.db_handle().await?,
65        crate::db::compartment::Kind::Tool,
66        &request.owner,
67        &request.name,
68        &request.version,
69    )
70    .await?;
71
72    let mut cmd = Command::new(&program);
73    cmd.args(argv)
74        .current_dir(&cwd)
75        .env("OBJECTIVEAI_STATE_DIR", &state_dir)
76        .env("OBJECTIVEAI_BIN_DIR", &cwd)
77        .env("OBJECTIVEAI_POSTGRES_URL", postgres_url)
78        .stdin(Stdio::null())
79        .stdout(Stdio::piped())
80        .stderr(Stdio::piped());
81    crate::spawn::apply_config_env(&mut cmd, &ctx.config);
82
83    let mut child = cmd.spawn().map_err(Error::ToolSpawn)?;
84    let stdout = child.stdout.take().expect("stdout was piped");
85    let stderr = child.stderr.take().expect("stderr was piped");
86
87    let mut events = spawn_pipe_reader(stdout, stderr);
88
89    let stream = async_stream::stream! {
90        while let Some(event) = events.recv().await {
91            match event {
92                PipeEvent::Stdout(line) => {
93                    yield Ok(ResponseItem::Stdout(line));
94                }
95                PipeEvent::Stderr(line) => {
96                    yield Ok(ResponseItem::Stderr(CliError {
97                        r#type: ErrorType::Error,
98                        level: None,
99                        fatal: None,
100                        message: serde_json::Value::String(line),
101                    }));
102                }
103                PipeEvent::StdoutEof | PipeEvent::StderrEof => {}
104                PipeEvent::StdoutErr(e) | PipeEvent::StderrErr(e) => {
105                    yield Err(Error::ToolRead(e));
106                    return;
107                }
108            }
109        }
110        // Both pipes closed — child has either exited or is about to.
111        // Wait for the exit code and surface non-zero as a stream
112        // `Err`, the same way legacy `dispatch_tool` returns
113        // `Err(Error::ToolExit(code))`.
114        match child.wait().await {
115            Ok(status) if status.success() => {}
116            Ok(status) => {
117                yield Err(Error::ToolExit(status.code().unwrap_or(1)));
118            }
119            Err(e) => yield Err(Error::ToolRead(e)),
120        }
121    };
122
123    Ok(Box::pin(stream))
124}
125
126pub mod request_schema {
127    use objectiveai_sdk::cli::command::tools::run as sdk;
128    use objectiveai_sdk::cli::command::tools::run::request_schema::{Request, Response};
129
130    use crate::context::Context;
131    use crate::error::Error;
132
133    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
134        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::Request)))
135    }
136}
137
138pub mod response_schema {
139    use objectiveai_sdk::cli::command::tools::run as sdk;
140    use objectiveai_sdk::cli::command::tools::run::response_schema::{Request, Response};
141
142    use crate::context::Context;
143    use crate::error::Error;
144
145    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
146        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::ResponseItem)))
147    }
148}
objectiveai_cli/command/tools/run.rs

objectiveai_cli/command/tools/
run.rs
