• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

dcdpr / jp / 24145397254

08 Apr 2026 04:02PM UTC coverage: 62.411% (-1.9%) from 64.307%
24145397254

Pull #521

github

web-flow
Merge 5cf24214c into 9ea8e05a0
Pull Request #521: feat(plugin): Add command plugin system (RFD 072)

137 of 1172 new or added lines in 16 files covered. (11.69%)

5 existing lines in 2 files now uncovered.

20195 of 32358 relevant lines covered (62.41%)

369.35 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

11.5
/crates/jp_cli/src/cmd/plugin/dispatch.rs
1
//! Host-side plugin message loop.
2
//!
3
//! Spawns the plugin binary, sends `init`, and relays workspace queries until
4
//! the plugin sends `exit` or the process terminates.
5

6
use std::{
7
    collections::HashSet,
8
    io::{BufRead, BufReader, Write},
9
    process::{Command, Stdio},
10
    sync::{
11
        Arc, Mutex,
12
        atomic::{AtomicBool, Ordering},
13
    },
14
    thread,
15
};
16

17
use camino::{Utf8Path, Utf8PathBuf};
18
use jp_config::{
19
    AppConfig,
20
    plugins::{
21
        PluginsConfig,
22
        command::{CommandPluginConfig, RunPolicy},
23
    },
24
};
25
use jp_plugin::{
26
    PROTOCOL_VERSION,
27
    message::{
28
        ConfigResponse, ConversationSummary, ConversationsResponse, DescribeResponse,
29
        ErrorResponse, EventsResponse, HostToPlugin, InitMessage, LogMessage, PathsInfo,
30
        PluginToHost, WorkspaceInfo,
31
    },
32
};
33
use jp_workspace::Workspace;
34
use serde_json::Value;
35
use tracing::{debug, error, trace, warn};
36

37
use super::registry;
38
use crate::{cmd, signals::SignalPair};
39

