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

dcdpr / jp / 24128084162

08 Apr 2026 09:24AM UTC coverage: 62.329% (-2.0%) from 64.307%
24128084162

Pull #521

github

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

118 of 1184 new or added lines in 15 files covered. (9.97%)

5 existing lines in 2 files now uncovered.

20176 of 32370 relevant lines covered (62.33%)

256.72 hits per line

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

10.99
/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
    path::Path,
10
    process::{Command, Stdio},
11
    sync::{
12
        Arc, Mutex,
13
        atomic::{AtomicBool, Ordering},
14
    },
15
    thread,
16
};
17

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

34
use super::registry;
35
use crate::{cmd, signals::SignalPair};
36

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

NEW
52
    let storage_path = workspace
×
NEW
53
        .storage_path()
×
NEW
54
        .ok_or_else(|| cmd::Error::from("workspace has no storage configured"))?;
×
55

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

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

NEW
75
    debug!(binary = %binary.display(), "Spawning plugin.");
×
76

NEW
77
    let mut cmd = Command::new(binary);
×
NEW
78
    cmd.stdin(Stdio::piped())
×
NEW
79
        .stdout(Stdio::piped())
×
NEW
80
        .stderr(Stdio::piped());
×
81

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

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

NEW
95
    let child_stdin = child.stdin.take().expect("stdin piped");
×
NEW
96
    let stdout = child.stdout.take().expect("stdout piped");
×
NEW
97
    let stderr = child.stderr.take().expect("stderr piped");
×
98

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

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

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

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

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

NEW
144
        kill_child(child_id);
×
NEW
145
    });
×
146

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

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

158
    // Always clean up, even on error.
NEW
159
    drop(child.wait());
×
NEW
160
    drop(stderr_handle.join());
×
NEW
161
    drop(shutdown_handle);
×
162

NEW
163
    result
×
NEW
164
}
×
165

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

178
        if line.trim().is_empty() {
2✔
NEW
179
            continue;
×
180
        }
2✔
181

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

185
        trace!(?msg, "Received plugin message.");
2✔
186

187
        let mut writer = stdin.lock().expect("stdin lock poisoned");
2✔
188

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

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

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

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

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

NEW
218
            PluginToHost::Log(log) => {
×
NEW
219
                emit_log(&log);
×
NEW
220
            }
×
221

222
            PluginToHost::Describe(_) => {
NEW
223
                debug!("Ignoring describe in message loop.");
×
224
            }
225

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

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

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

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

NEW
264
    HostToPlugin::Conversations(ConversationsResponse { id: req_id, data })
×
NEW
265
}
×
266

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

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

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

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

NEW
318
    for value in &mut event_values {
×
NEW
319
        jp_conversation::decode_event_value(value);
×
NEW
320
    }
×
321

NEW
322
    HostToPlugin::Events(EventsResponse {
×
NEW
323
        id: req_id,
×
NEW
324
        conversation: conversation_id.to_owned(),
×
NEW
325
        data: event_values,
×
NEW
326
    })
×
NEW
327
}
×
328

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

354
    HostToPlugin::Config(ConfigResponse {
2✔
355
        id: req_id,
2✔
356
        path,
2✔
357
        data,
2✔
358
    })
2✔
359
}
3✔
360

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

NEW
534
    Ok(())
×
NEW
535
}
×
536

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

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

555
    // Only handle command plugins.
NEW
556
    if plugin.kind != jp_plugin::registry::PluginKind::Command {
×
NEW
557
        return Ok(None);
×
NEW
558
    }
×
559

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

NEW
664
            drop(writeln!(
×
NEW
665
                std::io::stderr(),
×
666
                "  \u{2192} Found jp-{name} on $PATH ({})",
NEW
667
                binary_path.display()
×
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: &Path) -> 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, std::path::PathBuf)> {
×
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.as_std_path(), &mut seen, &mut plugins);
×
NEW
735
    }
×
736

NEW
737
    for dir in std::env::split_paths(&path_var) {
×
NEW
738
        scan_dir_for_plugins(&dir, &mut seen, &mut plugins);
×
NEW
739
    }
×
740

NEW
741
    plugins.sort_by(|a, b| a.0.cmp(&b.0));
×
NEW
742
    plugins
×
NEW
743
}
×
744

NEW
745
fn scan_dir_for_plugins(
×
NEW
746
    dir: &std::path::Path,
×
NEW
747
    seen: &mut HashSet<String>,
×
NEW
748
    plugins: &mut Vec<(String, std::path::PathBuf)>,
×
NEW
749
) {
×
NEW
750
    let Ok(entries) = std::fs::read_dir(dir) else {
×
NEW
751
        return;
×
752
    };
NEW
753
    for entry in entries.flatten() {
×
NEW
754
        let file_name = entry.file_name();
×
NEW
755
        let Some(name) = file_name.to_str() else {
×
NEW
756
            continue;
×
757
        };
NEW
758
        let Some(subcommand) = name.strip_prefix("jp-") else {
×
NEW
759
            continue;
×
760
        };
761

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

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

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

784
#[cfg(test)]
785
#[path = "dispatch_tests.rs"]
786
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