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

objectiveai_cli/command/plugins/
run.rs