40
/// Run a plugin binary, handling the full protocol lifecycle.
41
///
42
/// `binary` is the path to the plugin executable. `args` are the remaining CLI
43
/// arguments to forward.
NEW
44
pub(crate) fn run_plugin(
×
NEW
45
    binary: &Utf8Path,
×
NEW
46
    args: &[String],
×
NEW
47
    workspace: &Workspace,
×
NEW
48
    config: &Arc<AppConfig>,
×
NEW
49
    signals: &SignalPair,
×
NEW
50
    log_level: u8,
×
NEW
51
) -> Result<(), cmd::Error> {
×
NEW
52
    let config_json = serde_json::to_value(config.as_ref().to_partial())
×
NEW
53
        .map_err(|e| cmd::Error::from(format!("failed to serialize config: {e}")))?;
×
54

NEW
55
    let storage_path = workspace
×
NEW
56
        .storage_path()
×
NEW
57
        .ok_or_else(|| cmd::Error::from("workspace has no storage configured"))?;
×
58

NEW
59
    let home = std::env::home_dir().and_then(|p| camino::Utf8PathBuf::from_path_buf(p).ok());
×
60

NEW
61
    let init = HostToPlugin::Init(InitMessage {
×
NEW
62
        version: PROTOCOL_VERSION,
×
NEW
63
        workspace: WorkspaceInfo {
×
NEW
64
            root: workspace.root().to_owned(),
×
NEW
65
            storage: storage_path.to_owned(),
×
NEW
66
            id: workspace.id().to_string(),
×
NEW
67
        },
×
NEW
68
        paths: PathsInfo {
×
NEW
69
            user_data: jp_workspace::user_data_dir().ok(),
×
NEW
70
            user_config: jp_config::fs::user_global_config_path(home.as_deref()),
×
NEW
71
            user_workspace: workspace.user_storage_path().map(ToOwned::to_owned),
×
NEW
72
        },
×
NEW
73
        config: config_json.clone(),
×
NEW
74
        args: args.to_vec(),
×
NEW
75
        log_level,
×
NEW
76
    });
×
77

NEW
78
    debug!(%binary, "Spawning plugin.");
×
79

NEW
80
    let mut cmd = Command::new(binary);
×
NEW
81
    cmd.stdin(Stdio::piped())
×
NEW
82
        .stdout(Stdio::piped())
×
NEW
83
        .stderr(Stdio::piped());
×
84

85
    // Prevent the child from receiving SIGINT/SIGTERM directly. The host
86
    // sends `Shutdown` over the protocol instead, giving the plugin a
87
    // chance to exit gracefully.
88
    #[cfg(unix)]
89
    {
90
        use std::os::unix::process::CommandExt as _;
NEW
91
        cmd.process_group(0);
×
92
    }
93

NEW
94
    let mut child = cmd
×
NEW
95
        .spawn()
×
NEW
96
        .map_err(|e| cmd::Error::from(format!("failed to spawn plugin: {e}")))?;
×
97

NEW
98
    let child_stdin = child.stdin.take().expect("stdin piped");
×
NEW
99
    let stdout = child.stdout.take().expect("stdout piped");
×
NEW
100
    let stderr = child.stderr.take().expect("stderr piped");
×
101

102
    // Wrap stdin so the shutdown thread can write to it too.
NEW
103
    let stdin = Arc::new(Mutex::new(child_stdin));
×
104

105
    // Forward stderr to tracing in a background thread.
NEW
106
    let stderr_handle = thread::spawn(move || {
×
NEW
107
        let reader = BufReader::new(stderr);
×
NEW
108
        for line in reader.lines() {
×
NEW
109
            match line {
×
NEW
110
                Ok(line) => trace!(target: "plugin::stderr", "{}", line),
×
NEW
111
                Err(e) => {
×
NEW
112
                    warn!("Error reading plugin stderr: {e}");
×
NEW
113
                    break;
×
114
                }
115
            }
116
        }
NEW
117
    });
×
118

119
    // Shutdown thread: sends `Shutdown` directly to the plugin's stdin
120
    // when a signal arrives. If the plugin doesn't exit within the grace
121
    // period, sends SIGKILL.
NEW
122
    let shutdown_sent = Arc::new(AtomicBool::new(false));
×
NEW
123
    let shutdown_writer = stdin.clone();
×
NEW
124
    let shutdown_flag = shutdown_sent.clone();
×
NEW
125
    let child_id = child.id();
×
NEW
126
    let mut shutdown_rx = signals.receiver.resubscribe();
×
NEW
127
    let shutdown_handle = thread::spawn(move || {
×
NEW
128
        if futures::executor::block_on(shutdown_rx.recv()).is_err() {
×
NEW
129
            return;
×
NEW
130
        }
×
131

132
        // Send Shutdown over the protocol.
NEW
133
        if let Ok(mut writer) = shutdown_writer.lock() {
×
NEW
134
            drop(write_message(&mut *writer, &HostToPlugin::Shutdown));
×
NEW
135
        }
×
NEW
136
        shutdown_flag.store(true, Ordering::Release);
×
137

138
        // Grace period: wait in short intervals so we don't block cleanup
139
        // if the plugin exits promptly.
NEW
140
        for _ in 0..50 {
×
NEW
141
            thread::sleep(std::time::Duration::from_millis(100));
×
NEW
142
            if !is_process_alive(child_id) {
×
NEW
143
                return;
×
NEW
144
            }
×
145
        }
146

NEW
147
        kill_child(child_id);
×
NEW
148
    });
×
149

150
    // Send init.
151
    {
NEW
152
        let mut writer = stdin.lock().expect("stdin lock poisoned");
×
NEW
153
        write_message(&mut *writer, &init)
×
NEW
154
            .map_err(|e| cmd::Error::from(format!("failed to send init: {e}")))?;
×
155
    }
156

157
    // Read messages from plugin.
NEW
158
    let reader = BufReader::new(stdout);
×
NEW
159
    let result = message_loop(reader, &stdin, workspace, &config_json, &shutdown_sent);
×
160

161
    // Always clean up, even on error.
NEW
162
    drop(child.wait());
×
NEW
163
    drop(stderr_handle.join());
×
NEW
164
    drop(shutdown_handle);
×
165

NEW
166
    result
×
NEW
167
}
×
168

