From 02dd14394b13679a33ec6a43945663e9e5ceb173 Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Mon, 27 Jul 2026 17:48:31 -0600 Subject: [PATCH] feat: implement ACP session/prompt --- src/acp/server.rs | 186 +++++++++++++++++++++++++++++++++++++++++-- src/client/common.rs | 4 +- src/main.rs | 10 +-- 3 files changed, 187 insertions(+), 13 deletions(-) diff --git a/src/acp/server.rs b/src/acp/server.rs index c5ffab5..237bd6f 100644 --- a/src/acp/server.rs +++ b/src/acp/server.rs @@ -1,13 +1,47 @@ use super::types::{METHOD_NOT_FOUND, PARSE_ERROR, Request, Response}; +use crate::client::call_chat_completions_streaming; +use crate::config::{Input, RenderMode, RequestContext}; +use crate::utils::AbortSignal; use anyhow::Result; -use serde_json::json; +use serde_json::{Value, json}; +use std::sync::Arc; use tokio::io::{AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader}; -pub async fn run_acp_server() -> Result<()> { - run_acp_server_on(tokio::io::stdin(), tokio::io::stdout()).await +pub(crate) struct AcpServerState { + ctx: Option, + abort: AbortSignal, + session_active: bool, } -pub(crate) async fn run_acp_server_on(reader: R, mut writer: W) -> Result<()> +pub async fn run_acp_server(ctx: RequestContext, abort: AbortSignal) -> Result<()> { + let state = AcpServerState { + ctx: Some(ctx), + abort, + session_active: false, + }; + run_acp_server_with_state(tokio::io::stdin(), tokio::io::stdout(), state).await +} + +#[cfg(test)] +pub(crate) async fn run_acp_server_on(reader: R, writer: W) -> Result<()> +where + R: tokio::io::AsyncRead + Unpin, + W: AsyncWrite + Unpin, +{ + use crate::utils::create_abort_signal; + let state = AcpServerState { + ctx: None, + abort: create_abort_signal(), + session_active: false, + }; + run_acp_server_with_state(reader, writer, state).await +} + +async fn run_acp_server_with_state( + reader: R, + mut writer: W, + mut state: AcpServerState, +) -> Result<()> where R: tokio::io::AsyncRead + Unpin, W: AsyncWrite + Unpin, @@ -20,7 +54,7 @@ where if line.is_empty() { continue; } - if let Some(response) = dispatch(&line) { + if let Some(response) = dispatch(&line, &mut state).await { emit(&mut writer, &response).await?; } } @@ -28,7 +62,7 @@ where Ok(()) } -fn dispatch(raw: &str) -> Option { +async fn dispatch(raw: &str, state: &mut AcpServerState) -> Option { let req: Request = match serde_json::from_str(raw) { Ok(r) => r, Err(_) => return Some(Response::err(None, PARSE_ERROR, "Parse error")), @@ -38,6 +72,8 @@ fn dispatch(raw: &str) -> Option { Some(match req.method.as_str() { "initialize" => handle_initialize(req), + "session/new" => handle_session_new(req, state), + "session/prompt" => handle_session_prompt(req, state).await, _ => Response::err( req.id, METHOD_NOT_FOUND, @@ -57,6 +93,64 @@ fn handle_initialize(req: Request) -> Response { ) } +fn handle_session_new(req: Request, state: &mut AcpServerState) -> Response { + if state.session_active { + return Response::err( + req.id, + -32000, + "Session already active; this server supports one session per process", + ); + } + state.session_active = true; + Response::ok(req.id, json!({ "sessionId": "default" })) +} + +async fn handle_session_prompt(req: Request, state: &mut AcpServerState) -> Response { + if !state.session_active { + return Response::err(req.id, -32000, "No active session; call session/new first"); + } + + let text = match req + .params + .as_ref() + .and_then(|p| p.get("text")) + .and_then(Value::as_str) + { + Some(t) => t.to_string(), + None => return Response::err(req.id, -32602, "Missing params.text"), + }; + + let ctx = match state.ctx.as_mut() { + Some(c) => c, + None => return Response::err(req.id, -32000, "Server not configured with a context"), + }; + + let abort = state.abort.clone(); + match run_prompt_turn(ctx, &text, abort).await { + Ok(output) => Response::ok( + req.id, + json!({ "output": output, "stopReason": "end_turn" }), + ), + Err(e) => Response::err(req.id, -32000, format!("Prompt failed: {e}")), + } +} + +async fn run_prompt_turn( + ctx: &mut RequestContext, + text: &str, + abort: AbortSignal, +) -> Result { + ctx.render_mode = RenderMode::Silent; + let input = Input::from_str(ctx, text, None)?; + ctx.before_chat_completion(&input)?; + let client = input.create_client()?; + let (output, tool_results) = + call_chat_completions_streaming(&input, client.as_ref(), ctx, abort).await?; + let app = Arc::clone(&ctx.app.config); + ctx.after_chat_completion(app.as_ref(), &input, &output, &tool_results)?; + Ok(output) +} + async fn emit(writer: &mut W, response: &Response) -> Result<()> { let mut line = serde_json::to_string(response)?; line.push('\n'); @@ -147,4 +241,84 @@ mod tests { assert_eq!(v["result"]["name"], "coyote"); assert!(v["result"]["version"].is_string()); } + + #[tokio::test] + async fn session_new_returns_session_id() { + let input = concat!( + r#"{"jsonrpc":"2.0","id":10,"method":"session/new","params":{}}"#, + "\n", + ); + let mut output = Vec::new(); + run_acp_server_on(input.as_bytes(), &mut output) + .await + .unwrap(); + + let s = String::from_utf8(output).unwrap(); + let v: serde_json::Value = serde_json::from_str(s.trim()).unwrap(); + assert_eq!(v["id"], 10); + assert_eq!(v["result"]["sessionId"], "default"); + } + + #[tokio::test] + async fn session_new_twice_errors() { + let input = concat!( + r#"{"jsonrpc":"2.0","id":1,"method":"session/new","params":{}}"#, + "\n", + r#"{"jsonrpc":"2.0","id":2,"method":"session/new","params":{}}"#, + "\n", + ); + let mut output = Vec::new(); + run_acp_server_on(input.as_bytes(), &mut output) + .await + .unwrap(); + + let s = String::from_utf8(output).unwrap(); + let mut lines = s.lines().filter(|l| !l.is_empty()); + let first: serde_json::Value = serde_json::from_str(lines.next().unwrap()).unwrap(); + let second: serde_json::Value = serde_json::from_str(lines.next().unwrap()).unwrap(); + assert!(first["result"]["sessionId"].is_string()); + assert_eq!(second["error"]["code"], -32000); + } + + #[tokio::test] + async fn session_prompt_without_session_errors() { + let input = concat!( + r#"{"jsonrpc":"2.0","id":3,"method":"session/prompt","params":{"text":"hello"}}"#, + "\n", + ); + let mut output = Vec::new(); + run_acp_server_on(input.as_bytes(), &mut output) + .await + .unwrap(); + + let s = String::from_utf8(output).unwrap(); + let v: serde_json::Value = serde_json::from_str(s.trim()).unwrap(); + assert_eq!(v["id"], 3); + assert_eq!(v["error"]["code"], -32000); + } + + #[tokio::test] + async fn session_prompt_with_no_context_returns_error() { + let input = concat!( + r#"{"jsonrpc":"2.0","id":1,"method":"session/new","params":{}}"#, + "\n", + r#"{"jsonrpc":"2.0","id":2,"method":"session/prompt","params":{"text":"hello"}}"#, + "\n", + ); + let mut output = Vec::new(); + run_acp_server_on(input.as_bytes(), &mut output) + .await + .unwrap(); + + let s = String::from_utf8(output).unwrap(); + let mut lines = s.lines().filter(|l| !l.is_empty()); + let first: serde_json::Value = serde_json::from_str(lines.next().unwrap()).unwrap(); + let second: serde_json::Value = serde_json::from_str(lines.next().unwrap()).unwrap(); + assert_eq!(first["result"]["sessionId"], "default"); + assert_eq!(second["id"], 2); + assert!( + second["error"].is_object(), + "expected error response when no ctx" + ); + } } diff --git a/src/client/common.rs b/src/client/common.rs index cd68220..c74e3ce 100644 --- a/src/client/common.rs +++ b/src/client/common.rs @@ -502,7 +502,7 @@ pub async fn call_chat_completions_streaming( let (text, tool_calls, thinking) = handler.take(); match send_ret { Ok(_) => { - if !text.is_empty() && !text.ends_with('\n') { + if !silent && !text.is_empty() && !text.ends_with('\n') { println!(); } let mut tool_results = eval_tool_calls(ctx, tool_calls).await?; @@ -515,7 +515,7 @@ pub async fn call_chat_completions_streaming( Ok((text, tool_results)) } Err(err) => { - if !text.is_empty() { + if !silent && !text.is_empty() { println!(); } Err(err) diff --git a/src/main.rs b/src/main.rs index 23a97fc..995c7ba 100644 --- a/src/main.rs +++ b/src/main.rs @@ -77,7 +77,7 @@ async fn main() -> Result<()> { } let text = cli.text()?; - let working_mode = if text.is_none() && cli.file.is_empty() { + let working_mode = if !cli.acp_server && text.is_none() && cli.file.is_empty() { WorkingMode::Repl } else { WorkingMode::Cmd @@ -110,10 +110,6 @@ async fn main() -> Result<()> { let log_path = setup_logger(cli.acp_server)?; - if cli.acp_server { - return acp::run_acp_server().await; - } - if let Some(version) = &cli.update { let version = version.clone(); let force = cli.force; @@ -238,6 +234,10 @@ async fn main() -> Result<()> { ctx.render_mode = RenderMode::Silent; } + if cli.acp_server { + return acp::run_acp_server(ctx, abort_signal).await; + } + if let Err(err) = run(ctx, cli, text, abort_signal).await { render_error(err); process::exit(1);