objectiveai_cli/command/plugins/
run.rs

1//! `plugins run` — bare-naked port of legacy `dispatch_external`.
2//!
3//! Resolves the installed plugin's exec command (the tools model:
4//! the manifest's per-OS argv plus the caller's args, run with CWD =
5//! the plugin's `cli/` folder), spawns it with piped
6//! stdin/stdout/stderr, and yields each parsed line from the plugin's
7//! stdout as a [`ResponseItem`] as it arrives. The bidirectional
8//! protocol — plugin emits a `Command` request, the host runs it and
9//! streams the result back into the plugin's stdin wrapped in a
10//! `PluginCommandResponse` envelope, terminated by a
11//! `CommandComplete` marker — stays internal to the leaf. Consumers
12//! observe Command requests as stream items but the actual execution
13//! and stdin write-back happens here.
14
15use std::pin::Pin;
16use std::process::Stdio;
17use std::sync::Arc;
18
19use futures::{Stream, StreamExt};
20use objectiveai_sdk::cli::command::plugins::run::{Request, ResponseItem};
21use objectiveai_sdk::cli::plugins::Output as PluginOutput;
22use objectiveai_sdk::cli::{Error as CliError, ErrorType as CliErrorType};
23use serde::Serialize;
24use tokio::io::AsyncWriteExt;
25use tokio::process::{ChildStdin, Command};
26use tokio::sync::Mutex;
27use tokio::task::JoinHandle;
28
29use crate::child_io::{PipeEvent, spawn_pipe_reader};
30use crate::context::Context;
31use crate::error::Error;
32
33type ItemStream = Pin<Box<dyn Stream<Item = Result<ResponseItem, Error>> + Send>>;
34
35pub async fn execute(ctx: &Context, request: Request) -> Result<ItemStream, Error> {
36    let coord = format!("{}/{}/{}", request.owner, request.name, request.version);
37    let (exec, cli_dir) = ctx
38        .filesystem
39        .resolve_plugin(&request.owner, &request.name, &request.version)
40        .await
41        .ok_or_else(|| Error::PluginNotFound(coord.clone()))?;
42
43    // The command is the plugin's exec vector merged with the
44    // caller's args, verbatim. The first element is the program; the
45    // rest are its arguments. CWD is the plugin's `cli/` folder —
46    // always — with the same relative-path program resolution tools
47    // use.
48    let mut argv = exec;
49    argv.extend(request.args.iter().cloned());
50    let mut argv = argv.into_iter();
51    let program = argv
52        .next()
53        .ok_or_else(|| Error::PluginNotFound(format!("{coord} (empty exec)")))?;
54    let program = crate::spawn::resolve_program(program, &cli_dir);
55    // The cli dir may be empty/absent for exec-only plugins (nothing
56    // was zipped into it) — the CWD still has to exist to spawn.
57    tokio::fs::create_dir_all(&cli_dir)
58        .await
59        .map_err(Error::PluginSpawn)?;
60
61    // Per-plugin scratch space inside the (transient) state tree —
62    // plugins that persist files write here, never into their own
63    // (possibly committed) install folder.
64    let state_dir = ctx
65        .filesystem
66        .state_dir()
67        .join("plugins")
68        .join(&request.owner)
69        .join(&request.name)
70        .join(&request.version);
71    tokio::fs::create_dir_all(&state_dir)
72        .await
73        .map_err(Error::PluginSpawn)?;
74
75    // Per-plugin database compartment: an owned schema plus readonly
76    // access to the base objectiveai tables, handed to the child as
77    // a role-scoped connection URL. Provisioning is idempotent; a
78    // failure fails the run loudly rather than spawning a child with
79    // a silently missing database.
80    let postgres_url = crate::db::compartment::ensure(
81        ctx.db_handle().await?,
82        crate::db::compartment::Kind::Plugin,
83        &request.owner,
84        &request.name,
85        &request.version,
86    )
87    .await?;
88
89    // Context for nested (plugin-originated) commands: this caller's
90    // ctx, stamped with the plugin coordinate. `ctx.plugin` is set so
91    // a nested command knows which plugin it runs on behalf of, and
92    // `config.plugin_*` is set so any subprocess that nested command
93    // itself spawns inherits the coordinate via `apply_config_env`.
94    // No `reset_api_client()` here: plugin coords aren't on the
95    // `HttpClient`, so the memoized API client deliberately stays
96    // shared with the caller's ctx.
97    let mut nested_ctx = ctx.clone();
98    nested_ctx.config.plugin_owner = Some(request.owner.clone());
99    nested_ctx.config.plugin_repository = Some(request.name.clone());
100    nested_ctx.config.plugin_version = Some(request.version.clone());
101    nested_ctx.plugin = Some(crate::plugin_path::PluginPath {
102        owner: request.owner.clone(),
103        repository: request.name.clone(),
104        version: request.version.clone(),
105    });
106
107    let mut cmd = Command::new(&program);
108    cmd.args(argv)
109        .current_dir(&cli_dir)
110        .env("OBJECTIVEAI_STATE_DIR", &state_dir)
111        .env("OBJECTIVEAI_POSTGRES_URL", postgres_url)
112        .stdin(Stdio::piped())
113        .stdout(Stdio::piped())
114        .stderr(Stdio::piped())
115        .kill_on_drop(true);
116    crate::spawn::apply_config_env(&mut cmd, &nested_ctx.config);
117
118    let mut child = cmd.spawn().map_err(Error::PluginSpawn)?;
119    let stdout = child.stdout.take().expect("stdout was piped");
120    let stderr = child.stderr.take().expect("stderr was piped");
121    let stdin = child.stdin.take().expect("stdin was piped");
122    let plugin_stdin: Arc<Mutex<ChildStdin>> = Arc::new(Mutex::new(stdin));
123
124    let mut events = spawn_pipe_reader(stdout, stderr);
125
126    let stream = async_stream::stream! {
127        let mut command_tasks: Vec<(Option<String>, JoinHandle<i32>)> = Vec::new();
128        while let Some(event) = events.recv().await {
129            match event {
130                PipeEvent::Stderr(_) => {
131                    // Bare anonymous error — no level, no fatal, no
132                    // message. Stops at "something went wrong on
133                    // stderr" by deliberate host policy.
134                    yield Ok(ResponseItem::Error(CliError {
135                        r#type: CliErrorType::Error,
136                        level: None,
137                        fatal: None,
138                        message: serde_json::Value::Null,
139                    }));
140                }
141                PipeEvent::Stdout(trimmed) => {
142                    match serde_json::from_str::<PluginOutput>(&trimmed) {
143                        Ok(PluginOutput::Error(e)) => {
144                            yield Ok(ResponseItem::Error(e));
145                        }
146                        Ok(PluginOutput::Mcp(mcp)) => {
147                            yield Ok(ResponseItem::Mcp(mcp));
148                        }
149                        Ok(PluginOutput::Command(c)) => {
150                            // Command requests are host-internal —
151                            // the CLI intercepts them to drive the
152                            // bidirectional protocol back into the
153                            // plugin's stdin and does NOT surface
154                            // them on the user-visible `ResponseItem`
155                            // stream.
156                            let task_id = Some(c.id);
157                            let task = run_nested_command(
158                                nested_ctx.clone(),
159                                c.command,
160                                plugin_stdin.clone(),
161                                task_id.clone(),
162                            );
163                            command_tasks.push((task_id, task));
164                        }
165                        Ok(PluginOutput::Notification(value)) => {
166                            yield Ok(ResponseItem::Notification(value));
167                        }
168                        Err(_) => {
169                            // Legacy fallback: surface the raw line
170                            // as a notification so unparseable plugin
171                            // output is at least observable rather
172                            // than silently dropped.
173                            yield Ok(ResponseItem::Notification(
174                                serde_json::Value::String(trimmed),
175                            ));
176                        }
177                    }
178                }
179                PipeEvent::StdoutEof | PipeEvent::StderrEof => {}
180                PipeEvent::StdoutErr(e) | PipeEvent::StderrErr(e) => {
181                    yield Err(Error::PluginRead(e));
182                    return;
183                }
184            }
185        }
186
187        // Drain any in-flight Command tasks the plugin queued before
188        // its stdout EOF. Each task gets a terminal `CommandComplete`
189        // written to plugin stdin so the plugin sees the run boundary
190        // even when it didn't mint a correlation id.
191        for (id, task) in command_tasks {
192            let exit_code = task.await.unwrap_or(-1);
193            let envelope = PluginCommandResponse {
194                id: id.as_deref(),
195                value: CommandComplete {
196                    kind: "command_complete",
197                    exit_code,
198                },
199            };
200            let _ = write_envelope(&plugin_stdin, &envelope).await;
201        }
202
203        // Drop our reference to plugin stdin so the kernel pipe closes
204        // and a polite plugin sees EOF on its stdin read.
205        drop(plugin_stdin);
206
207        match child.wait().await {
208            Ok(status) if status.success() => {}
209            Ok(status) => {
210                yield Err(Error::PluginExit(status.code().unwrap_or(1)));
211            }
212            Err(e) => {
213                yield Err(Error::PluginRead(e));
214            }
215        }
216    };
217
218    Ok(Box::pin(stream))
219}
220
221/// Dispatch a plugin-originated command IN-PROCESS — no subprocess, no
222/// Postgres re-bootstrap. `command` arrives as an already-tokenized
223/// argv vector (the plugin executor carries it structured), so it runs
224/// through the very same `crate::run` entry the cli binary uses without
225/// any re-tokenization — an argument value containing whitespace stays a
226/// single token. Dispatched against `ctx` (which already carries this
227/// caller's identity plus the plugin coordinate). The body is a mirror
228/// of `main.rs::run_command`: every line that binary would write to
229/// stdout is instead forwarded into `plugin_stdin` wrapped in a
230/// [`PluginCommandResponse`]. Returns an exit code for the terminal
231/// `CommandComplete` (the tool's code on a `ToolExit`, else 0/1).
232fn run_nested_command(
233    ctx: Context,
234    command: Vec<String>,
235    plugin_stdin: Arc<Mutex<ChildStdin>>,
236    id: Option<String>,
237) -> JoinHandle<i32> {
238    tokio::spawn(async move {
239        let id = id.as_deref();
240        // Argv is already tokenized by the plugin executor — one
241        // element per argument. No `split_whitespace`, so a value like
242        // `--simple "a b c"` is not re-split into separate tokens.
243        let tokens: Vec<String> = command;
244
245        // A plugin may not invoke `plugins` or `tools` commands — no
246        // running another plugin, no running a tool. Forward the same
247        // error line the cli would emit and stop here.
248        let forbidden = match tokens.first().map(String::as_str) {
249            Some("plugins") => Some("plugins"),
250            Some("tools") => Some("tools"),
251            _ => None,
252        };
253        if let Some(kind) = forbidden {
254            let _ = forward_error(
255                &plugin_stdin,
256                id,
257                &Error::PluginCommandForbidden(kind),
258                Some(true),
259            )
260            .await;
261            return 1;
262        }
263
264        // `args[0]` is the program name, which `crate::run` strips
265        // unconditionally; the plugin's command is just the subcommand
266        // tokens, so prepend a placeholder.
267        let mut args: Vec<String> = vec!["objectiveai-cli".to_string()];
268        args.extend(tokens);
269
270        // A mirror of `main.rs::run_command`: drive the same `crate::run`
271        // stream, but forward each line into the plugin's stdin (wrapped
272        // in a `PluginCommandResponse`) instead of writing it to stdout.
273        let run_stream = match crate::run(args, Some(ctx)).await {
274            Ok(s) => s,
275            Err(e) => {
276                if let Error::ClapParse(ref clap_err) = e {
277                    if crate::is_informational(clap_err) {
278                        let _ = forward_help(&plugin_stdin, id, &clap_err.to_string()).await;
279                        return 0;
280                    }
281                }
282                let _ = forward_error(&plugin_stdin, id, &e, Some(true)).await;
283                return match e {
284                    Error::ToolExit(code) => code,
285                    _ => 1,
286                };
287            }
288        };
289        // Both arms forward each item to the plugin's stdin as a line;
290        // `drain` is generic over the item type (typed root items vs
291        // post-transform JSON).
292        let last_tool_exit = match run_stream {
293            crate::RunStream::Execute(stream) => drain(&plugin_stdin, id, stream).await,
294            crate::RunStream::ExecuteTransform(stream) => drain(&plugin_stdin, id, stream).await,
295        };
296        last_tool_exit.unwrap_or(0)
297    })
298}
299
300/// Drain a run stream into the plugin's stdin — one
301/// [`PluginCommandResponse`] line per item. Returns the last `ToolExit`
302/// code seen (surfaced as an `Err` item). Generic over the item type so
303/// it serves both `RunStream` arms. Stops early if the plugin's stdin
304/// closes.
305async fn drain<S, T>(
306    plugin_stdin: &Arc<Mutex<ChildStdin>>,
307    id: Option<&str>,
308    mut stream: S,
309) -> Option<i32>
310where
311    S: Stream<Item = Result<T, Error>> + Unpin,
312    T: Serialize,
313{
314    let mut last_tool_exit: Option<i32> = None;
315    while let Some(item) = stream.next().await {
316        let written = match item {
317            Ok(value) => forward_line(plugin_stdin, id, &value).await,
318            Err(e) => {
319                if let Error::ToolExit(code) = &e {
320                    last_tool_exit = Some(*code);
321                }
322                forward_error(plugin_stdin, id, &e, None).await
323            }
324        };
325        if written.is_err() {
326            // Plugin's stdin is gone; abandon the run.
327            break;
328        }
329    }
330    last_tool_exit
331}
332
333/// Mirror of `main.rs::write_line`: serialize `value` and forward it to
334/// the plugin's stdin in a [`PluginCommandResponse`] envelope.
335async fn forward_line<T: Serialize>(
336    plugin_stdin: &Arc<Mutex<ChildStdin>>,
337    id: Option<&str>,
338    value: &T,
339) -> std::io::Result<()> {
340    write_envelope(plugin_stdin, &PluginCommandResponse { id, value }).await
341}
342
343/// Mirror of `main.rs::write_error_line`.
344async fn forward_error(
345    plugin_stdin: &Arc<Mutex<ChildStdin>>,
346    id: Option<&str>,
347    e: &Error,
348    fatal: Option<bool>,
349) -> std::io::Result<()> {
350    let payload = CliError {
351        r#type: CliErrorType::Error,
352        level: Some(objectiveai_sdk::cli::Level::Error),
353        fatal,
354        message: e.output_message(),
355    };
356    forward_line(plugin_stdin, id, &payload).await
357}
358
359/// Mirror of `main.rs::write_help_line`.
360async fn forward_help(
361    plugin_stdin: &Arc<Mutex<ChildStdin>>,
362    id: Option<&str>,
363    help: &str,
364) -> std::io::Result<()> {
365    let payload = serde_json::json!({ "type": "help", "help": help });
366    forward_line(plugin_stdin, id, &payload).await
367}
368
369async fn write_envelope<T: Serialize>(
370    stdin: &Arc<Mutex<ChildStdin>>,
371    envelope: &T,
372) -> std::io::Result<()> {
373    let line = serde_json::to_string(envelope).expect("envelope serializes");
374    let mut guard = stdin.lock().await;
375    guard.write_all(line.as_bytes()).await?;
376    guard.write_all(b"\n").await?;
377    guard.flush().await?;
378    Ok(())
379}
380
381/// Wire envelope for nested-command output streamed back to plugin
382/// stdin. Matches `cli.plugins.PluginCommandResponse.json`. Defined
383/// locally rather than in the SDK because the SDK's `cli/output`
384/// module that hosts the canonical type is currently torn-up.
385#[derive(Serialize)]
386struct PluginCommandResponse<'a, T> {
387    #[serde(skip_serializing_if = "Option::is_none")]
388    id: Option<&'a str>,
389    value: T,
390}
391
392/// Terminal marker written to plugin stdin after each nested command
393/// finishes. Matches `cli.output.notification.CommandComplete.json`.
394#[derive(Serialize)]
395struct CommandComplete {
396    #[serde(rename = "type")]
397    kind: &'static str,
398    exit_code: i32,
399}
400
401pub mod request_schema {
402    use objectiveai_sdk::cli::command::plugins::run as sdk;
403    use objectiveai_sdk::cli::command::plugins::run::request_schema::{Request, Response};
404
405    use crate::context::Context;
406    use crate::error::Error;
407
408    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
409        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::Request)))
410    }
411}
412
413pub mod response_schema {
414    use objectiveai_sdk::cli::command::plugins::run as sdk;
415    use objectiveai_sdk::cli::command::plugins::run::response_schema::{Request, Response};
416
417    use crate::context::Context;
418    use crate::error::Error;
419
420    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
421        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::ResponseItem)))
422    }
423}
objectiveai_cli/command/plugins/run.rs

objectiveai_cli/command/plugins/
run.rs