169
/// The main message loop: reads plugin requests and sends responses.
170
fn message_loop(
1✔
171
    reader: BufReader<impl std::io::Read>,
1✔
172
    stdin: &Mutex<impl Write>,
1✔
173
    workspace: &Workspace,
1✔
174
    config_json: &Value,
1✔
175
    shutdown_sent: &AtomicBool,
1✔
176
) -> Result<(), cmd::Error> {
1✔
177
    for line in reader.lines() {
2✔
178
        let line =
2✔
179
            line.map_err(|e| cmd::Error::from(format!("failed to read from plugin: {e}")))?;
2✔
180

181
        if line.trim().is_empty() {
2✔
NEW
182
            continue;
×
183
        }
2✔
184

185
        let msg: PluginToHost = serde_json::from_str(&line)
2✔
186
            .map_err(|e| cmd::Error::from(format!("invalid plugin message: {e}: {line}")))?;
2✔
187

188
        trace!(?msg, "Received plugin message.");
2✔
189

190
        let mut writer = stdin.lock().expect("stdin lock poisoned");
2✔
191

192
        match msg {
2✔
193
            PluginToHost::Ready => {
194
                debug!("Plugin signaled ready.");
1✔
195
            }
196

NEW
197
            PluginToHost::ListConversations(req) => {
×
NEW
198
                let response = handle_list_conversations(workspace, req.id);
×
NEW
199
                write_message(&mut *writer, &response)?;
×
200
            }
201

NEW
202
            PluginToHost::ReadEvents(req) => {
×
NEW
203
                let response = handle_read_events(workspace, &req.conversation, req.id);
×
NEW
204
                write_message(&mut *writer, &response)?;
×
205
            }
206

NEW
207
            PluginToHost::ReadConfig(req) => {
×
NEW
208
                let response = handle_read_config(config_json, req.path, req.id);
×
NEW
209
                write_message(&mut *writer, &response)?;
×
210
            }
211

NEW
212
            PluginToHost::Print(print) => {
×
NEW
213
                // In Phase 1, write to stdout directly. Full printer
×
NEW
214
                // integration comes later when we thread through &Printer.
×
NEW
215
                let stdout = std::io::stdout();
×
NEW
216
                let mut handle = stdout.lock();
×
NEW
217
                drop(handle.write_all(print.text.as_bytes()));
×
NEW
218
                drop(handle.flush());
×
NEW
219
            }
×
220

NEW
221
            PluginToHost::Log(log) => {
×
NEW
222
                emit_log(&log);
×
NEW
223
            }
×
224

225
            PluginToHost::Describe(_) => {
NEW
226
                debug!("Ignoring describe in message loop.");
×
227
            }
228

229
            PluginToHost::Exit(exit) => {
1✔
230
                debug!(code = exit.code, "Plugin exited.");
1✔
231
                if exit.code == 0 {
1✔
232
                    return Ok(());
1✔
NEW
233
                }
×
NEW
234
                return match exit.reason {
×
NEW
235
                    Some(reason) => Err(cmd::Error::from((exit.code, reason))),
×
NEW
236
                    None => Err(cmd::Error::from(exit.code)),
×
237
                };
238
            }
239
        }
240
    }
241

242
    // Plugin's stdout closed without an `exit` message. If we sent a
243
    // shutdown, this is expected (the child exited after receiving it).
NEW
244
    if shutdown_sent.load(Ordering::Acquire) {
×
NEW
245
        debug!("Plugin exited after shutdown.");
×
NEW
246
        return Ok(());
×
NEW
247
    }
×
248

NEW
249
    error!("Plugin exited without sending exit message.");
×
NEW
250
    Err(cmd::Error::from((
×
NEW
251
        1u8,
×
NEW
252
        "plugin exited unexpectedly without sending exit message",
×
NEW
253
    )))
×
254
}
1✔
255

NEW
256
fn handle_list_conversations(workspace: &Workspace, req_id: Option<String>) -> HostToPlugin {
×
NEW
257
    let data: Vec<ConversationSummary> = workspace
×
NEW
258
        .conversations()
×
NEW
259
        .map(|(id, meta)| ConversationSummary {
×
NEW
260
            id: id.as_deciseconds().to_string(),
×
NEW
261
            title: meta.title.clone(),
×
NEW
262
            last_activated_at: meta.last_activated_at,
×
NEW
263
            events_count: meta.events_count,
×
NEW
264
        })
×
NEW
265
        .collect();
×
266

NEW
267
    HostToPlugin::Conversations(ConversationsResponse { id: req_id, data })
×
NEW
268
}
×
269

