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

objectiveai_cli/command/plugins/
run.rs
