diff --git a/.gitignore b/.gitignore index 96466e4..143d31e 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,7 @@ coverage/ playwright-report/ test-results/ target/ +packages/zed/target/ .multi-agent/ *.env *.env.local diff --git a/CHANGELOG.md b/CHANGELOG.md index f564e01..2af4b09 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,8 +4,9 @@ - Session access control: host approval, required display names, repo/email/domain gates (`session.policy` / `auth.pending` / `host.approve|deny`) - Configurable git remote prefixes and host rewrites; layered `chorus.json` org settings -- Hybrid OpenCode security e2e, VS Code adapter e2e, and Cloud Agent install script +- Hybrid OpenCode security e2e, VS Code/Zed adapter e2e, and Cloud Agent install script - VS Code extension (`packages/vscode`) — share/join/chat; Session sidebar; approve/deny +- Zed joiner extension (`packages/zed`) + `chorus-zed-helper` (MCP/CLI) - Extract `@chorus/client` (`JoinClient` + `RelayServer`) for reuse by OpenCode, VS Code, and future adapters - Real-time shared transcript for all agents: fan out collaborator prompts to every joiner, abort joiner local LLM, mirror `[Host]`/`[name]`/`[AI]` (prefer web UI for live view) - Joiner mirrors host prompts and AI replies into its OpenCode session transcript (`[Host]:` / `[AI]:`), including history replay on join (replaces toast-only viewing) diff --git a/Cargo.lock b/Cargo.lock index e14f344..abf8ceb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -175,6 +175,22 @@ dependencies = [ "url", ] +[[package]] +name = "chorus-zed-helper" +version = "0.1.0" +dependencies = [ + "chorus-relay", + "clap", + "futures-util", + "serde", + "serde_json", + "tokio", + "tokio-tungstenite 0.26.2", + "tracing", + "tracing-subscriber", + "url", +] + [[package]] name = "clap" version = "4.6.5" diff --git a/Cargo.toml b/Cargo.toml index 265d60d..a9cad15 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,8 @@ [workspace] resolver = "2" -members = ["crates/chorus-relay"] +members = ["crates/chorus-relay", "crates/chorus-zed-helper"] +exclude = ["packages/zed"] +# packages/zed is a standalone Zed extension (wasm32-wasip2) built by Zed / cargo --target. [workspace.package] edition = "2021" diff --git a/README.md b/README.md index 1567ca3..e3a780f 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@ Collaborative OpenCode session sharing. Pair-program a live AI session from anot ## How it works -Chorus is an [OpenCode](https://github.com/sst/opencode) plugin (plus a VS Code adapter) and a **Rust WebSocket relay** (`chorus-relay`). +Chorus is an [OpenCode](https://github.com/sst/opencode) plugin (plus VS Code / other adapters) and a **Rust WebSocket relay** (`chorus-relay`). ``` Host runs opencode → plugin loads @@ -52,22 +52,27 @@ One monorepo, two ecosystems, one wire contract: | `packages/plugin` | npm `@chorus/plugin` | OpenCode plugin — tools, hooks, spawns/manages relay | | `packages/client` | npm `@chorus/client` | Shared `JoinClient` + `RelayServer` for host adapters | | `packages/vscode` | VS Code extension `chorus` | Share/join Chorus sessions from VS Code | +| `packages/zed` | Zed extension `chorus` | Joiner adapter (WASM + MCP → `chorus-zed-helper`) | | `packages/shared` | npm `@chorus/shared` | TypeScript types + codecs for joiner and host-control protocols | | `crates/chorus-relay` | `chorus-relay` binary | Rust WebSocket relay (`/ws` joiners, `/host` control plane) | +| `crates/chorus-zed-helper` | `chorus-zed-helper` binary | Native Chorus join client (CLI + MCP) for Zed | | `protocol/` | fixtures (not published) | Canonical JSON examples both TS and Rust must deserialize | -Root `package.json` scripts are the only task entry (`build`, `test`, `typecheck`). Bun workspaces own `packages/*`; Cargo owns `crates/*`. +Root `package.json` scripts are the only task entry (`build`, `test`, `typecheck`). Bun workspaces own `packages/*`; Cargo owns `crates/*` (Zed’s `packages/zed` is built separately for `wasm32-wasip2`). ## Development ```sh bun install -bun run build # release relay + TS packages -bun run test # relay tests + TS/Bun tests (includes protocol fixtures) +bun run build # release relay + helper + TS packages +bun run test # relay/helper tests + TS/Bun tests (includes protocol fixtures) bun run typecheck cargo test -p chorus-relay +cargo test -p chorus-zed-helper ``` +Zed joiner (optional): see [packages/zed/README.md](packages/zed/README.md) for Install Dev Extension + `chorus-zed-helper` setup. + ### Multi-agent local testing Run several isolated OpenCode servers on one machine (each with Chorus loaded) and drive share/join over the HTTP API: diff --git a/crates/chorus-zed-helper/Cargo.toml b/crates/chorus-zed-helper/Cargo.toml new file mode 100644 index 0000000..ce6f57f --- /dev/null +++ b/crates/chorus-zed-helper/Cargo.toml @@ -0,0 +1,29 @@ +[package] +name = "chorus-zed-helper" +version = "0.1.0" +edition.workspace = true +license.workspace = true +authors.workspace = true +description = "Native Chorus join/host helper for the Zed extension (CLI + MCP)" +publish = false + +[[bin]] +name = "chorus-zed-helper" +path = "src/main.rs" + +[dependencies] +chorus-relay = { path = "../chorus-relay" } +clap = { version = "4", features = ["derive", "env"] } +futures-util = "0.3" +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tokio = { version = "1", features = ["full"] } +tokio-tungstenite = "0.26" +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } +url = "2" + +[dev-dependencies] +chorus-relay = { path = "../chorus-relay" } +tokio-tungstenite = "0.26" +futures-util = "0.3" diff --git a/crates/chorus-zed-helper/src/client.rs b/crates/chorus-zed-helper/src/client.rs new file mode 100644 index 0000000..5b6ab9f --- /dev/null +++ b/crates/chorus-zed-helper/src/client.rs @@ -0,0 +1,390 @@ +use std::sync::Arc; +use std::time::Duration; + +use chorus_relay::protocol::{ + ChatMessage, ClientMessage, ConnectedUser, ServerMessage, SessionEvent, +}; +use futures_util::{SinkExt, StreamExt}; +use serde::{Deserialize, Serialize}; +use tokio::sync::{mpsc, RwLock}; +use tokio::task::JoinHandle; +use tokio_tungstenite::{connect_async, tungstenite::Message}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum JoinStatus { + Disconnected, + Connecting, + Pending, + Connected, + Error, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SessionSnapshot { + pub status: JoinStatus, + pub host: String, + pub display_name: String, + pub session_id: Option, + pub users: Vec, + pub recent_events: Vec, + pub recent_chat: Vec, + pub last_error: Option, +} + +impl SessionSnapshot { + fn new(host: &str, display_name: &str) -> Self { + Self { + status: JoinStatus::Disconnected, + host: host.to_string(), + display_name: display_name.to_string(), + session_id: None, + users: Vec::new(), + recent_events: Vec::new(), + recent_chat: Vec::new(), + last_error: None, + } + } +} + +/// Live joiner WebSocket client for the Chorus `/ws` surface. +pub struct JoinClient { + inner: Arc>, + outbound: mpsc::UnboundedSender, + reader: Option>, +} + +struct Inner { + snapshot: SessionSnapshot, +} + +impl JoinClient { + /// Connect to `ws://host/ws` (or `wss://`), authenticate, and wait for + /// `session.history` (active) or `auth.pending` (awaiting host approval). + pub async fn connect( + host: &str, + token: &str, + display_name: &str, + repo_remote: Option<&str>, + email: Option<&str>, + ) -> Result { + let name = display_name.trim(); + if name.is_empty() { + return Err("display name is required".into()); + } + let ws_url = normalize_ws_url(host)?; + let (ws, _) = connect_async(&ws_url) + .await + .map_err(|e| format!("connect failed: {e}"))?; + + let (mut write, mut read) = ws.split(); + let auth = ClientMessage::Auth { + token: token.to_string(), + display_name: name.to_string(), + repo_remote: repo_remote + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string), + email: email + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_ascii_lowercase), + }; + write + .send(Message::Text( + serde_json::to_string(&auth) + .map_err(|e| e.to_string())? + .into(), + )) + .await + .map_err(|e| format!("auth send failed: {e}"))?; + + let mut snapshot = SessionSnapshot::new(host, name); + snapshot.status = JoinStatus::Connecting; + + // Wait until session.history (connected), auth.pending, deny/error, or timeout. + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + if tokio::time::Instant::now() > deadline { + return Err("timed out waiting for session.history or auth.pending".into()); + } + let next = tokio::time::timeout(Duration::from_secs(5), read.next()) + .await + .map_err(|_| "timed out waiting for relay message".to_string())?; + let msg = match next { + Some(Ok(Message::Text(t))) => t, + Some(Ok(Message::Close(frame))) => { + return Err(format!("connection closed before auth: {frame:?}")); + } + Some(Ok(_)) => continue, + Some(Err(e)) => return Err(format!("ws error: {e}")), + None => return Err("connection closed before auth".into()), + }; + let server: ServerMessage = + serde_json::from_str(&msg).map_err(|e| format!("bad server json: {e}"))?; + apply_server_message(&mut snapshot, &server); + match server { + ServerMessage::SessionHistory { .. } | ServerMessage::AuthPending { .. } => break, + ServerMessage::AuthDenied { message } => { + return Err(message); + } + ServerMessage::Error { message, .. } => { + return Err(message); + } + _ => continue, + } + } + + let (outbound_tx, mut outbound_rx) = mpsc::unbounded_channel::(); + let inner = Arc::new(RwLock::new(Inner { snapshot })); + + let writer_inner = Arc::clone(&inner); + let reader = tokio::spawn(async move { + loop { + tokio::select! { + outbound = outbound_rx.recv() => { + let Some(msg) = outbound else { break; }; + let Ok(text) = serde_json::to_string(&msg) else { continue; }; + if write.send(Message::Text(text.into())).await.is_err() { + let mut guard = writer_inner.write().await; + guard.snapshot.status = JoinStatus::Disconnected; + break; + } + } + incoming = read.next() => { + match incoming { + Some(Ok(Message::Text(t))) => { + if let Ok(server) = serde_json::from_str::(&t) { + let mut guard = writer_inner.write().await; + apply_server_message(&mut guard.snapshot, &server); + if matches!(server, ServerMessage::SessionClosed) { + break; + } + } + } + Some(Ok(Message::Close(_))) | None => { + let mut guard = writer_inner.write().await; + guard.snapshot.status = JoinStatus::Disconnected; + break; + } + Some(Ok(_)) => {} + Some(Err(e)) => { + let mut guard = writer_inner.write().await; + guard.snapshot.status = JoinStatus::Error; + guard.snapshot.last_error = Some(e.to_string()); + break; + } + } + } + } + } + }); + + Ok(Self { + inner, + outbound: outbound_tx, + reader: Some(reader), + }) + } + + pub async fn snapshot(&self) -> SessionSnapshot { + self.inner.read().await.snapshot.clone() + } + + pub fn send_chat(&self, content: &str) -> Result<(), String> { + // Pending joiners are blocked by the relay; fail closed client-side too. + // Snapshot check is best-effort (async); relay enforces the real gate. + self.outbound + .send(ClientMessage::ChatSend { + content: content.to_string(), + }) + .map_err(|_| "not connected".to_string()) + } + + pub fn send_prompt(&self, content: &str) -> Result<(), String> { + self.outbound + .send(ClientMessage::CollabInput { + content: content.to_string(), + }) + .map_err(|_| "not connected".to_string()) + } + + pub fn send_typing(&self) -> Result<(), String> { + self.outbound + .send(ClientMessage::Typing) + .map_err(|_| "not connected".to_string()) + } + + pub async fn disconnect(mut self) { + drop(self.outbound); + if let Some(handle) = self.reader.take() { + let _ = tokio::time::timeout(Duration::from_secs(1), handle).await; + } + } +} + +fn normalize_ws_url(host: &str) -> Result { + let trimmed = host.trim().trim_end_matches('/'); + if trimmed.is_empty() { + return Err("host is empty".into()); + } + if trimmed.starts_with("ws://") || trimmed.starts_with("wss://") { + if trimmed.ends_with("/ws") { + return Ok(trimmed.to_string()); + } + return Ok(format!("{trimmed}/ws")); + } + // Bare host:port or host + Ok(format!("ws://{trimmed}/ws")) +} + +fn apply_server_message(snapshot: &mut SessionSnapshot, msg: &ServerMessage) { + match msg { + ServerMessage::SessionHistory { events } => { + snapshot.status = JoinStatus::Connected; + snapshot.recent_events = events.clone(); + if let Some(first) = events.first() { + snapshot.session_id = Some(first.session_id.clone()); + } + snapshot.last_error = None; + } + ServerMessage::AuthPending { .. } => { + snapshot.status = JoinStatus::Pending; + snapshot.last_error = None; + } + ServerMessage::AuthDenied { message } => { + snapshot.status = JoinStatus::Error; + snapshot.last_error = Some(message.clone()); + } + ServerMessage::SessionEvent { event } => { + if snapshot.session_id.is_none() { + snapshot.session_id = Some(event.session_id.clone()); + } + snapshot.recent_events.push(event.clone()); + if snapshot.recent_events.len() > 50 { + let drain = snapshot.recent_events.len() - 50; + snapshot.recent_events.drain(0..drain); + } + // Approval often arrives as history/events after pending. + if snapshot.status == JoinStatus::Pending { + snapshot.status = JoinStatus::Connected; + } + } + ServerMessage::ChatMessage { message } => { + snapshot.recent_chat.push(message.clone()); + if snapshot.recent_chat.len() > 50 { + let drain = snapshot.recent_chat.len() - 50; + snapshot.recent_chat.drain(0..drain); + } + } + ServerMessage::UserList { users } => { + snapshot.users = users.clone(); + } + ServerMessage::UserJoined { user } => { + snapshot.users.push(user.clone()); + } + ServerMessage::UserLeft { user_id } => { + snapshot.users.retain(|u| u.user_id != *user_id); + } + ServerMessage::UserRoleChanged { user_id, role } => { + for u in &mut snapshot.users { + if u.user_id == *user_id { + u.role = role.clone(); + } + } + } + ServerMessage::SessionClosed => { + snapshot.status = JoinStatus::Disconnected; + } + ServerMessage::Error { message, .. } => { + snapshot.status = JoinStatus::Error; + snapshot.last_error = Some(message.clone()); + } + ServerMessage::UserTyping { .. } => {} + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn normalize_urls() { + assert_eq!( + normalize_ws_url("127.0.0.1:7742").unwrap(), + "ws://127.0.0.1:7742/ws" + ); + assert_eq!( + normalize_ws_url("ws://127.0.0.1:7742").unwrap(), + "ws://127.0.0.1:7742/ws" + ); + assert_eq!( + normalize_ws_url("ws://127.0.0.1:7742/ws").unwrap(), + "ws://127.0.0.1:7742/ws" + ); + } + + #[test] + fn apply_history_and_chat() { + let mut snap = SessionSnapshot::new("h", "Alice"); + apply_server_message( + &mut snap, + &ServerMessage::SessionHistory { + events: vec![SessionEvent { + id: "e1".into(), + session_id: "sess".into(), + event_type: "message.created".into(), + payload: json!({"text": "hi"}), + timestamp: 1, + }], + }, + ); + assert_eq!(snap.status, JoinStatus::Connected); + assert_eq!(snap.session_id.as_deref(), Some("sess")); + + apply_server_message( + &mut snap, + &ServerMessage::ChatMessage { + message: ChatMessage { + id: "c1".into(), + session_id: "sess".into(), + user_id: "u1".into(), + display_name: Some("Bob".into()), + content: "yo".into(), + timestamp: 2, + }, + }, + ); + assert_eq!(snap.recent_chat.len(), 1); + } + + #[test] + fn client_messages_match_fixtures_shape() { + let auth = serde_json::to_value(ClientMessage::Auth { + token: "abc123".into(), + display_name: "Alice".into(), + repo_remote: Some("https://github.com/acme/app.git".into()), + email: Some("alice@acme.com".into()), + }) + .unwrap(); + assert_eq!(auth["type"], "auth"); + assert_eq!(auth["token"], "abc123"); + assert_eq!(auth["displayName"], "Alice"); + assert_eq!(auth["repoRemote"], "https://github.com/acme/app.git"); + assert_eq!(auth["email"], "alice@acme.com"); + + let chat = serde_json::to_value(ClientMessage::ChatSend { + content: "hello chat".into(), + }) + .unwrap(); + assert_eq!(chat["type"], "chat.send"); + + let input = serde_json::to_value(ClientMessage::CollabInput { + content: "fix the bug".into(), + }) + .unwrap(); + assert_eq!(input["type"], "collab.input"); + } +} diff --git a/crates/chorus-zed-helper/src/control.rs b/crates/chorus-zed-helper/src/control.rs new file mode 100644 index 0000000..794b377 --- /dev/null +++ b/crates/chorus-zed-helper/src/control.rs @@ -0,0 +1,150 @@ +//! JSON-line control protocol between CLI subcommands and the background daemon. + +use std::path::PathBuf; + +use serde::{Deserialize, Serialize}; + +use crate::client::{JoinStatus, SessionSnapshot}; +use crate::host::ShareSnapshot; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "cmd", rename_all = "snake_case")] +pub enum ControlRequest { + Ping, + Join { + host: String, + token: String, + #[serde(default = "default_display_name")] + display_name: String, + #[serde(default)] + repo_remote: Option, + #[serde(default)] + email: Option, + }, + Leave, + Chat { + content: String, + }, + Prompt { + content: String, + }, + Share { + #[serde(default)] + port: Option, + #[serde(default)] + role: Option, + #[serde(default)] + display_name: Option, + #[serde(default)] + require_approval: Option, + #[serde(default)] + repo_remote: Option, + #[serde(default)] + allowed_email_domain: Option, + #[serde(default)] + public_host: Option, + }, + Stop, + Approve { + user_id: String, + }, + Deny { + user_id: String, + }, + Publish { + content: String, + #[serde(default)] + event_type: Option, + }, + Status, + Shutdown, +} + +fn default_display_name() -> String { + "Zed".to_string() +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ControlResponse { + pub ok: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub snapshot: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub share: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub message: Option, +} + +impl ControlResponse { + pub fn ok_msg(message: impl Into) -> Self { + Self { + ok: true, + error: None, + snapshot: None, + share: None, + message: Some(message.into()), + } + } + + pub fn ok_status(snapshot: SessionSnapshot) -> Self { + Self { + ok: true, + error: None, + snapshot: Some(snapshot), + share: None, + message: None, + } + } + + pub fn ok_share(share: ShareSnapshot) -> Self { + let message = share.join_command.clone(); + Self { + ok: true, + error: None, + snapshot: None, + share: Some(share), + message: Some(message), + } + } + + pub fn err(error: impl Into) -> Self { + Self { + ok: false, + error: Some(error.into()), + snapshot: None, + share: None, + message: None, + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +#[serde(rename_all = "camelCase")] +pub struct DaemonState { + pub status: Option, + pub host: Option, +} + +pub fn runtime_dir() -> PathBuf { + if let Ok(dir) = std::env::var("CHORUS_ZED_RUNTIME_DIR") { + return PathBuf::from(dir); + } + if let Ok(dir) = std::env::var("XDG_RUNTIME_DIR") { + return PathBuf::from(dir).join("chorus-zed"); + } + std::env::temp_dir().join(format!( + "chorus-zed-{}", + std::env::var("USER").unwrap_or_else(|_| "user".into()) + )) +} + +pub fn socket_path() -> PathBuf { + runtime_dir().join("control.sock") +} + +pub fn pid_path() -> PathBuf { + runtime_dir().join("daemon.pid") +} diff --git a/crates/chorus-zed-helper/src/host.rs b/crates/chorus-zed-helper/src/host.rs new file mode 100644 index 0000000..503434c --- /dev/null +++ b/crates/chorus-zed-helper/src/host.rs @@ -0,0 +1,555 @@ +//! Host/share client: in-process `chorus-relay` plus `/host` control WebSocket. +//! +//! Same contract as `@chorus/client` `RelayServer` / the VS Code adapter. Zed +//! still does not drive an OpenCode LLM loop — joiner `collab.input` is recorded +//! on the host snapshot for the agent/CLI to act on. + +use std::io::Read; +use std::net::UdpSocket; +use std::path::Path; +use std::process::Command; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use chorus_relay::protocol::{ + ChatMessage, ConnectedUser, HostToRelay, RelayToHost, SessionEvent, SessionToken, UserRole, +}; +use chorus_relay::server::{serve, RelayConfig}; +use futures_util::{SinkExt, StreamExt}; +use serde::{Deserialize, Serialize}; +use tokio::sync::{mpsc, oneshot, RwLock}; +use tokio::task::JoinHandle; +use tokio_tungstenite::{connect_async, tungstenite::Message}; + +use crate::client::JoinClient; + +const DEFAULT_PORT: u16 = 7742; +const TOKEN_TIMEOUT: Duration = Duration::from_secs(5); + +#[derive(Debug, Clone)] +pub struct ShareOpts { + pub port: u16, + pub role: UserRole, + pub display_name: String, + pub require_approval: bool, + pub repo_remote: Option, + pub allowed_email_domain: Option, + pub public_host: Option, +} + +impl Default for ShareOpts { + fn default() -> Self { + Self { + port: DEFAULT_PORT, + role: UserRole::Edit, + display_name: default_host_name(), + require_approval: true, + repo_remote: None, + allowed_email_domain: None, + public_host: None, + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ShareSnapshot { + pub sharing: bool, + pub port: u16, + pub host: String, + pub display_name: String, + pub session_id: String, + pub join_command: String, + pub require_approval: bool, + pub pending_users: Vec, + pub users: Vec, + pub recent_inputs: Vec, + pub recent_chat: Vec, + pub last_error: Option, +} + +pub struct HostSession { + inner: Arc>, + outbound: mpsc::UnboundedSender, + reader: Option>, + server: Option>, +} + +struct Inner { + snapshot: ShareSnapshot, + pending_token: Option>>, +} + +pub enum LiveSession { + Idle, + Joined(JoinClient), + Sharing(HostSession), +} + +impl LiveSession { + pub async fn shutdown(&mut self) { + match std::mem::replace(self, LiveSession::Idle) { + LiveSession::Idle => {} + LiveSession::Joined(client) => client.disconnect().await, + LiveSession::Sharing(host) => host.stop().await, + } + } +} + +impl HostSession { + pub async fn start(mut opts: ShareOpts) -> Result { + let name = opts.display_name.trim(); + if name.is_empty() { + return Err("display name is required".into()); + } + opts.display_name = name.to_string(); + if opts.port == 0 { + opts.port = DEFAULT_PORT; + } + if opts.repo_remote.is_none() { + opts.repo_remote = detect_repo_remote(None); + } + + let host_token = random_hex(32)?; + let session_id = format!("zed-{}", random_hex(6)?); + let config = RelayConfig { + port: opts.port, + host_token: host_token.clone(), + bind: "0.0.0.0".into(), + }; + let server = tokio::spawn(async move { + if let Err(e) = serve(config).await { + tracing::error!("chorus-relay serve failed: {e}"); + } + }); + + wait_for_port(opts.port).await.map_err(|e| { + server.abort(); + e + })?; + + let ws_url = format!("ws://127.0.0.1:{}/host", opts.port); + let (ws, _) = connect_async(&ws_url).await.map_err(|e| { + server.abort(); + format!("host control connect failed: {e}") + })?; + let (mut write, mut read) = ws.split(); + write + .send(Message::Text( + serde_json::to_string(&HostToRelay::HostAuth { token: host_token }) + .map_err(|e| e.to_string())? + .into(), + )) + .await + .map_err(|e| format!("host.auth send failed: {e}"))?; + + let deadline = tokio::time::Instant::now() + Duration::from_secs(8); + loop { + if tokio::time::Instant::now() > deadline { + server.abort(); + return Err("timed out waiting for host.ready".into()); + } + let next = tokio::time::timeout(Duration::from_secs(5), read.next()) + .await + .map_err(|_| "timed out waiting for host.ready".to_string())?; + match next { + Some(Ok(Message::Text(t))) => { + let msg: RelayToHost = + serde_json::from_str(&t).map_err(|e| format!("bad host json: {e}"))?; + if matches!(msg, RelayToHost::HostReady { .. }) { + break; + } + } + Some(Ok(Message::Close(frame))) => { + server.abort(); + return Err(format!("host control closed before ready: {frame:?}")); + } + Some(Ok(_)) => continue, + Some(Err(e)) => { + server.abort(); + return Err(format!("host ws error: {e}")); + } + None => { + server.abort(); + return Err("host control closed before ready".into()); + } + } + } + + let advertised = opts + .public_host + .clone() + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| format!("{}:{}", lan_ip(), opts.port)); + + let inner = Arc::new(RwLock::new(Inner { + snapshot: ShareSnapshot { + sharing: true, + port: opts.port, + host: advertised.clone(), + display_name: opts.display_name.clone(), + session_id: session_id.clone(), + join_command: String::new(), + require_approval: opts.require_approval, + pending_users: Vec::new(), + users: Vec::new(), + recent_inputs: Vec::new(), + recent_chat: Vec::new(), + last_error: None, + }, + pending_token: None, + })); + + let (outbound_tx, mut outbound_rx) = mpsc::unbounded_channel::(); + let reader_inner = Arc::clone(&inner); + let reader = tokio::spawn(async move { + loop { + tokio::select! { + outbound = outbound_rx.recv() => { + let Some(msg) = outbound else { break; }; + let Ok(text) = serde_json::to_string(&msg) else { continue; }; + if write.send(Message::Text(text.into())).await.is_err() { + break; + } + } + incoming = read.next() => { + match incoming { + Some(Ok(Message::Text(t))) => { + if let Ok(msg) = serde_json::from_str::(&t) { + apply_host_message(&reader_inner, msg).await; + } + } + Some(Ok(Message::Close(_))) | None => break, + Some(Ok(_)) => {} + Some(Err(e)) => { + let mut guard = reader_inner.write().await; + guard.snapshot.last_error = Some(e.to_string()); + break; + } + } + } + } + } + }); + + let session = Self { + inner: Arc::clone(&inner), + outbound: outbound_tx, + reader: Some(reader), + server: Some(server), + }; + + session.send(HostToRelay::SessionPolicy { + require_approval: Some(opts.require_approval), + repo_remote: Some(opts.repo_remote.clone().unwrap_or_default()), + allowed_email_domain: opts.allowed_email_domain.clone(), + additional_repo_remote_prefixes: None, + repo_remote_rewrites: None, + })?; + + let token = session + .issue_token(&session_id, opts.role.clone()) + .await?; + let join_command = format!( + "/chorus-join token=\"{}\" host=\"{}\"", + token.token, advertised + ); + { + let mut guard = inner.write().await; + guard.snapshot.join_command = join_command; + } + + Ok(session) + } + + async fn issue_token(&self, session_id: &str, role: UserRole) -> Result { + let (tx, rx) = oneshot::channel(); + { + let mut guard = self.inner.write().await; + guard.pending_token = Some(tx); + } + self.send(HostToRelay::TokenIssue { + session_id: session_id.to_string(), + role: Some(role), + ttl_ms: None, + })?; + match tokio::time::timeout(TOKEN_TIMEOUT, rx).await { + Ok(Ok(result)) => result, + Ok(Err(_)) => Err("token.issue channel closed".into()), + Err(_) => Err("token.issue timed out".into()), + } + } + + fn send(&self, msg: HostToRelay) -> Result<(), String> { + self.outbound + .send(msg) + .map_err(|_| "host control not connected".to_string()) + } + + pub async fn snapshot(&self) -> ShareSnapshot { + self.inner.read().await.snapshot.clone() + } + + pub fn approve(&self, user_id: &str) -> Result<(), String> { + self.send(HostToRelay::HostApprove { + user_id: user_id.to_string(), + }) + } + + pub fn deny(&self, user_id: &str) -> Result<(), String> { + self.send(HostToRelay::HostDeny { + user_id: user_id.to_string(), + }) + } + + pub async fn send_chat_named(&self, content: &str) -> Result<(), String> { + let name = self.inner.read().await.snapshot.display_name.clone(); + self.send(HostToRelay::ChatSend { + content: content.to_string(), + display_name: Some(name), + }) + } + + pub async fn publish(&self, content: &str, event_type: &str) -> Result<(), String> { + let (session_id, id) = { + let snap = self.inner.read().await; + (snap.snapshot.session_id.clone(), format!("{}-{}", now_ms(), random_hex(4).unwrap_or_else(|_| "id".into()))) + }; + let ty = match event_type { + "assistant" | "ai" => "assistant", + _ => "user", + }; + self.send(HostToRelay::SessionEvent { + event: SessionEvent { + id, + session_id, + event_type: ty.into(), + payload: serde_json::Value::String(content.to_string()), + timestamp: now_ms(), + }, + }) + } + + pub async fn stop(mut self) { + let _ = self.send(HostToRelay::HostClose); + drop(self.outbound); + if let Some(handle) = self.reader.take() { + let _ = tokio::time::timeout(Duration::from_secs(1), handle).await; + } + if let Some(handle) = self.server.take() { + handle.abort(); + let _ = tokio::time::timeout(Duration::from_secs(1), handle).await; + } + } +} + +async fn apply_host_message(inner: &Arc>, msg: RelayToHost) { + let mut guard = inner.write().await; + match msg { + RelayToHost::TokenIssued { token } => { + if let Some(tx) = guard.pending_token.take() { + let _ = tx.send(Ok(token)); + } + } + RelayToHost::Error { message, .. } => { + if let Some(tx) = guard.pending_token.take() { + let _ = tx.send(Err(message.clone())); + } + guard.snapshot.last_error = Some(message); + } + RelayToHost::CollabInput { + user_id, + display_name, + content, + } => { + let label = display_name.unwrap_or_else(|| user_id.chars().take(8).collect()); + guard + .snapshot + .recent_inputs + .push(format!("[{label}]: {content}")); + if guard.snapshot.recent_inputs.len() > 50 { + let drain = guard.snapshot.recent_inputs.len() - 50; + guard.snapshot.recent_inputs.drain(0..drain); + } + } + RelayToHost::ChatMessage { message } => { + guard.snapshot.recent_chat.push(message); + if guard.snapshot.recent_chat.len() > 50 { + let drain = guard.snapshot.recent_chat.len() - 50; + guard.snapshot.recent_chat.drain(0..drain); + } + } + RelayToHost::UserPending { user } => { + guard + .snapshot + .pending_users + .retain(|u| u.user_id != user.user_id); + guard.snapshot.pending_users.push(user); + } + RelayToHost::UserJoined { user } => { + guard + .snapshot + .pending_users + .retain(|u| u.user_id != user.user_id); + guard + .snapshot + .users + .retain(|u| u.user_id != user.user_id); + guard.snapshot.users.push(user); + } + RelayToHost::UserLeft { user_id } => { + guard + .snapshot + .pending_users + .retain(|u| u.user_id != user_id); + guard.snapshot.users.retain(|u| u.user_id != user_id); + } + RelayToHost::UserList { users } => { + guard.snapshot.users = users + .iter() + .filter(|u| { + matches!(u.status, chorus_relay::protocol::UserStatus::Active) + }) + .cloned() + .collect(); + guard.snapshot.pending_users = users + .into_iter() + .filter(|u| { + matches!(u.status, chorus_relay::protocol::UserStatus::Pending) + }) + .collect(); + } + _ => {} + } +} + +fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +fn random_hex(bytes: usize) -> Result { + let mut buf = vec![0u8; bytes]; + std::fs::File::open("/dev/urandom") + .and_then(|mut f| f.read_exact(&mut buf)) + .map_err(|e| format!("entropy: {e}"))?; + Ok(buf.iter().map(|b| format!("{b:02x}")).collect()) +} + +fn default_host_name() -> String { + std::env::var("CHORUS_DISPLAY_NAME") + .ok() + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .or_else(|| std::env::var("USER").ok()) + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| "Zed".into()) +} + +pub fn parse_role(raw: &str) -> Result { + match raw.trim().to_ascii_lowercase().as_str() { + "edit" => Ok(UserRole::Edit), + "view" => Ok(UserRole::View), + "admin" => Ok(UserRole::Admin), + other => Err(format!("unknown role '{other}' (edit|view|admin)")), + } +} + +pub fn lan_ip() -> String { + if let Ok(sock) = UdpSocket::bind("0.0.0.0:0") { + if sock.connect("8.8.8.8:80").is_ok() { + if let Ok(addr) = sock.local_addr() { + return addr.ip().to_string(); + } + } + } + "127.0.0.1".into() +} + +pub fn detect_repo_remote(cwd: Option<&Path>) -> Option { + let mut cmd = Command::new("git"); + cmd.args(["remote", "get-url", "origin"]); + cmd.stdin(std::process::Stdio::null()); + cmd.stderr(std::process::Stdio::null()); + if let Some(dir) = cwd { + cmd.current_dir(dir); + } + let out = cmd.output().ok()?; + if !out.status.success() { + return None; + } + let remote = String::from_utf8(out.stdout).ok()?.trim().to_string(); + if remote.is_empty() { + None + } else { + Some(remote) + } +} + +async fn wait_for_port(port: u16) -> Result<(), String> { + let deadline = tokio::time::Instant::now() + Duration::from_secs(8); + while tokio::time::Instant::now() < deadline { + if tokio::net::TcpStream::connect(("127.0.0.1", port)) + .await + .is_ok() + { + return Ok(()); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + Err(format!( + "chorus-relay did not become ready on port {port} (is it already in use?)" + )) +} + +pub fn format_share_status(snap: &ShareSnapshot) -> String { + let mut out = String::new(); + out.push_str("status: sharing\n"); + out.push_str(&format!("host: {}\n", snap.host)); + out.push_str(&format!("port: {}\n", snap.port)); + out.push_str(&format!("displayName: {}\n", snap.display_name)); + out.push_str(&format!("sessionId: {}\n", snap.session_id)); + out.push_str(&format!("requireApproval: {}\n", snap.require_approval)); + out.push_str(&format!("joinCommand: {}\n", snap.join_command)); + if let Some(err) = &snap.last_error { + out.push_str(&format!("error: {err}\n")); + } + out.push_str(&format!("pending ({})\n", snap.pending_users.len())); + for u in &snap.pending_users { + out.push_str(&format!( + " - {} [{}] {:?}\n", + u.display_name, u.user_id, u.role + )); + } + out.push_str(&format!("users ({})\n", snap.users.len())); + for u in &snap.users { + out.push_str(&format!( + " - {} [{}] {:?}\n", + u.display_name, u.user_id, u.role + )); + } + out.push_str(&format!( + "recentInputs: {} | recentChat: {}\n", + snap.recent_inputs.len(), + snap.recent_chat.len() + )); + for line in snap.recent_inputs.iter().rev().take(5).collect::>().into_iter().rev() + { + out.push_str(&format!(" prompt {line}\n")); + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_roles() { + assert!(matches!(parse_role("edit"), Ok(UserRole::Edit))); + assert!(matches!(parse_role("ADMIN"), Ok(UserRole::Admin))); + assert!(parse_role("owner").is_err()); + } +} diff --git a/crates/chorus-zed-helper/src/lib.rs b/crates/chorus-zed-helper/src/lib.rs new file mode 100644 index 0000000..a4302c5 --- /dev/null +++ b/crates/chorus-zed-helper/src/lib.rs @@ -0,0 +1,12 @@ +//! Chorus join + host helper used by the Zed extension (CLI + MCP). +//! +//! Speaks `/ws` (joiner) and `/host` (share) from `crates/chorus-relay` / `protocol/`. + +pub mod client; +pub mod control; +pub mod host; +pub mod mcp; + +pub use client::{JoinClient, JoinStatus, SessionSnapshot}; +pub use control::{ControlRequest, ControlResponse, DaemonState}; +pub use host::{HostSession, LiveSession, ShareOpts, ShareSnapshot}; diff --git a/crates/chorus-zed-helper/src/main.rs b/crates/chorus-zed-helper/src/main.rs new file mode 100644 index 0000000..0b556bd --- /dev/null +++ b/crates/chorus-zed-helper/src/main.rs @@ -0,0 +1,580 @@ +use std::fs; +use std::path::PathBuf; +use std::process::{Command, Stdio}; +use std::time::Duration; + +use clap::{Parser, Subcommand}; +use chorus_zed_helper::client::JoinClient; +use chorus_zed_helper::control::{ + pid_path, runtime_dir, socket_path, ControlRequest, ControlResponse, +}; +use chorus_zed_helper::host::{ + format_share_status, parse_role, HostSession, LiveSession, ShareOpts, +}; +use chorus_zed_helper::mcp::{self, format_status}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{UnixListener, UnixStream}; +use tokio::sync::Mutex; + +#[derive(Parser, Debug)] +#[command( + name = "chorus-zed-helper", + about = "Chorus join/host client for Zed (CLI + MCP)." +)] +struct Args { + #[command(subcommand)] + command: Commands, +} + +#[derive(Subcommand, Debug)] +enum Commands { + /// Start the background daemon that holds the WebSocket session + Daemon, + /// Connect to a Chorus relay as a joiner + Join { + /// Relay host:port (e.g. 192.168.1.10:7742) or ws:// URL + #[arg(long, env = "CHORUS_RELAY_HOST")] + host: String, + /// Join token from the host share command + #[arg(long)] + token: String, + /// Display name shown to collaborators (required non-empty) + #[arg(long, default_value = "Zed", env = "CHORUS_DISPLAY_NAME")] + name: String, + /// Optional git remote when the host enabled a same-repo gate + #[arg(long, env = "CHORUS_REPO_REMOTE")] + repo_remote: Option, + /// Optional email when the host enabled an allowedEmailDomain gate + #[arg(long, env = "CHORUS_EMAIL")] + email: Option, + }, + /// Disconnect from the current joined session + Leave, + /// Host a session (spawn relay, issue join token) + Share { + /// Relay listen port + #[arg(long, default_value_t = 7742, env = "CHORUS_PORT")] + port: u16, + /// Join-token role: edit, view, or admin + #[arg(long, default_value = "edit")] + role: String, + /// Host display name + #[arg(long, default_value = "Zed", env = "CHORUS_DISPLAY_NAME")] + name: String, + /// Auto-admit joiners (default is host approval required) + #[arg(long, default_value_t = false)] + auto_admit: bool, + /// Optional git origin for the same-repo gate + #[arg(long, env = "CHORUS_REPO_REMOTE")] + repo_remote: Option, + /// Optional email domain gate + #[arg(long)] + allowed_email_domain: Option, + /// Advertised host:port in the join command + #[arg(long, env = "CHORUS_PUBLIC_HOST")] + public_host: Option, + }, + /// Stop hosting and tear down the local relay + Stop, + /// Approve a pending joiner + Approve { + user_id: String, + }, + /// Deny a pending joiner + Deny { + user_id: String, + }, + /// Publish a host/AI line to joiners + Publish { + content: String, + #[arg(long, default_value = "user")] + r#type: String, + }, + /// Send a side-channel chat message + Chat { + /// Message text + content: String, + }, + /// Forward a prompt into the host session (`collab.input`) + Prompt { + /// Prompt text + content: String, + }, + /// Print connection status + Status, + /// Run as an MCP stdio server for Zed + Mcp, + /// Shut down the background daemon + Shutdown, +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let args = Args::parse(); + match args.command { + Commands::Daemon => run_daemon().await?, + Commands::Mcp => { + let handle = tokio::runtime::Handle::current(); + tokio::task::spawn_blocking(move || mcp::run_stdio(handle)) + .await + .map_err(|e| e.to_string())??; + } + Commands::Join { + host, + token, + name, + repo_remote, + email, + } => { + ensure_daemon().await?; + let resp = request(ControlRequest::Join { + host, + token, + display_name: name, + repo_remote, + email, + }) + .await?; + print_response(&resp); + if !resp.ok { + std::process::exit(1); + } + } + Commands::Share { + port, + role, + name, + auto_admit, + repo_remote, + allowed_email_domain, + public_host, + } => { + ensure_daemon().await?; + let resp = request(ControlRequest::Share { + port: Some(port), + role: Some(role), + display_name: Some(name), + require_approval: Some(!auto_admit), + repo_remote, + allowed_email_domain, + public_host, + }) + .await?; + print_response(&resp); + if !resp.ok { + std::process::exit(1); + } + } + Commands::Stop => { + let resp = request(ControlRequest::Stop).await?; + print_response(&resp); + if !resp.ok { + std::process::exit(1); + } + } + Commands::Approve { user_id } => { + let resp = request(ControlRequest::Approve { user_id }).await?; + print_response(&resp); + if !resp.ok { + std::process::exit(1); + } + } + Commands::Deny { user_id } => { + let resp = request(ControlRequest::Deny { user_id }).await?; + print_response(&resp); + if !resp.ok { + std::process::exit(1); + } + } + Commands::Publish { content, r#type } => { + let resp = request(ControlRequest::Publish { + content, + event_type: Some(r#type), + }) + .await?; + print_response(&resp); + if !resp.ok { + std::process::exit(1); + } + } + Commands::Leave => { + let resp = request(ControlRequest::Leave).await?; + print_response(&resp); + if !resp.ok { + std::process::exit(1); + } + } + Commands::Chat { content } => { + let resp = request(ControlRequest::Chat { content }).await?; + print_response(&resp); + if !resp.ok { + std::process::exit(1); + } + } + Commands::Prompt { content } => { + let resp = request(ControlRequest::Prompt { content }).await?; + print_response(&resp); + if !resp.ok { + std::process::exit(1); + } + } + Commands::Status => match request(ControlRequest::Status).await { + Ok(resp) => { + print_response(&resp); + if !resp.ok { + std::process::exit(1); + } + } + Err(_) => { + println!("status: idle\n(daemon not running)"); + } + }, + Commands::Shutdown => match request(ControlRequest::Shutdown).await { + Ok(resp) => print_response(&resp), + Err(_) => println!("daemon not running"), + }, + } + Ok(()) +} + +fn print_response(resp: &ControlResponse) { + if let Some(share) = &resp.share { + print!("{}", format_share_status(share)); + } else if let Some(snap) = &resp.snapshot { + print!("{}", format_status(snap)); + } else if let Some(msg) = &resp.message { + println!("{msg}"); + } else if let Some(err) = &resp.error { + eprintln!("error: {err}"); + } else if resp.ok { + println!("ok"); + } +} + +async fn ensure_daemon() -> Result<(), String> { + if ping_daemon().await.is_ok() { + return Ok(()); + } + let dir = runtime_dir(); + fs::create_dir_all(&dir).map_err(|e| e.to_string())?; + let sock = socket_path(); + if sock.exists() { + let _ = fs::remove_file(&sock); + } + + let exe = std::env::current_exe().map_err(|e| e.to_string())?; + let child = Command::new(&exe) + .arg("daemon") + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .map_err(|e| format!("failed to spawn daemon: {e}"))?; + + fs::write(pid_path(), child.id().to_string()).map_err(|e| e.to_string())?; + + for _ in 0..50 { + if ping_daemon().await.is_ok() { + return Ok(()); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + Err("daemon did not become ready".into()) +} + +async fn ping_daemon() -> Result<(), String> { + let resp = request(ControlRequest::Ping).await?; + if resp.ok { + Ok(()) + } else { + Err(resp.error.unwrap_or_else(|| "ping failed".into())) + } +} + +async fn request(req: ControlRequest) -> Result { + let path = socket_path(); + let mut stream = UnixStream::connect(&path) + .await + .map_err(|e| format!("connect control socket: {e}"))?; + let line = serde_json::to_string(&req).map_err(|e| e.to_string())?; + stream + .write_all(line.as_bytes()) + .await + .map_err(|e| e.to_string())?; + stream + .write_all(b"\n") + .await + .map_err(|e| e.to_string())?; + + let mut reader = BufReader::new(stream); + let mut response = String::new(); + reader + .read_line(&mut response) + .await + .map_err(|e| e.to_string())?; + serde_json::from_str(response.trim()).map_err(|e| format!("bad control response: {e}")) +} + +async fn run_daemon() -> Result<(), Box> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("warn")), + ) + .with_writer(std::io::stderr) + .init(); + + let dir = runtime_dir(); + fs::create_dir_all(&dir)?; + let sock: PathBuf = socket_path(); + if sock.exists() { + fs::remove_file(&sock)?; + } + + let listener = UnixListener::bind(&sock)?; + let session: Mutex = Mutex::new(LiveSession::Idle); + + loop { + let (stream, _) = listener.accept().await?; + let mut reader = BufReader::new(stream); + let mut line = String::new(); + if reader.read_line(&mut line).await? == 0 { + continue; + } + let req: ControlRequest = match serde_json::from_str(line.trim()) { + Ok(r) => r, + Err(e) => { + let mut stream = reader.into_inner(); + let resp = ControlResponse::err(format!("bad request: {e}")); + let body = serde_json::to_string(&resp)?; + stream.write_all(body.as_bytes()).await?; + stream.write_all(b"\n").await?; + continue; + } + }; + + let (resp, shutdown) = handle_control(&session, req).await; + let mut stream = reader.into_inner(); + let body = serde_json::to_string(&resp)?; + stream.write_all(body.as_bytes()).await?; + stream.write_all(b"\n").await?; + if shutdown { + break; + } + } + + let _ = fs::remove_file(sock); + let _ = fs::remove_file(pid_path()); + Ok(()) +} + +async fn handle_control( + session: &Mutex, + req: ControlRequest, +) -> (ControlResponse, bool) { + match req { + ControlRequest::Ping => (ControlResponse::ok_msg("pong"), false), + ControlRequest::Shutdown => { + let mut guard = session.lock().await; + guard.shutdown().await; + (ControlResponse::ok_msg("shutting down"), true) + } + ControlRequest::Join { + host, + token, + display_name, + repo_remote, + email, + } => { + { + let guard = session.lock().await; + if matches!(*guard, LiveSession::Sharing(_)) { + return ( + ControlResponse::err("stop sharing before joining another session"), + false, + ); + } + } + match JoinClient::connect( + &host, + &token, + &display_name, + repo_remote.as_deref(), + email.as_deref(), + ) + .await + { + Ok(client) => { + let snap = client.snapshot().await; + let mut guard = session.lock().await; + guard.shutdown().await; + *guard = LiveSession::Joined(client); + (ControlResponse::ok_status(snap), false) + } + Err(e) => (ControlResponse::err(e), false), + } + } + ControlRequest::Share { + port, + role, + display_name, + require_approval, + repo_remote, + allowed_email_domain, + public_host, + } => { + let role = match parse_role(role.as_deref().unwrap_or("edit")) { + Ok(r) => r, + Err(e) => return (ControlResponse::err(e), false), + }; + let name = display_name + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| { + std::env::var("USER").unwrap_or_else(|_| "Zed".into()) + }); + match HostSession::start(ShareOpts { + port: port.unwrap_or(7742), + role, + display_name: name, + require_approval: require_approval.unwrap_or(true), + repo_remote, + allowed_email_domain, + public_host, + }) + .await + { + Ok(host) => { + let snap = host.snapshot().await; + let mut guard = session.lock().await; + guard.shutdown().await; + *guard = LiveSession::Sharing(host); + (ControlResponse::ok_share(snap), false) + } + Err(e) => (ControlResponse::err(e), false), + } + } + ControlRequest::Stop => { + let mut guard = session.lock().await; + match std::mem::replace(&mut *guard, LiveSession::Idle) { + LiveSession::Sharing(host) => { + host.stop().await; + (ControlResponse::ok_msg("stopped sharing"), false) + } + LiveSession::Joined(client) => { + client.disconnect().await; + (ControlResponse::ok_msg("left session (was not sharing)"), false) + } + LiveSession::Idle => (ControlResponse::ok_msg("not sharing"), false), + } + } + ControlRequest::Approve { user_id } => { + let guard = session.lock().await; + match &*guard { + LiveSession::Sharing(host) => match host.approve(&user_id) { + Ok(()) => (ControlResponse::ok_msg(format!("approved {user_id}")), false), + Err(e) => (ControlResponse::err(e), false), + }, + _ => (ControlResponse::err("share a session first"), false), + } + } + ControlRequest::Deny { user_id } => { + let guard = session.lock().await; + match &*guard { + LiveSession::Sharing(host) => match host.deny(&user_id) { + Ok(()) => (ControlResponse::ok_msg(format!("denied {user_id}")), false), + Err(e) => (ControlResponse::err(e), false), + }, + _ => (ControlResponse::err("share a session first"), false), + } + } + ControlRequest::Publish { + content, + event_type, + } => { + let guard = session.lock().await; + match &*guard { + LiveSession::Sharing(host) => { + match host + .publish(&content, event_type.as_deref().unwrap_or("user")) + .await + { + Ok(()) => (ControlResponse::ok_msg("published"), false), + Err(e) => (ControlResponse::err(e), false), + } + } + _ => (ControlResponse::err("share a session first"), false), + } + } + ControlRequest::Leave => { + let mut guard = session.lock().await; + match std::mem::replace(&mut *guard, LiveSession::Idle) { + LiveSession::Joined(client) => { + client.disconnect().await; + (ControlResponse::ok_msg("left session"), false) + } + LiveSession::Sharing(host) => { + host.stop().await; + (ControlResponse::ok_msg("stopped sharing"), false) + } + LiveSession::Idle => (ControlResponse::ok_msg("not connected"), false), + } + } + ControlRequest::Chat { content } => { + let guard = session.lock().await; + match &*guard { + LiveSession::Joined(client) => match client.send_chat(&content) { + Ok(()) => (ControlResponse::ok_msg("chat sent"), false), + Err(e) => (ControlResponse::err(e), false), + }, + LiveSession::Sharing(host) => match host.send_chat_named(&content).await { + Ok(()) => (ControlResponse::ok_msg("chat sent"), false), + Err(e) => (ControlResponse::err(e), false), + }, + LiveSession::Idle => (ControlResponse::err("not connected"), false), + } + } + ControlRequest::Prompt { content } => { + let guard = session.lock().await; + match &*guard { + LiveSession::Joined(client) => match client.send_prompt(&content) { + Ok(()) => (ControlResponse::ok_msg("prompt sent"), false), + Err(e) => (ControlResponse::err(e), false), + }, + LiveSession::Sharing(_) => ( + ControlResponse::err( + "you are hosting — joiners send prompts; use publish for host/AI lines", + ), + false, + ), + LiveSession::Idle => (ControlResponse::err("not connected"), false), + } + } + ControlRequest::Status => { + let guard = session.lock().await; + match &*guard { + LiveSession::Joined(client) => { + let snap = client.snapshot().await; + (ControlResponse::ok_status(snap), false) + } + LiveSession::Sharing(host) => { + let snap = host.snapshot().await; + (ControlResponse::ok_share(snap), false) + } + LiveSession::Idle => ( + ControlResponse::ok_status(chorus_zed_helper::client::SessionSnapshot { + status: chorus_zed_helper::client::JoinStatus::Disconnected, + host: String::new(), + display_name: String::new(), + session_id: None, + users: vec![], + recent_events: vec![], + recent_chat: vec![], + last_error: None, + }), + false, + ), + } + } + } +} diff --git a/crates/chorus-zed-helper/src/mcp.rs b/crates/chorus-zed-helper/src/mcp.rs new file mode 100644 index 0000000..c5e2b59 --- /dev/null +++ b/crates/chorus-zed-helper/src/mcp.rs @@ -0,0 +1,565 @@ +//! Minimal stdio MCP server exposing Chorus join tools to Zed's Agent Panel. + +use std::io::{BufRead, BufReader, Write}; +use std::sync::Arc; + +use serde_json::{json, Value}; +use tokio::runtime::Handle; +use tokio::sync::Mutex; + +use crate::client::{JoinClient, JoinStatus}; +use crate::host::{ + format_share_status, parse_role, HostSession, LiveSession, ShareOpts, +}; + +struct Session { + live: LiveSession, +} + +/// Run an MCP server on stdin/stdout (blocking; requires a Tokio handle). +pub fn run_stdio(handle: Handle) -> Result<(), String> { + let session = Arc::new(Mutex::new(Session { + live: LiveSession::Idle, + })); + let stdin = std::io::stdin(); + let mut stdout = std::io::stdout(); + let mut reader = BufReader::new(stdin.lock()); + + loop { + let msg = match read_message(&mut reader)? { + Some(v) => v, + None => break, + }; + + let id = msg.get("id").cloned(); + let method = msg.get("method").and_then(|m| m.as_str()).unwrap_or(""); + let params = msg.get("params").cloned().unwrap_or(json!({})); + + // Notifications (no id) — ignore after handling initialize side-effects if any. + if id.is_none() { + continue; + } + + let result = match method { + "initialize" => Ok(json!({ + "protocolVersion": "2024-11-05", + "capabilities": { "tools": {} }, + "serverInfo": { + "name": "chorus-zed", + "version": env!("CARGO_PKG_VERSION") + } + })), + "tools/list" => Ok(json!({ "tools": tool_defs() })), + "tools/call" => { + let name = params + .get("name") + .and_then(|n| n.as_str()) + .unwrap_or(""); + let args = params + .get("arguments") + .cloned() + .unwrap_or(json!({})); + handle.block_on(call_tool(session.clone(), name, args)) + } + "ping" => Ok(json!({})), + _ => Err(format!("method not found: {method}")), + }; + + let response = match result { + Ok(value) => json!({ + "jsonrpc": "2.0", + "id": id, + "result": value + }), + Err(err) => json!({ + "jsonrpc": "2.0", + "id": id, + "error": { "code": -32000, "message": err } + }), + }; + write_message(&mut stdout, &response)?; + } + Ok(()) +} + +fn tool_defs() -> Vec { + vec![ + json!({ + "name": "chorus_join", + "description": "Join an existing Chorus collaborative session as a joiner (connect to relay /ws).", + "inputSchema": { + "type": "object", + "properties": { + "host": { + "type": "string", + "description": "Relay host:port (e.g. 192.168.1.10:7742) or full ws:// URL" + }, + "token": { "type": "string", "description": "Join token from the host" }, + "display_name": { + "type": "string", + "description": "Required display name shown to collaborators" + }, + "repo_remote": { + "type": "string", + "description": "Optional git remote URL when the host enabled a same-repo gate" + }, + "email": { + "type": "string", + "description": "Optional email when the host enabled an allowedEmailDomain gate" + } + }, + "required": ["host", "token", "display_name"] + } + }), + json!({ + "name": "chorus_leave", + "description": "Leave the current Chorus session and disconnect from the relay.", + "inputSchema": { "type": "object", "properties": {} } + }), + json!({ + "name": "chorus_chat", + "description": "Send a side-channel chat message (does not become an LLM prompt).", + "inputSchema": { + "type": "object", + "properties": { + "content": { "type": "string", "description": "Chat message text" } + }, + "required": ["content"] + } + }), + json!({ + "name": "chorus_prompt", + "description": "Send a collaborative prompt (collab.input) into the host OpenCode session. Requires edit role.", + "inputSchema": { + "type": "object", + "properties": { + "content": { "type": "string", "description": "Prompt text for the host session" } + }, + "required": ["content"] + } + }), + json!({ + "name": "chorus_status", + "description": "Show join or share status, users, pending joiners, and recent events.", + "inputSchema": { "type": "object", "properties": {} } + }), + json!({ + "name": "chorus_share", + "description": "Host a Chorus session: start chorus-relay, set access policy, and return a /chorus-join command. Does not drive an OpenCode LLM loop; joiner prompts appear in chorus_status.", + "inputSchema": { + "type": "object", + "properties": { + "port": { + "type": "number", + "description": "Relay listen port (default 7742)" + }, + "role": { + "type": "string", + "description": "Join-token role: edit (default), view, or admin" + }, + "display_name": { + "type": "string", + "description": "Host display name shown to collaborators" + }, + "require_approval": { + "type": "boolean", + "description": "Hold joiners pending until chorus_approve (default true)" + }, + "repo_remote": { + "type": "string", + "description": "Optional origin URL for the same-repo gate (defaults to git origin)" + }, + "allowed_email_domain": { + "type": "string", + "description": "Optional email domain gate (e.g. acme.com)" + }, + "public_host": { + "type": "string", + "description": "Advertised host:port in the join command (defaults to LAN IP:port)" + } + } + } + }), + json!({ + "name": "chorus_stop", + "description": "Stop hosting and tear down the local chorus-relay.", + "inputSchema": { "type": "object", "properties": {} } + }), + json!({ + "name": "chorus_approve", + "description": "Approve a pending joiner by user_id (from chorus_status pending list).", + "inputSchema": { + "type": "object", + "properties": { + "user_id": { "type": "string", "description": "Pending joiner userId" } + }, + "required": ["user_id"] + } + }), + json!({ + "name": "chorus_deny", + "description": "Deny a pending joiner by user_id.", + "inputSchema": { + "type": "object", + "properties": { + "user_id": { "type": "string", "description": "Pending joiner userId" } + }, + "required": ["user_id"] + } + }), + json!({ + "name": "chorus_publish", + "description": "Fan out a host/AI line to joiners (session.event). Use while sharing.", + "inputSchema": { + "type": "object", + "properties": { + "content": { "type": "string", "description": "Message text" }, + "type": { + "type": "string", + "description": "user (default) or assistant" + } + }, + "required": ["content"] + } + }), + ] +} + +async fn call_tool( + session: Arc>, + name: &str, + args: Value, +) -> Result { + match name { + "chorus_join" => { + let host = args + .get("host") + .and_then(|v| v.as_str()) + .ok_or("host is required")? + .to_string(); + let token = args + .get("token") + .and_then(|v| v.as_str()) + .ok_or("token is required")? + .to_string(); + let display_name = args + .get("display_name") + .and_then(|v| v.as_str()) + .map(str::trim) + .filter(|s| !s.is_empty()) + .ok_or("display_name is required and must be non-empty")? + .to_string(); + let repo_remote = args + .get("repo_remote") + .and_then(|v| v.as_str()) + .map(str::trim) + .filter(|s| !s.is_empty()); + let email = args + .get("email") + .and_then(|v| v.as_str()) + .map(str::trim) + .filter(|s| !s.is_empty()); + + let guard = session.lock().await; + if matches!(guard.live, LiveSession::Sharing(_)) { + return Err("stop sharing (chorus_stop) before joining another session".into()); + } + drop(guard); + + let client = + JoinClient::connect(&host, &token, &display_name, repo_remote, email).await?; + let snap = client.snapshot().await; + let mut guard = session.lock().await; + if matches!(guard.live, LiveSession::Sharing(_)) { + client.disconnect().await; + return Err("stop sharing (chorus_stop) before joining another session".into()); + } + guard.live.shutdown().await; + guard.live = LiveSession::Joined(client); + Ok(tool_text(format!( + "Joined Chorus session on {} as {} (status={:?}, users={})", + snap.host, + snap.display_name, + snap.status, + snap.users.len() + ))) + } + "chorus_leave" => { + let mut guard = session.lock().await; + match std::mem::replace(&mut guard.live, LiveSession::Idle) { + LiveSession::Joined(client) => { + client.disconnect().await; + Ok(tool_text("Left Chorus session.".into())) + } + LiveSession::Sharing(host) => { + host.stop().await; + Ok(tool_text("Stopped sharing (use chorus_stop next time).".into())) + } + LiveSession::Idle => Ok(tool_text("Not connected.".into())), + } + } + "chorus_chat" => { + let content = args + .get("content") + .and_then(|v| v.as_str()) + .ok_or("content is required")?; + let guard = session.lock().await; + match &guard.live { + LiveSession::Joined(client) => { + client.send_chat(content)?; + Ok(tool_text(format!("Chat sent: {content}"))) + } + LiveSession::Sharing(host) => { + host.send_chat_named(content).await?; + Ok(tool_text(format!("Host chat sent: {content}"))) + } + LiveSession::Idle => Err("not connected — call chorus_join or chorus_share first".into()), + } + } + "chorus_prompt" => { + let content = args + .get("content") + .and_then(|v| v.as_str()) + .ok_or("content is required")?; + let guard = session.lock().await; + match &guard.live { + LiveSession::Joined(client) => { + client.send_prompt(content)?; + Ok(tool_text(format!("Prompt forwarded to host: {content}"))) + } + LiveSession::Sharing(_) => Err( + "you are hosting — joiners send prompts; use chorus_publish to fan out host/AI lines" + .into(), + ), + LiveSession::Idle => Err("not connected — call chorus_join first".into()), + } + } + "chorus_status" => { + let guard = session.lock().await; + match &guard.live { + LiveSession::Idle => Ok(tool_text( + "status: idle\n(no active Chorus join or share)".into(), + )), + LiveSession::Joined(client) => { + let snap = client.snapshot().await; + Ok(tool_text(format_status(&snap))) + } + LiveSession::Sharing(host) => { + let snap = host.snapshot().await; + Ok(tool_text(format_share_status(&snap))) + } + } + } + "chorus_share" => { + let port = args + .get("port") + .and_then(|v| v.as_u64()) + .map(|n| n as u16) + .unwrap_or(7742); + let role = args + .get("role") + .and_then(|v| v.as_str()) + .unwrap_or("edit"); + let display_name = args + .get("display_name") + .and_then(|v| v.as_str()) + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string) + .unwrap_or_else(|| { + std::env::var("CHORUS_DISPLAY_NAME") + .ok() + .filter(|s| !s.trim().is_empty()) + .or_else(|| std::env::var("USER").ok()) + .unwrap_or_else(|| "Zed".into()) + }); + let require_approval = args + .get("require_approval") + .and_then(|v| v.as_bool()) + .unwrap_or(true); + let repo_remote = args + .get("repo_remote") + .and_then(|v| v.as_str()) + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string); + let allowed_email_domain = args + .get("allowed_email_domain") + .and_then(|v| v.as_str()) + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string); + let public_host = args + .get("public_host") + .and_then(|v| v.as_str()) + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string); + + let host = HostSession::start(ShareOpts { + port, + role: parse_role(role)?, + display_name, + require_approval, + repo_remote, + allowed_email_domain, + public_host, + }) + .await?; + let snap = host.snapshot().await; + let mut guard = session.lock().await; + guard.live.shutdown().await; + guard.live = LiveSession::Sharing(host); + Ok(tool_text(format!( + "Sharing on {} (approval={}). Collaborator command:\n{}", + snap.host, snap.require_approval, snap.join_command + ))) + } + "chorus_stop" => { + let mut guard = session.lock().await; + match std::mem::replace(&mut guard.live, LiveSession::Idle) { + LiveSession::Sharing(host) => { + host.stop().await; + Ok(tool_text("Stopped sharing.".into())) + } + LiveSession::Joined(client) => { + client.disconnect().await; + Ok(tool_text("Left session (was joined, not sharing).".into())) + } + LiveSession::Idle => Ok(tool_text("Not sharing.".into())), + } + } + "chorus_approve" => { + let user_id = args + .get("user_id") + .and_then(|v| v.as_str()) + .ok_or("user_id is required")?; + let guard = session.lock().await; + match &guard.live { + LiveSession::Sharing(host) => { + host.approve(user_id)?; + Ok(tool_text(format!("Approved {user_id}"))) + } + _ => Err("share a session first (chorus_share)".into()), + } + } + "chorus_deny" => { + let user_id = args + .get("user_id") + .and_then(|v| v.as_str()) + .ok_or("user_id is required")?; + let guard = session.lock().await; + match &guard.live { + LiveSession::Sharing(host) => { + host.deny(user_id)?; + Ok(tool_text(format!("Denied {user_id}"))) + } + _ => Err("share a session first (chorus_share)".into()), + } + } + "chorus_publish" => { + let content = args + .get("content") + .and_then(|v| v.as_str()) + .ok_or("content is required")?; + let event_type = args + .get("type") + .and_then(|v| v.as_str()) + .unwrap_or("user"); + let guard = session.lock().await; + match &guard.live { + LiveSession::Sharing(host) => { + host.publish(content, event_type).await?; + Ok(tool_text(format!("Published {event_type} line: {content}"))) + } + _ => Err("share a session first (chorus_share)".into()), + } + } + other => Err(format!("unknown tool: {other}")), + } +} + +fn tool_text(text: String) -> Value { + json!({ + "content": [{ "type": "text", "text": text }], + "isError": false + }) +} + +pub fn format_status(snap: &crate::client::SessionSnapshot) -> String { + let mut out = String::new(); + out.push_str(&format!("status: {:?}\n", snap.status)); + out.push_str(&format!("host: {}\n", snap.host)); + out.push_str(&format!("displayName: {}\n", snap.display_name)); + if let Some(id) = &snap.session_id { + out.push_str(&format!("sessionId: {id}\n")); + } + if let Some(err) = &snap.last_error { + out.push_str(&format!("error: {err}\n")); + } + out.push_str(&format!("users ({})", snap.users.len())); + if snap.users.is_empty() { + out.push('\n'); + } else { + out.push('\n'); + for u in &snap.users { + out.push_str(&format!( + " - {} [{}] {:?} ({:?})\n", + u.display_name, u.user_id, u.role, u.status + )); + } + } + out.push_str(&format!( + "recentEvents: {} | recentChat: {}\n", + snap.recent_events.len(), + snap.recent_chat.len() + )); + if snap.status == JoinStatus::Connected || snap.status == JoinStatus::Pending { + for chat in snap.recent_chat.iter().rev().take(5).collect::>().into_iter().rev() + { + let name = chat.display_name.as_deref().unwrap_or(&chat.user_id); + out.push_str(&format!(" chat <{name}> {}\n", chat.content)); + } + } + out +} + +fn read_message(reader: &mut R) -> Result, String> { + let mut headers = String::new(); + loop { + let mut line = String::new(); + let n = reader + .read_line(&mut line) + .map_err(|e| format!("stdin read: {e}"))?; + if n == 0 { + return Ok(None); + } + if line == "\r\n" || line == "\n" { + break; + } + headers.push_str(&line); + } + + let mut content_length: Option = None; + for header in headers.lines() { + let lower = header.to_ascii_lowercase(); + if let Some(rest) = lower.strip_prefix("content-length:") { + content_length = rest.trim().parse().ok(); + } + } + let len = content_length.ok_or("missing Content-Length")?; + let mut buf = vec![0u8; len]; + reader + .read_exact(&mut buf) + .map_err(|e| format!("body read: {e}"))?; + let value = serde_json::from_slice(&buf).map_err(|e| format!("json: {e}"))?; + Ok(Some(value)) +} + +fn write_message(writer: &mut W, value: &Value) -> Result<(), String> { + let body = serde_json::to_vec(value).map_err(|e| e.to_string())?; + write!(writer, "Content-Length: {}\r\n\r\n", body.len()).map_err(|e| e.to_string())?; + writer.write_all(&body).map_err(|e| e.to_string())?; + writer.flush().map_err(|e| e.to_string())?; + Ok(()) +} diff --git a/crates/chorus-zed-helper/tests/host_session.rs b/crates/chorus-zed-helper/tests/host_session.rs new file mode 100644 index 0000000..ca39055 --- /dev/null +++ b/crates/chorus-zed-helper/tests/host_session.rs @@ -0,0 +1,92 @@ +//! Integration: HostSession share → pending join → approve → collab.input. + +use std::time::Duration; + +use chorus_zed_helper::client::{JoinClient, JoinStatus}; +use chorus_zed_helper::host::{HostSession, ShareOpts, parse_role}; +use chorus_relay::protocol::UserRole; + +#[tokio::test] +async fn share_pending_approve_and_prompt() { + let port = 17944; + let host = HostSession::start(ShareOpts { + port, + role: UserRole::Edit, + display_name: "ZedHost".into(), + require_approval: true, + repo_remote: Some(String::new()), + allowed_email_domain: None, + public_host: Some(format!("127.0.0.1:{port}")), + }) + .await + .expect("share"); + + let snap = host.snapshot().await; + assert!(snap.sharing); + assert!(snap.join_command.contains("token=")); + assert!(snap.join_command.contains(&format!("host=\"127.0.0.1:{port}\""))); + + let token = snap + .join_command + .split("token=\"") + .nth(1) + .and_then(|s| s.split('"').next()) + .expect("token") + .to_string(); + + let joiner = JoinClient::connect( + &format!("127.0.0.1:{port}"), + &token, + "ZedGuest", + None, + None, + ) + .await + .expect("join"); + assert_eq!(joiner.snapshot().await.status, JoinStatus::Pending); + + let mut user_id = String::new(); + for _ in 0..40 { + let pending = host.snapshot().await.pending_users; + if let Some(u) = pending.first() { + user_id = u.user_id.clone(); + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert!(!user_id.is_empty(), "host never saw pending joiner"); + host.approve(&user_id).unwrap(); + + for _ in 0..40 { + if joiner.snapshot().await.status == JoinStatus::Connected { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert_eq!(joiner.snapshot().await.status, JoinStatus::Connected); + + joiner.send_prompt("from-guest").unwrap(); + let mut saw = false; + for _ in 0..40 { + if host + .snapshot() + .await + .recent_inputs + .iter() + .any(|l| l.contains("from-guest")) + { + saw = true; + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert!(saw, "host did not record collab.input"); + + joiner.disconnect().await; + host.stop().await; +} + +#[test] +fn role_helper() { + assert_eq!(parse_role("edit").unwrap(), UserRole::Edit); +} diff --git a/crates/chorus-zed-helper/tests/join_client.rs b/crates/chorus-zed-helper/tests/join_client.rs new file mode 100644 index 0000000..6bc7cf9 --- /dev/null +++ b/crates/chorus-zed-helper/tests/join_client.rs @@ -0,0 +1,127 @@ +//! Integration: JoinClient against a live chorus-relay (auth, chat, collab.input). + +use std::time::Duration; + +use chorus_relay::protocol::{HostToRelay, RelayToHost, UserRole}; +use chorus_relay::server::{serve, RelayConfig}; +use chorus_zed_helper::client::{JoinClient, JoinStatus}; +use futures_util::{SinkExt, StreamExt}; +use tokio_tungstenite::{connect_async, tungstenite::Message}; + +type WsStream = + tokio_tungstenite::WebSocketStream>; + +async fn start_relay(port: u16, host_token: &str) { + let token = host_token.to_string(); + tokio::spawn(async move { + serve(RelayConfig { + port, + host_token: token, + bind: "127.0.0.1".into(), + }) + .await + .expect("relay failed"); + }); + for _ in 0..50 { + if tokio::net::TcpStream::connect(("127.0.0.1", port)) + .await + .is_ok() + { + return; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("relay not ready on {port}"); +} + +async fn connect_host(port: u16, host_token: &str) -> WsStream { + let (mut ws, _) = connect_async(format!("ws://127.0.0.1:{port}/host")) + .await + .expect("host connect"); + ws.send(Message::Text( + serde_json::to_string(&HostToRelay::HostAuth { + token: host_token.into(), + }) + .unwrap() + .into(), + )) + .await + .unwrap(); + let msg = recv_json::(&mut ws).await; + assert!(matches!(msg, RelayToHost::HostReady { .. })); + let _ = tokio::time::timeout(Duration::from_millis(100), ws.next()).await; + ws +} + +async fn recv_json(ws: &mut WsStream) -> T { + loop { + let msg = ws.next().await.expect("ended").expect("ws"); + match msg { + Message::Text(t) => return serde_json::from_str(&t).expect("json"), + Message::Close(frame) => panic!("close: {frame:?}"), + _ => continue, + } + } +} + +async fn issue_token(host: &mut WsStream) -> String { + host.send(Message::Text( + serde_json::to_string(&HostToRelay::TokenIssue { + session_id: "sess-zed".into(), + role: Some(UserRole::Edit), + ttl_ms: None, + }) + .unwrap() + .into(), + )) + .await + .unwrap(); + loop { + let msg = recv_json::(host).await; + if let RelayToHost::TokenIssued { token } = msg { + return token.token; + } + } +} + +#[tokio::test] +async fn join_client_auth_chat_and_prompt() { + let port = 17942; + let host_token = "zed-helper-test-token"; + start_relay(port, host_token).await; + let mut host = connect_host(port, host_token).await; + let token = issue_token(&mut host).await; + + let client = JoinClient::connect(&format!("127.0.0.1:{port}"), &token, "ZedTester", None, None) + .await + .expect("join"); + let snap = client.snapshot().await; + assert_eq!(snap.status, JoinStatus::Connected); + + client.send_chat("hello from zed helper").unwrap(); + // Host should receive chat.message + let mut saw_chat = false; + for _ in 0..20 { + let msg = recv_json::(&mut host).await; + if let RelayToHost::ChatMessage { message } = msg { + assert_eq!(message.content, "hello from zed helper"); + saw_chat = true; + break; + } + } + assert!(saw_chat, "host did not receive chat"); + + client.send_prompt("fix the bug").unwrap(); + let mut saw_prompt = false; + for _ in 0..20 { + let msg = recv_json::(&mut host).await; + if let RelayToHost::CollabInput { content, .. } = msg { + assert_eq!(content, "fix the bug"); + saw_prompt = true; + break; + } + } + assert!(saw_prompt, "host did not receive collab.input"); + + client.disconnect().await; +} diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index e292275..d46f91c 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -49,7 +49,7 @@ Recommended shape: - **v1:** OpenCode adapter = this repo’s plugin + local **Rust** `chorus-relay` - **v1.x:** keep `@chorus/shared` as the stable joiner + host-control contract; `@chorus/client` holds `JoinClient` / `RelayServer` for non-OpenCode hosts -- **v1.x adapters in-monorepo:** VS Code (`packages/vscode`) — same wire protocol, host-specific UI only. +- **v1.x adapters in-monorepo:** VS Code (`packages/vscode`) and Zed (`packages/zed` + `crates/chorus-zed-helper`) — same wire protocol, host-specific UI only. Zed v1 is joiner/MCP-only (no host share). - **v2:** optional further adapters (Codex app-server remains a strong candidate) once OpenCode UX is solid; split adapter packages into separate repos only when publish/CI ownership diverges **Do not** make Claude Code / Amp the primary host — large audiences, but proprietary control planes. Treat them as distribution adapters later, not the core. diff --git a/docs/STATUS.md b/docs/STATUS.md index 54b826a..2f898be 100644 --- a/docs/STATUS.md +++ b/docs/STATUS.md @@ -14,7 +14,7 @@ Chorus is an **OpenCode↔OpenCode** LAN collaboration stack: The browser companion (`packages/web`) was intentionally removed. Joiners now mirror the host transcript into their OpenCode session (`[Host]:` / `[AI]:` lines via `noReply` inject); side-channel chat/typing remain toasts. The in-process Bun relay has been replaced by `crates/chorus-relay`. -**Editor adapters (in progress):** VS Code (`packages/vscode`). +**Editor adapters (in progress):** VS Code (`packages/vscode`) and Zed (`packages/zed` + `crates/chorus-zed-helper`). Zed v1 is **joiner-only** via MCP/CLI — no host share and no OpenCode transcript injection. Differentiation vs nearby OpenCode plugins (`opencode-live`, `opencode-sessions`, `opencode-ensemble`, `opencode-relay`): those target **multi-agent / same-DB sync**. Chorus targets **multi-human** pair programming on one live AI session. @@ -71,6 +71,7 @@ Differentiation vs nearby OpenCode plugins (`opencode-live`, `opencode-sessions` - **OpenCode** (`packages/plugin`) — primary host; full LLM loop + transcript mirror. - **VS Code** (`packages/vscode`) — share/join via `@chorus/client`; sidebar transcript; joiner `collab.input` works against an OpenCode host. VS Code share does **not** drive OpenCode’s model — publish host lines manually or pair with OpenCode. +- **Zed** — joiner-first scaffold when present under `packages/zed` / related crates (WASM extension limits; helper binary pattern). ## Explicit non-goals (for now) diff --git a/package.json b/package.json index d8de85d..667d172 100644 --- a/package.json +++ b/package.json @@ -7,15 +7,17 @@ ], "scripts": { "build:relay": "cargo build -p chorus-relay --release", + "build:zed-helper": "cargo build -p chorus-zed-helper --release", "build:shared": "bun run --filter @chorus/shared build", "build:client": "bun run --filter @chorus/client build", "build:plugin": "bun run --filter @chorus/plugin build", "build:vscode": "bun run --filter chorus build", "build:ts": "bun run build:shared && bun run build:client && bun run build:plugin && bun run build:vscode", - "build": "bun run build:relay && bun run build:ts", + "build": "bun run build:relay && bun run build:zed-helper && bun run build:ts", "test:relay": "cargo test -p chorus-relay", + "test:zed-helper": "cargo test -p chorus-zed-helper", "test:ts": "bun run build:shared && bun run build:client && bun run --filter @chorus/shared test && bun run --filter @chorus/client test && bun run --filter @chorus/plugin test && bun run --filter chorus test", - "test": "bun run build:relay && bun run test:relay && bun run test:ts", + "test": "bun run build:relay && bun run test:relay && bun run build:zed-helper && bun run test:zed-helper && bun run test:ts", "typecheck": "bun run build:shared && bun run build:client && bun run --filter @chorus/shared typecheck && bun run --filter @chorus/client typecheck && bun run --filter @chorus/plugin typecheck && bun run --filter chorus typecheck", "clean": "bun run --filter @chorus/shared clean && bun run --filter @chorus/client clean && bun run --filter @chorus/plugin clean && bun run --filter chorus clean && cargo clean", "multi-agent": "bun scripts/multi-agent.ts", @@ -24,8 +26,9 @@ "test:relay-stress": "bun scripts/multi-agent.ts relay-stress --agents 5", "test:security-e2e": "bun scripts/security-e2e.ts", "test:vscode-e2e": "bun scripts/vscode-e2e.ts", + "test:zed-e2e": "bun scripts/zed-e2e.ts", "test:plugin-e2e": "bun run test:multi-agent && bun run test:security-e2e", - "test:adapters-e2e": "bun run test:vscode-e2e", + "test:adapters-e2e": "bun run test:vscode-e2e && bun run test:zed-e2e", "docker-agents": "bun scripts/docker-agents.ts", "test:docker-agents": "bun scripts/docker-agents.ts up --agents 2 && bun scripts/docker-agents.ts smoke && bun scripts/docker-agents.ts pair && bun scripts/docker-agents.ts down" }, diff --git a/packages/zed/Cargo.lock b/packages/zed/Cargo.lock new file mode 100644 index 0000000..7aec8ee --- /dev/null +++ b/packages/zed/Cargo.lock @@ -0,0 +1,829 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + +[[package]] +name = "anyhow" +version = "1.0.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" + +[[package]] +name = "auditable-serde" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c7bf8143dfc3c0258df908843e169b5cc5fcf76c7718bd66135ef4a9cd558c5" +dependencies = [ + "semver", + "serde", + "serde_json", + "topological-sort", +] + +[[package]] +name = "bitflags" +version = "2.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "chorus-zed" +version = "0.1.0" +dependencies = [ + "zed_extension_api", +] + +[[package]] +name = "crc32fast" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "displaydoc" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "equivalent" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + +[[package]] +name = "flate2" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" +dependencies = [ + "crc32fast", + "miniz_oxide", +] + +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + +[[package]] +name = "futures" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-channel" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" +dependencies = [ + "futures-core", + "futures-sink", +] + +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-executor" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" + +[[package]] +name = "futures-macro" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + +[[package]] +name = "futures-task" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" + +[[package]] +name = "futures-util" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" +dependencies = [ + "futures-channel", + "futures-core", + "futures-io", + "futures-macro", + "futures-sink", + "futures-task", + "memchr", + "pin-project-lite", + "slab", +] + +[[package]] +name = "hashbrown" +version = "0.15.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +dependencies = [ + "foldhash", +] + +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" + +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + +[[package]] +name = "icu_collections" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa68d21081c4a05d5a901a1c62add574c77048b6a1c67be3b50ce0b60d4ca513" +dependencies = [ + "displaydoc", + "potential_utf", + "utf8_iter", + "yoke", + "zerofrom", + "zerovec", +] + +[[package]] +name = "icu_locale_core" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d56e28588da92eee5c3201a6eff33fabdd49b62269c8938d4ff050ce4d900deb" +dependencies = [ + "displaydoc", + "litemap", + "tinystr", + "writeable", + "zerovec", +] + +[[package]] +name = "icu_normalizer" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12f9cf5f235641ed274641dd81c3f28d870e276763d0797aeeab72317b1c646f" +dependencies = [ + "icu_collections", + "icu_normalizer_data", + "icu_properties", + "icu_provider", + "smallvec", + "zerovec", +] + +[[package]] +name = "icu_normalizer_data" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1563da1ed3e0b3bf3d74c9b85917ac9c56464d2f57242270c09c9e752f8021a0" + +[[package]] +name = "icu_properties" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e7ca276ad3145661a65914e6daf131ca5120cd3dcee8f8f3214b8875184a148" +dependencies = [ + "displaydoc", + "icu_collections", + "icu_locale_core", + "icu_properties_data", + "icu_provider", + "zerotrie", + "zerovec", +] + +[[package]] +name = "icu_properties_data" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e590f038c1464a96894fd6d10127e90a8be4509f56ff7ecef851b15cee0b7caa" + +[[package]] +name = "icu_provider" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92a7ed671a6aad807a8651a2e1782a6598fda9ce5185dd8158549e95a91c6428" +dependencies = [ + "displaydoc", + "icu_locale_core", + "writeable", + "yoke", + "zerofrom", + "zerotrie", + "zerovec", +] + +[[package]] +name = "id-arena" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d3067d79b975e8844ca9eb072e16b31c3c1c36928edf9c6789548c524d0d954" + +[[package]] +name = "idna" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de" +dependencies = [ + "idna_adapter", + "smallvec", + "utf8_iter", +] + +[[package]] +name = "idna_adapter" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714" +dependencies = [ + "icu_normalizer", + "icu_properties", +] + +[[package]] +name = "indexmap" +version = "2.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" +dependencies = [ + "equivalent", + "hashbrown 0.17.1", + "serde", + "serde_core", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "leb128fmt" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" + +[[package]] +name = "litemap" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" + +[[package]] +name = "log" +version = "0.4.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "miniz_oxide" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316" +dependencies = [ + "adler2", + "simd-adler32", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "potential_utf" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661" +dependencies = [ + "zerovec", +] + +[[package]] +name = "prettyplease" +version = "0.2.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" +dependencies = [ + "proc-macro2", + "syn 2.0.119", +] + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "semver" +version = "1.0.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" +dependencies = [ + "serde", + "serde_core", +] + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "simd-adler32" +version = "0.3.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "smallvec" +version = "1.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" + +[[package]] +name = "spdx" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3e17e880bafaeb362a7b751ec46bdc5b61445a188f80e0606e68167cd540fa3" +dependencies = [ + "smallvec", +] + +[[package]] +name = "stable_deref_trait" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "synstructure" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "tinystr" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1e27c91459209c2986af3dcf603a5a74a4368754ce37414f59acc971167f643" +dependencies = [ + "displaydoc", + "zerovec", +] + +[[package]] +name = "topological-sort" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea68304e134ecd095ac6c3574494fc62b909f416c4fca77e440530221e549d3d" + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "unicode-xid" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" + +[[package]] +name = "url" +version = "2.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed" +dependencies = [ + "form_urlencoded", + "idna", + "percent-encoding", + "serde", +] + +[[package]] +name = "utf8_iter" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" + +[[package]] +name = "wasm-encoder" +version = "0.227.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "80bb72f02e7fbf07183443b27b0f3d4144abf8c114189f2e088ed95b696a7822" +dependencies = [ + "leb128fmt", + "wasmparser", +] + +[[package]] +name = "wasm-metadata" +version = "0.227.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce1ef0faabbbba6674e97a56bee857ccddf942785a336c8b47b42373c922a91d" +dependencies = [ + "anyhow", + "auditable-serde", + "flate2", + "indexmap", + "serde", + "serde_derive", + "serde_json", + "spdx", + "url", + "wasm-encoder", + "wasmparser", +] + +[[package]] +name = "wasmparser" +version = "0.227.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0f51cad774fb3c9461ab9bccc9c62dfb7388397b5deda31bf40e8108ccd678b2" +dependencies = [ + "bitflags", + "hashbrown 0.15.5", + "indexmap", + "semver", +] + +[[package]] +name = "wit-bindgen" +version = "0.41.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10fb6648689b3929d56bbc7eb1acf70c9a42a29eb5358c67c10f54dbd5d695de" +dependencies = [ + "wit-bindgen-rt", + "wit-bindgen-rust-macro", +] + +[[package]] +name = "wit-bindgen-core" +version = "0.41.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92fa781d4f2ff6d3f27f3cc9b74a73327b31ca0dc4a3ef25a0ce2983e0e5af9b" +dependencies = [ + "anyhow", + "heck", + "wit-parser", +] + +[[package]] +name = "wit-bindgen-rt" +version = "0.41.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db52a11d4dfb0a59f194c064055794ee6564eb1ced88c25da2cf76e50c5621" +dependencies = [ + "bitflags", + "futures", + "once_cell", +] + +[[package]] +name = "wit-bindgen-rust" +version = "0.41.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d0809dc5ba19e2e98661bf32fc0addc5a3ca5bf3a6a7083aa6ba484085ff3ce" +dependencies = [ + "anyhow", + "heck", + "indexmap", + "prettyplease", + "syn 2.0.119", + "wasm-metadata", + "wit-bindgen-core", + "wit-component", +] + +[[package]] +name = "wit-bindgen-rust-macro" +version = "0.41.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ad19eec017904e04c60719592a803ee5da76cb51c81e3f6fbf9457f59db49799" +dependencies = [ + "anyhow", + "prettyplease", + "proc-macro2", + "quote", + "syn 2.0.119", + "wit-bindgen-core", + "wit-bindgen-rust", +] + +[[package]] +name = "wit-component" +version = "0.227.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "635c3adc595422cbf2341a17fb73a319669cc8d33deed3a48368a841df86b676" +dependencies = [ + "anyhow", + "bitflags", + "indexmap", + "log", + "serde", + "serde_derive", + "serde_json", + "wasm-encoder", + "wasm-metadata", + "wasmparser", + "wit-parser", +] + +[[package]] +name = "wit-parser" +version = "0.227.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddf445ed5157046e4baf56f9138c124a0824d4d1657e7204d71886ad8ce2fc11" +dependencies = [ + "anyhow", + "id-arena", + "indexmap", + "log", + "semver", + "serde", + "serde_derive", + "serde_json", + "unicode-xid", + "wasmparser", +] + +[[package]] +name = "writeable" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" + +[[package]] +name = "yoke" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "709fe23a0424b6a435d82152b1bd3fdfb0833487d5fa90d05d42762a9891fef5" +dependencies = [ + "stable_deref_trait", + "yoke-derive", + "zerofrom", +] + +[[package]] +name = "yoke-derive" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de844c262c8848816172cef550288e7dc6c7b7814b4ee56b3e1553f275f1858e" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "synstructure", +] + +[[package]] +name = "zed_extension_api" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0729d50b4ca0a7e28e590bbe32e3ca0194d97ef654961451a424c661a366fca0" +dependencies = [ + "serde", + "serde_json", + "wit-bindgen", +] + +[[package]] +name = "zerofrom" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272" +dependencies = [ + "zerofrom-derive", +] + +[[package]] +name = "zerofrom-derive" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11532158c46691caf0f2593ea8358fed6bbf68a0315e80aae9bd41fbade684a1" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "synstructure", +] + +[[package]] +name = "zerotrie" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f" +dependencies = [ + "displaydoc", + "yoke", + "zerofrom", +] + +[[package]] +name = "zerovec" +version = "0.11.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94b5c6b5976d66c1d703c4fd17d3f5e43c8cedaacf604961b171adc7130896d8" +dependencies = [ + "yoke", + "zerofrom", + "zerovec-derive", +] + +[[package]] +name = "zerovec-derive" +version = "0.11.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47402523226a02bfe5230160dc3ccc089aa6f6f19e7fcbb4e6f824bbb1b4aa62" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/packages/zed/Cargo.toml b/packages/zed/Cargo.toml new file mode 100644 index 0000000..4db7fa4 --- /dev/null +++ b/packages/zed/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "chorus-zed" +version = "0.1.0" +edition = "2021" +license = "MIT" +publish = false +description = "Zed extension that launches the Chorus join helper as an MCP context server" + +# Standalone from the monorepo Cargo workspace (wasm32-wasip2 via Zed). +[workspace] + +[lib] +crate-type = ["cdylib"] +path = "src/lib.rs" + +[dependencies] +zed_extension_api = "0.7.0" diff --git a/packages/zed/README.md b/packages/zed/README.md new file mode 100644 index 0000000..a3c45c6 --- /dev/null +++ b/packages/zed/README.md @@ -0,0 +1,116 @@ +# Chorus Zed extension (v1 — join + share) + +Join or **host** a Chorus collaborative session from [Zed](https://zed.dev) via a **native helper** + **MCP context server**. + +Zed extensions run as `wasm32-wasip2` and cannot hold long-lived WebSocket UIs themselves. This package is the thin WASM adapter; `crates/chorus-zed-helper` speaks the Chorus `/ws` (join) and `/host` (share) protocols. + +## What works (v1) + +| Capability | How | +|---|---| +| Host / share | MCP `chorus_share` or CLI `chorus-zed-helper share` | +| Approve / deny joiners | `chorus_approve` / `chorus_deny` | +| Stop sharing | `chorus_stop` / `stop` | +| Publish host/AI lines | `chorus_publish` / `publish` | +| Join a host’s relay | MCP tool `chorus_join` or CLI `chorus-zed-helper join` | +| Leave | `chorus_leave` / `leave` | +| Side-channel chat | `chorus_chat` / `chat` | +| Forward prompts (`collab.input`) | `chorus_prompt` / `prompt` (needs `edit` role; joiner only) | +| Status / recent events | `chorus_status` / `status` | + +## What does **not** work (honest limits) + +- **OpenCode LLM loop** — Zed/VS Code share does not inject into OpenCode. Joiner prompts show up in `chorus_status` (and you can `chorus_publish` host/AI lines). Pair with the OpenCode plugin if you want the model to run those prompts automatically. +- **OpenCode transcript mirroring** — session events are visible via `chorus_status` / helper state, not injected into a local agent transcript. +- **Extension slash commands** — Zed removed extension slash commands; use MCP tools (or the CLI). + +## Access control + +Aligned with the OpenCode plugin / `chorus-relay` session gates: + +- **Required display name** (`display_name` / `--name`) +- **Host approval** — `chorus_share` defaults to `require_approval=true`; joiners stay `pending` until `chorus_approve` +- **Optional repo gate** — pass `repo_remote` / `--repo-remote` (defaults to `git remote get-url origin`) +- **Optional email domain** — `allowed_email_domain` + +## Architecture + +``` +Zed Agent Panel + └─ MCP (stdio) ──► chorus-zed-helper mcp + ├─ share: in-process chorus-relay + /host + └─ join: WebSocket ──► chorus-relay /ws +``` + +CLI subcommands talk to a short-lived background daemon over a Unix socket so `share` / `join` / `chat` / `status` share one connection. MCP holds its own in-process session (Agent Panel). + +## Install the native helper + +From the monorepo root: + +```sh +cargo install --path crates/chorus-zed-helper --force +``` + +Verify: + +```sh +chorus-zed-helper --help +``` + +If a previous daemon is running, `chorus-zed-helper shutdown` first so it picks up the new binary. + +## Install Dev Extension in Zed + +1. Install `chorus-zed-helper` on `PATH` (above). +2. Install the `wasm32-wasip2` target if needed: `rustup target add wasm32-wasip2`. +3. In Zed: command palette → **zed: extensions**. +4. Click **Install Dev Extension**. +5. Select this directory: `packages/zed` (the folder that contains `extension.toml`). +6. Enable the **Chorus** context server in Agent / MCP settings if it is not auto-enabled. +7. Confirm tools `chorus_share`, `chorus_join`, `chorus_approve`, `chorus_status`, … appear. + +If the context server fails to start, register a manual context server in `settings.json`: + +```json +{ + "context_servers": { + "chorus": { + "command": "/absolute/path/to/chorus-zed-helper", + "args": ["mcp"] + } + } +} +``` + +## CLI usage + +```sh +# Host from this machine +chorus-zed-helper share --name Steven +chorus-zed-helper status # includes join command + pending users +chorus-zed-helper approve +chorus-zed-helper publish "looking at the failing test" +chorus-zed-helper stop + +# Join someone else's share +chorus-zed-helper join --host 192.168.1.10:7742 --token --name Alex +chorus-zed-helper prompt "fix the flaky auth test" +chorus-zed-helper leave +``` + +`share` defaults to host approval. Pass `--auto-admit` to skip it. + +## Develop / test + +```sh +# Helper unit + integration tests (in-process chorus-relay) +cargo test -p chorus-zed-helper + +# Extension WASM (optional local check; Zed also builds on Install Dev Extension) +cd packages/zed +rustup target add wasm32-wasip2 +cargo build --target wasm32-wasip2 +``` + +Protocol fixtures live in `/protocol`; Rust types are shared via `chorus-relay::protocol`. diff --git a/packages/zed/extension.toml b/packages/zed/extension.toml new file mode 100644 index 0000000..7218b35 --- /dev/null +++ b/packages/zed/extension.toml @@ -0,0 +1,11 @@ +id = "chorus" +name = "Chorus" +description = "Join or host Chorus collaborative AI sessions from Zed (native helper + MCP)" +version = "0.1.0" +schema_version = 1 +authors = ["Steven Roomberg"] +repository = "https://github.com/sroomberg/chorus" + +[context_servers.chorus] +name = "Chorus" +description = "Share or join Chorus sessions, approve joiners, chat, and collab prompts" diff --git a/packages/zed/src/lib.rs b/packages/zed/src/lib.rs new file mode 100644 index 0000000..dcb7d0d --- /dev/null +++ b/packages/zed/src/lib.rs @@ -0,0 +1,27 @@ +use zed_extension_api::{self as zed, Command, ContextServerId, Project, Result}; + +struct ChorusExtension; + +impl zed::Extension for ChorusExtension { + fn new() -> Self { + Self + } + + fn context_server_command( + &mut self, + _context_server_id: &ContextServerId, + _project: &Project, + ) -> Result { + // The WebSocket join client lives in the native `chorus-zed-helper` binary. + // Install it on PATH (`cargo install --path crates/chorus-zed-helper` from the repo, + // or `cargo build -p chorus-zed-helper --release` and copy `target/release/chorus-zed-helper`). + // Override the binary path via Zed settings `context_servers.chorus` env if needed. + Ok(Command { + command: "chorus-zed-helper".into(), + args: vec!["mcp".into()], + env: vec![], + }) + } +} + +zed::register_extension!(ChorusExtension); diff --git a/scripts/zed-e2e.ts b/scripts/zed-e2e.ts new file mode 100644 index 0000000..28e8d49 --- /dev/null +++ b/scripts/zed-e2e.ts @@ -0,0 +1,128 @@ +#!/usr/bin/env bun +/** + * Zed helper integration against a live chorus-relay: + * join (pending) → approve → chat + collab.input → leave. + */ +import { spawn } from "node:child_process"; +import { existsSync } from "node:fs"; +import { join, resolve, dirname } from "node:path"; +import { fileURLToPath } from "node:url"; +import { RelayServer } from "../packages/client/src/index.ts"; + +const REPO = resolve(dirname(fileURLToPath(import.meta.url)), ".."); +const PORT = parseInt(process.env["CHORUS_ZED_E2E_PORT"] ?? "18743", 10); +const RUNTIME = join(REPO, ".multi-agent", "zed-e2e-runtime"); + +function helperBin(): string { + if (process.env["CHORUS_ZED_HELPER_BIN"]) return process.env["CHORUS_ZED_HELPER_BIN"]; + for (const p of [ + join(REPO, "target/release/chorus-zed-helper"), + join(REPO, "target/debug/chorus-zed-helper"), + ]) { + if (existsSync(p)) return p; + } + throw new Error("chorus-zed-helper not found; cargo build -p chorus-zed-helper --release"); +} + +function runHelper(args: string[]): Promise<{ code: number; stdout: string; stderr: string }> { + return new Promise((resolveP, reject) => { + const child = spawn(helperBin(), args, { + env: { + ...process.env, + CHORUS_ZED_RUNTIME_DIR: RUNTIME, + }, + stdio: ["ignore", "pipe", "pipe"], + }); + let stdout = ""; + let stderr = ""; + child.stdout.on("data", (d) => { + stdout += String(d); + }); + child.stderr.on("data", (d) => { + stderr += String(d); + }); + child.on("error", reject); + child.on("close", (code) => resolveP({ code: code ?? 1, stdout, stderr })); + }); +} + +async function waitUntil(fn: () => boolean, label: string, ms = 8000): Promise { + const deadline = Date.now() + ms; + while (Date.now() < deadline) { + if (fn()) return; + await new Promise((r) => setTimeout(r, 50)); + } + throw new Error(`timeout: ${label}`); +} + +async function main() { + const relay = new RelayServer(PORT); + await relay.start(); + console.log(`✓ relay on :${PORT}`); + + const received: string[] = []; + const pending: string[] = []; + relay.setInputHandler(async (content) => { + received.push(content); + }); + relay.setUserPendingHandler((u) => pending.push(u.userId)); + relay.setSessionPolicy({ requireApproval: true }); + + try { + const token = (await relay.issueToken("zed-e2e", "edit")).token; + const host = `127.0.0.1:${PORT}`; + + console.log("→ chorus-zed-helper join (pending)"); + const joined = await runHelper(["join", "--host", host, "--token", token, "--name", "ZedE2E"]); + if (joined.code !== 0) { + throw new Error(`join failed: ${joined.stderr || joined.stdout}`); + } + console.log(joined.stdout.slice(0, 400)); + await waitUntil(() => pending.length > 0, "host pending"); + + const statusPending = await runHelper(["status"]); + if (!/Pending|pending/i.test(`${statusPending.stdout}${statusPending.stderr}`)) { + console.log(statusPending.stdout); + // Snapshot JSON may serialize as "pending" + if (!/"status"\s*:\s*"pending"/i.test(statusPending.stdout) && !/Pending/.test(statusPending.stdout)) { + throw new Error(`expected pending status, got:\n${statusPending.stdout}`); + } + } + console.log("✓ helper pending"); + + console.log("→ approve"); + relay.approveUser(pending[0]!); + await waitUntil(async () => { + /* poll via helper */ + return true; + }, "noop", 200); + for (let i = 0; i < 40; i++) { + const st = await runHelper(["status"]); + if (/Connected|connected/i.test(st.stdout) && !/Pending/.test(st.stdout)) break; + if (/"status"\s*:\s*"connected"/i.test(st.stdout)) break; + await new Promise((r) => setTimeout(r, 100)); + if (i === 39) throw new Error(`never connected:\n${st.stdout}`); + } + console.log("✓ helper admitted"); + + const prompt = await runHelper(["prompt", "from-zed-helper"]); + if (prompt.code !== 0) throw new Error(`prompt failed: ${prompt.stderr || prompt.stdout}`); + await waitUntil(() => received.includes("from-zed-helper"), "collab.input"); + console.log("✓ collab.input"); + + const chat = await runHelper(["chat", "zed-side-channel"]); + if (chat.code !== 0) throw new Error(`chat failed: ${chat.stderr || chat.stdout}`); + console.log("✓ chat"); + + await runHelper(["leave"]); + console.log("\nZed helper e2e passed."); + } finally { + await runHelper(["leave"]).catch(() => {}); + relay.stop(); + } +} + +main().catch((e) => { + console.error(e); + process.exit(1); +});