NEW
270
fn handle_read_events(
×
NEW
271
    workspace: &Workspace,
×
NEW
272
    conversation_id: &str,
×
NEW
273
    req_id: Option<String>,
×
NEW
274
) -> HostToPlugin {
×
NEW
275
    let conv_id = match jp_conversation::ConversationId::try_from_deciseconds_str(conversation_id) {
×
NEW
276
        Ok(id) => id,
×
NEW
277
        Err(e) => {
×
NEW
278
            return HostToPlugin::Error(ErrorResponse {
×
NEW
279
                id: req_id,
×
NEW
280
                request: Some("read_events".to_owned()),
×
NEW
281
                message: format!("invalid conversation ID: {e}"),
×
NEW
282
            });
×
283
        }
284
    };
285

NEW
286
    let handle = match workspace.acquire_conversation(&conv_id) {
×
NEW
287
        Ok(h) => h,
×
NEW
288
        Err(e) => {
×
NEW
289
            return HostToPlugin::Error(ErrorResponse {
×
NEW
290
                id: req_id,
×
NEW
291
                request: Some("read_events".to_owned()),
×
NEW
292
                message: format!("conversation not found: {e}"),
×
NEW
293
            });
×
294
        }
295
    };
296

NEW
297
    let events = match workspace.events(&handle) {
×
NEW
298
        Ok(stream) => stream,
×
NEW
299
        Err(e) => {
×
NEW
300
            return HostToPlugin::Error(ErrorResponse {
×
NEW
301
                id: req_id,
×
NEW
302
                request: Some("read_events".to_owned()),
×
NEW
303
                message: format!("failed to load events: {e}"),
×
NEW
304
            });
×
305
        }
306
    };
307

308
    // Serialize events to JSON values, then decode base64-encoded storage
309
    // fields so plugins receive plain text.
NEW
310
    let (_, mut event_values) = match events.to_parts() {
×
NEW
311
        Ok(parts) => parts,
×
NEW
312
        Err(e) => {
×
NEW
313
            return HostToPlugin::Error(ErrorResponse {
×
NEW
314
                id: req_id,
×
NEW
315
                request: Some("read_events".to_owned()),
×
NEW
316
                message: format!("failed to serialize events: {e}"),
×
NEW
317
            });
×
318
        }
319
    };
320

NEW
321
    for value in &mut event_values {
×
NEW
322
        jp_conversation::decode_event_value(value);
×
NEW
323
    }
×
324

NEW
325
    HostToPlugin::Events(EventsResponse {
×
NEW
326
        id: req_id,
×
NEW
327
        conversation: conversation_id.to_owned(),
×
NEW
328
        data: event_values,
×
NEW
329
    })
×
NEW
330
}
×
331

332
fn handle_read_config(
3✔
333
    config_json: &Value,
3✔
334
    path: Option<String>,
3✔
335
    req_id: Option<String>,
3✔
336
) -> HostToPlugin {
3✔
337
    let data = match &path {
3✔
338
        Some(path) => {
2✔
339
            let mut current = config_json;
2✔
340
            for segment in path.split('.') {
3✔
341
                match current.get(segment) {
3✔
342
                    Some(v) => current = v,
2✔
343
                    None => {
344
                        return HostToPlugin::Error(ErrorResponse {
1✔
345
                            id: req_id,
1✔
346
                            request: Some("read_config".to_owned()),
1✔
347
                            message: format!("config path not found: {path}"),
1✔
348
                        });
1✔
349
                    }
350
                }
351
            }
352
            current.clone()
1✔
353
        }
354
        None => config_json.clone(),
1✔
355
    };
356

357
    HostToPlugin::Config(ConfigResponse {
2✔
358
        id: req_id,
2✔
359
        path,
2✔
360
        data,
2✔
361
    })
2✔
362
}
3✔
363

NEW
364
fn emit_log(log: &LogMessage) {
×
NEW
365
    match log.level.as_str() {
×
NEW
366
        "trace" => trace!(target: "plugin", message = %log.message),
×
NEW
367
        "debug" => debug!(target: "plugin", message = %log.message),
×
NEW
368
        "info" => tracing::info!(target: "plugin", message = %log.message),
×
NEW
369
        "warn" => warn!(target: "plugin", message = %log.message),
×
NEW
370
        "error" => error!(target: "plugin", message = %log.message),
×
371
        _ => {
NEW
372
            warn!(target: "plugin", level = %log.level, message = %log.message, "unknown log level");
×
373
        }
374
    }
NEW
375
}
×
376

NEW
377
fn write_message(writer: &mut impl Write, msg: &HostToPlugin) -> Result<(), cmd::Error> {
×
NEW
378
    let json = serde_json::to_string(msg)
×
NEW
379
        .map_err(|e| cmd::Error::from(format!("failed to serialize message: {e}")))?;
×
NEW
380
    writeln!(writer, "{json}")
×
NEW
381
        .map_err(|e| cmd::Error::from(format!("failed to write to plugin stdin: {e}")))?;
×
NEW
382
    writer
×
NEW
383
        .flush()
×
NEW
384
        .map_err(|e| cmd::Error::from(format!("failed to flush plugin stdin: {e}")))?;
×
NEW
385
    Ok(())
×
NEW
386
}
×
387

388
/// Check if a process is still alive by PID.
389
#[cfg(unix)]
NEW
390
fn is_process_alive(pid: u32) -> bool {
×
391
    // kill with signal 0 checks existence without sending a signal.
NEW
392
    unsafe { libc::kill(libc::pid_t::from(pid.cast_signed()), 0) == 0 }
×
NEW
393
}
×
394

