diff --git a/crates/jp_cli/src/cmd/plugin/dispatch.rs b/crates/jp_cli/src/cmd/plugin/dispatch.rs index 8802b88a3..95bc86a78 100644 --- a/crates/jp_cli/src/cmd/plugin/dispatch.rs +++ b/crates/jp_cli/src/cmd/plugin/dispatch.rs @@ -4,55 +4,230 @@ //! the plugin sends `exit` or the process terminates. use std::{ - collections::HashSet, + collections::{BTreeSet, HashSet}, + fmt::Write as _, + fs, io::{BufRead, BufReader, Write}, - process::{Command, Stdio}, + process::{Child, ChildStdin, ChildStdout, Command, Stdio}, sync::{ Arc, Mutex, atomic::{AtomicBool, Ordering}, }, - thread, + thread::{self, JoinHandle}, }; use camino::{Utf8Path, Utf8PathBuf}; use jp_config::{ AppConfig, + fs::user_global_config_dir, + interrupt::{StreamingInterruptAction, ToolInterruptAction}, plugins::{ PluginsConfig, command::{CommandPluginConfig, RunPolicy}, }, + util::list_configs_in_load_path, }; -use jp_inquire::{InlineOption, InlineSelect}; +use jp_conversation::{ConversationId, ConversationStream, event::ChatRequest}; +use jp_editor::{EditOutcome, EditorBackend}; +use jp_inquire::{InlineOption, InlineReply, InlineSelect, ReplyOutcome}; use jp_plugin::{ PROTOCOL_VERSION, message::{ - ConfigResponse, ConversationSummary, ConversationsResponse, DescribeResponse, - ErrorResponse, EventsResponse, HostToPlugin, InitMessage, LogMessage, PathsInfo, - PluginToHost, WorkspaceInfo, + ComposeMode, ComposeOption, ComposeRequest, ComposeResponse, ConfigEntry, ConfigResponse, + ConfigsResponse, ConversationSummary, ConversationsResponse, CreatedResponse, + DescribeResponse, DoneResponse, DraftResponse, ErrorResponse, EventsResponse, HostToPlugin, + InitMessage, LogMessage, PathsInfo, PluginToHost, QueryCompleteResponse, QueryRequest, + SetTitleRequest, WorkspaceInfo, WriteDraftRequest, }, }; -use jp_workspace::Workspace; +use jp_printer::Printer; +use jp_storage::backend::FsStorageBackend; +use jp_workspace::{ConversationLock, LockResult, Workspace, session::Session}; +use relative_path::RelativePath; use serde_json::Value; -use tracing::{debug, error, trace, warn}; +use sha2::{Digest as _, Sha256}; +use tokio::sync::mpsc; +use tracing::{debug, error, info, trace, warn}; use super::registry; -use crate::{Ctx, cmd, signals::SignalRouter}; +use crate::{ + Ctx, KeyValueOrPath, cmd, + cmd::query::{PendingStreamTrim, TurnInputs}, + config_pipeline::build_partial_over, + editor::report_editor_failure, + signals::SignalRouter, +}; + +/// Runs the prompts a plugin asks for. +/// +/// Composition lives on this side of the protocol because the host owns both +/// ends of it: the plugin's stdin carries the protocol, so it has no terminal +/// to read keys from, and only the host knows which editor `Ctrl+X` opens. +/// +/// Holds a handle on the printer rather than a borrow, so the message loop it +/// belongs to can take the context mutably. +pub(crate) struct Composer { + printer: Arc, + editor: Option>, + is_tty: bool, +} + +impl Composer { + /// Collect what the request asks for, or nothing if the user declines. + fn compose(&self, request: &ComposeRequest) -> ComposeResponse { + let mut response = ComposeResponse { + id: request.id.clone(), + text: None, + values: vec![], + }; + + if !self.is_tty { + debug!("Plugin asked to compose without a terminal to ask on."); + return response; + } + + match &request.mode { + ComposeMode::MultiSelect { options } => { + response.values = self.ask_many(request, options); + } + _ => response.text = self.ask(request), + } + + response + } + + /// Pick any number of the offered options. + fn ask_many(&self, request: &ComposeRequest, options: &[ComposeOption]) -> Vec { + if options.is_empty() { + return vec![]; + } + + let labels: Vec<&str> = options.iter().map(|o| o.label.as_str()).collect(); + let mut writer = self.printer.prompt_writer(); + let mut prompt = inquire::MultiSelect::new(&request.message, labels); + if let Some(help) = &request.help { + prompt = prompt.with_help_message(help); + } + + let Ok(chosen) = prompt.raw_prompt_with_writer(&mut writer) else { + return vec![]; + }; + + chosen + .into_iter() + .filter_map(|item| options.get(item.index).map(|o| o.value.clone())) + .collect() + } + + fn ask(&self, request: &ComposeRequest) -> Option { + match &request.mode { + ComposeMode::Line { default } => { + let mut writer = self.printer.prompt_writer(); + let mut prompt = inquire::Text::new(&request.message); + if let Some(default) = default { + prompt = prompt.with_initial_value(default); + } + if let Some(help) = &request.help { + prompt = prompt.with_help_message(help); + } + + prompt.prompt_with_writer(&mut writer).ok() + } + // Answered by `ask_many`, which `compose` routes to first. + ComposeMode::MultiSelect { .. } => None, + ComposeMode::Buffer { initial_text } => { + self.buffer(request, initial_text.as_deref().unwrap_or_default()) + } + ComposeMode::Select { options, default } => { + if options.is_empty() { + return None; + } + + let labels: Vec<&str> = options.iter().map(|o| o.label.as_str()).collect(); + let start = default + .as_ref() + .and_then(|value| options.iter().position(|o| &o.value == value)) + .unwrap_or(0); + + let mut writer = self.printer.prompt_writer(); + let mut prompt = + inquire::Select::new(&request.message, labels).with_starting_cursor(start); + if let Some(help) = &request.help { + prompt = prompt.with_help_message(help); + } + + let chosen = prompt.raw_prompt_with_writer(&mut writer).ok()?; + + options.get(chosen.index).map(|o| o.value.clone()) + } + } + } + + /// The multi-line widget, looping so the editor escape returns to it. + fn buffer(&self, request: &ComposeRequest, initial: &str) -> Option { + let mut buffer = initial.to_owned(); + + loop { + let mut reply = InlineReply::new(request.message.as_str()) + .with_initial_text(buffer.as_str()) + .with_editor_escape(self.editor.is_some()); + if let Some(help) = &request.help { + reply = reply.with_help_message(help.as_str()); + } + + match reply.prompt(Box::new(self.printer.owned_prompt_writer())) { + Ok(ReplyOutcome::Submit(text)) => return Some(text), + Ok(ReplyOutcome::Cancelled) => return None, + Ok(ReplyOutcome::OpenEditor { current_text }) => { + // Whatever was typed before `Ctrl+X` seeds the editor, and + // whatever comes back seeds the widget again. + buffer = current_text; + let Some(editor) = self.editor.as_ref() else { + continue; + }; + match editor.edit_text(&buffer) { + Ok((EditOutcome::Saved, edited)) => buffer = edited, + Ok((EditOutcome::Cancelled, _)) => {} + Err(error) => report_editor_failure( + &self.printer, + &error, + "Continuing with the inline editor.", + ), + } + } + Err(error) => { + warn!(%error, "Inline composition failed."); + return None; + } + } + } + } +} /// Run a plugin binary, handling the full protocol lifecycle. /// /// `binary` is the path to the plugin executable. /// `args` are the remaining CLI arguments to forward. -pub(crate) fn run_plugin( +pub(crate) async fn run_plugin( name: &str, binary: &Utf8Path, args: &[String], - workspace: &Workspace, - storage_path: Option<&Utf8Path>, - user_storage_path: Option<&Utf8Path>, - config: &Arc, - signals: &SignalRouter, - log_level: u8, + ctx: &mut Ctx, ) -> Result<(), cmd::Error> { + let config = ctx.config(); + let log_level = ctx.term.args.verbose; + + // Owned before the workspace is borrowed mutably: both read through `&self`, + // so they would otherwise hold a borrow of all of `ctx`. + let storage_path = ctx.storage_path().map(ToOwned::to_owned); + let user_storage_path = ctx.user_storage_path().map(ToOwned::to_owned); + + let composer = Composer { + printer: ctx.printer.clone(), + editor: crate::editor::build_editor_backend(&config.editor), + is_tty: ctx.term.is_tty, + }; + let config_json = serde_json::to_value(config.as_ref().to_partial()) .map_err(|e| cmd::Error::from(format!("failed to serialize config: {e}")))?; @@ -67,26 +242,106 @@ pub(crate) fn run_plugin( let storage_path = storage_path.ok_or("workspace has no storage configured")?; - let home = std::env::home_dir().and_then(|p| camino::Utf8PathBuf::from_path_buf(p).ok()); - let init = HostToPlugin::Init(InitMessage { version: PROTOCOL_VERSION, workspace: WorkspaceInfo { - root: workspace.root().to_owned(), - storage: storage_path.to_owned(), - id: workspace.id().to_string(), - }, - paths: PathsInfo { - user_data: jp_workspace::user_data_dir().ok(), - user_config: jp_config::fs::user_global_config_dir(home.as_deref()), - user_workspace: user_storage_path.map(ToOwned::to_owned), + root: ctx.workspace.root().to_owned(), + storage: storage_path.clone(), + id: ctx.workspace.id().to_string(), }, + paths: well_known_paths(user_storage_path.as_deref()), config: config_json.clone(), options, args: args.to_vec(), log_level, }); + let PluginProcess { + mut child, + stdin, + stdout, + stderr_handle, + } = spawn_plugin(binary)?; + + // Shutdown thread: sends `Shutdown` directly to the plugin's stdin when + // an interrupt or a graceful shutdown request arrives. If the plugin + // doesn't exit within the grace period, sends SIGKILL. + // + // The guard drops when this function returns; the thread then sees the + // notification channel close and exits. + let (_interrupt_guard, mut interrupt_rx) = ctx.signals.push_handler(); + let shutdown_token = ctx.signals.shutdown_token(); + let shutdown_sent = Arc::new(AtomicBool::new(false)); + let shutdown_writer = stdin.clone(); + let shutdown_flag = shutdown_sent.clone(); + let child_id = child.id(); + let shutdown_handle = thread::spawn(move || { + let interrupted = futures::executor::block_on(async { + tokio::select! { + notified = interrupt_rx.recv() => notified.is_some(), + () = shutdown_token.cancelled() => true, + } + }); + + // The plugin run completed and deregistered its handler. + if !interrupted { + return; + } + + stop_plugin(&shutdown_writer, &shutdown_flag, child_id); + }); + + // Send init. + { + let mut writer = stdin.lock().expect("stdin lock poisoned"); + write_message(&mut *writer, &init) + .map_err(|e| cmd::Error::from(format!("failed to send init: {e}")))?; + } + + // Read on a thread of its own, so a turn awaiting the provider cannot stop + // the host from noticing what the plugin says next. + let (mut requests, reader_thread) = spawn_reader(stdout); + + let result = message_loop( + &mut requests, + &stdin, + ctx, + &config_json, + &shutdown_sent, + &composer, + ) + .await; + + // Always clean up, even on error. + drop(child.wait()); + drop(stderr_handle.join()); + drop(reader_thread); + drop(shutdown_handle); + + result +} + +/// How long a plugin gets to exit on its own after being told to stop. +const SHUTDOWN_GRACE_MS: u64 = 5_000; + +/// A spawned plugin process and its wired-up pipes. +struct PluginProcess { + child: Child, + + /// Shared, because the shutdown thread writes to it as well as the message + /// loop. + stdin: Arc>, + stdout: ChildStdout, + + /// Joins when the plugin's stderr closes. + stderr_handle: thread::JoinHandle<()>, +} + +/// Spawn the plugin, wiring its three pipes. +/// +/// Its stderr is forwarded to tracing from a thread, so a plugin that logs +/// heavily cannot fill the pipe and block on a write nobody is draining. +fn spawn_plugin(binary: &Utf8Path) -> Result { debug!(%binary, "Spawning plugin."); let mut cmd = Command::new(binary); @@ -111,10 +366,6 @@ pub(crate) fn run_plugin( let stdout = child.stdout.take().expect("stdout piped"); let stderr = child.stderr.take().expect("stderr piped"); - // Wrap stdin so the shutdown thread can write to it too. - let stdin = Arc::new(Mutex::new(child_stdin)); - - // Forward stderr to tracing in a background thread. let stderr_handle = thread::spawn(move || { let reader = BufReader::new(stderr); for line in reader.lines() { @@ -128,80 +379,96 @@ pub(crate) fn run_plugin( } }); - // Shutdown thread: sends `Shutdown` directly to the plugin's stdin when - // an interrupt or a graceful shutdown request arrives. If the plugin - // doesn't exit within the grace period, sends SIGKILL. - // - // The guard drops when this function returns; the thread then sees the - // notification channel close and exits. - let (_interrupt_guard, mut interrupt_rx) = signals.push_handler(); - let shutdown_token = signals.shutdown_token(); - let shutdown_sent = Arc::new(AtomicBool::new(false)); - let shutdown_writer = stdin.clone(); - let shutdown_flag = shutdown_sent.clone(); - let child_id = child.id(); - let shutdown_handle = thread::spawn(move || { - let interrupted = futures::executor::block_on(async { - tokio::select! { - notified = interrupt_rx.recv() => notified.is_some(), - () = shutdown_token.cancelled() => true, - } - }); + Ok(PluginProcess { + child, + stdin: Arc::new(Mutex::new(child_stdin)), + stdout, + stderr_handle, + }) +} - // The plugin run completed and deregistered its handler. - if !interrupted { - return; - } +/// Ask the plugin to stop, and kill it if it will not. +/// +/// `shutdown_sent` is raised once the request is out, so a stdout that closes +/// without an `exit` is read as the plugin obeying rather than as a crash. +/// It is raised after the write for that reason: before it, a closed stdout +/// still means something went wrong. +fn stop_plugin(stdin: &Mutex, shutdown_sent: &AtomicBool, child_id: u32) { + if let Ok(mut writer) = stdin.lock() { + drop(write_message(&mut *writer, &HostToPlugin::Shutdown)); + } + shutdown_sent.store(true, Ordering::Release); - // Send Shutdown over the protocol. - if let Ok(mut writer) = shutdown_writer.lock() { - drop(write_message(&mut *writer, &HostToPlugin::Shutdown)); + // Polled in short intervals so a prompt exit doesn't hold up cleanup. + let interval = std::time::Duration::from_millis(100); + for _ in 0..(SHUTDOWN_GRACE_MS / 100) { + thread::sleep(interval); + if !is_process_alive(child_id) { + return; } - shutdown_flag.store(true, Ordering::Release); + } - // Grace period: wait in short intervals so we don't block cleanup - // if the plugin exits promptly. - for _ in 0..50 { - thread::sleep(std::time::Duration::from_millis(100)); - if !is_process_alive(child_id) { - return; - } - } + kill_child(child_id); +} - kill_child(child_id); - }); +/// The JP directories a plugin is told about, so it needs no platform logic of +/// its own. +fn well_known_paths(user_storage_path: Option<&Utf8Path>) -> PathsInfo { + let home = std::env::home_dir().and_then(|p| camino::Utf8PathBuf::from_path_buf(p).ok()); - // Send init. - { - let mut writer = stdin.lock().expect("stdin lock poisoned"); - write_message(&mut *writer, &init) - .map_err(|e| cmd::Error::from(format!("failed to send init: {e}")))?; + PathsInfo { + user_data: jp_workspace::user_data_dir().ok(), + user_config: jp_config::fs::user_global_config_dir(home.as_deref()), + user_workspace: user_storage_path.map(ToOwned::to_owned), } +} - // Read messages from plugin. - let reader = BufReader::new(stdout); - let result = message_loop(reader, &stdin, workspace, &config_json, &shutdown_sent); - - // Always clean up, even on error. - drop(child.wait()); - drop(stderr_handle.join()); - drop(shutdown_handle); +/// Read plugin messages on a dedicated thread, forwarding each line. +/// +/// The channel closes when the plugin's stdout does, which ends the message +/// loop. +/// A full channel blocks the reader rather than dropping messages: the plugin +/// is waiting on replies to most of what it sends, so losing one would hang it. +fn spawn_reader( + stdout: impl std::io::Read + Send + 'static, +) -> (mpsc::Receiver, JoinHandle<()>) { + let (tx, rx) = mpsc::channel(64); + + let handle = thread::spawn(move || { + let reader = BufReader::new(stdout); + for line in reader.lines() { + match line { + Ok(line) => { + if tx.blocking_send(line).is_err() { + break; + } + } + Err(error) => { + warn!(%error, "Error reading from plugin."); + break; + } + } + } + }); - result + (rx, handle) } /// The main message loop: reads plugin requests and sends responses. -fn message_loop( - reader: BufReader, - stdin: &Mutex, - workspace: &Workspace, +/// +/// Async because a `query` runs a turn, which takes as long as the assistant +/// needs. +/// The turn goes to a task of its own, and this keeps answering reads while it +/// runs. +async fn message_loop( + requests: &mut mpsc::Receiver, + stdin: &Arc>, + ctx: &mut Ctx, config_json: &Value, shutdown_sent: &AtomicBool, + composer: &Composer, ) -> Result<(), cmd::Error> { - for line in reader.lines() { - let line = - line.map_err(|e| cmd::Error::from(format!("failed to read from plugin: {e}")))?; - + while let Some(line) = requests.recv().await { if line.trim().is_empty() { continue; } @@ -211,54 +478,47 @@ fn message_loop( trace!(?msg, "Received plugin message."); - let mut writer = stdin.lock().expect("stdin lock poisoned"); - + // Both of these block for as long as a person or a provider takes, so + // they run before the lock is taken: the shutdown thread needs that same + // lock to deliver `Shutdown` if the user interrupts partway. match msg { - PluginToHost::Ready => { - debug!("Plugin signaled ready."); - } - - PluginToHost::ListConversations(req) => { - let response = handle_list_conversations(workspace, req.id); - write_message(&mut *writer, &response)?; - } - - PluginToHost::ReadEvents(req) => { - let response = handle_read_events(workspace, &req.conversation, req.id); - write_message(&mut *writer, &response)?; + PluginToHost::Compose(request) => { + let response = composer.compose(&request); + let mut writer = stdin.lock().expect("stdin lock poisoned"); + write_message(&mut *writer, &HostToPlugin::Composed(response)).map_err(|e| { + cmd::Error::from(format!("failed to answer a compose request: {e}")) + })?; } - PluginToHost::ReadConfig(req) => { - let response = handle_read_config(config_json, req.path, req.id); - write_message(&mut *writer, &response)?; - } - - PluginToHost::Print(print) => { - // In Phase 1, write to stdout directly. Full printer - // integration comes later when we thread through &Printer. - let stdout = std::io::stdout(); - let mut handle = stdout.lock(); - drop(handle.write_all(print.text.as_bytes())); - drop(handle.flush()); - } - - PluginToHost::Log(log) => { - emit_log(&log); - } - - PluginToHost::Describe(_) => { - debug!("Ignoring describe in message loop."); + PluginToHost::Query(request) => { + // `None` means the turn is running and will answer for itself. + if let Some(response) = run_query(ctx, request, stdin).await { + let mut writer = stdin.lock().expect("stdin lock poisoned"); + write_message(&mut *writer, &response) + .map_err(|e| cmd::Error::from(format!("failed to answer a query: {e}")))?; + } } - PluginToHost::Exit(exit) => { - debug!(code = exit.code, "Plugin exited."); - if exit.code == 0 { + msg => { + let config = ctx.config(); + let fs_backend = ctx.fs_backend.clone(); + let session = ctx.session.clone(); + let signals = ctx.signals.clone(); + let mut writer = stdin.lock().expect("stdin lock poisoned"); + + if handle_request( + msg, + &mut *writer, + &mut ctx.workspace, + config_json, + session.as_ref(), + fs_backend.as_deref(), + &config, + &signals, + )? == Flow::Stop + { return Ok(()); } - return match exit.reason { - Some(reason) => Err(cmd::Error::from((exit.code, reason))), - None => Err(cmd::Error::from(exit.code)), - }; } } } @@ -277,11 +537,803 @@ fn message_loop( ))) } +/// Run a turn on the plugin's behalf. +/// +/// The turn is the same one `jp query` runs: it locks the conversation, calls +/// the provider, executes tools, and persists the events. +/// +/// `None` means the turn is under way and will send its own reply, correlated +/// by the request's id. +/// Anything returned is a failure that happened before the turn started. +async fn run_query( + ctx: &mut Ctx, + request: QueryRequest, + stdin: &Arc>, +) -> Option { + let reply_id = request.id.clone(); + let failed = |message: String| Some(query_error(reply_id.clone(), message)); + + if request.content.trim().is_empty() { + return failed("query content is empty".to_owned()); + } + + let new = request.new; + let lock = match lock_for_query(ctx, &request) { + Ok(lock) => lock, + Err(error) => return failed(error), + }; + + // The turn runs under the conversation's own config, not the host's. + // + // A host resolved its config once at startup with no conversation in view, so + // its persona, skills and enabled tools are whatever the bare workspace has. + // The conversation carries all of that in its stored deltas, and running a + // turn under anything else silently answers with the wrong model and no + // tools. + let config = match conversation_config(ctx, &lock, &request.cfg) { + Ok(mut config) => { + // A delegated turn has no terminal, so nothing can answer a prompt. + // + // An interrupt runs the same escalation as Ctrl-C, and the default + // streaming action is to show the interrupt menu. With no keyboard + // attached that menu blocks on a read nobody can satisfy: the turn + // keeps the conversation locked, the next request is refused as + // already-locked, and the host's terminal sits at a prompt meant for + // someone who is elsewhere. + // + // Stopping is the only interpretation available here, so it is the + // configured one. The reply and abort variants need a person. + config.interrupt.streaming.action = StreamingInterruptAction::Stop; + config.interrupt.tool_call.action = ToolInterruptAction::Stop; + Arc::new(config) + } + Err(error) => return failed(error), + }; + + debug!( + conversation = %lock.id(), + model = %config.assistant.model.id.resolved(), + "Running a delegated query.", + ); + + // Swapped around collecting only, because that is the part that reads the + // context. The turn itself carries the config it was given. + let host_config = ctx.swap_config(Arc::clone(&config)); + let prepared = prepare_turn(ctx, config, &lock, request.content).await; + ctx.swap_config(host_config); + + let (inputs, stream) = match prepared { + Ok(prepared) => prepared, + Err(error) => return failed(error.to_string()), + }; + + // Read from the lock, not the request: a new conversation was named by the + // host, and the plugin has no other way to learn its id. + let conversation = lock.id().to_string(); + + // Answered before the turn, because a caller that only needs somewhere to + // send the user cannot wait minutes to find out where that is. + if new { + let mut writer = stdin.lock().expect("stdin lock poisoned"); + drop(write_message( + &mut *writer, + &HostToPlugin::Created(CreatedResponse { + id: reply_id.clone(), + conversation: conversation.clone(), + }), + )); + } + + // Hand the turn to its own task. It owns everything it needs and the lock owns + // itself, so nothing here is borrowed for the minutes a turn can take, which + // is what keeps the message loop answering reads while it runs. + let stdin = Arc::clone(stdin); + + tokio::spawn(async move { + let outcome = inputs.run(&lock, stream).await; + + // Reported through tracing rather than to the terminal. These are facts + // about the host, not content: the turn's output belongs to the + // conversation, which is where whoever asked for it is reading. + let reply = match outcome { + Ok(()) => { + info!(%conversation, "A delegated turn finished."); + HostToPlugin::QueryComplete(QueryCompleteResponse { + id: reply_id, + conversation, + }) + } + Err(error) => { + // The full chain, not the outermost label: `cmd::Error` renders as + // "LLM error" and everything that matters hangs off its sources. + let detail = error_chain(&error); + warn!(%conversation, %detail, "A delegated turn failed."); + query_error(reply_id, detail) + } + }; + + let mut writer = stdin.lock().expect("stdin lock poisoned"); + drop(write_message(&mut *writer, &reply)); + }); + + None +} + +/// Take exclusive hold of the conversation a query names, creating it if asked. +/// +/// The lock comes before anything else a turn needs: it is the turn's proof of +/// exclusive access, and it owns what it needs, which is what lets the turn run +/// away from the message loop. +fn lock_for_query(ctx: &mut Ctx, request: &QueryRequest) -> Result { + if request.new { + // Resolved before the conversation exists, so a name that does not resolve + // leaves nothing behind to clean up. + let config = new_conversation_config(ctx, &request.cfg)?; + + let conversation = jp_conversation::Conversation { + title: request.title.clone().filter(|t| !t.trim().is_empty()), + ..jp_conversation::Conversation::default() + }; + + return ctx + .workspace + .create_and_lock_conversation(conversation, config, ctx.session.as_ref()) + .map_err(|error| format!("failed to create the conversation: {error}")); + } + + let id = parse_conversation_id(&request.conversation)?; + + let handle = ctx + .workspace + .acquire_conversation(&id) + .map_err(|error| format!("conversation {}: {error}", request.conversation))?; + + match ctx + .workspace + .lock_conversation(handle, ctx.session.as_ref()) + { + Ok(LockResult::Acquired(lock)) => Ok(lock), + Ok(LockResult::AlreadyLocked(_)) => { + Err("another process is working on this conversation".to_owned()) + } + Err(error) => Err(format!("failed to lock the conversation: {error}")), + } +} + +/// The configuration an existing conversation's next turn runs under. +/// +/// The conversation's own configuration with any named `cfg` layered over it. +/// With nothing named, this is what the stream already resolves to. +/// +/// The base layer is the one frozen when the conversation was created, so edits +/// to config files since then are not picked up. +/// Layering stored deltas over freshly read files needs the config pipeline, +/// which belongs with the caller that owns startup. +fn conversation_config( + ctx: &Ctx, + lock: &ConversationLock, + cfg: &[String], +) -> Result { + let stored = lock + .events() + .config() + .map_err(|error| format!("the conversation's config is invalid: {error}"))?; + + if cfg.is_empty() { + return Ok(stored); + } + + let args = cfg + .iter() + .map(|arg| arg.parse::()) + .collect::, _>>() + .map_err(|error| format!("invalid configuration argument: {error}"))?; + + let partial = build_partial_over( + stored.to_partial(), + &args, + Some(&ctx.workspace), + ctx.fs_backend.as_deref(), + ) + .map_err(|error| error.to_string())?; + + jp_config::util::build(partial) + .map_err(|error| format!("the resolved configuration is invalid: {error}")) +} + +/// The configuration a conversation created by a query starts from. +/// +/// The files layer, every config file and its `extends` chain plus the +/// environment, read fresh, with the named configurations on top. +/// A host that stays up for hours should start a conversation from the +/// configuration as it is now, not as it was when the process booted. +fn new_conversation_config(ctx: &Ctx, cfg: &[String]) -> Result, String> { + let base = crate::load_base_partial(ctx.fs_backend.as_deref()) + .map_err(|error| format!("failed to read the workspace configuration: {error}"))?; + + let args = cfg + .iter() + .map(|arg| arg.parse::()) + .collect::, _>>() + .map_err(|error| format!("invalid configuration argument: {error}"))?; + + let partial = build_partial_over(base, &args, Some(&ctx.workspace), ctx.fs_backend.as_deref()) + .map_err(|error| error.to_string())?; + + jp_config::util::build(partial) + .map(Arc::new) + .map_err(|error| format!("the resolved configuration is invalid: {error}")) +} + +/// Collect what the turn needs, which is the only part that needs the context. +/// +/// Fast on purpose: the message loop is blocked for exactly this long, and +/// every other request waits on it. +/// Starting the MCP servers is a spawn, not a wait; the waiting happens inside +/// [`TurnInputs::run`], on the turn's own task. +async fn prepare_turn( + ctx: &mut Ctx, + config: Arc, + lock: &ConversationLock, + content: String, +) -> Result<(TurnInputs, ConversationStream), cmd::Error> { + let mcp_servers = ctx.configure_active_mcp_servers().await?; + + let chat_request = ChatRequest { + content, + author: config.user.name.clone(), + ..ChatRequest::default() + }; + + // The message has moved from draft to request, so the draft is done. Clearing + // it here rather than from the caller gives it one owner: a client that + // cleared its own draft would be racing its debounced save, and losing. + if let Some(path) = draft_path(ctx.fs_backend.as_deref(), &ctx.workspace, &lock.id(), false) + && path.exists() + && let Err(error) = fs::remove_file(&path) + { + warn!(%error, "Failed to clear the query draft."); + } + + // Sending a message counts as using the conversation. + // + // Only on this path. At a terminal, activating a conversation is a deliberate + // act with its own command, and inferring it from a message would overwrite + // what the user said. A caller reaching in over the protocol has no such act + // to offer, so the last message is the best evidence of when the conversation + // was last used, which is what orders the list. + lock.as_mut() + .update_metadata(|meta| meta.last_activated_at = chrono::Utc::now()); + + // Recorded before the turn, so the conversation carries the change that + // produced it rather than a turn whose configuration came from nowhere. + match crate::cmd::query::get_config_delta_from_cli(&config, lock) { + Ok(Some(delta)) => { + lock.as_mut() + .update_events(|events| events.add_config_delta(delta)); + } + Ok(None) => {} + Err(error) => return Err(cmd::Error::from(error.to_string())), + } + + let stream = lock.events().clone(); + + let inputs = TurnInputs::collect( + ctx, + config, + lock, + chat_request, + PendingStreamTrim::default(), + mcp_servers, + // A sink, because this turn has no reader at this terminal. Nobody typed + // it here: it was asked for from somewhere else, and its output belongs to + // the conversation, which is where whoever asked is reading. Rendering it + // here would also braid two concurrent turns into one stream with no way + // to tell them apart. + Arc::new(Printer::sink()), + ) + .await?; + + Ok((inputs, stream)) +} + +/// Flatten an error and its sources into one line. +/// +/// JP's error types label a category and carry the cause underneath, so the +/// outermost message alone says "LLM error" where the source says what actually +/// went wrong. +/// A reader across a protocol has no way to ask for the rest, so send all of +/// it. +fn error_chain(error: &dyn std::error::Error) -> String { + let mut out = error.to_string(); + let mut source = error.source(); + + while let Some(cause) = source { + let text = cause.to_string(); + // Wrappers often restate their source; saying it twice helps nobody. + if !out.contains(&text) { + let _ = write!(out, ": {text}"); + } + source = cause.source(); + } + + out +} + +/// An error response to a `query` request. +fn query_error(id: Option, message: String) -> HostToPlugin { + HostToPlugin::Error(ErrorResponse { + id, + request: Some("query".to_owned()), + message, + }) +} + +/// Whether the message loop carries on after a request. +#[derive(Debug, PartialEq)] +enum Flow { + Continue, + Stop, +} + +/// Answer one request from the plugin. +/// +/// Runs with the writer lock held, so everything here has to be quick: anything +/// that blocks on the user is answered by the caller, before the lock is taken. +fn handle_request( + msg: PluginToHost, + writer: &mut impl Write, + workspace: &mut Workspace, + config_json: &Value, + session: Option<&Session>, + fs_backend: Option<&FsStorageBackend>, + config: &AppConfig, + signals: &SignalRouter, +) -> Result { + match msg { + PluginToHost::Ready(ready) => { + // The plugin states what it needs, so a mismatch is caught here + // rather than several messages later, when the host hits something it + // cannot parse and the plugin blocks on the reply. + if ready.protocol > PROTOCOL_VERSION { + return Err(cmd::Error::from(format!( + "this plugin needs protocol {}, and this `jp` speaks {PROTOCOL_VERSION}. \ + Reinstall the two together.", + ready.protocol, + ))); + } + debug!(protocol = ready.protocol, "Plugin signaled ready."); + } + + PluginToHost::ListConversations(req) => { + refresh_conversations(workspace); + let response = handle_list_conversations(workspace, req.id); + write_message(writer, &response)?; + } + + PluginToHost::ReadEvents(req) => { + refresh_conversations(workspace); + let response = handle_read_events(workspace, &req.conversation, req.id); + write_message(writer, &response)?; + } + + PluginToHost::ReadConfig(req) => { + let response = handle_read_config(config_json, req.path, req.id); + write_message(writer, &response)?; + } + + PluginToHost::ArchiveConversation(req) => { + let response = handle_archive(workspace, session, &req.conversation, req.id); + write_message(writer, &response)?; + } + + PluginToHost::SetTitle(req) => { + let response = handle_set_title(workspace, session, req); + write_message(writer, &response)?; + } + + PluginToHost::ReadDraft(req) => { + let response = handle_read_draft(fs_backend, workspace, &req.conversation, req.id); + write_message(writer, &response)?; + } + + PluginToHost::WriteDraft(req) => { + let response = handle_write_draft(fs_backend, workspace, req); + write_message(writer, &response)?; + } + + PluginToHost::Interrupt(req) => { + // Aimed at the named conversation, not at whatever is topmost. + // + // Several turns can be running at once, and the request already said + // which one it means. Falling back to the untargeted path would stop + // an arbitrary other turn, which is worse than stopping nothing. + // + // Nothing to answer: what the interrupt did lands in the + // conversation, and the turn's own outcome is still the reply to its + // `query`. + let reached = parse_conversation_id(&req.conversation) + .is_ok_and(|id| signals.interrupt_scope(id)); + + debug!( + conversation = %req.conversation, + reached, + "Interrupting on a plugin's behalf." + ); + } + + PluginToHost::ListConfigs(req) => { + let response = handle_list_configs(config, workspace, fs_backend, req.id); + write_message(writer, &response)?; + } + + PluginToHost::Print(print) => { + // In Phase 1, write to stdout directly. Full printer + // integration comes later when we thread through &Printer. + let stdout = std::io::stdout(); + let mut handle = stdout.lock(); + drop(handle.write_all(print.text.as_bytes())); + drop(handle.flush()); + } + + PluginToHost::Log(log) => { + emit_log(&log); + } + + PluginToHost::Describe(_) => { + debug!("Ignoring describe in message loop."); + } + + // Answered by the caller, before the lock this runs under is taken. + PluginToHost::Compose(_) | PluginToHost::Query(_) => { + unreachable!("answered before the lock") + } + + PluginToHost::Exit(exit) => { + debug!(code = exit.code, "Plugin exited."); + if exit.code != 0 { + return match exit.reason { + Some(reason) => Err(cmd::Error::from((exit.code, reason))), + None => Err(cmd::Error::from(exit.code)), + }; + } + return Ok(Flow::Stop); + } + } + + Ok(Flow::Continue) +} + +/// An error against a request that only reports whether it worked. +fn action_failed(id: Option, request: &str, message: String) -> HostToPlugin { + HostToPlugin::Error(ErrorResponse { + id, + request: Some(request.to_owned()), + message, + }) +} + +/// Take the lock a mutation needs, or say why it could not be had. +fn lock_for_action( + workspace: &Workspace, + session: Option<&Session>, + conversation: &str, +) -> Result { + let id = parse_conversation_id(conversation)?; + + let handle = workspace + .acquire_conversation(&id) + .map_err(|error| format!("conversation {conversation}: {error}"))?; + + match workspace.lock_conversation(handle, session) { + Ok(LockResult::Acquired(lock)) => Ok(lock), + Ok(LockResult::AlreadyLocked(_)) => { + Err("another process is working on this conversation".to_owned()) + } + Err(error) => Err(format!("failed to lock the conversation: {error}")), + } +} + +/// Move a conversation to the archive. +/// +/// Held under a lock like any other mutation: archiving moves the conversation +/// on disk, and doing that under a running turn would pull the files out from +/// beneath it. +fn handle_archive( + workspace: &mut Workspace, + session: Option<&Session>, + conversation: &str, + req_id: Option, +) -> HostToPlugin { + let lock = match lock_for_action(workspace, session, conversation) { + Ok(lock) => lock, + Err(message) => return action_failed(req_id, "archive_conversation", message), + }; + + workspace.archive_conversation(lock.into_mut()); + + HostToPlugin::Done(DoneResponse { id: req_id }) +} + +/// Rename a conversation. +/// +/// An empty title clears it, which leaves the conversation eligible for a +/// generated one again rather than naming it the empty string. +fn handle_set_title( + workspace: &Workspace, + session: Option<&Session>, + req: SetTitleRequest, +) -> HostToPlugin { + let lock = match lock_for_action(workspace, session, &req.conversation) { + Ok(lock) => lock, + Err(message) => return action_failed(req.id, "set_title", message), + }; + + let title = req + .title + .map(|title| title.trim().to_owned()) + .filter(|title| !title.is_empty()); + + lock.as_mut().update_metadata(|meta| meta.title = title); + + // Written out when the lock drops, at the end of this function. + HostToPlugin::Done(DoneResponse { id: req.id }) +} + +/// Read a conversation named by a plugin. +/// +/// The canonical spelling is the one JP prints and a user can paste back into +/// it, `jp-c17000000000`. +/// Bare deciseconds are accepted as well, since that is what the wire carried +/// before the two agreed. +fn parse_conversation_id(conversation: &str) -> Result { + conversation + .parse() + .or_else(|_| ConversationId::try_from_deciseconds_str(conversation)) + .map_err(|error| format!("invalid conversation ID `{conversation}`: {error}")) +} + +/// The query draft's fingerprint: a short hash of its content. +/// +/// Content rather than modification time, so a rewrite with identical text is +/// not mistaken for someone else's edit. +fn draft_revision(content: &str) -> String { + let digest = Sha256::digest(content.as_bytes()); + digest[..8].iter().fold(String::new(), |mut acc, byte| { + let _ = write!(acc, "{byte:02x}"); + acc + }) +} + +/// Where a conversation's query draft lives, if user-local storage is +/// configured. +/// +/// The same file `jp query` seeds an editor from and writes on interrupt. +/// It is deliberately user-local and never projected into the workspace tree, +/// so a half-written message is not something a teammate can end up with. +/// +/// With `create`, a path is derived for a conversation that has no directory +/// yet; without it, an absent directory means an absent draft. +fn draft_path( + fs_backend: Option<&FsStorageBackend>, + workspace: &Workspace, + id: &ConversationId, + create: bool, +) -> Option { + let fs = fs_backend?; + + if let Some(dir) = fs.find_user_local_conversation_dir(id) { + return Some(dir.join(crate::editor::QUERY_FILENAME)); + } + + if !create { + return None; + } + + // No directory yet, so derive the one `jp query` would use, which puts the + // conversation's title in the name. + let title = workspace.acquire_conversation(id).ok().and_then(|handle| { + workspace + .metadata(&handle) + .ok() + .and_then(|meta| meta.title.clone()) + }); + + Some( + fs.build_conversation_dir(id, title.as_deref(), true) + .join(crate::editor::QUERY_FILENAME), + ) +} + +/// Read a conversation's query draft. +/// +/// An absent draft is not an error: most conversations do not have one, and the +/// answer is an empty draft with no revision. +fn handle_read_draft( + fs_backend: Option<&FsStorageBackend>, + workspace: &Workspace, + conversation: &str, + req_id: Option, +) -> HostToPlugin { + let id = match parse_conversation_id(conversation) { + Ok(id) => id, + Err(message) => { + return HostToPlugin::Error(ErrorResponse { + id: req_id, + request: Some("read_draft".to_owned()), + message, + }); + } + }; + + let content = draft_path(fs_backend, workspace, &id, false) + .filter(|path| path.exists()) + .and_then(|path| fs::read_to_string(path).ok()); + + HostToPlugin::Draft(DraftResponse { + id: req_id, + conversation: conversation.to_owned(), + revision: content.as_deref().map(draft_revision), + content: content.unwrap_or_default(), + conflict: false, + }) +} + +/// Replace a conversation's query draft. +/// +/// The `revision` names the version the caller edited. +/// A draft that has moved on since is reported back rather than overwritten: +/// the other writer's text is exactly what the caller has not seen. +fn handle_write_draft( + fs_backend: Option<&FsStorageBackend>, + workspace: &Workspace, + req: WriteDraftRequest, +) -> HostToPlugin { + let failed = |message: String| { + HostToPlugin::Error(ErrorResponse { + id: req.id.clone(), + request: Some("write_draft".to_owned()), + message, + }) + }; + + let id = match parse_conversation_id(&req.conversation) { + Ok(id) => id, + Err(message) => return failed(message), + }; + + let Some(path) = draft_path(fs_backend, workspace, &id, true) else { + return failed("this workspace has no user-local storage for drafts".to_owned()); + }; + + let current = fs::read_to_string(&path).ok(); + let current_revision = current.as_deref().map(draft_revision); + + if current_revision != req.revision { + return HostToPlugin::Draft(DraftResponse { + id: req.id, + conversation: req.conversation, + content: current.unwrap_or_default(), + revision: current_revision, + conflict: true, + }); + } + + // An empty draft is no draft: a blank file left behind would have the CLI + // seed an editor with nothing and treat it as a recovery copy. + if req.content.is_empty() { + if path.exists() + && let Err(error) = fs::remove_file(&path) + { + return failed(format!("failed to remove the draft: {error}")); + } + + return HostToPlugin::Draft(DraftResponse { + id: req.id, + conversation: req.conversation, + content: String::new(), + revision: None, + conflict: false, + }); + } + + if let Some(parent) = path.parent() + && let Err(error) = fs::create_dir_all(parent) + { + return failed(format!("failed to create the draft directory: {error}")); + } + + if let Err(error) = fs::write(&path, &req.content) { + return failed(format!("failed to write the draft: {error}")); + } + + HostToPlugin::Draft(DraftResponse { + id: req.id, + conversation: req.conversation, + revision: Some(draft_revision(&req.content)), + content: req.content, + conflict: false, + }) +} + +/// Every configuration a query can name, from the same roots `--cfg` searches. +/// +/// Roots are searched independently, and a segment present in more than one is +/// still one selectable thing, because naming it merges all of them. +/// Sorted and deduplicated for that reason. +fn handle_list_configs( + config: &AppConfig, + workspace: &Workspace, + fs_backend: Option<&FsStorageBackend>, + req_id: Option, +) -> HostToPlugin { + let mut roots: Vec = Vec::new(); + + if let Some(dir) = user_global_config_dir(None) { + roots.push(dir); + } + roots.push(workspace.root().to_owned()); + if let Some(path) = + fs_backend.and_then(|fs| fs.user_storage_with_path(RelativePath::new("config"))) + { + roots.push(path); + } + + let mut segments = BTreeSet::new(); + for root in &roots { + for load_path in &config.config_load_paths { + let Ok(dir) = Utf8PathBuf::try_from(load_path.to_path(root)) else { + continue; + }; + + segments.extend(list_configs_in_load_path(&dir)); + } + } + + let data = segments + .into_iter() + .map(|segment| { + let (namespace, name) = match segment.rsplit_once('/') { + Some((namespace, name)) => (namespace.to_owned(), name.to_owned()), + None => (String::new(), segment.clone()), + }; + + ConfigEntry { + segment, + namespace, + name, + } + }) + .collect(); + + HostToPlugin::Configs(ConfigsResponse { id: req_id, data }) +} + +/// Re-read the conversation index, dropping what this process has cached. +/// +/// A plugin host is long-lived and does not own the store: a `jp query` in +/// another terminal, or another plugin, appends events to a conversation whose +/// metadata and stream this process loaded once and would otherwise keep +/// serving forever. +/// Re-reading the index clears both caches, so the read that follows comes from +/// disk, and conversations created or deleted since startup appear and +/// disappear. +/// +/// The scan is a directory listing per storage root; metadata and streams stay +/// lazy, so only what the request actually reads is loaded again. +/// +/// Deliberately not a sanitize pass: that repairs a store on startup and can +/// move broken conversations aside, which is not a thing a page view should do. +fn refresh_conversations(workspace: &mut Workspace) { + trace!("Re-reading the conversation index for a plugin request."); + workspace.load_conversation_index(); +} + fn handle_list_conversations(workspace: &Workspace, req_id: Option) -> HostToPlugin { let data: Vec = workspace .conversations() .map(|(id, meta)| ConversationSummary { - id: id.as_deciseconds().to_string(), + id: id.to_string(), title: meta.title.clone(), last_activated_at: meta.last_activated_at, events_count: meta.events_count, @@ -296,13 +1348,13 @@ fn handle_read_events( conversation_id: &str, req_id: Option, ) -> HostToPlugin { - let conv_id = match jp_conversation::ConversationId::try_from_deciseconds_str(conversation_id) { + let conv_id = match parse_conversation_id(conversation_id) { Ok(id) => id, - Err(e) => { + Err(message) => { return HostToPlugin::Error(ErrorResponse { id: req_id, request: Some("read_events".to_owned()), - message: format!("invalid conversation ID: {e}"), + message, }); } }; @@ -865,13 +1917,20 @@ fn unknown_subcommand_error(name: &str) -> cmd::Error { /// /// Resolves the plugin binary, then runs the protocol loop. /// Called from `Commands::run()` after the normal startup flow. -pub(crate) async fn run_external(args: &[String], ctx: &Ctx) -> cmd::Output { +pub(crate) async fn run_external(args: &[String], ctx: &mut Ctx) -> cmd::Output { let (subcommand, plugin_args) = args .split_first() .ok_or("no subcommand provided for plugin dispatch")?; - // Handle help without downloading or approval. - if plugin_args.iter().any(|a| a == "-h" || a == "--help") { + // A bare `jp --help` is answered from the plugin's self-description, + // without downloading or approving anything. + // + // Help for something *within* the plugin (`jp add --help`) is the + // plugin's own to render, and only it knows its subcommands, so that goes + // through normal dispatch below. + let bare_help = + !plugin_args.is_empty() && plugin_args.iter().all(|a| a == "-h" || a == "--help"); + if bare_help { let binary = find_any_plugin_binary(subcommand).ok_or_else(|| { cmd::Error::from(format!( "plugin `{subcommand}` not found. No installed plugin or `jp-{subcommand}` binary \ @@ -889,17 +1948,7 @@ pub(crate) async fn run_external(args: &[String], ctx: &Ctx) -> cmd::Output { debug!(%binary, subcommand, "Dispatching to plugin."); - run_plugin( - subcommand, - &binary, - plugin_args, - &ctx.workspace, - ctx.storage_path(), - ctx.user_storage_path(), - &config, - &ctx.signals, - ctx.term.args.verbose, - )?; + run_plugin(subcommand, &binary, plugin_args, ctx).await?; Ok(()) } diff --git a/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs b/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs index 7f14eaea5..28a6a03fa 100644 --- a/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs +++ b/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs @@ -1,7 +1,426 @@ +use camino_tempfile::{Utf8TempDir, tempdir}; +use jp_conversation::{Conversation, ConversationId}; +use jp_plugin::message::{ExitMessage, ReadyMessage}; +use jp_storage::backend::FsStorageBackend; use serde_json::json; use super::*; +/// A workspace no request in these tests reaches into, so it needs no storage. +fn bare_workspace() -> Workspace { + Workspace::new("/tmp/jp-test-plugin") +} + +/// A router with no signal source, for requests that never reach one. +/// +/// Must be called inside a tokio runtime, which is why the tests using it are +/// `#[tokio::test]` despite `handle_request` being synchronous. +fn router() -> SignalRouter { + crate::signals::testing::detached_router() +} + +/// How a conversation is spelled on the wire, matching `list_conversations`. +fn wire_id(id: ConversationId) -> String { + id.to_string() +} + +/// A workspace holding one conversation already on disk. +/// +/// The temp dir comes back so the caller keeps it alive; dropping it takes the +/// storage with it. +fn workspace_with_conversation() -> (Workspace, ConversationId, Utf8TempDir) { + let (ws, id, _fs, tmp) = workspace_with_drafts(); + (ws, id, tmp) +} + +/// The same, with user-local storage configured so drafts have somewhere to go. +fn workspace_with_drafts() -> ( + Workspace, + ConversationId, + Arc, + Utf8TempDir, +) { + let tmp = tempdir().unwrap(); + let fs = Arc::new( + FsStorageBackend::new(&tmp.path().join(".jp")) + .unwrap() + .with_user_storage(&tmp.path().join("user"), None, "test-workspace") + .unwrap(), + ); + + let id = ConversationId::try_from( + chrono::DateTime::::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000), + ) + .unwrap(); + fs.write_test_conversation(&id, &Conversation::default()); + + let mut workspace = Workspace::new(tmp.path()).with_backend(fs.clone()); + workspace.load_conversation_index(); + + (workspace, id, fs, tmp) +} + +/// Unwrap a draft response, or say what came back instead. +fn draft(response: HostToPlugin) -> jp_plugin::message::DraftResponse { + match response { + HostToPlugin::Draft(draft) => draft, + other => panic!("expected a draft response, got {other:?}"), + } +} + +/// A conversation with no draft reads back empty rather than as an error: most +/// conversations never have one. +#[test] +fn an_absent_draft_reads_as_empty() { + let (ws, id, fs, _tmp) = workspace_with_drafts(); + + let read = draft(handle_read_draft(Some(&fs), &ws, &wire_id(id), None)); + + assert_eq!(read.content, ""); + assert_eq!(read.revision, None); + assert!(!read.conflict); +} + +#[test] +fn a_written_draft_reads_back_with_its_revision() { + let (ws, id, fs, _tmp) = workspace_with_drafts(); + + let written = draft(handle_write_draft(Some(&fs), &ws, WriteDraftRequest { + id: Some("w1".to_owned()), + conversation: wire_id(id), + content: "half a thought".to_owned(), + revision: None, + })); + + assert_eq!(written.id.as_deref(), Some("w1")); + assert_eq!(written.content, "half a thought"); + assert!(!written.conflict); + + let read = draft(handle_read_draft(Some(&fs), &ws, &wire_id(id), None)); + + assert_eq!(read.content, "half a thought"); + assert_eq!( + read.revision, written.revision, + "the revision a write reports is the one a read gives back" + ); +} + +/// The whole point of the revision: a write based on a version that has since +/// moved on is refused, and the caller is handed what it had not seen. +#[test] +fn a_write_against_a_stale_revision_is_refused() { + let (ws, id, fs, _tmp) = workspace_with_drafts(); + + let first = draft(handle_write_draft(Some(&fs), &ws, WriteDraftRequest { + id: None, + conversation: wire_id(id), + content: "what the other writer typed".to_owned(), + revision: None, + })); + + // A second writer that started from no draft at all, and so never saw the + // first write. + let refused = draft(handle_write_draft(Some(&fs), &ws, WriteDraftRequest { + id: Some("w2".to_owned()), + conversation: wire_id(id), + content: "what I typed".to_owned(), + revision: None, + })); + + assert!(refused.conflict); + assert_eq!(refused.id.as_deref(), Some("w2")); + assert_eq!( + refused.content, "what the other writer typed", + "the refusal carries the text on disk, not the text submitted" + ); + assert_eq!(refused.revision, first.revision); + + // And nothing was overwritten. + let read = draft(handle_read_draft(Some(&fs), &ws, &wire_id(id), None)); + assert_eq!(read.content, "what the other writer typed"); +} + +/// An empty write removes the draft rather than leaving a blank file, which the +/// CLI would otherwise seed an editor from and treat as a recovery copy. +#[test] +fn an_empty_write_removes_the_draft() { + let (ws, id, fs, _tmp) = workspace_with_drafts(); + + let written = draft(handle_write_draft(Some(&fs), &ws, WriteDraftRequest { + id: None, + conversation: wire_id(id), + content: "to be discarded".to_owned(), + revision: None, + })); + + let cleared = draft(handle_write_draft(Some(&fs), &ws, WriteDraftRequest { + id: None, + conversation: wire_id(id), + content: String::new(), + revision: written.revision, + })); + + assert_eq!(cleared.content, ""); + assert_eq!(cleared.revision, None); + assert!(!cleared.conflict); + + let path = draft_path(Some(&fs), &ws, &id, false); + assert!( + path.is_none_or(|path| !path.exists()), + "the draft file is gone, not blank" + ); +} + +/// Without user-local storage there is nowhere a draft may live, and saying so +/// beats writing it into the workspace where a teammate would see it. +#[test] +fn writing_a_draft_without_user_local_storage_fails() { + let (ws, id, _tmp) = workspace_with_conversation(); + + let response = handle_write_draft(None, &ws, WriteDraftRequest { + id: Some("w3".to_owned()), + conversation: wire_id(id), + content: "nowhere to go".to_owned(), + revision: None, + }); + + match response { + HostToPlugin::Error(error) => { + assert_eq!(error.request.as_deref(), Some("write_draft")); + assert!(error.message.contains("user-local storage"), "{error:?}"); + } + other => panic!("expected an error, got {other:?}"), + } +} + +#[test] +fn set_title_names_a_conversation() { + let (ws, id, _tmp) = workspace_with_conversation(); + + let response = handle_set_title(&ws, None, SetTitleRequest { + id: Some("r1".to_owned()), + conversation: wire_id(id), + title: Some(" Tool call header misaligns ".to_owned()), + }); + + assert_eq!( + response, + HostToPlugin::Done(DoneResponse { + id: Some("r1".to_owned()) + }) + ); + + let handle = ws.acquire_conversation(&id).unwrap(); + assert_eq!( + ws.metadata(&handle).unwrap().title.as_deref(), + Some("Tool call header misaligns"), + "the title is stored trimmed" + ); +} + +/// A blank title clears the name rather than storing an empty one, which leaves +/// the conversation eligible for a generated title again. +#[test] +fn a_blank_set_title_clears_the_name() { + let (ws, id, _tmp) = workspace_with_conversation(); + + handle_set_title(&ws, None, SetTitleRequest { + id: None, + conversation: wire_id(id), + title: Some("Named".to_owned()), + }); + + handle_set_title(&ws, None, SetTitleRequest { + id: None, + conversation: wire_id(id), + title: Some(" ".to_owned()), + }); + + let handle = ws.acquire_conversation(&id).unwrap(); + assert_eq!(ws.metadata(&handle).unwrap().title, None); +} + +#[test] +fn archiving_takes_a_conversation_out_of_the_index() { + let (mut ws, id, _tmp) = workspace_with_conversation(); + + let response = handle_archive(&mut ws, None, &wire_id(id), Some("r2".to_owned())); + + assert_eq!( + response, + HostToPlugin::Done(DoneResponse { + id: Some("r2".to_owned()) + }) + ); + assert!( + ws.acquire_conversation(&id).is_err(), + "an archived conversation is no longer in the index" + ); +} + +/// A failure names the request it belongs to, so a plugin with several in +/// flight can tell which one it answers. +#[test] +fn an_unknown_conversation_fails_against_its_request() { + let mut ws = bare_workspace(); + + let response = handle_archive(&mut ws, None, "not-an-id", Some("r3".to_owned())); + + match response { + HostToPlugin::Error(error) => { + assert_eq!(error.id.as_deref(), Some("r3")); + assert_eq!(error.request.as_deref(), Some("archive_conversation")); + assert!( + error.message.contains("invalid conversation ID"), + "{error:?}" + ); + } + other => panic!("expected an error, got {other:?}"), + } +} + +/// A long-running host sees what another process wrote after it started. +/// +/// The host loads the index once at startup. +/// Without re-reading it, a plugin asking for the conversation list is served +/// that snapshot for the life of the process, so a conversation started in a +/// terminal never appears. +#[tokio::test] +async fn a_conversation_written_after_startup_is_listed() { + let (mut ws, first, fs, tmp) = workspace_with_drafts(); + let mut sink: Vec = Vec::new(); + + // The host's view, taken at startup. + ws.load_conversation_index(); + assert_eq!(ws.conversations().count(), 1); + + // Another process writes a second conversation. Same store, its own handle, + // which is what a `jp query` in a terminal amounts to. + let second = ConversationId::try_from( + chrono::DateTime::::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_001), + ) + .unwrap(); + fs.write_test_conversation(&second, &Conversation::default()); + + let response = handle_request( + PluginToHost::ListConversations(jp_plugin::message::OptionalId { id: None }), + &mut sink, + &mut ws, + &json!({}), + None, + None, + &AppConfig::new_test(), + &router(), + ) + .unwrap(); + assert_eq!(response, Flow::Continue); + + let listed: Vec = ws.conversations().map(|(id, _)| id.to_string()).collect(); + + assert!( + listed.contains(&second.to_string()), + "a conversation written after startup must be listed: {listed:?}" + ); + assert!( + listed.contains(&first.to_string()), + "and the one from startup is still there: {listed:?}" + ); + + drop(tmp); +} + +/// The canonical spelling is what JP prints, and bare deciseconds still resolve +/// because that is what the wire carried before. +#[test] +fn a_conversation_is_named_the_way_jp_prints_it() { + let (ws, id, _tmp) = workspace_with_conversation(); + + assert_eq!( + wire_id(id), + "jp-c17000000000", + "a plugin is handed the spelling a user can paste back into `jp`" + ); + + assert_eq!(parse_conversation_id("jp-c17000000000").unwrap(), id); + assert_eq!(parse_conversation_id("17000000000").unwrap(), id); + assert!(parse_conversation_id("not-an-id").is_err()); + + // Asserted exactly, not round-tripped: both spellings parse, so a + // round-trip would pass whichever one the host emitted. + let HostToPlugin::Conversations(listed) = handle_list_conversations(&ws, None) else { + panic!("expected a conversations response"); + }; + let [summary] = listed.data.as_slice() else { + panic!("expected exactly one conversation"); + }; + assert_eq!(summary.id, "jp-c17000000000"); +} + +#[tokio::test] +async fn a_ready_carries_on_and_a_clean_exit_stops() { + let mut ws = bare_workspace(); + let config = json!({}); + let mut sink: Vec = Vec::new(); + + assert_eq!( + handle_request( + PluginToHost::Ready(ReadyMessage { protocol: 1 }), + &mut sink, + &mut ws, + &config, + None, + None, + &AppConfig::new_test(), + &router(), + ) + .unwrap(), + Flow::Continue + ); + + assert_eq!( + handle_request( + PluginToHost::Exit(ExitMessage { + code: 0, + reason: None, + }), + &mut sink, + &mut ws, + &config, + None, + None, + &AppConfig::new_test(), + &router(), + ) + .unwrap(), + Flow::Stop + ); +} + +/// A non-zero exit is the plugin's failure, so it surfaces as one rather than +/// ending the run quietly. +#[tokio::test] +async fn a_failing_exit_carries_its_code_and_reason() { + let mut ws = bare_workspace(); + let mut sink: Vec = Vec::new(); + + let error = handle_request( + PluginToHost::Exit(ExitMessage { + code: 3, + reason: Some("no such ticket".to_owned()), + }), + &mut sink, + &mut ws, + &json!({}), + None, + None, + &AppConfig::new_test(), + &router(), + ) + .expect_err("a non-zero exit is an error"); + + assert!(error.to_string().contains("no such ticket"), "{error}"); +} + #[test] fn handle_read_config_full() { let config = json!({"assistant": {"name": "JP"}, "style": {"code": {}}}); @@ -41,23 +460,100 @@ fn handle_read_config_invalid_path() { assert!(matches!(resp, HostToPlugin::Error(_))); } +/// An error whose `Display` says one thing and whose source says another. +#[derive(Debug)] +struct Layered { + message: &'static str, + source: Option>, +} + +impl std::fmt::Display for Layered { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.message) + } +} + +impl std::error::Error for Layered { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + self.source + .as_deref() + .map(|source| source as &(dyn std::error::Error + 'static)) + } +} + +fn layered(message: &'static str, source: Option) -> Layered { + Layered { + message, + source: source.map(Box::new), + } +} + +/// The outermost message is a category, so sending it alone tells the reader +/// nothing. +/// A plugin cannot ask for the sources, so they are flattened in. #[test] -fn message_loop_ready_then_exit() { - use std::io::{BufReader, Cursor}; +fn an_error_is_reported_with_its_causes() { + let error = layered( + "LLM error", + Some(layered( + "request failed", + Some(layered( + "prompt is too long: 1000327 tokens > 1000000 maximum", + None, + )), + )), + ); - let plugin_output = [r#"{"type":"ready"}"#, r#"{"type":"exit","code":0}"#].join("\n"); + assert_eq!( + error_chain(&error), + "LLM error: request failed: prompt is too long: 1000327 tokens > 1000000 maximum" + ); +} - let reader = BufReader::new(Cursor::new(plugin_output)); - let sink: Mutex> = Mutex::new(Vec::new()); - let config = json!({}); - let shutdown_sent = AtomicBool::new(false); +/// A wrapper that already quotes its source should not say it twice. +#[test] +fn a_restated_cause_is_not_repeated() { + let error = layered( + "config error: no such model `haiku`", + Some(layered("no such model `haiku`", None)), + ); - // We can't easily construct a Workspace for a unit test without a temp dir, - // but this test only exercises ready + exit (no workspace queries). We - // construct a minimal in-memory workspace. - let ws = jp_workspace::Workspace::new("/tmp/jp-test-plugin"); + assert_eq!(error_chain(&error), "config error: no such model `haiku`"); +} + +#[test] +fn a_lone_error_is_reported_as_itself() { + assert_eq!( + error_chain(&layered("nothing underneath", None)), + "nothing underneath" + ); +} + +/// A plugin needing a newer protocol than this host is refused on its `ready`, +/// before it can send anything the host would fail to parse. +#[tokio::test] +async fn a_plugin_needing_a_newer_protocol_is_refused() { + let mut ws = bare_workspace(); + let mut sink: Vec = Vec::new(); - message_loop(reader, &sink, &ws, &config, &shutdown_sent).unwrap(); + let error = handle_request( + PluginToHost::Ready(ReadyMessage { + protocol: PROTOCOL_VERSION + 1, + }), + &mut sink, + &mut ws, + &json!({}), + None, + None, + &AppConfig::new_test(), + &router(), + ) + .expect_err("a plugin needing a newer protocol must be refused"); + + assert!( + error.to_string().contains("Reinstall the two together"), + "{error}" + ); } #[test] diff --git a/crates/jp_cli/src/cmd/query.rs b/crates/jp_cli/src/cmd/query.rs index e79122f00..62e7568d2 100644 --- a/crates/jp_cli/src/cmd/query.rs +++ b/crates/jp_cli/src/cmd/query.rs @@ -557,61 +557,20 @@ impl Query { } } - // Wait for all MCP servers to finish loading, showing a timer line - // when the wait takes long enough to be noticeable. - await_mcp_servers( + let inputs = TurnInputs::collect( + ctx, + cfg.clone(), + &lock, + chat_request, + pending_trim, mcp_servers_handle, - cfg.style.mcp_startup.clone(), + // Typed at this terminal, so it renders here. ctx.printer.clone(), - ctx.term.is_tty, - ctx.term.width, ) .await?; - let forced_tool = cfg.assistant.tool_choice.function_name(); - let tools = - tool_definitions(cfg.conversation.tools.iter(), &ctx.mcp_client, forced_tool).await?; - - let attachment_urls: Vec<_> = cfg - .conversation - .attachments - .iter() - .map(jp_config::conversation::attachment::AttachmentConfig::to_url) - .collect::, _>>()?; - let attachments = load_conversation_attachments(ctx, attachment_urls).await?; - - debug!(count = attachments.len(), "Attachments loaded."); - - let thread = build_thread(stream, attachments, &cfg.assistant, !tools.is_empty())?; - let root = ctx.workspace.root().to_path_buf(); - let approvals = Arc::new(load_approval_store(ctx.fs_backend.as_deref())); - - // Sanitize any structural issues (orphaned tool calls, missing - // user messages, etc.) before sending the stream to the provider. - lock.as_mut().update_events(ConversationStream::sanitize); - - let invocation = InvocationContext { - workspace_id: ctx.workspace.id().to_string(), - conversation_id: lock.id().to_string(), - }; - - let turn_result = self - .handle_turn( - &cfg, - &ctx.signals, - &ctx.mcp_client, - root, - ctx.term.is_tty, - &thread.attachments, - &lock, - cfg.assistant.tool_choice.clone(), - &tools, - ctx.printer.clone(), - approvals, - chat_request, - invocation, - pending_trim, - ) + let turn_result = inputs + .run(&lock, stream) .await .map_err(|error| cmd::Error::from(error).with_persistence(true)); @@ -897,10 +856,15 @@ impl Query { Ok((QuerySource::Editor, editor_provided_config)) } - /// Handle a single turn of conversation with the LLM. + /// Run one turn of a conversation against the LLM. + /// + /// Takes no CLI state: everything it needs is an argument, so a caller that + /// isn't the `query` command can drive a turn too. + /// + /// The `lock` is proof the conversation is held for the duration (RFD 020), + /// and it owns what it needs, so nothing here borrows the workspace. #[expect(clippy::too_many_arguments)] - async fn handle_turn( - &self, + pub(crate) async fn run_turn( cfg: &AppConfig, signals: &SignalRouter, mcp_client: &jp_mcp::Client, @@ -1101,6 +1065,143 @@ impl Query { } } +/// Everything a turn needs, owned. +/// +/// Collecting reads the context; running borrows nothing from it. +/// That split is the point: a caller can start a turn and then get on with +/// other work, because the running half holds nothing belonging to whoever +/// started it. +pub(crate) struct TurnInputs { + config: Arc, + + /// A handle onto the process router, not a second router, so a handler this + /// turn pushes is one the signal task reaches. + signals: SignalRouter, + mcp_client: jp_mcp::Client, + root: Utf8PathBuf, + is_tty: bool, + width: Option, + attachments: Vec, + printer: Arc, + approvals: Arc, + chat_request: ChatRequest, + invocation: InvocationContext, + pending_trim: PendingStreamTrim, + + /// MCP servers starting in the background, awaited by [`TurnInputs::run`]. + mcp_servers: StartupSet, +} + +impl TurnInputs { + /// Collect what a turn needs from the context. + /// + /// Deliberately does nothing slow. + /// Everything that waits, MCP servers coming up and the provider, happens + /// in [`Self::run`], which borrows nothing from the context. + /// A caller serving other work off the same task is blocked for exactly as + /// long as this takes. + /// + /// Attachment loading is the one wait that has to stay here, because it + /// reads conversations out of the workspace. + /// Local reads, not process startup. + /// + /// `printer` is where the turn's output goes. + /// A turn typed at a terminal renders to it; a turn asked for from + /// elsewhere has no reader there, and a sink is how it stops writing into a + /// session it does not belong to. + pub(crate) async fn collect( + ctx: &Ctx, + config: Arc, + lock: &ConversationLock, + chat_request: ChatRequest, + pending_trim: PendingStreamTrim, + mcp_servers: StartupSet, + printer: Arc, + ) -> Result { + let attachment_urls: Vec<_> = config + .conversation + .attachments + .iter() + .map(jp_config::conversation::attachment::AttachmentConfig::to_url) + .collect::, _>>()?; + let attachments = load_conversation_attachments(ctx, attachment_urls).await?; + + debug!(count = attachments.len(), "Attachments loaded."); + + Ok(Self { + root: ctx.workspace.root().to_path_buf(), + approvals: Arc::new(load_approval_store(ctx.fs_backend.as_deref())), + invocation: InvocationContext { + workspace_id: ctx.workspace.id().to_string(), + conversation_id: lock.id().to_string(), + }, + signals: ctx.signals.clone(), + mcp_client: ctx.mcp_client.clone(), + printer, + is_tty: ctx.term.is_tty, + width: ctx.term.width, + attachments, + mcp_servers, + chat_request, + pending_trim, + config, + }) + } + + /// Finish preparing, then run the turn. + /// + /// Borrows only the lock, which owns itself, so this can run on a task of + /// its own, which is where the waiting belongs. + /// `stream` is the snapshot the thread is assembled from. + pub(crate) async fn run( + self, + lock: &ConversationLock, + stream: ConversationStream, + ) -> Result<()> { + let cfg = &self.config; + + // Wait for all MCP servers to finish loading, showing a timer line when the + // wait takes long enough to be noticeable. Starting a server can mean + // compiling one, so this is not a wait to hold anything else up for. + await_mcp_servers( + self.mcp_servers, + cfg.style.mcp_startup.clone(), + self.printer.clone(), + self.is_tty, + self.width, + ) + .await?; + + let forced_tool = cfg.assistant.tool_choice.function_name(); + let tools = + tool_definitions(cfg.conversation.tools.iter(), &self.mcp_client, forced_tool).await?; + + let thread = build_thread(stream, self.attachments, &cfg.assistant, !tools.is_empty())?; + + // Sanitize any structural issues (orphaned tool calls, missing user + // messages, etc.) before sending the stream to the provider. + lock.as_mut().update_events(ConversationStream::sanitize); + + Query::run_turn( + cfg, + &self.signals, + &self.mcp_client, + self.root, + self.is_tty, + &thread.attachments, + lock, + cfg.assistant.tool_choice.clone(), + &tools, + self.printer, + self.approvals, + self.chat_request, + self.invocation, + self.pending_trim, + ) + .await + } +} + /// Wait for background MCP server startups to complete. /// /// Shows a single aggregate timer line on stderr once the wait exceeds the @@ -1661,7 +1762,13 @@ fn apply_title_override(lock: &ConversationLock, title: Option<&str>, no_title: } } -fn get_config_delta_from_cli( +/// What a conversation would have to record to arrive at `cfg`. +/// +/// The difference between the configuration the stream already resolves to and +/// the one a turn is about to run under, which is what `--cfg` amounts to: not +/// a setting for one turn, but a change from this point on. +/// `None` when the two agree and there is nothing to record. +pub(crate) fn get_config_delta_from_cli( cfg: &AppConfig, lock: &ConversationLock, ) -> Result> { diff --git a/crates/jp_cli/src/cmd/query/turn_loop.rs b/crates/jp_cli/src/cmd/query/turn_loop.rs index 3d3183b9b..c1961d6c1 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop.rs @@ -202,7 +202,13 @@ pub(super) async fn run_turn_loop( // thread building, response processing) and receives interrupts the inner // streaming/tool handlers decline. Its notifications are consumed at the // top of each phase-loop iteration; the guard drops when the turn ends. - let (_turn_interrupt_guard, mut turn_interrupt_rx) = signals.push_handler(); + // + // Registered under the conversation, so an interrupt that names one reaches + // that turn rather than whichever happens to be topmost. Several turns can be + // in flight at once when something other than a terminal is driving them, and + // a Ctrl-C's "whatever is in front of me" is the wrong guess for a request + // that already said which. + let (_turn_interrupt_guard, mut turn_interrupt_rx) = signals.push_handler_for(lock.id()); let mut turn_state = TurnState::default(); let mut stream_retry = StreamRetryState::new(cfg.assistant.request, is_tty); @@ -571,6 +577,11 @@ pub(super) async fn run_turn_loop( let is_finished = matches!(event, Event::Finished(_)); + // A `Flush` is the provider saying a content block is + // final, which makes it the point where the stream is + // consistent enough to persist. + let commits_content = matches!(event, Event::Flush { .. }); + // `handle_llm_event` returns turn control plus // any newly committed event that needs immediate // shell handling. We dispatch on the committed @@ -585,6 +596,26 @@ pub(super) async fn run_turn_loop( &mut stream_retry, ) }); + + // Persist each completed block and each tool call as it + // lands, rather than once the phase ends. + // + // Durability is one reason: a process that dies mid-turn + // keeps everything the assistant had finished saying, + // instead of losing the whole phase. The other is that + // anything reading the conversation from outside this + // process sees the turn progress, which is what lets a + // second frontend watch a turn rather than only its + // result. + // + // A failed write is not fatal: the phase-end flush will + // try again, and the in-memory stream is still correct. + let commits = commits_content + || matches!(committed, CommittedEvent::ToolCallRequest(_)); + if commits && let Err(error) = conv.flush() { + warn!(%error, "Failed to persist mid-turn; will retry."); + } + match action { LoopAction::Continue => {} LoopAction::Break => break, diff --git a/crates/jp_cli/src/cmd/query/turn_loop_tests.rs b/crates/jp_cli/src/cmd/query/turn_loop_tests.rs index 3fe664f9c..e83e06e8c 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop_tests.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop_tests.rs @@ -415,6 +415,97 @@ async fn test_interrupt_stop_during_streaming_persists_content() { assert!(test_result.is_ok(), "Test timed out after 10 seconds"); } +/// A block the provider has finished is on disk before the turn ends. +/// +/// Read while the turn is still running, which is the only window where this +/// differs from persisting at the end of the phase: the provider commits a +/// block, flushes it, and then parks, and the store is read at that point +/// rather than afterwards. +/// Anything watching the conversation from another process sees the same thing. +#[tokio::test(flavor = "multi_thread")] +async fn a_completed_block_is_persisted_before_the_turn_ends() { + let test_result = Box::pin(timeout(Duration::from_secs(10), async { + let tmp = tempdir().unwrap(); + let root = tmp.path(); + let storage = root.join(".jp"); + + let config = AppConfig::new_test(); + let fs = Arc::new(FsStorageBackend::new(&storage).expect("failed to create backend")); + let mut workspace = Workspace::new(root).with_backend(fs.clone()); + + let lock = workspace + .create_and_lock_conversation(Conversation::default(), config.clone().into(), None) + .unwrap(); + let conv_id = lock.id(); + + let stalled = Arc::new(Notify::new()); + let provider: Arc = Arc::new( + StallingMockProvider::with_message("The answer is 4.").notify_when_stalled(&stalled), + ); + let model = provider + .model_details(&"test-model".parse().unwrap()) + .await + .unwrap(); + + let (printer, _out, _err) = Printer::memory(OutputFormat::TextPretty); + let printer = Arc::new(printer); + let mcp_client = jp_mcp::Client::default(); + let (router, signals) = test_router(); + let router = Arc::new(router); + + // Stop, so the turn ends once the store has been read. + let backend = MockPromptBackend::new().with_inline_responses(['s']); + + // The window: the stream has flushed its block and parked, and the turn + // has not ended. Read the store here, then let the turn finish. + let observer = tokio::spawn({ + let stalled = Arc::clone(&stalled); + let fs = Arc::clone(&fs); + async move { + stalled.notified().await; + let seen = fs.read_test_events_raw(&conv_id); + signals.interrupt().await; + seen + } + }); + + let result = run_turn_loop( + Arc::clone(&provider), + &model, + &config, + &router, + &mcp_client, + root, + false, // is_tty + &[], // attachments + &lock, + ToolChoice::Auto, + &[], // tools + printer.clone(), + Arc::new(backend), + ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), + ChatRequest::from("What is 2+2?"), + InvocationContext::default(), + PendingStreamTrim::default(), + ) + .await; + + let mid_turn = observer.await.unwrap(); + + assert!(result.is_ok(), "Turn loop should complete: {result:?}"); + + let mid_turn = mid_turn.expect("the store is written before the turn ends"); + assert!( + mid_turn.contains("The answer is 4."), + "A flushed block should already be persisted while the turn runs.\nFile \ + contents:\n{mid_turn}" + ); + })) + .await; + + assert!(test_result.is_ok(), "Test timed out after 10 seconds"); +} + /// Cancelling the streaming interrupt menu (a second Ctrl-C) escalates: the /// partial content is committed, a graceful shutdown begins, and the turn ends /// with the interrupt error. diff --git a/crates/jp_cli/src/config_pipeline.rs b/crates/jp_cli/src/config_pipeline.rs index 23847e0b2..ce565382f 100644 --- a/crates/jp_cli/src/config_pipeline.rs +++ b/crates/jp_cli/src/config_pipeline.rs @@ -238,6 +238,26 @@ pub(crate) fn build_partial_from_cfg_args( apply_cfg_args(PartialAppConfig::empty(), &resolved) } +/// Layer raw `--cfg` arguments over a partial, keeping what the base already +/// says. +/// +/// The same resolution [`build_partial_from_cfg_args`] does, applied onto +/// `base` rather than onto nothing. +/// That distinction is the whole difference between "what did these arguments +/// ask for" and "what does this configuration look like once they are applied". +/// +/// Unlike that function, no arguments is not an error: layering nothing over a +/// configuration leaves the configuration. +pub(crate) fn build_partial_over( + base: PartialAppConfig, + args: &[KeyValueOrPath], + workspace: Option<&Workspace>, + fs: Option<&FsStorageBackend>, +) -> Result { + let resolved = resolve_cfg_args(args, &base, workspace, fs)?; + apply_cfg_args(base, &resolved) +} + #[cfg(test)] #[path = "config_pipeline_tests.rs"] mod tests; diff --git a/crates/jp_cli/src/ctx.rs b/crates/jp_cli/src/ctx.rs index 9f76e4e74..7b7135952 100644 --- a/crates/jp_cli/src/ctx.rs +++ b/crates/jp_cli/src/ctx.rs @@ -162,6 +162,23 @@ impl Ctx { self.config.clone() } + /// Install a resolved config for one scoped run, returning the previous + /// one. + /// + /// A `jp` invocation resolves config once and treats it as fixed, which is + /// why there is otherwise no way to change it here. + /// A long-running host is the exception: it serves many conversations from + /// one process, and a turn has to run under the config of the conversation + /// it belongs to, its persona, its enabled tools, its model. + /// Swap before the turn and swap back after. + /// + /// Takes an already-resolved [`AppConfig`], so this stays a substitution + /// rather than a mutation: assembling a config is still the partial API's + /// job. + pub(crate) fn swap_config(&mut self, config: Arc) -> Arc { + std::mem::replace(&mut self.config, config) + } + /// Get a runtime handle. pub(crate) fn handle(&self) -> &Handle { self.runtime.handle() diff --git a/crates/jp_cli/src/lib.rs b/crates/jp_cli/src/lib.rs index 09a525ae6..fb2e38aeb 100644 --- a/crates/jp_cli/src/lib.rs +++ b/crates/jp_cli/src/lib.rs @@ -829,7 +829,7 @@ pub(crate) fn resolve_config( /// No `--cfg` args or per-conversation config. /// /// See: -fn load_base_partial(fs: Option<&FsStorageBackend>) -> Result { +pub(crate) fn load_base_partial(fs: Option<&FsStorageBackend>) -> Result { let partials = load_partial_configs_from_files(fs, absolute_utf8(".").ok())?; let partial = load_partials_with_inheritance(partials)?; diff --git a/crates/jp_cli/src/signals.rs b/crates/jp_cli/src/signals.rs index e8a7590d0..b75e57e99 100644 --- a/crates/jp_cli/src/signals.rs +++ b/crates/jp_cli/src/signals.rs @@ -34,6 +34,7 @@ use std::{ }; use futures::{Stream, StreamExt as _}; +use jp_conversation::ConversationId; use tokio::{ runtime::{Handle, Runtime}, sync::mpsc::{self, error::TrySendError}, @@ -81,13 +82,21 @@ enum Routed { /// /// Created once at application startup; the embedded signal task lives for the /// duration of the process. +/// +/// Cloning shares one router rather than making a second: every clone reads the +/// same handler stack and shutdown token, which is what lets work running off +/// the starting thread register a handler the signal task will actually reach. +#[derive(Clone)] pub struct SignalRouter { inner: Arc, /// Keeps the signal-consuming task attached to the router. /// The task runs until the signal source ends (never, for the OS-backed /// source); the handle is never awaited or aborted. - _signal_task: JoinHandle<()>, + /// + /// Shared rather than duplicated, since there is one task however many + /// handles onto the router exist. + _signal_task: Arc>, } impl SignalRouter { @@ -144,7 +153,7 @@ impl SignalRouter { Self { inner, - _signal_task: signal_task, + _signal_task: Arc::new(signal_task), } } @@ -168,7 +177,25 @@ impl SignalRouter { /// receiver fires. #[must_use] pub fn push_handler(&self) -> (InterruptGuard, mpsc::Receiver<()>) { - self.inner.push_handler() + self.inner.push_handler(None) + } + + /// Register an interrupt handler scope that can also be interrupted by + /// name. + /// + /// Behaves as [`push_handler`] for a Ctrl-C, which still goes to whichever + /// handler is topmost. + /// The scope only matters to [`interrupt_scope`], for interrupts that + /// arrive from somewhere with no notion of "topmost". + /// + /// [`interrupt_scope`]: Self::interrupt_scope + /// [`push_handler`]: Self::push_handler + #[must_use] + pub fn push_handler_for( + &self, + conversation: ConversationId, + ) -> (InterruptGuard, mpsc::Receiver<()>) { + self.inner.push_handler(Some(conversation)) } /// Called by a handler's event loop when it declines to handle the current @@ -179,6 +206,24 @@ impl SignalRouter { pub fn decline(&self) { self.inner.notify_next_or_shutdown(); } + + /// Interrupt one named scope, leaving every other handler alone. + /// + /// For interrupts that arrive with a target rather than from a keyboard. + /// A Ctrl-C means "whatever I am looking at", and the topmost handler is + /// the right guess; a request naming a conversation means that + /// conversation, and with several turns running the topmost handler is very + /// likely the wrong one. + /// + /// Deliberately outside the escalation ladder: a repeat is a repeat of the + /// same targeted request, not a signal to give up on the process. + /// + /// Returns whether a handler was found. + /// `false` means nothing is registered for that scope, so there was nothing + /// to interrupt. + pub fn interrupt_scope(&self, conversation: ConversationId) -> bool { + self.inner.notify_scope(conversation) + } } /// Deregisters its interrupt handler from the router's stack on drop. @@ -201,6 +246,21 @@ struct HandlerId(u64); struct RegisteredHandler { id: HandlerId, + /// The conversation this handler is a scope for, when something other than + /// a keypress might want to interrupt it specifically. + /// + /// A Ctrl-C is aimed at whatever the user is looking at, which is the + /// topmost handler, so the terminal path ignores this. + /// An interrupt arriving over a protocol names its target instead: several + /// turns can be running at once, and stopping the wrong one is worse than + /// stopping nothing. + /// + /// The id itself, not a rendering of it. + /// An id has more than one spelling, and comparing one spelling to another + /// matches nothing while looking exactly like a stop button that does not + /// work. + scope: Option, + /// Notifies the handler's event loop that SIGINT arrived. /// The event loop runs the interrupt logic; the router never does. notify_tx: mpsc::Sender<()>, @@ -343,13 +403,20 @@ impl RouterInner { /// Register a handler scope: push a fresh notification channel onto the /// stack and return the deregistration guard plus the receiver. - fn push_handler(self: &Arc) -> (InterruptGuard, mpsc::Receiver<()>) { + fn push_handler( + self: &Arc, + scope: Option, + ) -> (InterruptGuard, mpsc::Receiver<()>) { let (notify_tx, notify_rx) = mpsc::channel(1); let id = HandlerId(self.next_handler_id.fetch_add(1, Ordering::Relaxed)); self.stack .lock() .expect("handler stack lock poisoned") - .push(RegisteredHandler { id, notify_tx }); + .push(RegisteredHandler { + id, + scope, + notify_tx, + }); ( InterruptGuard { @@ -360,6 +427,31 @@ impl RouterInner { ) } + /// Notify the handler registered for `scope`, if one still is. + /// + /// Searched from the top down, so the innermost handler for a conversation + /// is the one reached, matching how a Ctrl-C finds the innermost handler + /// overall. + /// + /// Returns whether anything was notified. + /// `false` means the scope has no handler, usually because its work already + /// finished, and is not an error: there was nothing left to interrupt. + fn notify_scope(&self, scope: ConversationId) -> bool { + let tx = self + .stack + .lock() + .expect("handler stack lock poisoned") + .iter() + .rev() + .find(|handler| handler.scope == Some(scope)) + .map(|handler| handler.notify_tx.clone()); + + match tx { + Some(tx) => tx.try_send(()).is_ok(), + None => false, + } + } + /// Remove a handler by id. /// /// Id-based rather than positional so guards can drop in any order: early diff --git a/crates/jp_cli/src/signals_tests.rs b/crates/jp_cli/src/signals_tests.rs index 37ab94801..cd0e74e00 100644 --- a/crates/jp_cli/src/signals_tests.rs +++ b/crates/jp_cli/src/signals_tests.rs @@ -6,7 +6,84 @@ use super::*; /// Push a handler scope onto the router state. fn push_handler(inner: &Arc) -> (InterruptGuard, mpsc::Receiver<()>) { - inner.push_handler() + inner.push_handler(None) +} + +/// A fixed conversation id, distinct per `secs`. +fn conversation(secs: u64) -> ConversationId { + ConversationId::try_from( + chrono::DateTime::::UNIX_EPOCH + Duration::from_secs(secs), + ) + .unwrap() +} + +/// A targeted interrupt reaches the named scope and nothing else. +/// +/// The failure this guards against is stopping the wrong turn: with several +/// running, the topmost handler is very likely not the one that was asked for. +#[test] +fn a_scoped_interrupt_notifies_only_its_own_scope() { + let inner = RouterInner::new(Duration::from_secs(2)); + let wanted = conversation(1_700_000_000); + let other = conversation(1_700_000_001); + + let (_guard_wanted, mut rx_wanted) = inner.push_handler(Some(wanted)); + // Pushed after, so it is topmost and would be the one a Ctrl-C reached. + let (_guard_other, mut rx_other) = inner.push_handler(Some(other)); + let (_guard_plain, mut rx_plain) = push_handler(&inner); + + assert!(inner.notify_scope(wanted)); + + assert_eq!(rx_wanted.try_recv(), Ok(())); + assert_eq!(rx_other.try_recv(), Err(TryRecvError::Empty)); + assert_eq!(rx_plain.try_recv(), Err(TryRecvError::Empty)); +} + +/// A scope with no handler is not an error: its work already finished, so there +/// was nothing left to interrupt. +#[test] +fn interrupting_an_unknown_scope_reports_that_nothing_was_reached() { + let inner = RouterInner::new(Duration::from_secs(2)); + let (_guard, mut rx) = push_handler(&inner); + + assert!(!inner.notify_scope(conversation(1_700_000_000))); + + assert_eq!( + rx.try_recv(), + Err(TryRecvError::Empty), + "an unscoped handler is not a fallback target" + ); +} + +/// A dropped guard takes its scope with it, so a later interrupt finds nothing +/// rather than a stale channel. +#[test] +fn a_finished_scope_is_no_longer_reachable() { + let inner = RouterInner::new(Duration::from_secs(2)); + let id = conversation(1_700_000_000); + + let (guard, _rx) = inner.push_handler(Some(id)); + assert!(inner.notify_scope(id)); + + drop(guard); + assert!(!inner.notify_scope(id)); +} + +/// A Ctrl-C still goes to whichever handler is topmost, scoped or not. +/// +/// The scope is extra information for targeted interrupts, not a change to how +/// the keyboard path chooses. +#[test] +fn a_scope_does_not_change_where_a_keypress_lands() { + let inner = RouterInner::new(Duration::from_secs(2)); + + let (_guard_bottom, mut rx_bottom) = inner.push_handler(Some(conversation(1_700_000_000))); + let (_guard_top, mut rx_top) = inner.push_handler(Some(conversation(1_700_000_001))); + + assert_eq!(inner.route_interrupt(Instant::now()), Routed::Handler); + + assert_eq!(rx_top.try_recv(), Ok(())); + assert_eq!(rx_bottom.try_recv(), Err(TryRecvError::Empty)); } #[test] @@ -219,6 +296,37 @@ fn escalation_counter_bumps_and_resets() { assert_eq!(state.bump(now + Duration::from_secs(10)), 1); } +/// A clone is a second handle onto one router, not a second router. +/// +/// Work running away from the thread that started it holds a clone, and a +/// handler it registers has to be one the signal task reaches. +/// Were the state duplicated instead of shared, the press would find an empty +/// stack and fall through to requesting shutdown, which is what this pins. +#[tokio::test] +async fn a_cloned_router_shares_the_handler_stack() { + let (router, signals) = super::testing::test_router(); + let clone = router.clone(); + + // Registered through the clone, delivered through the original's source. + let (_guard, mut interrupt_rx) = clone.push_handler(); + + signals.interrupt().await; + + // Bounded, because the failure this guards against is a notification that + // never arrives: the guard keeps the sender alive, so an unshared stack would + // leave `recv` blocked rather than closed, and the test would hang instead of + // failing. + tokio::time::timeout(Duration::from_secs(5), interrupt_rx.recv()) + .await + .expect("a handler pushed on a clone is reached by the signal task") + .expect("the notification channel stayed open"); + + // The press stopped at the handler, so neither of these ran. + assert!(!router.shutdown_token().is_cancelled()); + assert!(!clone.shutdown_token().is_cancelled()); + assert!(signals.exit_codes().is_empty()); +} + /// Drives the full Ctrl-C escalation ladder through the real signal task — /// handler notification, shutdown, process exit — without ending the test /// process: the injected exit action records the code instead of exiting. diff --git a/crates/jp_config/src/util.rs b/crates/jp_config/src/util.rs index 75593ca19..dd101783d 100644 --- a/crates/jp_config/src/util.rs +++ b/crates/jp_config/src/util.rs @@ -155,6 +155,62 @@ pub fn find_file_in_load_path( None } +/// List every segment [`find_file_in_load_path`] can resolve within a load +/// path. +/// +/// The inverse of that lookup: instead of asking where one name lives, this +/// walks the load path and reports every configuration file as the segment that +/// selects it, the path relative to `load_path` without its extension. +/// Directories are part of that segment, so `/skill/rfd.toml` is +/// reported as `skill/rfd`, and passing that back to `--cfg` finds the same +/// file. +/// +/// Segments are sorted, and a load path that does not exist yields nothing +/// rather than an error: an absent directory is a load path with nothing in it. +pub fn list_configs_in_load_path(load_path: &dyn AsRef) -> Vec { + let root = load_path.as_ref(); + let mut segments = Vec::new(); + + collect_config_segments(root, root, &mut segments); + segments.sort(); + segments.dedup(); + segments +} + +/// Recurse `dir`, pushing each configuration file's segment relative to `root`. +fn collect_config_segments(root: &Path, dir: &Path, out: &mut Vec) { + let Ok(entries) = fs::read_dir(dir) else { + return; + }; + + for entry in entries.flatten() { + let path = entry.path(); + + if path.is_dir() { + collect_config_segments(root, &path, out); + continue; + } + + let is_config = path + .extension() + .and_then(OsStr::to_str) + .is_some_and(|ext| VALID_CONFIG_FILE_EXTS.contains(&ext)); + + if !is_config { + continue; + } + + // A name that does not survive the round trip through UTF-8 could not + // have been typed as a `--cfg` argument either, so dropping it loses + // nothing selectable. + if let Ok(relative) = path.strip_prefix(root) + && let Some(segment) = relative.with_extension("").to_str() + { + out.push(segment.replace('\\', "/")); + } + } +} + /// Load a partial configuration from a file at `path`, if it exists. /// /// This loads either the file directly, or tries to load a file with the same diff --git a/crates/jp_config/src/util_tests.rs b/crates/jp_config/src/util_tests.rs index 195964ce6..f8b9bc134 100644 --- a/crates/jp_config/src/util_tests.rs +++ b/crates/jp_config/src/util_tests.rs @@ -613,6 +613,59 @@ fn test_load_partial_at_path_recursive() { } } +/// The inverse of `find_file_in_load_path`: every segment it could resolve. +/// +/// Nested directories are part of the segment, since that is what selects the +/// file, and non-config files are not selectable at all. +#[test] +fn test_list_configs_in_load_path() { + let tmp = tempdir().unwrap(); + let root = tmp.path(); + + write_config(&root.join("default.toml"), ""); + write_config(&root.join("skill/rfd.toml"), ""); + write_config(&root.join("skill/web.yaml"), ""); + write_config(&root.join("persona/deep/nested.json"), ""); + + // Neither is a configuration file, so neither is selectable. + fs::write(root.join("README.md"), "").unwrap(); + fs::write(root.join("skill/notes.txt"), "").unwrap(); + + assert_eq!(list_configs_in_load_path(&root), vec![ + "default".to_owned(), + "persona/deep/nested".to_owned(), + "skill/rfd".to_owned(), + "skill/web".to_owned(), + ]); +} + +/// A load path that isn't there holds nothing, which is not an error: a +/// workspace need not have every directory the load path names. +#[test] +fn test_list_configs_in_missing_load_path() { + let tmp = tempdir().unwrap(); + + assert!(list_configs_in_load_path(&tmp.path().join("absent")).is_empty()); +} + +/// Each segment is what `find_file_in_load_path` resolves back to the same +/// file, which is the contract that makes a listed segment usable as `--cfg`. +#[test] +fn test_listed_segments_resolve_back_to_their_files() { + let tmp = tempdir().unwrap(); + let root = tmp.path(); + + write_config(&root.join("default.toml"), ""); + write_config(&root.join("skill/rfd.toml"), ""); + + for segment in list_configs_in_load_path(&root) { + assert!( + find_file_in_load_path(&segment, &root).is_some(), + "`{segment}` was listed but does not resolve" + ); + } +} + #[test] fn test_load_partial_at_path_self_extending_cycle() { let tmp = tempdir().unwrap(); diff --git a/crates/jp_md/src/buffer/fixup.rs b/crates/jp_md/src/buffer/fixup.rs index c586cd171..2f0d94961 100644 --- a/crates/jp_md/src/buffer/fixup.rs +++ b/crates/jp_md/src/buffer/fixup.rs @@ -33,7 +33,10 @@ use super::Event; /// or pass them unchanged. /// Fixups may hold state across events (e.g. remembering properties of the /// previous block). -pub trait EventFixup { +/// Implementors must be `Send`: a renderer built from these travels to +/// whichever task or thread is running a turn, and a turn does not always run +/// on the one that started it. +pub trait EventFixup: Send { /// Process a single event. /// Returns `None` to suppress the event, or `Some(event)` (possibly /// modified) to pass it through. diff --git a/crates/jp_plugin/src/lib.rs b/crates/jp_plugin/src/lib.rs index aaa41c567..7e3c8f412 100644 --- a/crates/jp_plugin/src/lib.rs +++ b/crates/jp_plugin/src/lib.rs @@ -11,7 +11,7 @@ mod protocol; pub mod registry; pub use message::{HostToPlugin, PluginToHost}; -pub use protocol::{Error, PROTOCOL_VERSION}; +pub use protocol::{Error, PROTOCOL_VERSION, ready}; #[cfg(test)] #[path = "lib_tests.rs"] diff --git a/crates/jp_plugin/src/lib_tests.rs b/crates/jp_plugin/src/lib_tests.rs index f6a53d7ec..b8dd8611f 100644 --- a/crates/jp_plugin/src/lib_tests.rs +++ b/crates/jp_plugin/src/lib_tests.rs @@ -1,6 +1,22 @@ use serde_json::json; -use crate::message::*; +use crate::{message::*, ready}; + +#[test] +fn the_handshake_refuses_a_host_that_is_too_old() { + assert_eq!(ready(2, 2), Ok(ReadyMessage { protocol: 2 })); + assert_eq!(ready(1, 2), Ok(ReadyMessage { protocol: 1 })); + + let exit = ready(2, 1).unwrap_err(); + assert_eq!(exit.code, 1); + assert!( + exit.reason + .as_ref() + .is_some_and(|reason| reason.contains("protocol 2") && reason.contains("speaks 1")), + "{:?}", + exit.reason + ); +} #[test] fn conversations_response_serializes_without_null_id() { diff --git a/crates/jp_plugin/src/message.rs b/crates/jp_plugin/src/message.rs index 8ba34bd8f..4525260d9 100644 --- a/crates/jp_plugin/src/message.rs +++ b/crates/jp_plugin/src/message.rs @@ -56,6 +56,36 @@ pub enum HostToPlugin { /// Response to `read_config`. Config(ConfigResponse), + /// Response to `compose`. + Composed(ComposeResponse), + + /// A request that only reports whether it worked, worked. + /// + /// The answer to `archive_conversation` and `set_title`. + /// A failure comes back as [`HostToPlugin::Error`] instead, naming which + /// request it was. + Done(DoneResponse), + + /// Response to `read_draft` and `write_draft`. + Draft(DraftResponse), + + /// Response to `list_configs`. + Configs(ConfigsResponse), + + /// A delegated turn finished. + QueryComplete(QueryCompleteResponse), + + /// A conversation asked for by `query` with `new` exists. + /// + /// Sent as soon as it has been created and locked, before its first turn + /// runs. + /// A caller that only needs somewhere to send the user cannot wait for the + /// turn: it can take minutes, and the conversation is usable immediately. + /// + /// The first of two replies to such a request; `query_complete` follows + /// when the turn ends. + Created(CreatedResponse), + /// An error response to any plugin request. Error(ErrorResponse), @@ -74,8 +104,8 @@ pub enum HostToPlugin { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(tag = "type", rename_all = "snake_case")] pub enum PluginToHost { - /// Acknowledge successful initialization. - Ready, + /// Acknowledge successful initialization, and state what the plugin needs. + Ready(ReadyMessage), /// Request a list of conversations. ListConversations(OptionalId), @@ -86,6 +116,30 @@ pub enum PluginToHost { /// Request the resolved config (or a subtree). ReadConfig(ReadConfigRequest), + /// Ask the host to collect text from the user. + Compose(ComposeRequest), + + /// Move a conversation to the archive. + ArchiveConversation(ConversationRequest), + + /// Rename a conversation, or clear its name. + SetTitle(SetTitleRequest), + + /// Read a conversation's query draft. + ReadDraft(ConversationRequest), + + /// Replace a conversation's query draft. + WriteDraft(WriteDraftRequest), + + /// List the configurations a query can name. + ListConfigs(OptionalId), + + /// Ask the host to run a turn on a conversation. + Query(QueryRequest), + + /// Ask the host to interrupt the turn running on a conversation. + Interrupt(InterruptRequest), + /// Print user-facing output through JP's printer. Print(PrintMessage), @@ -99,6 +153,40 @@ pub enum PluginToHost { Exit(ExitMessage), } +impl PluginToHost { + /// The correlation ID this message carries, if any. + /// + /// `None` covers two cases that need no distinguishing here: a request that + /// omitted its optional id, and a message that is not a request at all. + /// Neither has an answer to correlate. + /// + /// Exists so a host can answer a request it could not otherwise inspect, + /// notably when handling it failed and the failure has to be reported + /// against the right request rather than sent into the void for the plugin + /// to time out on. + #[must_use] + pub fn id(&self) -> Option<&str> { + match self { + Self::ListConversations(m) | Self::ListConfigs(m) => m.id.as_deref(), + Self::ReadEvents(m) => m.id.as_deref(), + Self::ReadConfig(m) => m.id.as_deref(), + Self::Compose(m) => m.id.as_deref(), + Self::Query(m) => m.id.as_deref(), + Self::ArchiveConversation(m) | Self::ReadDraft(m) => m.id.as_deref(), + Self::SetTitle(m) => m.id.as_deref(), + Self::WriteDraft(m) => m.id.as_deref(), + + // Not requests: nothing is waiting on an answer to any of these. + Self::Ready(_) + | Self::Interrupt(_) + | Self::Print(_) + | Self::Log(_) + | Self::Describe(_) + | Self::Exit(_) => None, + } + } +} + // --- Host-to-Plugin messages --- /// The `init` message sent to the plugin on startup. @@ -210,6 +298,189 @@ pub struct ConfigResponse { pub data: Value, } +/// A request that only reports whether it worked, worked. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct DoneResponse { + /// Optional request correlation ID. + #[serde(skip_serializing_if = "Option::is_none")] + pub id: Option, +} + +/// A conversation's query draft. +/// +/// The answer to both `read_draft` and `write_draft`, so a write reports back +/// what the draft now holds without a second read. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct DraftResponse { + /// Optional request correlation ID. + #[serde(skip_serializing_if = "Option::is_none")] + pub id: Option, + + /// The conversation the draft belongs to. + pub conversation: String, + + /// The draft text, empty when there is no draft. + pub content: String, + + /// A fingerprint of `content`, absent when there is no draft. + /// + /// Passed back in the next `write_draft` to say which version was edited. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub revision: Option, + + /// Whether a `write_draft` was refused because the draft had moved on. + /// + /// When true, `content` and `revision` describe what is on disk, not what + /// was submitted, and the write did not happen. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub conflict: bool, +} + +/// Replace a conversation's query draft. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct WriteDraftRequest { + /// Optional request correlation ID. + #[serde(skip_serializing_if = "Option::is_none")] + pub id: Option, + + /// The conversation the draft belongs to. + pub conversation: String, + + /// The new draft text. + /// Empty removes the draft. + pub content: String, + + /// The `revision` the edit was based on, from an earlier draft response. + /// + /// Absent means "there was no draft when I started". + /// A mismatch against what is on disk is refused rather than overwritten. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub revision: Option, +} + +/// Ask the host to run a turn on a conversation. +/// +/// The host owns the agent loop: it locks the conversation, appends the +/// request, calls the provider, runs whatever tools the assistant asks for, and +/// persists the result. +/// A plugin doing this itself would need the user's credentials, the tool +/// registry, and the MCP servers, and would end up a second implementation of +/// the turn loop. +/// +/// The host answers with [`HostToPlugin::QueryComplete`] once the turn has +/// finished, or [`HostToPlugin::Error`] if it could not be started. +/// A turn runs for as long as the assistant needs, so a plugin awaiting the +/// reply should allow minutes, not seconds. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct QueryRequest { + /// Optional request correlation ID. + #[serde(skip_serializing_if = "Option::is_none")] + pub id: Option, + + /// The conversation to add the turn to. + /// + /// Ignored when `new` is set, which makes the host create one. + #[serde(default)] + pub conversation: String, + + /// What the user said. + pub content: String, + + /// Start a new conversation rather than adding to an existing one. + /// + /// The host replies twice: [`HostToPlugin::Created`] as soon as the + /// conversation exists, carrying the id it assigned, then + /// [`HostToPlugin::QueryComplete`] when the turn ends. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub new: bool, + + /// A title for a conversation created by this request. + /// + /// Left unset, a new conversation is untitled until the title generator + /// names it from the first turn. + /// Ignored without `new`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub title: Option, + + /// Configurations to layer over what this turn would otherwise run under, + /// as `--cfg` takes them. + /// + /// Not scoped to the turn. + /// The host records the difference as a config event, so the choice holds + /// for the turns after it and the stream carries the reason, which is what + /// `jp q --cfg` does. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub cfg: Vec, +} + +/// Ask the host to interrupt the turn running on a conversation. +/// +/// Reaches the turn the same way a Ctrl-C from a terminal would, so it +/// escalates on repeat exactly as the terminal does: the first asks the turn to +/// wrap up, and pressing on abandons it. +/// +/// Fire-and-forget: the host sends no acknowledgement, because what the +/// interrupt did shows up in the conversation itself. +/// The outcome of the turn still arrives as the reply to the original `query`. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct InterruptRequest { + /// The conversation whose turn should stop. + /// + /// Required, and not a convenience: a host can be running several turns at + /// once, so there is no "the" turn to infer. + pub conversation: String, +} + +/// Response to `query`, sent once the turn has finished. +/// +/// Carries no transcript: the events are persisted, and the plugin reads them +/// back with [`PluginToHost::ReadEvents`]. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct QueryCompleteResponse { + /// Optional request correlation ID. + #[serde(skip_serializing_if = "Option::is_none")] + pub id: Option, + + /// The conversation the turn ran on. + pub conversation: String, +} + +/// The conversation a `query` with `new` created. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct CreatedResponse { + /// Optional request correlation ID. + #[serde(skip_serializing_if = "Option::is_none")] + pub id: Option, + + /// The id the host assigned, which the caller has no other way to learn. + pub conversation: String, +} + +/// One configuration a query can name. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct ConfigEntry { + /// What to pass to select it, e.g. `skill/rfd`. + pub segment: String, + + /// The directories above it, e.g. `skill`. + /// Empty at the top level. + pub namespace: String, + + /// The last part of the segment, e.g. `rfd`. + pub name: String, +} + +/// Response to `list_configs`. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct ConfigsResponse { + /// Optional request correlation ID. + #[serde(skip_serializing_if = "Option::is_none")] + pub id: Option, + + /// Every selectable configuration, sorted by segment. + pub data: Vec, +} + /// An error response. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct ErrorResponse { @@ -227,6 +498,26 @@ pub struct ErrorResponse { // --- Plugin-to-Host messages --- +/// The plugin's answer to `init`. +/// +/// Carrying the required protocol version here rather than leaving each plugin +/// to check for itself means the host can refuse a plugin it is too old to +/// serve, and a plugin cannot forget to ask: the field has no Rust default, so +/// it has to be named at every construction site. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct ReadyMessage { + /// The lowest protocol version this plugin can work with. + /// + /// Defaults to 1 on the wire, so a plugin built before this field existed + /// still parses, and is taken at its word. + #[serde(default = "legacy_protocol")] + pub protocol: u32, +} + +const fn legacy_protocol() -> u32 { + 1 +} + /// A message with only an optional correlation ID. #[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] pub struct OptionalId { @@ -246,6 +537,35 @@ pub struct ReadEventsRequest { pub conversation: String, } +/// A request naming one conversation and nothing else. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct ConversationRequest { + /// Optional request correlation ID. + #[serde(skip_serializing_if = "Option::is_none")] + pub id: Option, + + /// The conversation to act on. + pub conversation: String, +} + +/// Rename a conversation, or clear its name. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct SetTitleRequest { + /// Optional request correlation ID. + #[serde(skip_serializing_if = "Option::is_none")] + pub id: Option, + + /// The conversation to rename. + pub conversation: String, + + /// The new title. + /// + /// Absent, or blank, clears it: the conversation is then eligible for a + /// generated title again rather than being named the empty string. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub title: Option, +} + /// Request to read config. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct ReadConfigRequest { @@ -258,6 +578,96 @@ pub struct ReadConfigRequest { pub path: Option, } +/// Ask the host to collect text from the user. +/// +/// Composition happens on the host rather than in the plugin because the host +/// owns both ends of it: a plugin's stdin carries this protocol, so it has no +/// terminal to read keys from, and only the host knows which editor the +/// `Ctrl+X` escape should open. +/// +/// The host answers with [`HostToPlugin::Composed`]. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct ComposeRequest { + /// Optional request correlation ID. + #[serde(skip_serializing_if = "Option::is_none")] + pub id: Option, + + /// Rendered before the input, naming what is being asked for. + pub message: String, + + /// What kind of input to collect. + pub mode: ComposeMode, + + /// Help text rendered alongside the prompt. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub help: Option, +} + +/// What a [`ComposeRequest`] asks for. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ComposeMode { + /// A single line, pre-filled with `default`. + Line { + #[serde(default, skip_serializing_if = "Option::is_none")] + default: Option, + }, + + /// A multi-line buffer seeded with `initial_text`, offering the `Ctrl+X` + /// escape to the user's configured editor. + Buffer { + #[serde(default, skip_serializing_if = "Option::is_none")] + initial_text: Option, + }, + + /// One of a fixed set of choices. + /// + /// The response carries the chosen option's `value`, not its label. + Select { + options: Vec, + + /// The `value` to start the selection on. + #[serde(default, skip_serializing_if = "Option::is_none")] + default: Option, + }, + + /// Any number of a fixed set of choices. + /// + /// The response carries the chosen `value`s in [`ComposeResponse::values`]. + MultiSelect { options: Vec }, +} + +/// One choice in a [`ComposeMode::Select`]. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct ComposeOption { + /// What the plugin gets back when this is chosen. + pub value: String, + + /// What the user reads. + pub label: String, +} + +/// Response to `compose`. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct ComposeResponse { + /// Optional request correlation ID. + #[serde(skip_serializing_if = "Option::is_none")] + pub id: Option, + + /// What the user wrote, or the single value they chose. + /// + /// `None` when they cancelled, when there was no terminal to ask on, or + /// when the request was a multi-select. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub text: Option, + + /// The values chosen from a [`ComposeMode::MultiSelect`]. + /// + /// Empty for every other mode, and for a cancelled or unanswerable one. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub values: Vec, +} + /// Print user-facing output through JP's printer. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct PrintMessage { diff --git a/crates/jp_plugin/src/message_tests.rs b/crates/jp_plugin/src/message_tests.rs index c0dc65673..7acfe17c2 100644 --- a/crates/jp_plugin/src/message_tests.rs +++ b/crates/jp_plugin/src/message_tests.rs @@ -62,14 +62,23 @@ fn paths_info_omits_none_fields() { #[test] fn plugin_ready_roundtrip() { - let msg = PluginToHost::Ready; + let msg = PluginToHost::Ready(ReadyMessage { protocol: 2 }); let json = serde_json::to_string(&msg).unwrap(); - assert_eq!(json, r#"{"type":"ready"}"#); + assert_eq!(json, r#"{"type":"ready","protocol":2}"#); let parsed: PluginToHost = from_str(&json).unwrap(); assert_eq!(msg, parsed); } +/// A plugin built before the field existed sends a bare `ready`, and is taken +/// at its word: it can only have been written against protocol 1. +#[test] +fn plugin_ready_without_a_protocol_reads_as_the_first_version() { + let parsed: PluginToHost = from_str(r#"{"type":"ready"}"#).unwrap(); + + assert_eq!(parsed, PluginToHost::Ready(ReadyMessage { protocol: 1 })); +} + #[test] fn plugin_list_conversations_roundtrip() { let msg = PluginToHost::ListConversations(OptionalId { id: None }); diff --git a/crates/jp_plugin/src/protocol.rs b/crates/jp_plugin/src/protocol.rs index 08b4d73cd..9d4df5292 100644 --- a/crates/jp_plugin/src/protocol.rs +++ b/crates/jp_plugin/src/protocol.rs @@ -1,7 +1,43 @@ //! Protocol constants and helpers. +use crate::message::{ExitMessage, ReadyMessage}; + /// Current protocol version. -pub const PROTOCOL_VERSION: u32 = 1; +/// +/// Bumped whenever the host gains a message or a message variant a plugin might +/// send. +/// +/// | Version | Adds | +/// | ------- | --------------------------------------------------------------- | +/// | 1 | The initial protocol. | +/// | 2 | `compose` / `composed`, for host-run prompts. | +/// | 3 | `archive_conversation` and `set_title`, answered with `done`. | +/// | 4 | `read_draft` / `write_draft`, for a conversation's query draft. | +/// | 5 | `list_configs`, naming the configurations a query can select. | +/// | 6 | `query`, with `created` and `query_complete` in reply. | +/// | 7 | `interrupt`, for stopping a turn the host is running. | +pub const PROTOCOL_VERSION: u32 = 7; + +/// Answer a host's `init`, refusing it when it is too old to serve this plugin. +/// +/// `required` is the lowest protocol version the plugin can work with; `host` +/// is the version from [`crate::message::InitMessage`]. +/// The `Err` case is the [`ExitMessage`] to send instead of going any further: +/// a plugin that carried on would send messages the host cannot read, and then +/// block on replies that never come. +pub fn ready(required: u32, host: u32) -> Result { + if host < required { + return Err(ExitMessage { + code: 1, + reason: Some(format!( + "this plugin needs `jp` protocol {required}, and this `jp` speaks {host}. \ + Reinstall the two together." + )), + }); + } + + Ok(ReadyMessage { protocol: required }) +} /// Errors that can occur during plugin protocol communication. #[derive(Debug, thiserror::Error)] diff --git a/crates/plugins/command/path/src/main.rs b/crates/plugins/command/path/src/main.rs index 160569115..18cbefc34 100644 --- a/crates/plugins/command/path/src/main.rs +++ b/crates/plugins/command/path/src/main.rs @@ -13,6 +13,11 @@ use jp_plugin::message::{ DescribeResponse, ExitMessage, HostToPlugin, InitMessage, PluginToHost, PrintMessage, }; +/// The protocol version this plugin needs from the host. +/// +/// It only reads paths out of `init`, so anything that can spawn it will do. +const REQUIRED_PROTOCOL: u32 = 1; + const HELP_TEXT: &str = "\ Print JP directory paths. @@ -60,7 +65,10 @@ fn run(mut stdin: impl BufRead, mut stdout: impl Write) -> Result<(), String> { match first_msg { HostToPlugin::Describe => send_describe(&mut stdout), HostToPlugin::Init(init) => { - send(&mut stdout, &PluginToHost::Ready)?; + match jp_plugin::ready(REQUIRED_PROTOCOL, init.version) { + Ok(ready) => send(&mut stdout, &PluginToHost::Ready(ready))?, + Err(exit) => return send(&mut stdout, &PluginToHost::Exit(exit)), + } handle_command(&init, &mut stdout) } other => Err(format!("expected init or describe, got: {other:?}")), diff --git a/crates/plugins/command/serve-web/src/client.rs b/crates/plugins/command/serve-web/src/client.rs index 8bab72427..ce99bd975 100644 --- a/crates/plugins/command/serve-web/src/client.rs +++ b/crates/plugins/command/serve-web/src/client.rs @@ -241,6 +241,17 @@ fn reader_loop(reader: impl BufRead, inner: &Inner, shutdown_tx: &watch::Sender< warn!("Unexpected message after startup"); } + // This plugin only reads, so neither of these answers a request it + // sent: they belong to something that isn't ours. + HostToPlugin::Composed(_) + | HostToPlugin::Done(_) + | HostToPlugin::Draft(_) + | HostToPlugin::Configs(_) + | HostToPlugin::QueryComplete(_) + | HostToPlugin::Created(_) => { + warn!(?msg, "Received a response to a request we never sent"); + } + // Response messages — dispatch to the pending request. msg @ (HostToPlugin::Conversations(_) | HostToPlugin::Events(_) diff --git a/crates/plugins/command/serve-web/src/main.rs b/crates/plugins/command/serve-web/src/main.rs index 03d881736..45082bfa7 100644 --- a/crates/plugins/command/serve-web/src/main.rs +++ b/crates/plugins/command/serve-web/src/main.rs @@ -26,6 +26,12 @@ use crate::{ log_layer::{ProtocolLogHandle, ProtocolLogLayer}, }; +/// The protocol version this plugin needs from the host. +/// +/// It reads conversations, events, and config, all of which the first version +/// carries. +const REQUIRED_PROTOCOL: u32 = 1; + const HELP_TEXT: &str = "\ Start the read-only web interface for browsing JP conversations. @@ -141,7 +147,10 @@ fn run_server( } // Send early protocol messages before sharing stdout. - send(&mut stdout, &PluginToHost::Ready)?; + match jp_plugin::ready(REQUIRED_PROTOCOL, init.version) { + Ok(ready) => send(&mut stdout, &PluginToHost::Ready(ready))?, + Err(exit) => return send(&mut stdout, &PluginToHost::Exit(exit)), + } send( &mut stdout, &PluginToHost::Print(PrintMessage { diff --git a/crates/plugins/command/ticket/src/main.rs b/crates/plugins/command/ticket/src/main.rs index 332f1bde4..3a57bba24 100644 --- a/crates/plugins/command/ticket/src/main.rs +++ b/crates/plugins/command/ticket/src/main.rs @@ -18,8 +18,8 @@ use chrono::{Local, SecondsFormat, Utc}; use clap::{CommandFactory, Parser, Subcommand}; use jp_github::models::issues::{Comment as IssueComment, Issue}; use jp_plugin::message::{ - DescribeResponse, ExitMessage, HostToPlugin, InitMessage, LogMessage, PluginToHost, - PrintMessage, + ComposeMode, ComposeOption, ComposeRequest, DescribeResponse, ExitMessage, HostToPlugin, + InitMessage, LogMessage, PluginToHost, PrintMessage, }; use serde::Serialize; use serde_json::Value; @@ -243,14 +243,29 @@ fn run(mut stdin: impl BufRead, mut stdout: impl Write) -> Result<(), String> { match read_message(&mut stdin)? { HostToPlugin::Describe => send_describe(&mut stdout), HostToPlugin::Init(init) => { - send(&mut stdout, &PluginToHost::Ready)?; - handle_command(&init, &mut stdout) + match jp_plugin::ready(REQUIRED_PROTOCOL, init.version) { + Ok(ready) => send(&mut stdout, &PluginToHost::Ready(ready))?, + Err(exit) => return send(&mut stdout, &PluginToHost::Exit(exit)), + } + handle_command(&init, &mut stdin, &mut stdout) } other => Err(format!("expected init or describe, got: {other:?}")), } } -fn handle_command(init: &InitMessage, stdout: &mut impl Write) -> Result<(), String> { +/// The protocol version this plugin needs from the host. +/// +/// Composition (`compose` / `composed`) arrived in 2. +/// Running against an older `jp` means every interactive path sends a message +/// the host cannot read, so the handshake refuses it up front rather than +/// discovering it mid-prompt. +const REQUIRED_PROTOCOL: u32 = 2; + +fn handle_command( + init: &InitMessage, + stdin: &mut impl BufRead, + stdout: &mut impl Write, +) -> Result<(), String> { let parsed = Args::try_parse_from( std::iter::once("jp ticket".to_owned()).chain(with_show_alias(&init.args)), ); @@ -267,7 +282,10 @@ fn handle_command(init: &InitMessage, stdout: &mut impl Write) -> Result<(), Str let dir = resolve_dir(&init.workspace.root, args.dir.as_deref()); - let command = args.command; + let command = match compose_missing(&dir, args.command, stdin, stdout) { + Ok(command) => command, + Err(message) => return send_exit(stdout, 1, Some(&message)), + }; match execute(&dir, command, &init.config) { Ok(output) => { @@ -290,6 +308,339 @@ fn handle_command(init: &InitMessage, stdout: &mut impl Write) -> Result<(), Str } } +/// How composed text divides into a title and a body. +#[derive(Debug, PartialEq, Eq)] +enum Composition { + /// Nothing to file. + Empty, + /// One line: a title on its own. + Title(String), + /// A title, a blank line, and the rest. + TitleAndBody { title: String, body: String }, + /// Prose that runs from the first line into the second, so all of it is + /// body and the title has to be asked for separately. + Body(String), +} + +impl Composition { + /// Read composed text the way a commit message reads: subject, blank line, + /// then the rest. + /// + /// Text that runs straight on from the first line has no subject, so it is + /// all body. + /// Trailing blank lines never change the reading. + fn read(text: &str) -> Self { + let text = text.trim_end(); + let mut lines = text.lines(); + + let Some(first) = lines.next().map(str::trim).filter(|line| !line.is_empty()) else { + return Self::Empty; + }; + + match lines.next() { + None => Self::Title(first.to_owned()), + Some(second) if !second.trim().is_empty() => Self::Body(text.to_owned()), + Some(_) => { + let body = lines.collect::>().join("\n").trim().to_owned(); + if body.is_empty() { + Self::Title(first.to_owned()) + } else { + Self::TitleAndBody { + title: first.to_owned(), + body, + } + } + } + } + } +} + +/// Title used when the composed text is all body and the user names nothing. +const UNTITLED: &str = "untitled"; + +/// Ask the user for anything the command line didn't carry. +/// +/// Composition runs through the host: the plugin's stdin is this protocol, so +/// the host owns the terminal and the editor the `Ctrl+X` escape opens. +fn compose_missing( + dir: &Utf8Path, + command: Command, + stdin: &mut impl BufRead, + stdout: &mut impl Write, +) -> Result { + match command { + Command::Add { + kind, + title, + author, + body, + implements, + } if kind.is_none() || title.is_none() => { + // Kind first: it frames what you're about to write. The title is + // read out of the composed text, or asked for last. + let kind = match kind { + Some(kind) => kind, + None => pick_kind(stdin, stdout)?, + }; + let (title, body) = match title { + Some(title) => (title, body), + None => compose_ticket(Some(kind), body, stdin, stdout)?, + }; + + Ok(Command::Add { + kind: Some(kind), + title: Some(title), + author, + body, + implements, + }) + } + + Command::Comment { + id: None, + author, + re, + body, + } => compose_missing( + dir, + Command::Comment { + id: Some(pick_ticket(dir, stdin, stdout, "Comment on", false)?), + author, + re, + body, + }, + stdin, + stdout, + ), + + // A comment has no title, so the whole buffer is its body. + Command::Comment { + id: Some(id), + author, + re, + body: None, + } => { + let body = compose(stdin, stdout, ComposeRequest { + id: None, + message: format!("Comment on {id}"), + mode: ComposeMode::Buffer { initial_text: None }, + help: None, + })?; + + Ok(Command::Comment { + id: Some(id), + author, + re, + body: Some(body), + }) + } + + // Closing offers only what is still open. + Command::Close { id: None } => Ok(Command::Close { + id: Some(pick_ticket(dir, stdin, stdout, "Close", true)?), + }), + + Command::Show { id: None, json } => Ok(Command::Show { + id: Some(pick_ticket(dir, stdin, stdout, "Show", false)?), + json, + }), + + Command::Edit { + id: None, + title, + body, + kind, + status, + } => Ok(Command::Edit { + id: Some(pick_ticket(dir, stdin, stdout, "Edit", false)?), + title, + body, + kind, + status, + }), + + Command::Delete { id: None } => Ok(Command::Delete { + id: Some(pick_ticket(dir, stdin, stdout, "Delete", false)?), + }), + + Command::Promote { id: None, to } => Ok(Command::Promote { + id: Some(pick_ticket(dir, stdin, stdout, "Promote", true)?), + to, + }), + + // Importing several at once is the common case after a triage sweep. + Command::Import { + numbers, + repo, + kind, + } if numbers.is_empty() => Ok(Command::Import { + numbers: pick_issues(&repo, stdin, stdout)?, + repo, + kind, + }), + + other => Ok(other), + } +} + +/// Ask which of a repository's open issues to import. +fn pick_issues( + repo: &str, + stdin: &mut impl BufRead, + stdout: &mut impl Write, +) -> Result, String> { + let (owner, name) = repo + .split_once('/') + .ok_or_else(|| format!("`{repo}` is not an `owner/name` pair."))?; + + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|error| format!("failed to start the async runtime: {error}"))?; + let issues = runtime.block_on(fetch_open(owner, name))?; + + let options: Vec = issues + .iter() + // A pull request is an issue on this endpoint, and isn't importable. + .filter(|issue| issue.pull_request.is_none()) + .map(|issue| ComposeOption { + value: issue.number.to_string(), + label: format!("#{:<5} {}", issue.number, issue.title), + }) + .collect(); + + if options.is_empty() { + return Err(format!("No open issues in {repo}.")); + } + + let chosen = compose_many(stdin, stdout, ComposeRequest { + id: None, + message: format!("Import from {repo}"), + mode: ComposeMode::MultiSelect { options }, + help: Some("Space to select, Enter to import.".to_owned()), + })?; + + chosen + .iter() + .map(|value| { + value + .parse() + .map_err(|_| format!("`{value}` is not an issue number.")) + }) + .collect() +} + +/// Ask which kind of work a ticket describes. +fn pick_kind(stdin: &mut impl BufRead, stdout: &mut impl Write) -> Result { + let chosen = compose(stdin, stdout, ComposeRequest { + id: None, + message: "Kind of work".to_owned(), + mode: ComposeMode::Select { + options: [Kind::Bug, Kind::Feature, Kind::Chore] + .into_iter() + .map(|kind| ComposeOption { + value: kind.to_string(), + label: kind.to_string(), + }) + .collect(), + default: None, + }, + help: None, + })?; + + chosen + .parse() + .map_err(|_| format!("`{chosen}` is not a kind.")) +} + +/// Compose a ticket's title and description, asking for the title separately +/// when the text has no subject line. +fn compose_ticket( + kind: Option, + initial: Option, + stdin: &mut impl BufRead, + stdout: &mut impl Write, +) -> Result<(String, Option), String> { + let composed = compose(stdin, stdout, ComposeRequest { + id: None, + message: kind.map_or_else( + || "New ticket".to_owned(), + |kind| format!("New {kind} ticket"), + ), + mode: ComposeMode::Buffer { + initial_text: initial, + }, + help: Some("First line is the title, then a blank line, then the description.".to_owned()), + })?; + + match Composition::read(&composed) { + Composition::Empty => Err("Nothing to file.".to_owned()), + Composition::Title(title) => Ok((title, None)), + Composition::TitleAndBody { title, body } => Ok((title, Some(body))), + Composition::Body(body) => { + let title = compose(stdin, stdout, ComposeRequest { + id: None, + message: "Title".to_owned(), + mode: ComposeMode::Line { + default: Some(UNTITLED.to_owned()), + }, + help: None, + })?; + let title = title.trim(); + + Ok(( + if title.is_empty() { UNTITLED } else { title }.to_owned(), + Some(body), + )) + } + } +} + +/// Ask which ticket to act on. +/// +/// `open_only` drops the closed ones, for the actions that only make sense on +/// live work. +fn pick_ticket( + dir: &Utf8Path, + stdin: &mut impl BufRead, + stdout: &mut impl Write, + verb: &str, + open_only: bool, +) -> Result { + let entries = store::list(dir).map_err(|error| error.to_string())?; + + let options: Vec = entries + .iter() + .filter_map(|entry| entry.ticket.as_ref().ok()) + .filter(|ticket| !open_only || ticket.metadata.status != Status::Done) + .map(|ticket| ComposeOption { + value: ticket.id.to_string(), + label: format!( + "{} {:<12} {}", + ticket.id, ticket.metadata.status, ticket.title + ), + }) + .collect(); + + if options.is_empty() { + return Err("No tickets to choose from.".to_owned()); + } + + let chosen = compose(stdin, stdout, ComposeRequest { + id: None, + message: format!("{verb} which ticket?"), + mode: ComposeMode::Select { + options, + default: None, + }, + help: None, + })?; + + chosen + .parse() + .map_err(|_| format!("`{chosen}` is not a ticket id.")) +} + /// Take a value the composer fills in interactively, or explain its absence. /// /// Reached only when there was no terminal to ask on, since every one of these @@ -298,6 +649,62 @@ fn required(value: Option, what: &str) -> Result { value.ok_or_else(|| format!("No {what} given, and no terminal to ask for one.")) } +/// Ask the host for several values at once, and wait for them. +fn compose_many( + stdin: &mut impl BufRead, + stdout: &mut impl Write, + request: ComposeRequest, +) -> Result, String> { + send(stdout, &PluginToHost::Compose(request))?; + + match read_message(stdin)? { + HostToPlugin::Composed(response) if response.values.is_empty() => { + Err("Nothing selected.".to_owned()) + } + HostToPlugin::Composed(response) => Ok(response.values), + HostToPlugin::Error(error) => Err(error.message), + HostToPlugin::Shutdown => Err("Interrupted.".to_owned()), + other => Err(format!("expected a composed response, got: {other:?}")), + } +} + +/// Read a repository's open issues. +async fn fetch_open(owner: &str, repo: &str) -> Result, String> { + let mut builder = jp_github::Octocrab::builder(); + if let Some(token) = token() { + builder = builder.personal_token(token); + } + let client = builder + .build() + .map_err(|error| format!("failed to create the GitHub client: {error}"))?; + + client + .issues(owner, repo) + .list() + .per_page(PER_PAGE) + .send() + .await + .map_err(|error| format!("failed to list issues in {owner}/{repo}: {error}")) +} + +/// Ask the host to collect text, and wait for it. +fn compose( + stdin: &mut impl BufRead, + stdout: &mut impl Write, + request: ComposeRequest, +) -> Result { + send(stdout, &PluginToHost::Compose(request))?; + + match read_message(stdin)? { + HostToPlugin::Composed(response) => response + .text + .ok_or_else(|| "Nothing composed; run it again with the text as arguments.".to_owned()), + HostToPlugin::Error(error) => Err(error.message), + HostToPlugin::Shutdown => Err("Interrupted.".to_owned()), + other => Err(format!("expected a composed response, got: {other:?}")), + } +} + /// Read `jp ticket 42` as `jp ticket show 42`. /// /// A bare id is the most common thing to type, and no subcommand name parses as diff --git a/crates/plugins/command/ticket/src/main_tests.rs b/crates/plugins/command/ticket/src/main_tests.rs index 554767ce8..47bb65056 100644 --- a/crates/plugins/command/ticket/src/main_tests.rs +++ b/crates/plugins/command/ticket/src/main_tests.rs @@ -195,6 +195,37 @@ fn the_author_falls_back_through_jp_then_git_then_the_environment() { // those two values as arguments instead of reading them itself. } +#[test] +fn composed_text_splits_into_a_title_and_a_body() { + // A single line, however many blank lines follow it, is a title alone. + assert_eq!( + Composition::read("Tool call header misaligned"), + Composition::Title("Tool call header misaligned".to_owned()) + ); + assert_eq!( + Composition::read("Tool call header misaligned\n\n\n"), + Composition::Title("Tool call header misaligned".to_owned()) + ); + + // Subject, blank line, body. + assert_eq!( + Composition::read("Header misaligned\n\nIt wraps one column early.\n"), + Composition::TitleAndBody { + title: "Header misaligned".to_owned(), + body: "It wraps one column early.".to_owned(), + } + ); + + // Prose running straight on from the first line has no subject. + assert_eq!( + Composition::read("The header wraps one column\nearly, below 80 columns."), + Composition::Body("The header wraps one column\nearly, below 80 columns.".to_owned()) + ); + + assert_eq!(Composition::read(""), Composition::Empty); + assert_eq!(Composition::read(" \n\n"), Composition::Empty); +} + #[test] fn list_filters_are_optional() { let args = parse(&["list", "--status", "In Progress"]).unwrap(); @@ -344,7 +375,7 @@ fn a_run_reports_ready_then_output_then_exit() { match messages.as_slice() { [ - PluginToHost::Ready, + PluginToHost::Ready(_), PluginToHost::Print(print), PluginToHost::Exit(exit), ] => { @@ -362,13 +393,36 @@ fn a_run_reports_ready_then_output_then_exit() { ); } +/// A plugin newer than its host cannot ask for anything interactive, so it says +/// so and stops, without claiming ready, rather than hanging on a reply the +/// host cannot produce. +#[test] +fn an_older_host_is_refused_up_front() { + let dir = Utf8TempDir::new().unwrap(); + let messages = exchange(&init_at(REQUIRED_PROTOCOL - 1, dir.path(), &["list"])); + + match messages.as_slice() { + [PluginToHost::Exit(exit)] => { + assert_eq!(exit.code, 1); + assert!( + exit.reason + .as_ref() + .is_some_and(|reason| reason.contains("needs `jp` protocol 2")), + "{:?}", + exit.reason + ); + } + other => panic!("unexpected exchange: {other:?}"), + } +} + #[test] fn a_bad_argument_exits_non_zero_with_a_reason() { let dir = Utf8TempDir::new().unwrap(); let messages = exchange(&init(dir.path(), &["close", "not-an-id"])); match messages.as_slice() { - [PluginToHost::Ready, PluginToHost::Exit(exit)] => { + [PluginToHost::Ready(_), PluginToHost::Exit(exit)] => { assert_eq!(exit.code, 1); assert!( exit.reason