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 as the working
4//! directory, and yield each stdout/stderr line as a [`ResponseItem`]
5//! as it arrives. A non-zero exit code surfaces as a final
6//! `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 version folder (where
34    // `objectiveai.json` lives) — always.
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_POSTGRES_URL", postgres_url)
77        .stdin(Stdio::null())
78        .stdout(Stdio::piped())
79        .stderr(Stdio::piped());
80    crate::spawn::apply_config_env(&mut cmd, &ctx.config);
81
82    let mut child = cmd.spawn().map_err(Error::ToolSpawn)?;
83    let stdout = child.stdout.take().expect("stdout was piped");
84    let stderr = child.stderr.take().expect("stderr was piped");
85
86    let mut events = spawn_pipe_reader(stdout, stderr);
87
88    let stream = async_stream::stream! {
89        while let Some(event) = events.recv().await {
90            match event {
91                PipeEvent::Stdout(line) => {
92                    yield Ok(ResponseItem::Stdout(line));
93                }
94                PipeEvent::Stderr(line) => {
95                    yield Ok(ResponseItem::Stderr(CliError {
96                        r#type: ErrorType::Error,
97                        level: None,
98                        fatal: None,
99                        message: serde_json::Value::String(line),
100                    }));
101                }
102                PipeEvent::StdoutEof | PipeEvent::StderrEof => {}
103                PipeEvent::StdoutErr(e) | PipeEvent::StderrErr(e) => {
104                    yield Err(Error::ToolRead(e));
105                    return;
106                }
107            }
108        }
109        // Both pipes closed — child has either exited or is about to.
110        // Wait for the exit code and surface non-zero as a stream
111        // `Err`, the same way legacy `dispatch_tool` returns
112        // `Err(Error::ToolExit(code))`.
113        match child.wait().await {
114            Ok(status) if status.success() => {}
115            Ok(status) => {
116                yield Err(Error::ToolExit(status.code().unwrap_or(1)));
117            }
118            Err(e) => yield Err(Error::ToolRead(e)),
119        }
120    };
121
122    Ok(Box::pin(stream))
123}
124
125pub mod request_schema {
126    use objectiveai_sdk::cli::command::tools::run as sdk;
127    use objectiveai_sdk::cli::command::tools::run::request_schema::{Request, Response};
128
129    use crate::context::Context;
130    use crate::error::Error;
131
132    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
133        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::Request)))
134    }
135}
136
137pub mod response_schema {
138    use objectiveai_sdk::cli::command::tools::run as sdk;
139    use objectiveai_sdk::cli::command::tools::run::response_schema::{Request, Response};
140
141    use crate::context::Context;
142    use crate::error::Error;
143
144    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
145        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::ResponseItem)))
146    }
147}
objectiveai_cli/command/tools/run.rs

objectiveai_cli/command/tools/
run.rs