395
#[cfg(windows)]
396
fn is_process_alive(pid: u32) -> bool {
397
    use windows_sys::Win32::{
398
        Foundation::{CloseHandle, STILL_ACTIVE},
399
        System::Threading::{GetExitCodeProcess, OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION},
400
    };
401

402
    unsafe {
403
        let handle = OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid);
404
        if handle.is_null() {
405
            return false;
406
        }
407
        let mut exit_code: u32 = 0;
408
        let ok = GetExitCodeProcess(handle, &mut exit_code);
409
        CloseHandle(handle);
410
        ok != 0 && (exit_code as i32) == STILL_ACTIVE
411
    }
412
}
413

414
/// Send SIGKILL to a child process by PID.
415
///
416
/// Used as a last resort when the plugin doesn't exit within the grace
417
/// period after receiving `Shutdown`.
418
#[cfg(unix)]
NEW
419
fn kill_child(pid: u32) {
×
420
    // SAFETY: We're sending a signal to a process we spawned.
NEW
421
    unsafe {
×
NEW
422
        libc::kill(libc::pid_t::from(pid.cast_signed()), libc::SIGKILL);
×
NEW
423
    }
×
NEW
424
    debug!(pid, "Sent SIGKILL to plugin after grace period.");
×
NEW
425
}
×
426

427
#[cfg(windows)]
428
fn kill_child(pid: u32) {
429
    use windows_sys::Win32::{
430
        Foundation::CloseHandle,
431
        System::Threading::{OpenProcess, PROCESS_TERMINATE, TerminateProcess},
432
    };
433

434
    // SAFETY: We're terminating a process we spawned.
435
    unsafe {
436
        let handle = OpenProcess(PROCESS_TERMINATE, 0, pid);
437
        if !handle.is_null() {
438
            TerminateProcess(handle, 1);
439
            CloseHandle(handle);
440
        }
441
    }
442
    debug!(pid, "Sent TerminateProcess to plugin after grace period.");
443
}
444

445
/// Search `$PATH` for a plugin binary matching the given subcommand segments.
446
///
447
/// For `["serve"]`, looks for `jp-serve`. For `["conversation", "export"]`,
448
/// looks for `jp-conversation-export`.
449
pub(crate) fn find_plugin_binary(segments: &[&str]) -> Option<Utf8PathBuf> {
1✔
450
    let name = format!("jp-{}", segments.join("-"));
1✔
451
    which::which(&name)
1✔
452
        .ok()
1✔
453
        .and_then(|p| Utf8PathBuf::from_path_buf(p).ok())
1✔
454
}
1✔
455

456
/// Find any existing plugin binary without downloading or prompting.
457
///
458
/// Checks the install directory first, then `$PATH`. Used for non-mutating
459
/// operations like help requests.
NEW
460
pub(crate) fn find_any_plugin_binary(name: &str) -> Option<Utf8PathBuf> {
×
NEW
461
    if let Some(path) = registry::find_installed(name) {
×
NEW
462
        return Some(path);
×
NEW
463
    }
×
NEW
464
    let segments: Vec<&str> = name.split('-').collect();
×
NEW
465
    find_plugin_binary(&segments)
×
NEW
466
}
×
467

468
/// Resolve a plugin binary through multiple sources:
469
///
470
/// 1. User-local install directory (previously installed plugins)
471
/// 2. Plugin registry (auto-install if official, prompt if third-party)
472
/// 3. `$PATH` (with approval check for unapproved plugins)
473
///
474
/// The `plugins_config` drives installation and execution policy. Per-plugin
475
/// settings override the defaults from the registry (official vs third-party).
NEW
476
pub(crate) fn resolve_plugin_binary(
×
NEW
477
    name: &str,
×
NEW
478
    runtime: &tokio::runtime::Runtime,
×
NEW
479
    plugins_config: &PluginsConfig,
×
NEW
480
    is_tty: bool,
×
NEW
481
) -> Result<Utf8PathBuf, cmd::Error> {
×
NEW
482
    let plugin_cfg = plugins_config.command.get(name);
×
483

484
    // Explicit deny in config.
NEW
485
    if plugin_cfg.is_some_and(|c| c.run == Some(RunPolicy::Deny)) {
×
NEW
486
        return Err(cmd::Error::from(format!(
×
NEW
487
            "plugin `{name}` is denied by configuration (plugins.command.{name}.run = \"deny\")"
×
NEW
488
        )));
×
NEW
489
    }
×
490

491
    // 1. Already installed locally.
NEW
492
    if let Some(path) = registry::find_installed(name) {
×
NEW
493
        debug!(name, %path, "Found installed plugin.");
×
NEW
494
        verify_checksum(name, &path, plugin_cfg)?;
×
NEW
495
        return Ok(path);
×
NEW
496
    }
×
497

498
    // 2. Check registry.
NEW
499
    if let Some(path) = try_registry_install(name, runtime, plugins_config, is_tty)? {
×
NEW
500
        return Ok(path);
×
NEW
501
    }
×
502

503
    // 3. Check $PATH with run policy.
NEW
504
    let segments: Vec<&str> = name.split('-').collect();
×
NEW
505
    if let Some(path) = find_plugin_binary(&segments) {
×
NEW
506
        check_run_policy(name, &path, plugin_cfg, is_tty)?;
×
NEW
507
        return Ok(path);
×
NEW
508
    }
×
509

NEW
510
    Err(cmd::Error::from(format!(
×
NEW
511
        "unknown command `{name}`. No built-in command, registry plugin, or `jp-{name}` binary \
×
NEW
512
         found on $PATH."
×
NEW
513
    )))
×
NEW
514
}
×
515

