diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 485d17272..de4111c59 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -97,10 +97,11 @@ is derived from it. **Process** is how to build, check and submit work.
### Reporting bugs and asking for features
Open an [issue](https://github.com/Tryanks/tcode/issues). For a bug, include
-your OS, which agent (Claude Code / Codex / which ACP agent), what you did, what
-happened and what you expected. A screenshot beats a paragraph for anything
-visual. If you are unsure whether something is a bug or intended, open an issue
-anyway — the answer is worth writing down either way.
+your OS, which agent (Claude Code, Codex, pi, OpenCode, Cursor, Grok, or which
+ACP agent), what you did, what happened and what you expected. A screenshot
+beats a paragraph for anything visual. If you are unsure whether something is a
+bug or intended, open an issue anyway — the answer is worth writing down either
+way.
### Building
@@ -121,8 +122,8 @@ Platform prerequisites:
`libwayland-dev`, `libxcb*`, `libssl-dev`, `libasound2-dev`, `libfontconfig-dev`).
The embedded preview browser is compiled out on Linux.
-Provider CLIs are resolved from `PATH` (`claude`, `codex`, `pi`, `opencode`) and
-can be overridden in **Settings → Providers**.
+Provider CLIs are resolved from `PATH` (`claude`, `codex`, `pi`, `opencode`,
+`cursor-agent`, `grok`) and can be overridden in **Settings → Providers**.
### Before you open a pull request
@@ -275,7 +276,8 @@ crates/ios, crates/android, crates/web
`tcode_ui::run_shell`
crates/platform/* GPUI platform backends for iOS and Android
crates/agent provider clients (no GPUI) — claude.rs, codex.rs, pi.rs,
- opencode.rs, acp.rs
+ opencode.rs, cursor.rs, grok.rs, acp.rs; the ACP
+ session they share is acp_session.rs
crates/term terminal implementation (PTY, host-side emulator)
crates/mcp-host shared authenticated loopback host for in-process MCP servers
crates/preview-mcp MCP server exposing the preview browser to the agent
diff --git a/Cargo.lock b/Cargo.lock
index 7664f492b..f3e76efe3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -153,6 +153,7 @@ dependencies = [
"log",
"serde",
"serde_json",
+ "sha2 0.11.0",
"smol",
"toml 1.1.6+spec-1.1.0",
"ureq",
diff --git a/README.md b/README.md
index 685acfd17..84c71c142 100644
--- a/README.md
+++ b/README.md
@@ -6,7 +6,7 @@
**A native GUI for the terminal coding agents you already use.**
-Claude Code, Codex, pi, OpenCode, and any agent that speaks ACP — one window,
+Claude Code, Codex, pi, OpenCode, Cursor, Grok, and any agent that speaks ACP — one window,
on your desktop, your phone, or in a browser.
[Download](https://github.com/Tryanks/tcode/releases) ·
@@ -78,6 +78,8 @@ and a live plan/task panel.
| [Codex](https://developers.openai.com/codex/cli) | `codex` on your `PATH` |
| [pi](https://github.com/earendil-works/pi) | `pi` on your `PATH` |
| [OpenCode](https://opencode.ai) | `opencode` on your `PATH` |
+| [Cursor](https://cursor.com/cli) | `cursor-agent` on your `PATH` |
+| [Grok](https://x.ai) | `grok` on your `PATH` |
**Everything else, over [ACP](https://agentclientprotocol.com).** Tcode ships a
marketplace backed by the official ACP registry.
@@ -179,7 +181,7 @@ also include a `SHA256SUMS.txt` file.
| Browser | Embedded in the headless release; open the HTTP link it prints (HTTPS only through your own tunnel). No separate signed app package |
**2. Have an agent installed.** Tcode drives the CLIs, it doesn't bundle them.
-Make sure `claude`, `codex`, `pi` or `opencode` is on your `PATH` — or install an ACP agent from
+Make sure `claude`, `codex`, `pi`, `opencode`, `cursor-agent` or `grok` is on your `PATH` — or install an ACP agent from
the marketplace once Tcode is running.
**3. Add a project and start a thread.** Point Tcode at a directory, type, send.
diff --git a/assets/icons/cursor.svg b/assets/icons/cursor.svg
new file mode 100644
index 000000000..9f1876a6e
--- /dev/null
+++ b/assets/icons/cursor.svg
@@ -0,0 +1,4 @@
+
diff --git a/assets/icons/grok.svg b/assets/icons/grok.svg
new file mode 100644
index 000000000..9fb5e2ea6
--- /dev/null
+++ b/assets/icons/grok.svg
@@ -0,0 +1,4 @@
+
diff --git a/assets/icons/puzzle.svg b/assets/icons/puzzle.svg
new file mode 100644
index 000000000..bc7ea8e9e
--- /dev/null
+++ b/assets/icons/puzzle.svg
@@ -0,0 +1,13 @@
+
diff --git a/crates/agent/Cargo.toml b/crates/agent/Cargo.toml
index e149771a2..f14022cbe 100644
--- a/crates/agent/Cargo.toml
+++ b/crates/agent/Cargo.toml
@@ -8,6 +8,7 @@ default = ["process"]
process = [
"dep:agent-client-protocol",
"dep:base64",
+ "dep:sha2",
"dep:smol",
"dep:ureq",
"dep:uuid",
@@ -22,6 +23,7 @@ log = "0.4"
base64 = { version = "0.23", optional = true }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
+sha2 = { version = "0.11", optional = true }
smol = { version = "2", optional = true }
ureq = { version = "2", default-features = false, features = ["tls", "gzip"], optional = true }
uuid = { version = "1", features = ["v4"], optional = true }
diff --git a/crates/agent/examples/probe.rs b/crates/agent/examples/probe.rs
index 308ab284e..52dd11d7b 100644
--- a/crates/agent/examples/probe.rs
+++ b/crates/agent/examples/probe.rs
@@ -1,17 +1,30 @@
//! Headless end-to-end probe for provider clients.
//!
-//! Catalog mode: `probe --list-models `.
+//! Catalog mode: `probe --list-models [--binary ]`.
+//! Plugin mode: `probe plugins [--home ] [--project ] [ …]`
+//! lists the native plugin catalog after running the optional operation:
+//! `install|update [--accept ]`, `uninstall|enable|disable `,
+//! `add-marketplace ` or `remove-marketplace `. `--home` isolates
+//! the provider's native state (`CODEX_HOME` for Codex, `GROK_HOME` for Grok;
+//! for Claude Code `HOME` and `CLAUDE_CONFIG_DIR`).
//! Turn mode: `probe [cwd] [approval] [acp-command args…] [flags]`.
-//! Flags are `--mode plan`, `--effort `, `--interrupt-after `,
-//! `--steer `, and `--image `. Only one of the last three may be used.
+//! Flags are `--binary `, `--model `, `--mode plan`, `--effort `,
+//! `--resume `, `--fork`, `--leave-questions` (user-input requests
+//! stay unanswered), `--mcp ` (an HTTP MCP server registered
+//! as tcode registers its own), `--linger ` (stay open until no turn has
+//! run for that long, to see turns the agent starts itself), `--follow-up
+//! ` (send another turn that long after the first completes,
+//! running turn or not), `--interrupt-after `, `--steer `, and
+//! `--image `. Only one of the last three may be used.
use std::path::PathBuf;
-use std::time::Duration;
+use std::time::{Duration, Instant};
use agent::{
AcpAgent, AcpLaunch, AgentEvent, ApprovalDecision, ApprovalMode, Attachment, InteractionMode,
- ItemContent, OptionSelection, ProviderKind, SessionCommand, SessionOptions, TurnOptions,
- TurnStatus, list_models, start_session,
+ ItemContent, LaunchEnv, McpRegistration, OptionSelection, PluginContext, PluginOp, PluginScope,
+ ProviderKind, ResumeCursor, SessionCommand, SessionOptions, TurnOptions, TurnStatus,
+ list_models, list_plugins, run_plugin_op, start_session,
};
use base64::Engine as _;
@@ -29,19 +42,121 @@ enum ProbeMode {
fn usage() -> ! {
eprintln!(
- "usage: probe [cwd] \
- [supervised|auto_edits|full_access] [acp-command args…] [flags]"
+ "usage: probe [cwd] \
+ [supervised|read_only|auto_edits|full_access] [acp-command args…] [flags]"
);
- eprintln!(" probe --list-models ");
+ eprintln!(
+ " probe --list-models [--binary ]"
+ );
+ eprintln!(" probe plugins [--home ] [--project ] [ …]");
std::process::exit(2);
}
+fn parse_scope(arg: Option) -> PluginScope {
+ match arg.as_deref() {
+ Some("user") => PluginScope::User,
+ Some("project") => PluginScope::Project,
+ Some("local") => PluginScope::Local,
+ Some("managed") => PluginScope::Managed,
+ _ => usage(),
+ }
+}
+
+fn plugins(mut args: impl Iterator- ) -> i32 {
+ let provider = parse_provider(args.next().as_deref());
+ let mut home = None;
+ let mut project = None;
+ let mut accept = None;
+ let mut positional = Vec::new();
+ while let Some(arg) = args.next() {
+ match arg.as_str() {
+ "--home" => home = args.next().map(PathBuf::from),
+ "--project" => project = args.next().map(PathBuf::from),
+ "--accept" => accept = args.next(),
+ _ => positional.push(arg),
+ }
+ }
+ let mut env = Vec::new();
+ if let (ProviderKind::ClaudeCode, Some(home)) = (provider, &home) {
+ env.push((
+ "CLAUDE_CONFIG_DIR".to_string(),
+ home.join(".claude").display().to_string(),
+ ));
+ }
+ let context = PluginContext {
+ binary_path: None,
+ launch_env: LaunchEnv {
+ env,
+ home: home.clone(),
+ },
+ cwd: project
+ .clone()
+ .or(home)
+ .unwrap_or_else(|| std::env::current_dir().unwrap()),
+ project: project.is_some(),
+ };
+ let mut positional = positional.into_iter();
+ let op = positional.next().map(|verb| {
+ let mut id = || positional.next().unwrap_or_else(|| usage());
+ match verb.as_str() {
+ "install" => PluginOp::Install {
+ id: id(),
+ scope: parse_scope(positional.next()),
+ accept_command: accept.clone(),
+ },
+ "update" => PluginOp::Update {
+ id: id(),
+ scope: parse_scope(positional.next()),
+ accept_command: accept.clone(),
+ },
+ "uninstall" => PluginOp::Uninstall {
+ id: id(),
+ scope: parse_scope(positional.next()),
+ },
+ "enable" | "disable" => PluginOp::SetEnabled {
+ id: id(),
+ scope: parse_scope(positional.next()),
+ enabled: verb == "enable",
+ },
+ "add-marketplace" => PluginOp::AddMarketplace { source: id() },
+ "remove-marketplace" => PluginOp::RemoveMarketplace { name: id() },
+ _ => usage(),
+ }
+ });
+ smol::block_on(async move {
+ if let Some(op) = op {
+ eprintln!("probe: {op:?} in {}", context.cwd.display());
+ match run_plugin_op(provider, &context, &op).await {
+ Ok(outcome) => println!("outcome: {outcome:#?}"),
+ Err(error) => println!("outcome: error: {error}"),
+ }
+ }
+ match list_plugins(provider, &context).await {
+ Ok(listing) => {
+ let listing = serde_json::json!({
+ "marketplaces": listing.marketplaces,
+ "marketplace_actions": listing.marketplace_actions,
+ "entries": listing.entries,
+ });
+ println!("{}", serde_json::to_string_pretty(&listing).unwrap());
+ 0
+ }
+ Err(error) => {
+ eprintln!("list_plugins failed: {error}");
+ 1
+ }
+ }
+ })
+}
+
fn parse_provider(arg: Option<&str>) -> ProviderKind {
match arg {
Some("codex") => ProviderKind::Codex,
Some("claude") => ProviderKind::ClaudeCode,
Some("pi") => ProviderKind::Pi,
Some("opencode") => ProviderKind::OpenCode,
+ Some("cursor") => ProviderKind::Cursor,
+ Some("grok") => ProviderKind::Grok,
Some("acp") => ProviderKind::Acp,
_ => usage(),
}
@@ -89,10 +204,18 @@ fn image_attachment(path: PathBuf) -> Attachment {
fn main() {
env_logger::init();
let args: Vec = std::env::args().skip(1).collect();
+ if args.first().map(String::as_str) == Some("plugins") {
+ std::process::exit(plugins(args.into_iter().skip(1)));
+ }
if args.first().map(String::as_str) == Some("--list-models") {
let provider = parse_provider(args.get(1).map(String::as_str));
+ let binary = match args.get(2).map(String::as_str) {
+ Some("--binary") => Some(args.get(3).map(PathBuf::from).unwrap_or_else(|| usage())),
+ Some(_) => usage(),
+ None => None,
+ };
let exit_code = smol::block_on(async move {
- match list_models(provider, None, Default::default(), Default::default()).await {
+ match list_models(provider, binary, Default::default(), Default::default()).await {
Ok(models) => {
println!("{}", serde_json::to_string_pretty(&models).unwrap());
0
@@ -107,7 +230,14 @@ fn main() {
}
let mut interaction_mode = InteractionMode::Build;
+ let mut binary = None;
+ let mut model = None;
let mut effort = None;
+ let mut resume = None;
+ let mut fork = false;
+ let mut leave_questions = false;
+ let mut mcp_servers = Vec::new();
+ let mut continuation = Continuation::default();
let mut probe_mode = ProbeMode::Standard;
let mut positional = Vec::new();
let mut args = args.into_iter();
@@ -124,7 +254,38 @@ fn main() {
None => usage(),
};
}
+ "--binary" => binary = Some(args.next().map(PathBuf::from).unwrap_or_else(|| usage())),
+ "--model" => model = Some(args.next().unwrap_or_else(|| usage())),
"--effort" => effort = Some(args.next().unwrap_or_else(|| usage())),
+ "--resume" => {
+ let cursor = args.next().unwrap_or_else(|| usage());
+ resume = Some(ResumeCursor(serde_json::from_str(&cursor).unwrap_or_else(
+ |error| {
+ eprintln!("--resume takes the JSON of a resume cursor: {error}");
+ std::process::exit(2);
+ },
+ )));
+ }
+ "--fork" => fork = true,
+ "--leave-questions" => leave_questions = true,
+ "--linger" => {
+ let seconds = args.next().and_then(|value| value.parse().ok());
+ continuation.linger = Some(Duration::from_secs(seconds.unwrap_or_else(|| usage())));
+ }
+ "--follow-up" => {
+ let seconds = args.next().and_then(|value| value.parse().ok());
+ let seconds = seconds.unwrap_or_else(|| usage());
+ let prompt = args.next().unwrap_or_else(|| usage());
+ continuation.follow_up = Some((Duration::from_secs(seconds), prompt));
+ }
+ "--mcp" => {
+ let mut next = || args.next().unwrap_or_else(|| usage());
+ mcp_servers.push(McpRegistration {
+ name: next(),
+ url: next(),
+ bearer_token: next(),
+ });
+ }
"--interrupt-after" => {
let seconds = args
.next()
@@ -162,15 +323,20 @@ fn main() {
});
let mut remaining: Vec = positional.collect();
let approval_mode = match remaining.first().map(String::as_str) {
- Some("supervised" | "auto_edits" | "full_access") => match remaining.remove(0).as_str() {
- "supervised" => ApprovalMode::Supervised,
- "auto_edits" => ApprovalMode::AutoAcceptEdits,
- _ => ApprovalMode::FullAccess,
- },
+ Some("supervised" | "read_only" | "auto_edits" | "full_access") => {
+ match remaining.remove(0).as_str() {
+ "supervised" => ApprovalMode::Supervised,
+ "read_only" => ApprovalMode::ReadOnly,
+ "auto_edits" => ApprovalMode::AutoAcceptEdits,
+ _ => ApprovalMode::FullAccess,
+ }
+ }
None => ApprovalMode::Supervised,
Some(_) if provider == ProviderKind::Acp => ApprovalMode::Supervised,
Some(other) => {
- eprintln!("unknown approval mode {other:?}; use supervised|auto_edits|full_access");
+ eprintln!(
+ "unknown approval mode {other:?}; use supervised|read_only|auto_edits|full_access"
+ );
std::process::exit(2);
}
};
@@ -210,10 +376,31 @@ fn main() {
effort,
probe_mode,
acp,
+ mcp_servers,
+ continuation,
+ Resumption {
+ resume,
+ fork,
+ leave_questions,
+ },
+ binary,
+ model,
));
std::process::exit(exit_code);
}
+struct Resumption {
+ resume: Option,
+ fork: bool,
+ leave_questions: bool,
+}
+
+#[derive(Default)]
+struct Continuation {
+ linger: Option,
+ follow_up: Option<(Duration, String)>,
+}
+
#[allow(clippy::too_many_arguments)]
async fn run_probe(
provider: ProviderKind,
@@ -224,28 +411,38 @@ async fn run_probe(
effort: Option,
probe_mode: ProbeMode,
acp: Option,
+ mcp_servers: Vec,
+ continuation: Continuation,
+ resumption: Resumption,
+ binary_path: Option,
+ model: Option,
) -> i32 {
+ // Grok's effort is its wire config option, persisted under its option id.
+ let effort_id = match provider {
+ ProviderKind::Grok => "acp:cfg:reasoning_effort",
+ _ => "reasoningEffort",
+ };
let option_selections = effort
.iter()
.map(|value| OptionSelection {
- id: "reasoningEffort".into(),
+ id: effort_id.into(),
value: serde_json::Value::String(value.clone()),
})
.collect();
- let model = match (provider, effort.is_some()) {
+ let model = model.or_else(|| match (provider, effort.is_some()) {
(ProviderKind::ClaudeCode, true) => Some("claude-opus-4-8".to_string()),
_ => None,
- };
+ });
let opts = SessionOptions {
cwd,
model,
- resume: None,
- fork: false,
- binary_path: None,
+ resume: resumption.resume,
+ fork: resumption.fork,
+ binary_path,
approval_mode,
option_selections,
interaction_mode,
- mcp_servers: Vec::new(),
+ mcp_servers,
launch_env: Default::default(),
extra_args: Vec::new(),
acp,
@@ -318,18 +515,46 @@ async fn run_probe(
let mut turns_completed = 0;
let mut first_status = None;
let mut steer_accepted = false;
+ let lingering = continuation
+ .linger
+ .or(continuation.follow_up.as_ref().map(|_| PHANTOM_TURN_GRACE));
+ let mut follow_up = continuation.follow_up;
+ let mut open_until = None;
+ let mut running = false;
+ let mut closing = false;
loop {
- let event = if matches!(probe_mode, ProbeMode::Steer(_)) && turns_completed > 0 {
- smol::future::or(handle.events.recv(), async {
- smol::Timer::after(PHANTOM_TURN_GRACE).await;
- Err(smol::channel::RecvError)
- })
- .await
- .ok()
- } else {
- handle.events.recv().await.ok()
+ let quiet = match lingering {
+ Some(linger) if turns_completed > 0 && !running && !closing => Some(linger),
+ _ if matches!(probe_mode, ProbeMode::Steer(_)) && turns_completed > 0 => {
+ Some(PHANTOM_TURN_GRACE)
+ }
+ _ => None,
+ };
+ let event = match quiet {
+ Some(quiet) => {
+ let event = smol::future::or(handle.events.recv(), async {
+ smol::Timer::after(quiet).await;
+ Err(smol::channel::RecvError)
+ })
+ .await
+ .ok();
+ if event.is_none() {
+ if open_until.is_none_or(|until| Instant::now() >= until) {
+ closing = true;
+ handle.commands.send(SessionCommand::Shutdown).await.ok();
+ }
+ continue;
+ }
+ event
+ }
+ None => handle.events.recv().await.ok(),
};
let Some(event) = event else { break };
+ match &event {
+ AgentEvent::TurnStarted { .. } => running = true,
+ AgentEvent::TurnCompleted { .. } => running = false,
+ _ => {}
+ }
if provider == ProviderKind::Acp {
match &event {
@@ -360,6 +585,9 @@ async fn run_probe(
.await
.ok();
}
+ AgentEvent::UserInputRequested { questions, .. } if resumption.leave_questions => {
+ eprintln!("probe: leaving {} question(s) unanswered", questions.len());
+ }
AgentEvent::UserInputRequested {
request_id,
questions,
@@ -440,7 +668,25 @@ async fn run_probe(
"probe: TurnCompleted {turn_id} status={status:?} (#{turns_completed})"
);
}
- if !matches!(probe_mode, ProbeMode::Steer(_)) {
+ if let Some((delay, text)) = follow_up.take() {
+ open_until = Some(Instant::now() + delay + Duration::from_secs(1));
+ let commands = handle.commands.clone();
+ smol::spawn(async move {
+ smol::Timer::after(delay).await;
+ eprintln!("probe: FOLLOW-UP -> {text:?}");
+ commands
+ .send(SessionCommand::SendTurn {
+ delivery_id: 1,
+ text,
+ options: None,
+ attachments: Vec::new(),
+ })
+ .await
+ .ok();
+ })
+ .detach();
+ }
+ if !matches!(probe_mode, ProbeMode::Steer(_)) && lingering.is_none() {
handle.commands.send(SessionCommand::Shutdown).await.ok();
}
}
diff --git a/crates/agent/src/acp.rs b/crates/agent/src/acp.rs
index 8016cc189..e828b1a77 100644
--- a/crates/agent/src/acp.rs
+++ b/crates/agent/src/acp.rs
@@ -1,97 +1,32 @@
//! Agent Client Protocol provider: any agent from the ACP registry.
//!
//! Registry-managed agents share this adapter; native providers retain their
-//! richer protocol-specific clients.
-//!
-//! Shape: one child process per session, JSON-RPC over its stdio, driven by an
-//! `agent_client_protocol::Client` connection builder. The whole connection
-//! lives on a dedicated thread running a `LocalExecutor`; the canonical
-//! [`AgentEvent`] stream and the [`SessionCommand`] channel are the only things
-//! that cross the thread boundary — exactly like `claude.rs` / `codex.rs`.
+//! richer protocol-specific clients. The protocol machinery lives in
+//! [`crate::acp_session`]; this module is the registry's dialect: launching a
+//! registry recipe and the generic policies for agents tcode knows nothing
+//! more about.
-use std::collections::{HashMap, VecDeque};
-use std::fmt::Write as _;
-use std::io;
-use std::path::{Path, PathBuf};
-use std::pin::Pin;
-use std::rc::Rc;
-use std::sync::{
- Arc, Mutex, MutexGuard,
- atomic::{AtomicBool, Ordering},
-};
-use std::task::{Context, Poll};
+use std::path::PathBuf;
use std::time::Duration;
-use agent_client_protocol::{self as sdk, schema::ProtocolVersion, schema::v1 as acp};
+use agent_client_protocol::schema::v1 as acp;
use serde_json::{Value, json};
-use smol::channel::{Receiver, Sender};
use smol::future;
-use smol::io::{AsyncRead, AsyncReadExt as _, AsyncWrite};
-use smol::prelude::*;
+use crate::acp_session::{
+ self, Dialect, Established, Launch, Session, Setup, State, TurnEnd, describe, is_auth_required,
+ prompt_status,
+};
use crate::{
- AcpAgent, AcpLaunch, AgentError, AgentEvent, ApprovalDecision, ApprovalKind, ApprovalMode,
- ApprovalOption, ApprovalOptionKind, ApprovalRequest, Attachment, DeltaKind, FileChange,
- FileChangeKind, InteractionMode, ItemContent, ItemStatus, McpRegistration, OptionDescriptor,
- OptionSelection, PlanStep, PlanStepStatus, ProviderCommand, ProviderCommandKind, ProviderKind,
- ResumeCursor, SelectOption, SessionCommand, SessionHandle, SessionOptions, ThreadItem,
- TokenUsage, TurnStatus,
+ AcpAgent, AcpLaunch, AgentError, AgentEvent, ApprovalMode, Attachment, InteractionMode,
+ ProviderKind, ResumeCursor, SessionHandle, SessionOptions, TokenUsage,
};
-/// Option-descriptor ids. The composer renders an ACP agent's own
-/// modes/models/config options through the existing traits picker, so each one
-/// needs a stable id that routes back to the right ACP method.
-const MODE_OPTION_ID: &str = "acp:mode";
-const MODEL_OPTION_ID: &str = "acp:model";
-const CONFIG_OPTION_PREFIX: &str = "acp:cfg:";
-
-/// Cap on captured terminal output when the agent sets none (1 MiB).
-const DEFAULT_TERMINAL_OUTPUT_LIMIT: u64 = 1 << 20;
-
-/// How long we wait for an agent to answer `initialize` before declaring it
-/// broken. Generous: an `npm exec` recipe may have to fetch the package first.
-const INITIALIZE_TIMEOUT: Duration = Duration::from_secs(120);
-
/// How long we let an agent's `authenticate` run before giving up and telling
/// the user to sign in with the agent's own CLI (Gemini's is a browser OAuth
/// flow that otherwise blocks session startup for five minutes).
const AUTH_TIMEOUT: Duration = Duration::from_secs(20);
-type AcpConnection = sdk::ConnectionTo;
-
-trait MutexExt {
- fn lock_recover(&self) -> MutexGuard<'_, T>;
-}
-
-impl MutexExt for Mutex {
- fn lock_recover(&self) -> MutexGuard<'_, T> {
- self.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
- }
-}
-
-macro_rules! request_handler {
- ($client:expr, $request:ty, $method:ident) => {{
- let client = $client.clone();
- async move |args: $request, responder, connection| {
- let client = client.clone();
- connection
- .spawn(async move { responder.respond_with_result(client.$method(args).await) })?;
- Ok(())
- }
- }};
- ($client:expr, $request:ty, $method:ident, connection) => {{
- let client = $client.clone();
- async move |args: $request, responder, connection| {
- let client = client.clone();
- let task_connection = connection.clone();
- connection.spawn(async move {
- responder.respond_with_result(client.$method(args, &task_connection).await)
- })?;
- Ok(())
- }
- }};
-}
-
/// Start (or resume) a session with an ACP agent.
pub async fn start(opts: SessionOptions) -> Result {
if opts.fork {
@@ -99,723 +34,208 @@ pub async fn start(opts: SessionOptions) -> Result {
"session fork is not supported for this provider".into(),
));
}
- let (commands_tx, commands_rx) = smol::channel::unbounded();
- let (events_tx, events_rx) = smol::channel::unbounded();
- let (ready_tx, ready_rx) = smol::channel::bounded(1);
-
- // Keep each ACP connection and all of its callbacks on one dedicated
- // executor thread. The 1.2 SDK requires Send handlers internally, but the
- // adapter still exposes only channels across this boundary.
- std::thread::Builder::new()
- .name("acp-session".into())
- .spawn(move || {
- let executor = Rc::new(smol::LocalExecutor::new());
- let task = run_actor(executor.clone(), opts, commands_rx, events_tx, ready_tx);
- smol::block_on(executor.run(task));
- })
- .map_err(|err| {
- AgentError::Spawn(format!("could not start the ACP session thread: {err}"))
- })?;
-
- ready_rx.recv().await.map_err(|_| {
- AgentError::Protocol("ACP actor exited before reporting startup status".into())
- })??;
-
- Ok(SessionHandle {
- provider: ProviderKind::Acp,
- commands: commands_tx,
- events: events_rx,
- })
-}
-
-/// The resolved command line for a launch recipe.
-///
-/// `Npx` becomes `npm exec --yes -- ` (the registry's own
-/// contract, and what zed runs); `Binary` / `Custom` run as given.
-fn launch_command(launch: &AcpLaunch) -> Result<(PathBuf, Vec), AgentError> {
- match launch {
- AcpLaunch::Npx { package, args, .. } => {
- let npm = crate::resolve_binary(None, "npm")?;
- let mut argv = vec![
- "exec".to_string(),
- "--yes".to_string(),
- "--".to_string(),
- package.clone(),
- ];
- argv.extend(args.iter().cloned());
- Ok((npm, argv))
- }
- AcpLaunch::Binary { command, args, .. } => Ok((command.clone(), args.clone())),
- AcpLaunch::Custom { command, args, .. } => {
- let binary = crate::resolve_binary(None, command)?;
- Ok((binary, args.clone()))
- }
- }
-}
-
-fn spawn_agent(
- agent: &AcpAgent,
- opts: &SessionOptions,
-) -> Result {
- let (program, mut args) = launch_command(&agent.launch)?;
- args.extend(opts.extra_args.iter().cloned());
-
- let mut cmd = crate::process::async_command(&program);
- cmd.args(&args)
- .current_dir(&opts.cwd)
- .stdin(std::process::Stdio::piped())
- .stdout(std::process::Stdio::piped())
- .stderr(std::process::Stdio::piped());
- // Recipe env first, the user's configured env last (the user always wins).
- let recipe_env = match &agent.launch {
- AcpLaunch::Npx { env, .. }
- | AcpLaunch::Binary { env, .. }
- | AcpLaunch::Custom { env, .. } => env,
+ let Some(agent) = opts.acp.clone() else {
+ return Err(AgentError::Protocol(
+ "no ACP agent selected for this session".into(),
+ ));
};
- for (key, value) in recipe_env {
- cmd.env(key, value);
- }
- for (key, value) in opts.launch_env.pairs(ProviderKind::Acp) {
- cmd.env(key, value);
- }
- log::debug!(
- "spawning ACP agent {}: {} {:?}",
- agent.id,
- program.display(),
- args
- );
- cmd.spawn().map_err(|err| {
- AgentError::Spawn(format!(
- "could not launch ACP agent `{}` ({}): {err}",
- agent.name,
- program.display()
- ))
- })
-}
-
-/// Reports an agent closing (or breaking) stdout while still handing the bytes
-/// to the SDK transport. The 1.2 `ByteStreams` component treats a clean EOF as
-/// a completed input stream, so observing it here preserves tcode's immediate
-/// `SessionClosed` behavior instead of leaving the command loop waiting.
-struct ObservedReader {
- inner: smol::process::ChildStdout,
- done: Sender,
-}
-
-impl ObservedReader {
- fn new(inner: smol::process::ChildStdout, done: Sender) -> Self {
- Self { inner, done }
- }
-}
-
-impl AsyncRead for ObservedReader {
- fn poll_read(
- mut self: Pin<&mut Self>,
- cx: &mut Context<'_>,
- buf: &mut [u8],
- ) -> Poll> {
- match Pin::new(&mut self.inner).poll_read(cx, buf) {
- Poll::Ready(Ok(0)) => {
- let _ = self
- .done
- .try_send("the ACP agent closed its stdio".to_string());
- Poll::Ready(Ok(0))
- }
- Poll::Ready(Err(err)) => {
- let _ = self.done.try_send(format!("ACP transport error: {err}"));
- Poll::Ready(Err(err))
- }
- other => other,
- }
- }
+ acp_session::start(ProviderKind::Acp, Registry { agent }, opts).await
}
-/// Reports a turn only after the SDK has written its complete JSON-RPC prompt
-/// line to the child. `ConnectionTo::send_request` merely queues internally and
-/// does not expose a synchronous enqueue failure, so it is not a delivery
-/// boundary on its own.
-struct ObservedWriter {
- inner: W,
- line: Vec,
- pending_deliveries: Arc>>,
- events: Sender,
+/// An agent from the registry (or a custom command), driven by the generic
+/// ACP policies.
+struct Registry {
+ agent: AcpAgent,
}
-impl ObservedWriter {
- fn new(
- inner: W,
- pending_deliveries: Arc>>,
- events: Sender,
- ) -> Self {
- Self {
- inner,
- line: Vec::new(),
- pending_deliveries,
- events,
- }
- }
-
- fn finish_line(&mut self) {
- let is_prompt = serde_json::from_slice::(&self.line)
- .ok()
- .is_some_and(|message| {
- message.get("method").and_then(Value::as_str) == Some("session/prompt")
- });
- self.line.clear();
- if !is_prompt {
- return;
- }
- if let Some(delivery_id) = self.pending_deliveries.lock_recover().pop_front() {
- let _ = self
- .events
- .try_send(AgentEvent::TurnAccepted { delivery_id });
- }
+impl Dialect for Registry {
+ fn name(&self) -> &str {
+ &self.agent.name
}
-}
-impl AsyncWrite for ObservedWriter {
- fn poll_write(
- mut self: Pin<&mut Self>,
- cx: &mut Context<'_>,
- buf: &[u8],
- ) -> Poll> {
- let written = match Pin::new(&mut self.inner).poll_write(cx, buf) {
- Poll::Ready(Ok(written)) => written,
- other => return other,
+ fn launch(&self, opts: &SessionOptions) -> Result {
+ let (program, mut args) = launch_command(&self.agent.launch)?;
+ args.extend(opts.extra_args.iter().cloned());
+ let env = match &self.agent.launch {
+ AcpLaunch::Npx { env, .. }
+ | AcpLaunch::Binary { env, .. }
+ | AcpLaunch::Custom { env, .. } => env.clone(),
};
- for byte in &buf[..written] {
- if *byte == b'\n' {
- self.finish_line();
- } else {
- self.line.push(*byte);
- }
- }
- Poll::Ready(Ok(written))
- }
-
- fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> {
- Pin::new(&mut self.inner).poll_flush(cx)
- }
-
- fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> {
- Pin::new(&mut self.inner).poll_close(cx)
+ Ok(Launch {
+ program,
+ args,
+ env,
+ client_meta: None,
+ })
}
-}
-
-async fn run_actor(
- executor: Rc>,
- opts: SessionOptions,
- commands: Receiver,
- events: Sender,
- ready: Sender>,
-) {
- let Some(agent) = opts.acp.clone() else {
- let _ = ready
- .send(Err(AgentError::Protocol(
- "no ACP agent selected for this session".into(),
- )))
- .await;
- return;
- };
-
- let mut child = match spawn_agent(&agent, &opts) {
- Ok(child) => child,
- Err(err) => {
- let _ = ready.send(Err(err)).await;
- return;
- }
- };
- let Some(stdin) = child.stdin.take() else {
- let _ = ready
- .send(Err(AgentError::Spawn(format!(
- "ACP agent `{}` started without piped stdin",
- agent.name
- ))))
- .await;
- return;
- };
- let Some(stdout) = child.stdout.take() else {
- let _ = ready
- .send(Err(AgentError::Spawn(format!(
- "ACP agent `{}` started without piped stdout",
- agent.name
- ))))
- .await;
- return;
- };
- let Some(stderr) = child.stderr.take() else {
- let _ = ready
- .send(Err(AgentError::Spawn(format!(
- "ACP agent `{}` started without piped stderr",
- agent.name
- ))))
- .await;
- return;
- };
- // The agent's stderr is its log channel: keep the tail so a startup failure
- // can be reported in the agent's own words.
- let stderr_tail = crate::process::StderrTail::default();
- executor
- .spawn({
- let tail = stderr_tail.clone();
- let id = agent.id.clone();
- async move {
- let mut lines = smol::io::BufReader::new(stderr).lines();
- while let Some(Ok(line)) = lines.next().await {
- log::debug!("acp[{id}] stderr: {line}");
- tail.push(line);
+ async fn establish(&self, setup: &Setup<'_>) -> Result {
+ let agent = &self.agent;
+ let opts = setup.opts;
+ let mcp_servers = setup.mcp_servers().await;
+ let resumed = opts
+ .resume
+ .as_ref()
+ .and_then(|cursor| cursor.0.get("acp_session_id"))
+ .and_then(Value::as_str)
+ .map(str::to_string)
+ .filter(|_| setup.init.agent_capabilities.load_session);
+
+ let mut loaded_session = None;
+ if let Some(session_id) = resumed {
+ // `session/load` replays the whole conversation as `session/update`
+ // notifications. Our JSONL log is the authoritative history and the
+ // UI has already folded it, so the replay is swallowed (see
+ // `session_update`); we only want the session live again.
+ let session_id = acp::SessionId::new(session_id);
+ let loaded = setup
+ .load(
+ acp::LoadSessionRequest::new(session_id.clone(), opts.cwd.clone())
+ .mcp_servers(mcp_servers.clone()),
+ )
+ .await;
+ match loaded {
+ Ok(loaded) => {
+ loaded_session = Some((session_id, loaded.modes, loaded.config_options));
}
- }
- })
- .detach();
-
- let state = Arc::new(Mutex::new(State::new(opts.cwd.clone())));
- let client = AcpClient {
- events: events.clone(),
- state: state.clone(),
- };
- let (io_done_tx, io_done) = smol::channel::bounded::(1);
- let pending_deliveries = Arc::new(Mutex::new(VecDeque::new()));
- let transport = sdk::ByteStreams::new(
- ObservedWriter::new(stdin, pending_deliveries.clone(), events.clone()),
- ObservedReader::new(stdout, io_done_tx),
- );
- let session_started = Arc::new(AtomicBool::new(false));
- let connection_result = sdk::Client
- .builder()
- .name(format!("tcode-acp-{}", agent.id))
- .on_receive_notification(
- {
- let client = client.clone();
- async move |args: acp::SessionNotification, _connection| {
- client.session_notification(args).await
+ Err(err) => {
+ log::warn!(
+ "acp[{}]: session/load failed ({}); starting a fresh session",
+ agent.id,
+ describe(&err)
+ );
+ setup
+ .warn(format!(
+ "{} could not resume the previous conversation; starting a new one",
+ agent.name
+ ))
+ .await;
}
- },
- sdk::on_receive_notification!(),
- )
- .on_receive_request(
- request_handler!(client, acp::RequestPermissionRequest, request_permission),
- sdk::on_receive_request!(),
- )
- .on_receive_request(
- request_handler!(client, acp::ReadTextFileRequest, read_text_file),
- sdk::on_receive_request!(),
- )
- .on_receive_request(
- request_handler!(client, acp::WriteTextFileRequest, write_text_file),
- sdk::on_receive_request!(),
- )
- .on_receive_request(
- request_handler!(
- client,
- acp::CreateTerminalRequest,
- create_terminal,
- connection
- ),
- sdk::on_receive_request!(),
- )
- .on_receive_request(
- request_handler!(client, acp::TerminalOutputRequest, terminal_output),
- sdk::on_receive_request!(),
- )
- .on_receive_request(
- request_handler!(
- client,
- acp::WaitForTerminalExitRequest,
- wait_for_terminal_exit
- ),
- sdk::on_receive_request!(),
- )
- .on_receive_request(
- request_handler!(client, acp::KillTerminalRequest, kill_terminal),
- sdk::on_receive_request!(),
- )
- .on_receive_request(
- request_handler!(client, acp::ReleaseTerminalRequest, release_terminal),
- sdk::on_receive_request!(),
- )
- .connect_with(transport, {
- let session_started = session_started.clone();
- let stderr_tail = stderr_tail.clone();
- let actor_events = events.clone();
- let actor_ready = ready.clone();
- async move |connection| {
- connected_actor(
- &executor,
- connection,
- &agent,
- &opts,
- &commands,
- &actor_events,
- &actor_ready,
- &state,
- &stderr_tail,
- &io_done,
- &session_started,
- &pending_deliveries,
- )
- .await
}
- })
- .await;
-
- // The connection closure owns the protocol shutdown; this is the hard
- // process boundary for both graceful shutdowns and broken transports.
- let _ = child.kill();
-
- if !session_started.load(Ordering::Acquire) {
- if let Err(err) = connection_result {
- let message = format!("ACP transport error: {}", describe(&err));
- let message = stderr_tail.append_to(message, "\n");
- let _ = ready.send(Err(AgentError::Protocol(message))).await;
- }
- return;
- }
-
- let close_reason = match connection_result {
- Ok(reason) => reason,
- Err(err) => Some(format!("ACP transport error: {}", describe(&err))),
- };
- let close_reason = close_reason.map(|reason| stderr_tail.append_to(reason, "\nstderr:\n"));
- let _ = events
- .send(AgentEvent::SessionClosed {
- reason: close_reason,
- })
- .await;
-}
-
-#[allow(clippy::too_many_arguments)]
-async fn connected_actor(
- executor: &Rc>,
- connection: AcpConnection,
- agent: &AcpAgent,
- opts: &SessionOptions,
- commands: &Receiver,
- events: &Sender,
- ready: &Sender>,
- state: &Arc>,
- stderr_tail: &crate::process::StderrTail,
- io_done: &Receiver,
- session_started: &AtomicBool,
- pending_deliveries: &Arc>>,
-) -> Result