516
/// Verify a binary's checksum against the config-pinned value, if any.
NEW
517
fn verify_checksum(
×
NEW
518
    name: &str,
×
NEW
519
    binary_path: &Utf8Path,
×
NEW
520
    plugin_cfg: Option<&CommandPluginConfig>,
×
NEW
521
) -> Result<(), cmd::Error> {
×
NEW
522
    let Some(checksum) = plugin_cfg.and_then(|c| c.checksum.as_ref()) else {
×
NEW
523
        return Ok(());
×
524
    };
525

NEW
526
    let actual = registry::sha256_file(binary_path)?;
×
NEW
527
    if actual != checksum.value {
×
NEW
528
        return Err(cmd::Error::from(format!(
×
NEW
529
            "plugin `{name}` binary checksum mismatch.\nexpected: {}\nactual:   {actual}\nThe \
×
NEW
530
             binary at {binary_path} has changed since it was pinned. Update \
×
NEW
531
             plugins.command.{name}.checksum.value in your config to accept the new binary.",
×
NEW
532
            checksum.value,
×
NEW
533
        )));
×
NEW
534
    }
×
535

NEW
536
    Ok(())
×
NEW
537
}
×
538

539
/// Try to install a plugin from the cached registry.
NEW
540
fn try_registry_install(
×
NEW
541
    name: &str,
×
NEW
542
    runtime: &tokio::runtime::Runtime,
×
NEW
543
    plugins_config: &PluginsConfig,
×
NEW
544
    is_tty: bool,
×
NEW
545
) -> Result<Option<Utf8PathBuf>, cmd::Error> {
×
NEW
546
    let Some(reg) = registry::load_cached() else {
×
NEW
547
        return Ok(None);
×
548
    };
549

550
    // Find the registry entry whose `id` matches the requested name.
551
    // In Phase 5, this will use the command path (registry key) for
552
    // multi-segment routing. For now, we match on `id`.
NEW
553
    let Some(plugin) = reg.plugins.values().find(|p| p.id == name) else {
×
NEW
554
        return Ok(None);
×
555
    };
556

557
    // Only handle command plugins.
NEW
558
    let jp_plugin::registry::PluginKind::Command { ref binaries, .. } = plugin.kind else {
×
NEW
559
        return Ok(None);
×
560
    };
561

NEW
562
    let target = registry::current_target();
×
NEW
563
    let Some(binary_info) = binaries.get(&target) else {
×
NEW
564
        return Ok(None);
×
565
    };
566

NEW
567
    let id = &plugin.id;
×
NEW
568
    let plugin_cfg = plugins_config.command.get(id);
×
569

570
    // Check if auto-install is allowed.
NEW
571
    let auto_install = plugin_cfg
×
NEW
572
        .and_then(|c| c.install)
×
NEW
573
        .unwrap_or(plugins_config.auto_install);
×
574

NEW
575
    if !auto_install && !plugin.official {
×
NEW
576
        return Ok(None);
×
NEW
577
    }
×
578

579
    // Determine run policy: config > registry default.
NEW
580
    let run_policy = plugin_cfg
×
NEW
581
        .and_then(|c| c.run)
×
NEW
582
        .unwrap_or(if plugin.official {
×
NEW
583
            RunPolicy::Unattended
×
584
        } else {
NEW
585
            RunPolicy::Ask
×
586
        });
587

NEW
588
    match run_policy {
×
589
        RunPolicy::Deny => {
NEW
590
            return Err(cmd::Error::from(format!(
×
NEW
591
                "plugin `{id}` is denied by configuration"
×
NEW
592
            )));
×
593
        }
594
        RunPolicy::Ask => {
NEW
595
            if !is_tty {
×
NEW
596
                return Err(cmd::Error::from(format!(
×
NEW
597
                    "plugin `{id}` requires approval. Run `jp plugin install {id}` first, or set \
×
NEW
598
                     plugins.command.{id}.run = \"unattended\" in config."
×
NEW
599
                )));
×
NEW
600
            }
×
601

NEW
602
            drop(writeln!(
×
NEW
603
                std::io::stderr(),
×
604
                "  \u{2192} Plugin `{id}` found in registry."
605
            ));
NEW
606
            let options = vec!["yes", "no"];
×
NEW
607
            let answer = inquire::Select::new("Install and run it?", options)
×
NEW
608
                .prompt()
×
NEW
609
                .map_err(|e| cmd::Error::from(format!("prompt failed: {e}")))?;
×
610

NEW
611
            if answer == "no" {
×
NEW
612
                return Err(cmd::Error::from("plugin execution cancelled"));
×
NEW
613
            }
×
614
        }
NEW
615
        RunPolicy::Unattended => {}
×
616
    }
617

NEW
618
    drop(writeln!(
×
NEW
619
        std::io::stderr(),
×
620
        "  \u{2192} Installing jp-{id} for {target}..."
621
    ));
NEW
622
    let data = runtime.block_on(async {
×
NEW
623
        let client = reqwest::Client::new();
×
NEW
624
        registry::download_and_verify(&client, binary_info).await
×
NEW
625
    })?;
×
626

NEW
627
    let path = registry::install_binary(id, &data)?;
×
NEW
628
    drop(writeln!(
×
NEW
629
        std::io::stderr(),
×
630
        "  \u{2192} Installed to {path}",
631
    ));
632

633
    // Verify against pinned checksum if configured.
NEW
634
    verify_checksum(id, &path, plugin_cfg)?;
×
635

NEW
636
    Ok(Some(path))
×
NEW
637
}
×
638

639
/// Check run policy for a `$PATH`-discovered plugin.
NEW
640
fn check_run_policy(
×
NEW
641
    name: &str,
×
NEW
642
    binary_path: &Utf8Path,
×
NEW
643
    plugin_cfg: Option<&CommandPluginConfig>,
×
NEW
644
    is_tty: bool,
×
NEW
645
) -> Result<(), cmd::Error> {
×
646
    // Verify pinned checksum first.
NEW
647
    verify_checksum(name, binary_path, plugin_cfg)?;
×
648

NEW
649
    let run_policy = plugin_cfg.and_then(|c| c.run).unwrap_or(RunPolicy::Ask);
×
650

NEW
651
    match run_policy {
×
NEW
652
        RunPolicy::Unattended => Ok(()),
×
NEW
653
        RunPolicy::Deny => Err(cmd::Error::from(format!(
×
NEW
654
            "plugin `{name}` is denied by configuration"
×
NEW
655
        ))),
×
656
        RunPolicy::Ask => {
NEW
657
            if !is_tty {
×
NEW
658
                return Err(cmd::Error::from(format!(
×
NEW
659
                    "plugin `jp-{name}` found on $PATH but requires approval. Set \
×
NEW
660
                     plugins.command.{name}.run = \"unattended\" in config, or run `jp {name}` in \
×
NEW
661
                     a terminal."
×
NEW
662
                )));
×
NEW
663
            }
×
664

NEW
665
            drop(writeln!(
×
NEW
666
                std::io::stderr(),
×
667
                "  \u{2192} Found jp-{name} on $PATH ({binary_path})",
668
            ));
NEW
669
            let options = vec!["yes", "no"];
×
NEW
670
            let answer = inquire::Select::new("Run it?", options)
×
NEW
671
                .prompt()
×
NEW
672
                .map_err(|e| cmd::Error::from(format!("prompt failed: {e}")))?;
×
673

NEW
674
            if answer == "no" {
×
NEW
675
                return Err(cmd::Error::from("plugin execution denied"));
×
NEW
676
            }
×
NEW
677
            Ok(())
×
678
        }
679
    }
NEW
680
}
×
681

682
/// Send a `Describe` request to a plugin and return its metadata.
683
///
684
/// Spawns the binary, sends `{"type":"describe"}`, reads one response line,
685
/// and returns the parsed [`DescribeResponse`]. Returns `None` if the plugin
686
/// doesn't support describe or fails to respond.
NEW
687
pub(crate) fn describe_plugin(binary: &Utf8Path) -> Option<DescribeResponse> {
×
NEW
688
    let mut child = Command::new(binary)
×
NEW
689
        .stdin(Stdio::piped())
×
NEW
690
        .stdout(Stdio::piped())
×
NEW
691
        .stderr(Stdio::null())
×
NEW
692
        .spawn()
×
NEW
693
        .ok()?;
×
694

NEW
695
    let mut child_stdin = child.stdin.take()?;
×
NEW
696
    let child_stdout = child.stdout.take()?;
×
697

698
    // Send describe request.
NEW
699
    let json = serde_json::to_string(&HostToPlugin::Describe).ok()?;
×
NEW
700
    writeln!(child_stdin, "{json}").ok()?;
×
NEW
701
    child_stdin.flush().ok()?;
×
NEW
702
    drop(child_stdin); // Signal no more messages.
×
703

704
    // Read one line response.
NEW
705
    let mut reader = BufReader::new(child_stdout);
×
NEW
706
    let mut line = String::new();
×
NEW
707
    reader.read_line(&mut line).ok()?;
×
708

NEW
709
    drop(child.wait());
×
710

NEW
711
    if line.trim().is_empty() {
×
NEW
712
        return None;
×
NEW
713
    }
×
714

NEW
715
    let msg: PluginToHost = serde_json::from_str(line.trim()).ok()?;
×
NEW
716
    match msg {
×
NEW
717
        PluginToHost::Describe(resp) => Some(resp),
×
NEW
718
        _ => None,
×
719
    }
NEW
720
}
×
721

722
/// Discover plugin binaries on `$PATH` and in the user-local install directory.
723
///
724
/// Returns `(subcommand_name, binary_path)` pairs, sorted by name.
725
/// For a binary named `jp-serve`, the subcommand name is `serve`.
726
/// Installed plugins take priority over `$PATH` duplicates.
NEW
727
pub(crate) fn discover_plugins() -> Vec<(String, Utf8PathBuf)> {
×
NEW
728
    let path_var = std::env::var_os("PATH").unwrap_or_default();
×
NEW
729
    let mut seen = HashSet::new();
×
NEW
730
    let mut plugins = Vec::new();
×
731

732
    // Scan install directory first so installed plugins take priority.
NEW
733
    if let Some(bin_dir) = registry::bin_dir() {
×
NEW
734
        scan_dir_for_plugins(&bin_dir, &mut seen, &mut plugins);
×
NEW
735
    }
×
736

NEW
737
    for dir in std::env::split_paths(&path_var) {
×
NEW
738
        let Some(dir) = Utf8Path::from_path(&dir) else {
×
NEW
739
            continue;
×
740
        };
741

NEW
742
        scan_dir_for_plugins(dir, &mut seen, &mut plugins);
×
743
    }
744

NEW
745
    plugins.sort_by(|a, b| a.0.cmp(&b.0));
×
NEW
746
    plugins
×
NEW
747
}
×
748

NEW
749
fn scan_dir_for_plugins(
×
NEW
750
    dir: &Utf8Path,
×
NEW
751
    seen: &mut HashSet<String>,
×
NEW
752
    plugins: &mut Vec<(String, Utf8PathBuf)>,
×
NEW
753
) {
×
NEW
754
    let Ok(entries) = dir.read_dir_utf8() else {
×
NEW
755
        return;
×
756
    };
NEW
757
    for entry in entries.flatten() {
×
NEW
758
        let name = entry.file_name();
×
NEW
759
        let Some(subcommand) = name.strip_prefix("jp-") else {
×
NEW
760
            continue;
×
761
        };
762

763
        // On Windows, strip the .exe extension.
764
        #[cfg(windows)]
765
        let subcommand = subcommand.strip_suffix(".exe").unwrap_or(subcommand);
766

767
        // On Unix, skip non-executable files.
768
        #[cfg(unix)]
769
        {
770
            use std::os::unix::fs::PermissionsExt as _;
NEW
771
            let Ok(meta) = entry.metadata() else {
×
NEW
772
                continue;
×
773
            };
NEW
774
            if meta.permissions().mode() & 0o111 == 0 {
×
NEW
775
                continue;
×
NEW
776
            }
×
777
        }
778

NEW
779
        if seen.insert(subcommand.to_owned()) {
×
NEW
780
            plugins.push((subcommand.to_owned(), entry.into_path()));
×
NEW
781
        }
×
782
    }
NEW
783
}
×
784

785
#[cfg(test)]
786
#[path = "dispatch_tests.rs"]
787
mod tests;
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE TRIAL · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc