diff --git a/Cargo.lock b/Cargo.lock index ff03f048..d6f08232 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2238,6 +2238,7 @@ dependencies = [ "tauri-plugin-window-state", "ureq", "url", + "uuid", "windows-sys 0.61.2", ] diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 24f74f07..4e06d4cc 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -20,6 +20,7 @@ tauri = { version = "2", features = ["protocol-asset", "macos-private-api"] } tauri-plugin-opener = "2" serde.workspace = true serde_json.workspace = true +uuid = { version = "1", features = ["v4"] } base64 = "0.22" tauri-plugin-dialog = "2" rusqlite = { version = "0.40.2", features = ["bundled"], default-features = false } diff --git a/src-tauri/src/control.rs b/src-tauri/src/control.rs new file mode 100644 index 00000000..20542e53 --- /dev/null +++ b/src-tauri/src/control.rs @@ -0,0 +1,552 @@ +//! Authenticated loopback transport. App windows own execution; callers never +//! receive arbitrary Tauri command access or direct database write access. +use std::collections::HashMap; +use std::io::{BufRead, BufReader, Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::path::{Component, Path}; +use std::process::Command; +use std::sync::{mpsc, Arc, Mutex}; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use tauri::{AppHandle, Emitter, Manager, State, WebviewWindow}; + +#[derive(Clone)] +struct Grant { + window: String, + session: String, + cwd: String, + token: String, +} +struct Pending { + window: String, + reply: mpsc::Sender, +} +struct ActiveTurn { + window: String, + cwd: String, +} +#[derive(Default)] +struct Inner { + grants: HashMap, + pending: HashMap, + workers: HashMap, + active: HashMap, +} +impl Inner { + fn window_sessions(&self, label: &str) -> Vec { + let leads: Vec = self + .grants + .values() + .filter(|grant| grant.window == label) + .map(|grant| grant.session.clone()) + .collect(); + let mut ids = leads.clone(); + ids.extend( + self.workers + .iter() + .filter(|(_, lead)| leads.contains(lead)) + .map(|(id, _)| id.clone()), + ); + ids.extend( + self.active + .iter() + .filter(|(_, turn)| turn.window == label) + .map(|(id, _)| id.clone()), + ); + ids.sort(); + ids.dedup(); + ids + } + fn close_window(&mut self, label: &str) -> Vec { + let ids = self.window_sessions(label); + self.grants.retain(|id, _| !ids.contains(id)); + self.workers.retain(|id, _| !ids.contains(id)); + self.active.retain(|id, _| !ids.contains(id)); + self.pending.retain(|_, pending| { + if pending.window != label { + return true; + } + let _ = pending + .reply + .send(json!({"ok":false,"error":"MonoCode window closed"})); + false + }); + ids + } +} +pub struct ControlHost { + endpoint: String, + inner: Arc>, +} + +fn paths_overlap(a: &str, b: &str) -> bool { + a == b || a.starts_with(&format!("{b}/")) || b.starts_with(&format!("{a}/")) +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct Request { + token: String, + action: String, + input: Value, + request_id: String, +} + +#[derive(Clone, Serialize)] +#[serde(rename_all = "camelCase")] +struct Event { + id: String, + session_id: String, + request_id: String, + action: String, + input: Value, +} + +pub fn init(app: &AppHandle) -> Result<(), String> { + let listener = TcpListener::bind("127.0.0.1:0").map_err(|e| e.to_string())?; + let endpoint = listener + .local_addr() + .map_err(|e| e.to_string())? + .to_string(); + let inner = Arc::new(Mutex::new(Inner::default())); + app.manage(ControlHost { + endpoint, + inner: inner.clone(), + }); + let app = app.clone(); + std::thread::spawn(move || { + // Limit concurrent readers, including unauthenticated sockets. + let (tx, rx) = mpsc::sync_channel::(32); + let rx = Arc::new(Mutex::new(rx)); + for _ in 0..8 { + let rx = rx.clone(); + let app = app.clone(); + let inner = inner.clone(); + std::thread::spawn(move || loop { + let stream = match rx.lock() { + Ok(rx) => rx.recv(), + Err(_) => return, + }; + let Ok(stream) = stream else { return }; + serve(stream, &app, &inner); + }); + } + for stream in listener.incoming().flatten() { + let _ = tx.try_send(stream); + } + }); + Ok(()) +} + +fn serve(mut stream: TcpStream, app: &AppHandle, inner: &Arc>) { + let _ = stream.set_read_timeout(Some(Duration::from_secs(3))); + let _ = stream.set_write_timeout(Some(Duration::from_secs(3))); + let result = (|| -> Result { + let mut raw = String::new(); + BufReader::new(&mut stream) + .take(262_145) + .read_line(&mut raw) + .map_err(|e| e.to_string())?; + if raw.len() > 262_144 { + return Err("Request exceeds 256 KiB".into()); + } + let request: Request = serde_json::from_str(&raw).map_err(|_| "Invalid control request")?; + if !request.input.is_object() + || request.request_id.is_empty() + || request.request_id.len() > 128 + { + return Err("Invalid input or request ID".into()); + } + let id = uuid::Uuid::new_v4().to_string(); + let (tx, rx) = mpsc::channel(); + let grant = { + let mut host = inner.lock().map_err(|_| "Control service unavailable")?; + let grant = host + .grants + .values() + .find(|g| g.token == request.token) + .cloned() + .ok_or("Connection revoked or unauthorized")?; + if host.pending.len() >= 24 { + return Err("Too many pending control requests".into()); + } + host.pending.insert( + id.clone(), + Pending { + window: grant.window.clone(), + reply: tx, + }, + ); + grant + }; + let event = Event { + id: id.clone(), + session_id: grant.session, + request_id: request.request_id, + action: request.action, + input: request.input, + }; + let delivered = app.emit_to(grant.window.as_str(), "monocode-control-request", event); + let result = if delivered.is_err() { + Err("MonoCode executor is unavailable".into()) + } else { + rx.recv_timeout(Duration::from_secs(35)) + .map_err(|_| "Control request timed out. Retry with the same request ID.".into()) + }; + if let Ok(mut host) = inner.lock() { + host.pending.remove(&id); + } + result + })(); + let response = result.unwrap_or_else(|error| json!({"ok": false, "error": error})); + let _ = writeln!(stream, "{response}"); +} + +#[tauri::command] +pub fn control_enable( + window: WebviewWindow, + host: State<'_, ControlHost>, + session_id: String, + cwd: String, +) -> Result { + let cwd = std::fs::canonicalize(crate::fs::expand_home(&cwd)).map_err(|e| e.to_string())?; + if !cwd.is_dir() { + return Err("Choose a project folder first".into()); + } + let cwd = cwd.to_string_lossy().replace('\\', "/").to_lowercase(); + let mut inner = host + .inner + .lock() + .map_err(|_| "Control service unavailable")?; + if inner + .active + .iter() + .any(|(id, turn)| id != &session_id && paths_overlap(&turn.cwd, &cwd)) + { + return Err( + "Another session is running in this checkout. Stop it before enabling orchestration." + .into(), + ); + } + if inner.grants.values().any(|g| { + paths_overlap(&g.cwd, &cwd) && (g.session != session_id || g.window != window.label()) + }) { + return Err("This checkout already has an orchestrator in another session".into()); + } + inner.grants.insert( + session_id.clone(), + Grant { + window: window.label().into(), + session: session_id, + cwd, + token: format!( + "{}{}", + uuid::Uuid::new_v4().simple(), + uuid::Uuid::new_v4().simple() + ), + }, + ); + let executable = std::env::current_exe().map_err(|e| e.to_string())?; + Ok(executable.to_string_lossy().into_owned()) +} + +#[tauri::command] +pub fn control_disable( + window: WebviewWindow, + host: State<'_, ControlHost>, + session_id: String, +) -> Result<(), String> { + let mut inner = host + .inner + .lock() + .map_err(|_| "Control service unavailable")?; + if inner + .grants + .get(&session_id) + .is_some_and(|g| g.window == window.label()) + { + inner.grants.remove(&session_id); + inner.workers.retain(|_, parent| parent != &session_id); + } + Ok(()) +} + +#[tauri::command] +pub fn control_attach_worker( + window: WebviewWindow, + host: State<'_, ControlHost>, + lead_id: String, + session_id: String, +) -> Result<(), String> { + let mut inner = host + .inner + .lock() + .map_err(|_| "Control service unavailable")?; + if !inner + .grants + .get(&lead_id) + .is_some_and(|grant| grant.window == window.label()) + { + return Err("Lead connection is inactive".into()); + } + inner.workers.insert(session_id, lead_id); + Ok(()) +} + +#[tauri::command] +pub fn control_authorize_turn( + window: WebviewWindow, + host: State<'_, ControlHost>, + session_id: String, + cwd: String, +) -> Result<(), String> { + let cwd = std::fs::canonicalize(crate::fs::expand_home(&cwd)) + .map_err(|e| e.to_string())? + .to_string_lossy() + .replace('\\', "/") + .to_lowercase(); + let mut inner = host + .inner + .lock() + .map_err(|_| "Control service unavailable")?; + if let Some(lead) = inner + .grants + .values() + .find(|grant| paths_overlap(&grant.cwd, &cwd)) + { + if lead.window != window.label() + || (lead.session != session_id && inner.workers.get(&session_id) != Some(&lead.session)) + { + return Err("This checkout is controlled by an orchestrator. Stop that run before starting independent work.".into()); + } + } + inner.active.insert( + session_id, + ActiveTurn { + window: window.label().to_string(), + cwd, + }, + ); + Ok(()) +} + +#[tauri::command] +pub fn control_turn_finished(host: State<'_, ControlHost>, session_id: String) { + if let Ok(mut inner) = host.inner.lock() { + inner.active.remove(&session_id); + } +} + +pub fn window_closed(app: &AppHandle, label: &str) { + let host = app.state::(); + let ids = { + let Ok(inner) = host.inner.lock() else { return }; + inner.window_sessions(label) + }; + for id in &ids { + let _ = crate::harness::harness_kill(app.state(), id.clone()); + } + if let Ok(mut inner) = host.inner.lock() { + inner.close_window(label); + }; +} + +pub fn configure_child(app: &AppHandle, session_id: &str, cmd: &mut Command) { + cmd.env_remove("MONOCODE_CONTROL_ENDPOINT") + .env_remove("MONOCODE_CONTROL_TOKEN"); + let Some(host) = app.try_state::() else { + return; + }; + if let Ok(inner) = host.inner.lock() { + if let Some(grant) = inner.grants.get(session_id) { + cmd.env("MONOCODE_CONTROL_ENDPOINT", &host.endpoint) + .env("MONOCODE_CONTROL_TOKEN", &grant.token); + } + }; +} + +#[tauri::command] +pub fn control_reply( + window: WebviewWindow, + host: State<'_, ControlHost>, + id: String, + response: Value, +) -> Result<(), String> { + let mut inner = host + .inner + .lock() + .map_err(|_| "Control service unavailable")?; + if inner + .pending + .get(&id) + .is_some_and(|p| p.window == window.label()) + { + if let Some(pending) = inner.pending.remove(&id) { + let _ = pending.reply.send(response); + } + } + Ok(()) +} + +#[tauri::command] +pub fn control_save( + store: State<'_, crate::session_store::SessionStore>, + lead_id: String, + state: String, +) -> Result<(), String> { + if state.len() > 8_000_000 { + return Err("Orchestration history is too large".into()); + } + let run: Value = serde_json::from_str(&state).map_err(|_| "Invalid run state")?; + let conn = store.lock_conn()?; + crate::session_store::save_orchestration(&conn, &lead_id, &run).map_err(|e| e.to_string())?; + Ok(()) +} + +#[tauri::command] +pub fn control_load( + store: State<'_, crate::session_store::SessionStore>, + lead_id: String, +) -> Result, String> { + use rusqlite::OptionalExtension; + store + .lock_conn()? + .query_row( + "SELECT state FROM orchestration_runs WHERE lead_id=?1", + [lead_id], + |r| r.get(0), + ) + .optional() + .map_err(|e| e.to_string()) +} + +fn resolve_scope(root: &Path, value: &str) -> Result { + let path = Path::new(value); + if value.is_empty() + || path.is_absolute() + || path + .components() + .any(|c| matches!(c, Component::ParentDir | Component::Prefix(_))) + { + return Err("Write scopes must be project-relative paths without '..'".into()); + } + let root = std::fs::canonicalize(root).map_err(|e| e.to_string())?; + let mut existing = root.join(path); + let mut missing = Vec::new(); + while !existing.exists() { + missing.push(existing.file_name().ok_or("Invalid scope")?.to_os_string()); + if !existing.pop() { + return Err("Invalid scope".into()); + } + } + existing = std::fs::canonicalize(existing).map_err(|e| e.to_string())?; + if !existing.starts_with(&root) { + return Err("Write scope points outside the project".into()); + } + for part in missing.into_iter().rev() { + existing.push(part); + } + Ok(existing.to_string_lossy().replace('\\', "/").to_lowercase()) +} + +#[tauri::command] +pub fn control_scopes(cwd: String, files: Vec) -> Result, String> { + if files.len() > 64 { + return Err("At most 64 write scopes per task".into()); + } + files + .iter() + .map(|file| resolve_scope(&crate::fs::expand_home(&cwd), file)) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn closing_a_window_releases_ordinary_turns_and_owned_orchestration() { + let mut inner = Inner::default(); + for (id, window) in [ + ("ordinary", "closing"), + ("lead", "closing"), + ("other", "open"), + ] { + inner.active.insert( + id.into(), + ActiveTurn { + window: window.into(), + cwd: format!("/{id}"), + }, + ); + } + for (id, window) in [("lead", "closing"), ("other", "open")] { + inner.grants.insert( + id.into(), + Grant { + window: window.into(), + session: id.into(), + cwd: format!("/{id}"), + token: id.into(), + }, + ); + } + inner.workers.insert("worker".into(), "lead".into()); + inner.workers.insert("other-worker".into(), "other".into()); + let (reply, response) = mpsc::channel(); + inner.pending.insert( + "pending".into(), + Pending { + window: "closing".into(), + reply, + }, + ); + let (reply, other_response) = mpsc::channel(); + inner.pending.insert( + "other-pending".into(), + Pending { + window: "open".into(), + reply, + }, + ); + + assert_eq!( + inner.close_window("closing"), + ["lead", "ordinary", "worker"] + ); + assert_eq!(inner.active.len(), 1); + assert_eq!(inner.active["other"].window, "open"); + assert_eq!(inner.grants.len(), 1); + assert!(inner.grants.contains_key("other")); + assert_eq!(inner.workers.len(), 1); + assert_eq!(inner.workers["other-worker"], "other"); + assert_eq!(response.try_recv().unwrap()["ok"], false); + assert!(other_response.try_recv().is_err()); + assert!(inner.pending.contains_key("other-pending")); + assert!(inner.close_window("closing").is_empty()); + } + + #[test] + fn scopes_reject_escape_and_resolve_new_files() { + let root = std::env::temp_dir().join(uuid::Uuid::new_v4().to_string()); + std::fs::create_dir_all(&root).unwrap(); + assert!(resolve_scope(&root, "../escape").is_err()); + assert!(resolve_scope(&root, "/absolute").is_err()); + assert!(resolve_scope(&root, "src/new.ts") + .unwrap() + .ends_with("/src/new.ts")); + #[cfg(unix)] + { + std::os::unix::fs::symlink(std::env::temp_dir(), root.join("outside")).unwrap(); + assert!(resolve_scope(&root, "outside/file").is_err()); + } + std::fs::remove_dir_all(root).unwrap(); + } + #[test] + fn checkout_reservations_include_nested_folders() { + assert!(paths_overlap("/repo", "/repo/src")); + assert!(paths_overlap("/repo/src", "/repo")); + assert!(!paths_overlap("/repo", "/repo2")); + } +} diff --git a/src-tauri/src/control_cli.rs b/src-tauri/src/control_cli.rs new file mode 100644 index 00000000..3856f5d4 --- /dev/null +++ b/src-tauri/src/control_cli.rs @@ -0,0 +1,373 @@ +//! The desktop executable also provides a small, JSON-only control client. +use std::io::{BufRead, BufReader, Read, Write}; +use std::net::{SocketAddr, TcpStream}; +use std::time::Duration; + +use serde_json::{json, Value}; + +const USAGE: &str = r#"MonoCode local control — supervise this orchestration run from the lead agent. + +Usage: {exe} control ACTION [--json JSON | --input FILE|-] [--request-id ID] + +Actions, with the JSON object each one takes: + list {} + The run, every task with its status and latest result, and the + harness/model IDs you may assign. + delegate {"title":"Short title","harness":"", + "model":"","prompt":"Self-contained instructions", + "files":["src/feature"],"dependsOn":[""]} + Queue a worker and return its taskId. "model" is optional and + defaults to the first model list allows for that harness. + "files" is the write scope: project-relative paths, where a + directory covers its descendants and ["."] reserves the whole + checkout. "dependsOn" holds taskIds that must be reviewed first. + get {"taskId":"..."} + One task, including its latest result. + wait {"timeoutSeconds":20} + Block until a task changes state, or until the timeout (0-25). + Returns at once when nothing is running or queued. + respond {"taskId":"...","requestId":7,"decision":"allow"|"deny"} + Answer an approval an agent is blocked on. Agents never prompt the + user; list, get and wait report the prompt as that task's + "needsInput", and it stays stopped until you decide. + answer {"taskId":"...","requestId":9,"answers":{"":[""]}} + Answer a question an agent asked, or pass "skip":true instead of + "answers". The question and its options come from needsInput. + steer {"taskId":"...","text":"..."} + Redirect an agent that is still running, without discarding the + work it has already done. Use this the moment you see it going + the wrong way; message only lands once it has stopped. + message {"taskId":"...","text":"..."} + Send a completed or failed worker another turn; it keeps its + session, scope and history. + cancel {"taskId":"..."} + Cancel a task, whether it is running or still queued. + review {"taskId":"..."} + Accept a completed task's result. + finish {} + End the run, once every task is accepted or cancelled. + +Usual loop: list -> delegate ... -> wait or get -> steer an agent that drifts, +unblock one with respond or answer -> inspect the changes yourself -> message +for corrections -> review each task -> finish. + +Output is one JSON line: {"ok":true,"result":...} or {"ok":false,"error":"..."}. +The exit code is 0 only when "ok" is true. + +Input must be a JSON object; unknown fields are rejected rather than ignored. +--json takes it inline, --input FILE reads a file, --input - reads stdin. + +Every call carries a request ID, and the run applies each ID at most once. A +failed response reports the ID it used whenever the outcome is unknown — a +timeout, say. Retry that exact call with --request-id ID; retrying a delegate +under a fresh ID instead would queue a second worker. + +Tasks run inside the MonoCode app, not in this process. Exiting this CLI, or a +failure here, never cancels a task that was already accepted. + +MonoCode sets MONOCODE_CONTROL_ENDPOINT and MONOCODE_CONTROL_TOKEN for the lead +agent's process only. They are already in your environment; never print them. +"#; + +const ACTIONS: [&str; 11] = [ + "list", "delegate", "get", "steer", "message", "cancel", "wait", "review", "finish", "respond", + "answer", +]; + +/// Quote for the shell the lead agent actually runs commands in, and only when +/// the path needs it. The path is absolute, so a leading slash means a POSIX +/// shell — where a backslash escapes rather than separates, and so is never +/// safe bare. +fn quoted(value: &str) -> String { + if !value.starts_with('/') { + return if value.contains([' ', '\t', '"']) { + format!("\"{}\"", value.replace('"', "")) + } else { + value.into() + }; + } + if value + .chars() + .all(|c| c.is_ascii_alphanumeric() || "._/-:".contains(c)) + { + value.into() + } else { + format!("'{}'", value.replace('\'', r"'\''")) + } +} + +pub fn help() -> String { + let exe = std::env::current_exe() + .map(|path| quoted(&path.to_string_lossy())) + .unwrap_or_else(|_| "monocode".into()); + USAGE.replace("{exe}", &exe) +} + +enum Parsed { + Help, + Call(String, Value, String), +} + +pub fn run(args: Vec) -> i32 { + let parsed = match parse_args(&args) { + Ok(parsed) => parsed, + Err(error) => { + println!("{}", json!({"ok": false, "error": error})); + return 1; + } + }; + let (action, input, request_id) = match parsed { + Parsed::Help => { + println!("{}", help()); + return 0; + } + Parsed::Call(action, input, request_id) => (action, input, request_id), + }; + match send(&action, &input, &request_id) { + Ok(mut value) => { + if value.get("ok").and_then(Value::as_bool) == Some(true) { + println!("{value}"); + return 0; + } + // MonoCode may have timed out waiting on its own executor, so a + // failure here is not proof the call was rejected either. + if let Some(object) = value.as_object_mut() { + object + .entry("requestId") + .or_insert_with(|| json!(request_id)); + object + .entry("retryWith") + .or_insert_with(|| json!(format!("--request-id {request_id}"))); + } + println!("{value}"); + 1 + } + Err(Failure { error, sent }) => { + // The call may have reached the run even though its answer was + // lost. Hand back the request ID so a retry cannot duplicate it. + let mut response = json!({"ok": false, "error": error}); + if sent { + response["requestId"] = json!(request_id); + response["retryWith"] = json!(format!("--request-id {request_id}")); + } + println!("{response}"); + 1 + } + } +} + +struct Failure { + error: String, + /// The request was already on the wire, so the run may have applied it. + sent: bool, +} +fn unsent(error: impl Into) -> Failure { + Failure { + error: error.into(), + sent: false, + } +} +fn sent(error: impl Into) -> Failure { + Failure { + error: error.into(), + sent: true, + } +} + +fn send(action: &str, input: &Value, request_id: &str) -> Result { + let endpoint = std::env::var("MONOCODE_CONTROL_ENDPOINT").map_err(|_| { + unsent("No MonoCode connection. Confirm the Orchestrator proposal in MonoCode first.") + })?; + let token = std::env::var("MONOCODE_CONTROL_TOKEN") + .map_err(|_| unsent("No MonoCode session credential. Start the lead from MonoCode."))?; + let address: SocketAddr = endpoint + .parse() + .map_err(|_| unsent("Invalid MonoCode endpoint"))?; + if !address.ip().is_loopback() { + return Err(unsent("MonoCode control only connects to localhost")); + } + let mut stream = TcpStream::connect_timeout(&address, Duration::from_secs(3)) + .map_err(|_| unsent("MonoCode is not running or this connection has expired."))?; + stream + .set_read_timeout(Some(Duration::from_secs(40))) + .map_err(|e| unsent(e.to_string()))?; + stream + .set_write_timeout(Some(Duration::from_secs(5))) + .map_err(|e| unsent(e.to_string()))?; + writeln!( + stream, + "{}", + json!({"token":token,"action":action,"input":input,"requestId":request_id}) + ) + .map_err(|e| sent(e.to_string()))?; + let mut line = String::new(); + BufReader::new(stream) + .take(2_000_001) + .read_line(&mut line) + .map_err(|e| sent(format!("No reply from MonoCode: {e}")))?; + if line.len() > 2_000_000 { + return Err(sent("MonoCode response is too large")); + } + serde_json::from_str(&line).map_err(|_| sent("MonoCode returned an invalid response")) +} + +fn read_capped(mut source: impl Read) -> Result { + let mut raw = String::new(); + source + .by_ref() + .take(262_145) + .read_to_string(&mut raw) + .map_err(|e| e.to_string())?; + if raw.len() > 262_144 { + return Err("Input exceeds 256 KiB".into()); + } + Ok(raw) +} + +fn parse_args(args: &[String]) -> Result { + let is_help = |value: &str| matches!(value, "help" | "--help" | "-h"); + let Some(action) = args.first() else { + return Ok(Parsed::Help); + }; + if is_help(action) { + return Ok(Parsed::Help); + } + let action = action.clone(); + if !ACTIONS.contains(&action.as_str()) { + return Err(format!( + "Unknown action: {action}. Use one of: {}. Run control --help.", + ACTIONS.join(", ") + )); + } + let mut input = None; + let mut request_id = uuid::Uuid::new_v4().to_string(); + let mut index = 1; + while index < args.len() { + let flag = &args[index]; + // `control delegate --help` should explain the command, not fail. + if is_help(flag) { + return Ok(Parsed::Help); + } + if !flag.starts_with("--") { + return Err(format!( + "Unexpected argument: {flag}. Pass the JSON object as --json ''." + )); + } + let value = args.get(index + 1).ok_or_else(|| { + format!("Missing value for {flag}. Run control --help for the argument list.") + })?; + match flag.as_str() { + "--request-id" => request_id = value.clone(), + "--json" | "--input" => { + if input.is_some() { + return Err("Supply only one input".into()); + } + let raw = if flag == "--json" { + if value.len() > 262_144 { + return Err("Input exceeds 256 KiB".into()); + } + value.clone() + } else if value == "-" { + read_capped(std::io::stdin())? + } else { + read_capped(std::fs::File::open(value).map_err(|e| e.to_string())?)? + }; + let parsed: Value = serde_json::from_str(&raw) + .map_err(|e| format!("Invalid JSON: {e}. Pass one JSON object, e.g. --json '{{\"taskId\":\"...\"}}'."))?; + if !parsed.is_object() { + return Err("Input must be a JSON object".into()); + } + input = Some(parsed); + } + _ => { + return Err(format!( + "Unknown option: {flag}. Supported: --json, --input, --request-id." + )) + } + } + index += 2; + } + if request_id.is_empty() || request_id.len() > 128 { + return Err("Invalid request ID".into()); + } + Ok(Parsed::Call( + action, + input.unwrap_or_else(|| json!({})), + request_id, + )) +} + +#[cfg(test)] +mod tests { + use super::*; + fn args(values: &[&str]) -> Vec { + values.iter().map(|s| s.to_string()).collect() + } + fn call(values: &[&str]) -> Result<(String, Value, String), String> { + match parse_args(&args(values))? { + Parsed::Call(action, input, id) => Ok((action, input, id)), + Parsed::Help => Err("help".into()), + } + } + #[test] + fn validates_inputs_without_invoking_a_shell() { + let (_, input, id) = call(&[ + "delegate", + "--json", + r#"{"prompt":"$(touch nope) `hello`\nnext"}"#, + "--request-id", + "retry-1", + ]) + .unwrap(); + assert_eq!(id, "retry-1"); + assert_eq!(input["prompt"], "$(touch nope) `hello`\nnext"); + assert!(call(&["delegate", "--json", "[]"]).is_err()); + assert!(call(&["delegate", "--json", "{}", "--json", "{}"]).is_err()); + assert!(call(&["unknown"]).is_err()); + } + #[test] + fn explains_help_and_malformed_invocations() { + assert!(matches!(parse_args(&args(&[])), Ok(Parsed::Help))); + assert!(matches!(parse_args(&args(&["--help"])), Ok(Parsed::Help))); + // Agents commonly probe a subcommand for its own usage text. + assert!(matches!( + parse_args(&args(&["delegate", "--help"])), + Ok(Parsed::Help) + )); + assert!(call(&["get", r#"{"taskId":"x"}"#]) + .unwrap_err() + .contains("--json")); + assert!(call(&["get", "--json"]).unwrap_err().contains("--help")); + assert!(call(&["get", "--taskId", "x"]) + .unwrap_err() + .contains("Unknown option")); + assert!(call(&["get", "--json", "{taskId}"]) + .unwrap_err() + .contains("Invalid JSON")); + } + #[test] + fn help_names_every_action_and_the_real_executable() { + let text = help(); + for action in ACTIONS { + assert!(text.contains(action), "help omits {action}"); + } + assert!(!text.contains("{exe}")); + assert!(text.contains("--request-id")); + } + #[test] + fn quotes_the_control_path_only_when_the_shell_needs_it() { + assert_eq!( + quoted("/Applications/MonoCode.app/Contents/MacOS/monocode"), + "/Applications/MonoCode.app/Contents/MacOS/monocode" + ); + assert_eq!(quoted("/Users/a b/MonoCode"), "'/Users/a b/MonoCode'"); + assert_eq!(quoted("C:\\Tools\\monocode.exe"), "C:\\Tools\\monocode.exe"); + assert_eq!( + quoted("C:\\Program Files\\MonoCode\\monocode.exe"), + "\"C:\\Program Files\\MonoCode\\monocode.exe\"" + ); + // A backslash escapes in a POSIX shell, so bare would rewrite the path. + assert_eq!(quoted("/Users/a\\b/MonoCode"), "'/Users/a\\b/MonoCode'"); + assert_eq!(quoted("/Users/it's/MonoCode"), r"'/Users/it'\''s/MonoCode'"); + } +} diff --git a/src-tauri/src/harness.rs b/src-tauri/src/harness.rs index b61fa28e..8c345fc9 100644 --- a/src-tauri/src/harness.rs +++ b/src-tauri/src/harness.rs @@ -348,6 +348,8 @@ pub fn harness_spawn( .stderr(Stdio::piped()); prepare_child(&mut cmd, &command); + crate::control::configure_child(&app, &session_id, &mut cmd); + let mut child = spawn_managed(&mut cmd).map_err(|e| format!("Failed to start {command}: {e}"))?; let pid = child.id(); diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 09b39dcf..07181a3c 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -2,6 +2,8 @@ use tauri::Manager; mod chat_background; mod checkpoint; +mod control; +pub mod control_cli; mod cursor_store; mod fs; mod gitlab; @@ -197,6 +199,7 @@ pub fn run() { .setup(|app| { harness::reap_orphaned_harness_processes(); session_store::init(app.handle())?; + control::init(app.handle())?; reminders::init(app.handle()); checkpoint::init(app.handle())?; menu::install(app.handle())?; @@ -220,6 +223,15 @@ pub fn run() { menu::dispatch(app, event.id().as_ref()); }) .invoke_handler(tauri::generate_handler![ + control::control_enable, + control::control_disable, + control::control_reply, + control::control_save, + control::control_load, + control::control_scopes, + control::control_attach_worker, + control::control_authorize_turn, + control::control_turn_finished, default_cwd, home_dir, notifications::notification_permission, @@ -403,6 +415,7 @@ pub fn run() { .. } => { let other_window = handle.webview_windows().keys().any(|name| name != &label); + control::window_closed(handle, &label); if !other_window { reap_harness_children(handle); } diff --git a/src-tauri/src/main.rs b/src-tauri/src/main.rs index 3b2b9e17..8af4da08 100644 --- a/src-tauri/src/main.rs +++ b/src-tauri/src/main.rs @@ -1,6 +1,11 @@ #![cfg_attr(not(debug_assertions), windows_subsystem = "windows")] fn main() { + if std::env::args().nth(1).as_deref() == Some("control") { + std::process::exit(monocode_lib::control_cli::run( + std::env::args().skip(2).collect(), + )); + } #[cfg(all(debug_assertions, target_os = "macos"))] monocode_lib::ensure_macos_dev_bundle(); monocode_lib::run() diff --git a/src-tauri/src/session_store.rs b/src-tauri/src/session_store.rs index 2090ad55..82e75647 100644 --- a/src-tauri/src/session_store.rs +++ b/src-tauri/src/session_store.rs @@ -4,7 +4,7 @@ use std::time::{SystemTime, UNIX_EPOCH}; use rusqlite::{params, Connection, OptionalExtension}; use serde::{Deserialize, Serialize}; -use serde_json::Value; +use serde_json::{json, Value}; use tauri::{AppHandle, Emitter, Manager, State}; const MIGRATION_V1: &str = r#" @@ -99,6 +99,10 @@ pub struct SessionUpsert { #[serde(rename_all = "camelCase")] pub struct SessionSummary { pub id: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub orchestration_lead_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub orchestration: Option, pub cwd: String, pub harness: String, pub model: String, @@ -126,6 +130,8 @@ pub struct SessionSummary { #[serde(rename_all = "camelCase")] pub struct SessionRecord { pub id: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub orchestration_lead_id: Option, pub cwd: String, pub harness: String, pub model: String, @@ -603,9 +609,128 @@ fn migrate(conn: &Connection) -> rusqlite::Result<()> { )?; crate::notes::ensure_notes_table(conn)?; crate::reminders::ensure_table(conn)?; + ensure_orchestration_history(conn)?; Ok(()) } +fn ensure_orchestration_history(conn: &Connection) -> rusqlite::Result<()> { + let indexed: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE name = 'orchestration_sidebar')", + [], + |row| row.get(0), + )?; + let tx = conn.unchecked_transaction()?; + tx.execute_batch( + "CREATE TABLE IF NOT EXISTS orchestration_runs (lead_id TEXT PRIMARY KEY, state TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS orchestration_sidebar (lead_id TEXT PRIMARY KEY, summary TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS orchestration_workers (session_id TEXT PRIMARY KEY, lead_id TEXT NOT NULL);", + )?; + if !indexed { + // One-time compatibility pass for the preview that listed workers as + // separate chats. Normal sidebar reads never scan transcripts/run blobs. + let runs = tx + .prepare("SELECT lead_id, state FROM orchestration_runs")? + .query_map([], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + })? + .collect::>>()?; + for (lead, state) in runs { + if let Ok(run) = serde_json::from_str::(&state) { + index_orchestration(&tx, &lead, &run)?; + } + } + let workers = tx.prepare("SELECT id, blocks_json FROM sessions WHERE blocks_json LIKE '%orchestrationLeadId%'")? + .query_map([], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)))? + .collect::>>()?; + for (id, blocks) in workers { + if let Ok(blocks) = serde_json::from_str::(&blocks) { + remember_worker_from_blocks(&tx, &id, &blocks)?; + } + } + } + tx.commit() +} + +fn remember_worker(conn: &Connection, id: &str, lead: &str) -> rusqlite::Result<()> { + if id != lead && validate_id(id, "Worker").is_ok() && validate_id(lead, "Lead").is_ok() { + // Keep earlier workers indexed when a lead starts a subsequent run. + conn.execute( + "INSERT OR IGNORE INTO orchestration_workers(session_id, lead_id) VALUES (?1, ?2)", + params![id, lead], + )?; + } + Ok(()) +} + +fn remember_worker_from_blocks( + conn: &Connection, + id: &str, + blocks: &Value, +) -> rusqlite::Result<()> { + if let Some(blocks) = blocks.as_array() { + for block in blocks { + if block["role"] == "user" { + if let Some(lead) = block["orchestrationLeadId"].as_str() { + remember_worker(conn, id, lead)?; + break; + } + } + } + } + Ok(()) +} + +fn index_orchestration(conn: &Connection, lead: &str, run: &Value) -> rusqlite::Result<()> { + let Some(tasks) = run["tasks"].as_array() else { + return Ok(()); + }; + let mut summaries = Vec::new(); + for task in tasks { + let Some(id) = task["sessionId"].as_str() else { + continue; + }; + remember_worker(conn, id, lead)?; + summaries.push(serde_json::json!({ + "sessionId": id, "title": task["title"], "harness": task["harness"], + "model": task["model"], "status": task["status"], + })); + } + let summary = serde_json::json!({ "status": run["status"], "tasks": summaries }); + conn.execute("INSERT INTO orchestration_sidebar(lead_id, summary) VALUES (?1, ?2) ON CONFLICT(lead_id) DO UPDATE SET summary = excluded.summary", params![lead, summary.to_string()])?; + Ok(()) +} + +pub(crate) fn save_orchestration( + conn: &Connection, + lead: &str, + run: &Value, +) -> rusqlite::Result<()> { + let tx = conn.unchecked_transaction()?; + tx.execute("INSERT INTO orchestration_runs(lead_id, state) VALUES (?1, ?2) ON CONFLICT(lead_id) DO UPDATE SET state = excluded.state", params![lead, run.to_string()])?; + index_orchestration(&tx, lead, run)?; + tx.commit() +} + +fn worker_parent(conn: &Connection, id: &str) -> rusqlite::Result> { + conn.query_row( + "SELECT lead_id FROM orchestration_workers WHERE session_id = ?1", + [id], + |row| row.get(0), + ) + .optional() +} + +fn orchestration_summary(conn: &Connection, id: &str) -> rusqlite::Result> { + Ok(optional_json( + conn.query_row( + "SELECT summary FROM orchestration_sidebar WHERE lead_id = ?1", + [id], + |row| row.get(0), + ) + .optional()?, + )) +} + fn upsert_session(conn: &Connection, session: &SessionUpsert) -> rusqlite::Result { let now = now_millis(); let model_settings = serde_json::to_string(&session.model_settings) @@ -708,8 +833,11 @@ fn upsert_session(conn: &Connection, session: &SessionUpsert) -> rusqlite::Resul ], )?; + remember_worker_from_blocks(conn, &session.id, &session.blocks)?; Ok(SessionSummary { id: session.id.clone(), + orchestration_lead_id: worker_parent(conn, &session.id)?, + orchestration: orchestration_summary(conn, &session.id)?, cwd: session.cwd.clone(), harness: session.harness.clone(), model: session.model.clone(), @@ -979,11 +1107,13 @@ fn list_by_project(conn: &Connection, cwd: &str) -> rusqlite::Result rusqlite::Result rusqlite::Result> { let mut statement = conn.prepare( "SELECT id, cwd, harness, model, runtime_mode, title, provider_session_id, created_at, updated_at, branch, archived, pinned, - linked_work_item_json + linked_work_item_json, + (SELECT summary FROM orchestration_sidebar WHERE lead_id = sessions.id) FROM sessions WHERE has_user_message = 1 AND linked_work_item_json IS NOT NULL AND id NOT IN (SELECT id FROM sessions WHERE inbox_ask IS NOT NULL) + AND id NOT IN (SELECT session_id FROM orchestration_workers) ORDER BY updated_at DESC, id ASC", )?; let rows = statement.query_map([], |row| { @@ -1029,6 +1163,8 @@ fn list_linked(conn: &Connection) -> rusqlite::Result> { let pinned: i64 = row.get(11)?; Ok(SessionSummary { id: row.get(0)?, + orchestration_lead_id: None, + orchestration: optional_json(row.get(13)?), cwd: row.get(1)?, harness: row.get(2)?, model: row.get(3)?, @@ -1075,8 +1211,114 @@ fn optional_json(raw: Option) -> Option { } fn delete_session(conn: &Connection, session_id: &str) -> rusqlite::Result<()> { - conn.execute("DELETE FROM sessions WHERE id = ?1", params![session_id])?; - Ok(()) + let tx = conn.unchecked_transaction()?; + let parent = worker_parent(&tx, session_id)?; + // Ownership is also carried in transcripts for older clients. Release + // that metadata along with the index so reopening a worker stays detached. + let workers = tx.prepare("SELECT id, blocks_json FROM sessions WHERE id IN (SELECT session_id FROM orchestration_workers WHERE lead_id = ?1)")? + .query_map([session_id], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)))? + .collect::>>()?; + for (id, raw) in workers { + let mut blocks: Value = serde_json::from_str(&raw).map_err(|e| { + rusqlite::Error::FromSqlConversionFailure(0, rusqlite::types::Type::Text, Box::new(e)) + })?; + if let Some(blocks) = blocks.as_array_mut() { + for block in blocks { + if block["orchestrationLeadId"] == session_id { + if let Some(block) = block.as_object_mut() { + block.remove("orchestrationLeadId"); + } + } + } + } + tx.execute( + "UPDATE sessions SET blocks_json = ?1 WHERE id = ?2", + params![blocks.to_string(), id], + )?; + } + tx.execute( + "DELETE FROM orchestration_runs WHERE lead_id = ?1", + [session_id], + )?; + tx.execute( + "DELETE FROM orchestration_sidebar WHERE lead_id = ?1", + [session_id], + )?; + if let Some(parent) = parent.filter(|id| id != session_id) { + let raw: Option = tx + .query_row( + "SELECT state FROM orchestration_runs WHERE lead_id = ?1", + [&parent], + |row| row.get(0), + ) + .optional()?; + if let Some(raw) = raw { + let mut run: Value = serde_json::from_str(&raw).map_err(|e| { + rusqlite::Error::FromSqlConversionFailure( + 0, + rusqlite::types::Type::Text, + Box::new(e), + ) + })?; + let mut removed = false; + if let Some(tasks) = run["tasks"].as_array_mut() { + let ids: Vec = tasks + .iter() + .filter(|task| task["sessionId"] == session_id) + .filter_map(|task| task["id"].as_str().map(str::to_string)) + .collect(); + let before = tasks.len(); + tasks.retain(|task| task["sessionId"] != session_id); + removed = before != tasks.len(); + if removed { + for task in tasks { + if let Some(deps) = task["dependsOn"].as_array_mut() { + deps.retain(|dep| !ids.iter().any(|id| dep == id)); + } + // Removing a prerequisite must not release queued work. + // The app stops the run before deleting any current member. + if matches!( + task["status"].as_str(), + Some("queued" | "running" | "cancelling") + ) { + task["status"] = json!("cancelled"); + task["accepted"] = json!(false); + task["delivered"] = json!(true); + } + } + } + } + if removed { + if matches!(run["status"].as_str(), Some("active" | "paused")) { + run["status"] = json!("stopped"); + run["error"] = + json!("A worker conversation was deleted. Start a new run to continue."); + } + run["requests"] = json!({}); + tx.execute( + "UPDATE orchestration_runs SET state = ?1 WHERE lead_id = ?2", + params![run.to_string(), parent], + )?; + index_orchestration(&tx, &parent, &run)?; + } + } + // Also handle summaries left by an older client without a run record. + if let Some(mut summary) = orchestration_summary(&tx, &parent)? { + if let Some(tasks) = summary["tasks"].as_array_mut() { + tasks.retain(|task| task["sessionId"] != session_id); + } + tx.execute( + "UPDATE orchestration_sidebar SET summary = ?1 WHERE lead_id = ?2", + params![summary.to_string(), parent], + )?; + } + } + tx.execute( + "DELETE FROM orchestration_workers WHERE session_id = ?1 OR lead_id = ?1", + [session_id], + )?; + tx.execute("DELETE FROM sessions WHERE id = ?1", [session_id])?; + tx.commit() } fn set_archived(conn: &Connection, session_id: &str, archived: bool) -> rusqlite::Result<()> { @@ -1123,6 +1365,7 @@ fn get_session(conn: &Connection, session_id: &str) -> rusqlite::Result(&raw).unwrap(), other); + migrate(&conn).unwrap(); + assert!(worker_parent(&conn, "current").unwrap().is_none()); + } + + #[test] + fn deleting_a_worker_prunes_parent_state_and_does_not_release_dependencies() { + let store = SessionStore::open_in_memory().unwrap(); + let conn = store.conn.lock().unwrap(); + for id in ["lead", "worker", "dependent", "earlier"] { + upsert_session(&conn, &sample(id, "/tmp/a", id)).unwrap(); + } + remember_worker(&conn, "earlier", "lead").unwrap(); + save_orchestration( + &conn, + "lead", + &json!({"status":"active", "requests":{"old":{"result":"worker"}}, "tasks":[ + {"id":"task", "sessionId":"worker", "status":"running", "dependsOn":[]}, + {"id":"next", "sessionId":"dependent", "status":"queued", "dependsOn":["task"]} + ]}), + ) + .unwrap(); + delete_session(&conn, "worker").unwrap(); + assert!(get_session(&conn, "worker").unwrap().is_none()); + assert!(worker_parent(&conn, "worker").unwrap().is_none()); + assert_eq!( + worker_parent(&conn, "earlier").unwrap().as_deref(), + Some("lead") + ); + let raw: String = conn + .query_row( + "SELECT state FROM orchestration_runs WHERE lead_id = 'lead'", + [], + |row| row.get(0), + ) + .unwrap(); + let run: Value = serde_json::from_str(&raw).unwrap(); + assert_eq!(run["status"], "stopped"); + assert_eq!(run["requests"], json!({})); + assert_eq!(run["tasks"].as_array().unwrap().len(), 1); + assert_eq!(run["tasks"][0]["sessionId"], "dependent"); + assert_eq!(run["tasks"][0]["status"], "cancelled"); + assert_eq!(run["tasks"][0]["dependsOn"], json!([])); + let summary = orchestration_summary(&conn, "lead").unwrap().unwrap(); + assert_eq!(summary["tasks"].as_array().unwrap().len(), 1); + assert_eq!(summary["tasks"][0]["sessionId"], "dependent"); + assert_eq!(summary["tasks"][0]["status"], "cancelled"); + } + + #[test] + fn failed_session_deletion_rolls_back_orchestration_cleanup() { + let store = SessionStore::open_in_memory().unwrap(); + let conn = store.conn.lock().unwrap(); + for id in ["lead", "worker"] { + upsert_session(&conn, &sample(id, "/tmp/a", id)).unwrap(); + } + save_orchestration( + &conn, + "lead", + &json!({"status":"stopped", "tasks":[{"id":"task", "sessionId":"worker"}]}), + ) + .unwrap(); + conn.execute_batch("CREATE TRIGGER reject_delete BEFORE DELETE ON sessions BEGIN SELECT RAISE(ABORT, 'test failure'); END;").unwrap(); + for id in ["lead", "worker"] { + assert!(delete_session(&conn, id).is_err()); + assert!(get_session(&conn, id).unwrap().is_some()); + assert_eq!( + worker_parent(&conn, "worker").unwrap().as_deref(), + Some("lead") + ); + assert_eq!( + orchestration_summary(&conn, "lead").unwrap().unwrap()["tasks"][0]["sessionId"], + "worker" + ); + } + } + #[test] fn migration_v6_adds_archived_column() { let store = SessionStore::open_in_memory().unwrap(); diff --git a/src/App.tsx b/src/App.tsx index c357e9e1..0526ce4c 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -1,4 +1,28 @@ import { invoke } from "@tauri-apps/api/core"; +import { orchestrator, type ControlOutcome } from "./lib/orchestration"; +import { modelsFor } from "./lib/models"; +import { isHarnessAvailable } from "./lib/harness/availability"; +import { + completeOrchestrationProposal, + orchestrationPlanningPrompt, + proposalBlock, + validateOrchestrationSettings, + withOrchestrationProposal, + type OrchestrationProposal, +} from "./lib/orchestrationPlan"; +import { discoverOrchestrationSettings } from "./lib/orchestrationCatalog"; +import { + attachOrchestrationWorkers, + consolidateOrchestrationTabs, + prepareOrchestrationWorkerDetails, + releaseOrchestrationWorker, +} from "./lib/orchestrationWorkspace"; +import { + OrchestrationActions, + OrchestrationWorkers, + type OrchestrationWorkerDetail, +} from "./chrome/OrchestrationActions"; +import { flushSync } from "react-dom"; import { listen } from "@tauri-apps/api/event"; import { getCurrentWindow } from "@tauri-apps/api/window"; import { ask, message } from "@tauri-apps/plugin-dialog"; @@ -75,6 +99,7 @@ import { nextTerminalTitle, openChangesTab, openCommitTab, + newAgentTab, openEditorTab, openSessionChangesTab, openTerminalTab, @@ -232,6 +257,7 @@ import { } from "./lib/workspaceTabGroups"; import { runSessionRemoval } from "./lib/sessionRemoval"; import { + HARNESSES, HARNESS_LABEL, HARNESS_TITLE, canReplaceSessionTitle, @@ -661,6 +687,20 @@ export default function App({ useState(null); const openingInboxSessions = useRef(new Map>()); const [notesViewOpen, setNotesViewOpen] = useState(false); + const [inspectedWorkerId, setInspectedWorkerId] = useState( + null, + ); + // Set while the lead's tab is still opening; the agent tab lands on the + // commit that brings it in. + const [workerDetailRequest, setWorkerDetailRequest] = useState<{ + leadId: string; + workers: OrchestrationWorkerDetail[]; + } | null>(null); + const orchestrationRuns = useSyncExternalStore( + orchestrator.subscribe, + orchestrator.snapshot, + orchestrator.snapshot, + ); const notesEnabled = useSyncExternalStore( subscribeNotesEnabled, loadNotesEnabled, @@ -827,6 +867,7 @@ export default function App({ const stopSessionForRemoval = useCallback( async (sessionId: string): Promise => { + await orchestrator.stopForSession(sessionId); const open = sessionsRef.current.find( (session) => session.id === sessionId, ); @@ -990,7 +1031,10 @@ export default function App({ const nextBusySessionIds = useMemo(() => { const ids = new Set(); for (const session of sessions) { - if (session.busy) ids.add(session.id); + if (session.busy) { + ids.add(session.id); + if (session.orchestrationLeadId) ids.add(session.orchestrationLeadId); + } } return ids; }, [sessions]); @@ -1036,7 +1080,10 @@ export default function App({ const nextApprovalSessionIds = useMemo(() => { const ids = new Set(); for (const session of sessions) { - if (sessionNeedsInput(session)) ids.add(session.id); + if (sessionNeedsInput(session)) { + ids.add(session.id); + if (session.orchestrationLeadId) ids.add(session.orchestrationLeadId); + } } return ids; }, [sessions]); @@ -1383,6 +1430,20 @@ export default function App({ for (const session of sessions) { if (session.inboxAsk) visibleIds.add(session.id); } + // Internal workers stay attached to the lead, even while idle between + // turns. They must not be discarded merely because they have no tab. + for (const session of sessions) { + if ( + session.orchestrationLeadId && + (visibleIds.has(session.orchestrationLeadId) || + orchestrationRuns.some( + (run) => + run.leadId === session.orchestrationLeadId && + ["active", "paused"].includes(run.status), + )) + ) + visibleIds.add(session.id); + } for (const sessionId of visibleIds) { openingSessionIds.current.delete(sessionId); loadedSessionCache.current.delete(sessionId); @@ -1416,12 +1477,13 @@ export default function App({ skipForgetSessionIds.current.has(session.id), ), ); - }, [sessions, tabs, persistSession, liveAgentsEnabled]); + }, [sessions, tabs, persistSession, liveAgentsEnabled, orchestrationRuns]); const activateTab = useCallback((id: string, paneId?: string) => { const tab = tabsRef.current.find((entry) => entry.id === id); const nextFocusedId = - tab && paneId && + tab && + paneId && (leafIds(tab.layout).includes(paneId) || tab.editorPanes.some((entry) => entry.id === paneId) || (tab.terminalPanes ?? []).some((entry) => entry.id === paneId)) @@ -3197,29 +3259,30 @@ export default function App({ const onSelectHistorySession = useCallback( async (sessionId: string) => { - const linkedUpdate = linkedSessionUpdatesRef.current.get(sessionId); - if (focusOpenSession(sessionId)) { - if (linkedUpdate) revealLinkedSessionUpdate(sessionId, linkedUpdate); - return; - } - const session = await ensureOpenSession(sessionId); + let session = await ensureOpenSession(sessionId); if (!session || session.inboxAsk) return; - // A previous race may have left this tab pointing at a parked session. - // Once the session is restored, reuse that tab instead of creating a - // duplicate and leave the broken pane behind. - if (focusOpenSession(sessionId)) { - if (linkedUpdate) revealLinkedSessionUpdate(sessionId, linkedUpdate); + const parentId = + session.orchestrationLeadId ?? + orchestrator.forSession(sessionId)?.leadId; + if (parentId && parentId !== sessionId) { + setInspectedWorkerId(sessionId); + session = await ensureOpenSession(parentId); + if (!session) return; + } + const linkedUpdate = linkedSessionUpdatesRef.current.get(session.id); + if (focusOpenSession(session.id)) { + if (linkedUpdate) revealLinkedSessionUpdate(session.id, linkedUpdate); return; } if (replaceBlankPaneWithSession(session)) { - if (linkedUpdate) revealLinkedSessionUpdate(sessionId, linkedUpdate); + if (linkedUpdate) revealLinkedSessionUpdate(session.id, linkedUpdate); return; } const tab = newTab(session.id); appendTab(tab, session.cwd); setActiveTabId(tab.id); setComposerFocused(true); - if (linkedUpdate) revealLinkedSessionUpdate(sessionId, linkedUpdate); + if (linkedUpdate) revealLinkedSessionUpdate(session.id, linkedUpdate); }, [ appendTab, @@ -3267,7 +3330,9 @@ export default function App({ const sessionReminders = useSessionReminders( openReminderSession, ensureReminderSessionsSaved, - sessions.filter((session) => !session.inboxAsk).map((session) => session.id), + sessions + .filter((session) => !session.inboxAsk) + .map((session) => session.id), ); const dismissNoticesForContinuedSession = useCallback( @@ -3445,6 +3510,10 @@ export default function App({ ); }, stop: async () => { + const run = + mode === "delete" ? orchestrator.forSession(sessionId) : undefined; + if (run && (run.status === "active" || run.status === "paused")) + await orchestrator.stopRun(run.leadId); await stopSessionForRemoval(sessionId); }, updateSession: (stopped) => { @@ -3457,7 +3526,33 @@ export default function App({ persist: async (latest) => { if (latest) await flushSessionCheckpoint(sessionId); if (mode === "delete") { - await deleteSession(sessionId); + await orchestrator.deleteSession(sessionId, () => + deleteSession(sessionId), + ); + const released = sessionsRef.current.map((session) => + releaseOrchestrationWorker(session, sessionId), + ); + sessionsRef.current = released; + setSessions(released); + for (const [id, cached] of loadedSessionCache.current) { + if (releaseOrchestrationWorker(cached, sessionId) !== cached) + invalidateLoadedSession(id); + } + // Pending reads may still carry the deleted lead's ownership. + for (const id of sessionLoads.current.keys()) + invalidateLoadedSession(id); + for (const [id, pending] of pendingPersist.current) { + pendingPersist.current.set( + id, + releaseOrchestrationWorker(pending, sessionId), + ); + } + const releaseSummary = (entry: SessionSummary) => + entry.orchestrationLeadId === sessionId + ? { ...entry, orchestrationLeadId: undefined } + : entry; + setHistory((current) => current.map(releaseSummary)); + setStoredLinkedSessions((current) => current.map(releaseSummary)); return; } if (latest && shouldPersistSession(latest)) { @@ -4210,8 +4305,36 @@ export default function App({ intent?: TurnIntent; planBlockId?: string; buildTarget?: PlanBuildTarget; + managed?: boolean; + onSettled?: (outcome: ControlOutcome) => void; }, ) => { + const controlError = orchestrator.submissionError( + sessionId, + options?.managed, + ); + if (controlError) { + enqueueHarnessEvent(sessionId, { type: "status", text: controlError }); + flushHarnessEvents(); + return; + } + if (options?.managed) { + const target = sessionsRef.current.find((s) => s.id === sessionId); + if ( + !target || + target.busy || + target.pendingSwitch || + isPreparingHandoff(target) || + removingSessionIds.current.has(sessionId) + ) { + options.onSettled?.({ + status: "failed", + text: "", + error: "Session is unavailable or already running", + }); + return; + } + } if (removingSessionIds.current.has(sessionId)) return; const storedCurrent = sessionsRef.current.find((s) => s.id === sessionId); if (!storedCurrent) return; @@ -4219,6 +4342,22 @@ export default function App({ ? withPlanBuildTarget(storedCurrent, options.buildTarget) : storedCurrent; const intent = options?.intent ?? "default"; + if (intent === "orchestrate") { + try { + const run = orchestrator.forSession(sessionId); + if (run && ["active", "paused"].includes(run.status)) + throw new Error( + "Stop the current orchestration run before preparing another proposal.", + ); + } catch (error) { + enqueueHarnessEvent(sessionId, { + type: "status", + text: error instanceof Error ? error.message : String(error), + }); + flushHarnessEvents(); + return; + } + } const approvedPlan = options?.planBlockId ? current.blocks.find( (block) => @@ -4263,7 +4402,7 @@ export default function App({ if (current.busy && !pendingSwitch) { const followUpBehavior = - intent === "plan" + intent === "plan" || intent === "orchestrate" ? "queue" : (options?.followUpBehavior ?? loadFollowUpBehavior()); if (followUpBehavior === "queue") { @@ -4363,6 +4502,26 @@ export default function App({ const gen = (turnGen.current.get(sessionId) ?? 0) + 1; turnGen.current.set(sessionId, gen); + const proposalId = + intent === "orchestrate" ? crypto.randomUUID() : undefined; + let proposalDraft: OrchestrationProposal | undefined = proposalId + ? { + version: 1, + leadId: sessionId, + cwd: current.cwd, + request: harnessText, + author: { + harness: current.harness, + model: current.model, + name: resolveModel(current.harness, current.model).name, + }, + settings: { choices: [], maxWorkers: 2 }, + status: "planning", + title: "Orchestration plan", + summary: "", + tasks: [], + } + : undefined; const isFirstTurn = current.blocks.length === 0; const placeholderTitle = canReplaceSessionTitle( current.title, @@ -4386,7 +4545,11 @@ export default function App({ : card ? SECOND_OPINION_TITLE : submittedText; - const cards = rawCommand ? undefined : userTurnCards(noteCard, card); + const cards = { + ...(rawCommand ? undefined : userTurnCards(noteCard, card)), + // The orchestrator writes these turns, not the user; hide them. + ...(options?.managed ? { internal: true } : {}), + }; const live = isLiveHarness(current.harness); const queuedHandoff = live && !pendingSwitch ? pendingHandoff(current) : null; @@ -4516,10 +4679,50 @@ export default function App({ if (pendingSwitch) { void forgetHarnessSession(pendingSwitch.from, sessionId); } + options?.onSettled?.({ + status: "failed", + text: "", + error: "Harness is not connected", + }); return; } + if (proposalId && proposalDraft) { + const draft = proposalDraft; + setSessions((prev) => + prev.map((session) => + session.id === sessionId + ? { + ...session, + blocks: [...session.blocks, proposalBlock(proposalId, draft)], + } + : session, + ), + ); + } + + let controlOutcome: ControlOutcome = { + status: "failed", + text: "", + error: "Turn did not complete", + }; + let controlText = ""; + let proposalText = ""; + let nativeProposalText = ""; void (async () => { + if (proposalDraft && proposalId) { + const settings = await discoverOrchestrationSettings(); + if (turnGen.current.get(sessionId) !== gen) return; + proposalDraft = { ...proposalDraft, settings }; + const discovering = proposalDraft; + setSessions((prev) => + prev.map((session) => + session.id === sessionId + ? withOrchestrationProposal(session, proposalId, discovering) + : session, + ), + ); + } let wrap = handoffCard ? { from: handoffCard.from, @@ -4571,6 +4774,22 @@ export default function App({ let providerFailureSeen = false; const routePlanEvent = (event: HarnessEvent): HarnessEvent | null => { if (event.type === "session.error") providerFailureSeen = true; + if (proposalDraft) { + if (event.type === "message.delta") { + proposalText = (proposalText + event.text).slice(-200_000); + return null; + } + if (event.type === "message.completed") { + proposalText += "\n"; + return null; + } + if (event.type === "plan") { + nativeProposalText = event.append + ? nativeProposalText + event.text + : event.text; + return null; + } + } if (intent !== "plan") return event; if (event.type === "plan") { nativePlanSeen = true; @@ -4582,7 +4801,7 @@ export default function App({ return event; }; - if (!current.inboxAsk) { + if (!current.inboxAsk && !orchestrator.forSession(sessionId)) { await beginSessionTurn(sessionId, workCwd).catch(() => undefined); } if (turnGen.current.get(sessionId) !== gen) return; @@ -4597,8 +4816,11 @@ export default function App({ sessionId, cwd: workCwd, }); - const turnPrompt = - intent === "plan" && !rawCommand ? planTurnPrompt(prompt) : prompt; + const turnPrompt = proposalDraft + ? orchestrationPlanningPrompt(prompt, proposalDraft.settings) + : intent === "plan" && !rawCommand + ? planTurnPrompt(prompt) + : prompt; const earlier = queuedHandoff ? userMessagesAfterHandoff(current) : []; @@ -4609,21 +4831,34 @@ export default function App({ model: current.model, modelSettings: current.modelSettings, runtimeMode: current.runtimeMode, - intent, - text: inboxAskPrompt( - rawCommand ? undefined : current.inboxAsk, - wrap && !rawCommand - ? wrapHandoffPrompt( - wrap.text, - wrap.from, - turnPrompt.trim() || CONTINUE_PROMPT, - earlier, - ) - : turnPrompt, + intent: intent === "orchestrate" ? "plan" : intent, + // A lead drives the control CLI over loopback; without this the + // harness sandbox denies the socket and it cannot supervise. + controlsAgents: orchestrator.run(sessionId)?.status === "active", + text: orchestrator.prompt( + sessionId, + inboxAskPrompt( + rawCommand ? undefined : current.inboxAsk, + wrap && !rawCommand + ? wrapHandoffPrompt( + wrap.text, + wrap.from, + turnPrompt.trim() || CONTINUE_PROMPT, + earlier, + ) + : turnPrompt, + ), ), attachments: prepared, onEvent: (event) => { if (turnGen.current.get(sessionId) !== gen) return; + orchestrator.observe(sessionId, event); + if (options?.onSettled && event.type === "message.delta") + controlText = (controlText + event.text).slice(-20_000); + if (options?.onSettled && event.type === "message.completed") + controlText += "\n"; + if (event.type === "session.error") + controlOutcome.error = event.message; if ( wrap && (event.type === "session.started" || @@ -4632,7 +4867,8 @@ export default function App({ revealHandoff(wrap.text); } nudgeOpenEditors(event, workCwd); - trackSessionEdits(sessionId, workCwd, event); + if (!orchestrator.forSession(sessionId)) + trackSessionEdits(sessionId, workCwd, event); const routed = routePlanEvent(event); if (routed) enqueueHarnessEvent(sessionId, routed); }, @@ -4658,6 +4894,7 @@ export default function App({ error instanceof Error ? error.message : `${current.harness} adapter failed`; + controlOutcome.error = message; if (!providerFailureSeen) { enqueueHarnessEvent(sessionId, { type: "session.error", @@ -4668,6 +4905,16 @@ export default function App({ } finally { if (turnGen.current.get(sessionId) !== gen) return; flushHarnessEvents(); + controlOutcome = { + status: + providerFailureSeen || + isProviderFailureText(controlText) || + !buildSucceeded + ? "failed" + : "completed", + text: controlText.trim(), + ...(providerFailureSeen ? { error: controlOutcome.error } : {}), + }; // A failed provider can leave its process alive with a dead event // stream or poisoned turn state. Park it now; the next prompt will // reconnect and resume through a fresh transport. @@ -4685,9 +4932,22 @@ export default function App({ providerFailureSeen || isProviderFailureText(lastAssistantTextInTurn(stopped)); const finalized = - intent === "plan" && !nativePlanSeen && !providerFailed - ? promoteLastAssistantToPlan(stopped, planEventKey) - : stopped; + proposalDraft && proposalId + ? withOrchestrationProposal( + stopped, + proposalId, + completeOrchestrationProposal( + proposalDraft, + nativeProposalText || proposalText, + providerFailed || !buildSucceeded + ? (controlOutcome.error ?? + "The lead could not finish planning.") + : undefined, + ), + ) + : intent === "plan" && !nativePlanSeen && !providerFailed + ? promoteLastAssistantToPlan(stopped, planEventKey) + : stopped; return approvedPlan && intent === "build" ? withPlanStatus( finalized, @@ -4717,7 +4977,45 @@ export default function App({ nudgeWatchedFiles(); window.setTimeout(() => nudgeWatchedFiles(), 150); } - })(); + })() + .catch((error: unknown) => { + controlOutcome = { + status: "failed", + text: controlText, + error: error instanceof Error ? error.message : String(error), + }; + if (turnGen.current.get(sessionId) === gen) { + enqueueHarnessEvent(sessionId, { + type: "session.error", + message: controlOutcome.error!, + }); + flushHarnessEvents(); + setSessions((prev) => + prev.map((session) => + session.id === sessionId + ? proposalId && proposalDraft + ? withOrchestrationProposal( + stopStreaming(session), + proposalId, + completeOrchestrationProposal( + proposalDraft, + "", + controlOutcome.error, + ), + ) + : stopStreaming(session) + : session, + ), + ); + } + }) + .finally(() => { + options?.onSettled?.( + turnGen.current.get(sessionId) !== gen + ? { status: "cancelled", text: controlText } + : controlOutcome, + ); + }); }, [ dismissNoticesForContinuedSession, @@ -4772,6 +5070,7 @@ export default function App({ !session || session.busy || block?.role !== "plan" || + !!block.orchestration || !block.text.trim() || block.plan?.status === "streaming" || block.plan?.status === "building" || @@ -4903,14 +5202,23 @@ export default function App({ ? queuedMessageForSubmit(session, messageId, "steer") : undefined; if (!session || !message) return; + if (message.intent === "orchestrate" && session.busy) { + enqueueHarnessEvent(sessionId, { + type: "status", + text: "Orchestration planning will start after the current turn finishes.", + }); + flushHarnessEvents(); + return; + } onSubmit(sessionId, message.text, message.attachments, { followUpBehavior: "steer", queuedMessageId: message.id, noteCard: message.noteCard, handoffCard: message.handoffCard, + intent: message.intent, }); }, - [onSubmit], + [onSubmit, enqueueHarnessEvent, flushHarnessEvents], ); const onResumeQueue = useCallback( @@ -5153,7 +5461,14 @@ export default function App({ ); const onStop = useCallback( - (sessionId: string) => { + (sessionId: string, managed = false) => { + if (!managed) { + const stopping = orchestrator.stopForSession(sessionId); + if (stopping) { + void stopping.catch(console.error); + return; + } + } const session = sessionsRef.current.find((s) => s.id === sessionId); turnGen.current.set(sessionId, (turnGen.current.get(sessionId) ?? 0) + 1); flushHarnessEvents(); @@ -5252,20 +5567,414 @@ export default function App({ const onQuestionInteraction = useCallback( (sessionId: string, requestId: number) => { const session = sessionsRef.current.find((s) => s.id === sessionId); - if (session) keepHarnessQuestionOpen(session.harness, sessionId, requestId); + if (session) + keepHarnessQuestionOpen(session.harness, sessionId, requestId); }, [], ); const onOpenApprovalSession = useCallback( (sessionId: string) => { - if (!focusOpenSession(sessionId)) { + const parentId = + sessionsRef.current.find((session) => session.id === sessionId) + ?.orchestrationLeadId ?? orchestrator.forSession(sessionId)?.leadId; + if (parentId && parentId !== sessionId) { + setInspectedWorkerId(sessionId); + if (!focusOpenSession(parentId)) void onSelectHistorySession(parentId); + } else if (!focusOpenSession(sessionId)) { void onSelectHistorySession(sessionId); } }, [focusOpenSession, onSelectHistorySession], ); + useEffect(() => { + setSessions((prev) => attachOrchestrationWorkers(prev, orchestrationRuns)); + }, [orchestrationRuns]); + + useEffect(() => { + const next = consolidateOrchestrationTabs( + tabs, + activeTabId, + orchestrationRuns, + ); + if (next.tabs !== tabs) setTabs(next.tabs); + if (next.activeTabId !== activeTabId) setActiveTabId(next.activeTabId); + }, [tabs, activeTabId, orchestrationRuns]); + + useLayoutEffect(() => { + orchestrator.bind({ + session: (id) => sessionsRef.current.find((session) => session.id === id), + sessions: () => sessionsRef.current, + choices: () => + HARNESSES.filter(isHarnessAvailable).map((harness) => ({ + harness, + models: modelsFor(harness).map(({ id, name }) => ({ id, name })), + })), + createWorker: async (run, task) => { + await invoke("control_attach_worker", { + leadId: run.leadId, + sessionId: task.sessionId, + }); + const existing = sessionsRef.current.find( + (session) => session.id === task.sessionId, + ); + const lead = sessionsRef.current.find( + (session) => session.id === run.leadId, + ); + if (!lead) throw new Error("Lead session is unavailable"); + if (existing) { + if ( + existing.harness !== task.harness || + existing.model !== task.model || + existing.cwd !== run.cwd + ) + throw new Error( + "This worker's configuration changed. Restore its approved harness, model and project before retrying.", + ); + // The lead's runtime mode governs its agents, including across a + // change mid-run: auto stays auto, supervised asks the lead. + if (existing.runtimeMode !== lead.runtimeMode) { + const synced = { ...existing, runtimeMode: lead.runtimeMode }; + await upsertSession(synced); + const next = sessionsRef.current.map((session) => + session.id === synced.id ? synced : session, + ); + sessionsRef.current = next; + setSessions(next); + } + return; + } + const restored = await getSession(task.sessionId); + if ( + restored && + (restored.harness !== task.harness || restored.model !== task.model) + ) + throw new Error( + "The saved worker no longer matches its approved model. Create a new assignment.", + ); + const base = restored + ? { + ...restored, + busy: false, + cwd: run.cwd, + worktreeCwd: undefined, + runtimeMode: lead.runtimeMode, + } + : { + ...newSession( + task.harness, + run.cwd, + task.model, + lead.runtimeMode, + ), + id: task.sessionId, + title: task.title, + }; + const worker = { ...base, orchestrationLeadId: run.leadId }; + if (worker.providerSessionId) + bindHarnessSession( + worker.harness, + worker.id, + worker.providerSessionId, + worker.cwd, + ); + await upsertSession(worker); + const next = [...sessionsRef.current, worker]; + sessionsRef.current = next; + setSessions(next); + // Workers belong to the lead's agent panel; no workspace tab is created. + }, + submit: (id, text, done) => { + // Commit the new turn before the scheduler or confirmation updates + // another session snapshot in the same event loop. + flushSync(() => + onSubmit(id, text, [], { managed: true, onSettled: done }), + ); + }, + steer: async (id, text) => { + const session = sessionsRef.current.find((entry) => entry.id === id); + if (!session) throw new Error("This agent is no longer available"); + if (!session.busy) + throw new Error( + "This agent is not running a turn; send it a fresh one with message.", + ); + if ( + !isLiveHarness(session.harness) || + !canSteerHarness(session.harness) + ) + throw new Error( + `${session.harness} cannot take guidance mid-turn. Wait for the turn to finish, then use message.`, + ); + // Record it on the worker before dispatch, so its own transcript shows + // why it changed course even if the harness call then fails. + const next = sessionsRef.current.map((entry) => + entry.id === id ? appendSteerUser(entry, text) : entry, + ); + sessionsRef.current = next; + setSessions(next); + await steerHarnessTurn({ + harness: session.harness, + sessionId: id, + cwd: sessionWorkCwd(session), + model: session.model, + modelSettings: session.modelSettings, + text, + }); + }, + respondApproval: (id, requestId, decision) => { + const session = sessionsRef.current.find((entry) => entry.id === id); + if (session) + respondHarnessApproval(session.harness, id, requestId, decision); + }, + answerQuestion: (id, requestId, reply) => { + const session = sessionsRef.current.find((entry) => entry.id === id); + if (session) + respondHarnessQuestion(session.harness, id, requestId, reply); + }, + stop: async (id) => { + const session = sessionsRef.current.find((entry) => entry.id === id); + onStop(id, true); + try { + if (session) + await Promise.all( + sessionChildHarnesses(session).map((harness) => + stopHarnessSession(harness, id), + ), + ); + } finally { + // Also reap processes left behind by a renderer reload, before the + // corresponding session has been restored in this window. + await invoke("harness_kill", { sessionId: id }); + await invoke("control_turn_finished", { sessionId: id }); + } + }, + }); + }, [onSubmit, onStop]); + + useEffect(() => { + orchestrator.sync(); + }, [sessions]); + + useEffect(() => { + const listening = listen<{ + id: string; + sessionId: string; + requestId: string; + action: string; + input: Record; + }>("monocode-control-request", ({ payload }) => { + void orchestrator + .handle( + payload.sessionId, + payload.requestId, + payload.action, + payload.input, + ) + .then( + (result) => + invoke("control_reply", { + id: payload.id, + response: { ok: true, result }, + }), + (error: unknown) => + invoke("control_reply", { + id: payload.id, + response: { + ok: false, + error: error instanceof Error ? error.message : String(error), + }, + }), + ) + .catch(console.error); + }); + return () => { + void listening.then((unlisten) => unlisten()); + }; + }, []); + + const confirmingOrchestration = useRef(new Set()); + const queueWorkerPanes = useCallback( + (workers: OrchestrationWorkerDetail[]) => { + // Finished workers are not open; load stored transcripts before the + // tabs appear so the pane does not flash the empty state. + void prepareOrchestrationWorkerDetails(workers, { + openLead: async (leadId) => { + if (!focusOpenSession(leadId)) await onSelectHistorySession(leadId); + }, + openWorker: ensureOpenSession, + hasSession: (id) => + sessionsRef.current.some((session) => session.id === id), + }) + .then((request) => { + if (request?.workers.length) setWorkerDetailRequest(request); + }) + .catch(console.error); + }, + [ensureOpenSession, focusOpenSession, onSelectHistorySession], + ); + const onOpenWorkerDetails = useCallback( + (worker: OrchestrationWorkerDetail) => { + setInspectedWorkerId(worker.sessionId); + queueWorkerPanes([worker]); + }, + [queueWorkerPanes], + ); + useEffect(() => { + if (!workerDetailRequest) return; + const { leadId, workers } = workerDetailRequest; + const tab = tabs.find((entry) => leafIds(entry.layout).includes(leadId)); + if (!tab) { + // Still opening: this runs again on the commit that lands the lead. If + // the lead never arrived at all, drop the request rather than let it + // fire against some later tab change. + if (!sessionsRef.current.some((entry) => entry.id === leadId)) { + setWorkerDetailRequest(null); + } + return; + } + setWorkerDetailRequest(null); + // Every agent of a run shares one pane, the way files do: `openEditorTab` + // focuses an open tab, adds to the pane already beside the lead, or splits + // one off when there is none. + const cwd = + sessionsRef.current.find((entry) => entry.id === leadId)?.cwd ?? + projectCwdRef.current; + const files = workers.map((worker) => + newAgentTab(worker.title, cwd, { + sessionId: worker.sessionId, + leadId, + harness: worker.harness, + }), + ); + setTabs((prev) => + prev.map((entry) => { + if (entry.id !== tab.id) return entry; + const opened = files.reduce( + (next, file) => openEditorTab(next, file), + entry, + ); + // Leave the first worker focused so View agents lands on the start + // of the run rather than the last tab added. + return files[0] ? openEditorTab(opened, files[0]) : opened; + }), + ); + setActiveTabId(tab.id); + setComposerFocused(false); + }, [tabs, workerDetailRequest]); + const orchestrationWorkers = useMemo( + () => ({ + selectedId: inspectedWorkerId, + inspect: setInspectedWorkerId, + openDetails: onOpenWorkerDetails, + }), + [inspectedWorkerId, onOpenWorkerDetails], + ); + const updateOrchestrationCard = useCallback( + (leadId: string, blockId: string, proposal: OrchestrationProposal) => { + const next = sessionsRef.current.map((session) => + session.id === leadId + ? withOrchestrationProposal(session, blockId, proposal) + : session, + ); + sessionsRef.current = next; + setSessions(next); + return next.find((session) => session.id === leadId); + }, + [], + ); + const orchestrationActions = useMemo( + () => ({ + open: onOpenApprovalSession, + openAgents: queueWorkerPanes, + update: ( + leadId: string, + blockId: string, + edited: OrchestrationProposal, + ) => { + const session = sessionsRef.current.find( + (entry) => entry.id === leadId, + ); + const proposal = session?.blocks.find( + (block) => block.id === blockId, + )?.orchestration; + if ( + !session || + session.busy || + proposal?.status !== "ready" || + confirmingOrchestration.current.has(leadId) + ) + return; + // Keep the discovered catalog authoritative while allowing task and parallelism edits. + const settings = validateOrchestrationSettings({ + ...proposal.settings, + maxWorkers: edited.settings.maxWorkers, + }); + updateOrchestrationCard(leadId, blockId, { + ...proposal, + settings, + tasks: edited.tasks, + }); + }, + confirm: async (leadId: string, blockId: string) => { + if (confirmingOrchestration.current.has(leadId)) return; + confirmingOrchestration.current.add(leadId); + let proposal: OrchestrationProposal | undefined; + try { + await orchestrator.hydrate(leadId); + const session = sessionsRef.current.find( + (entry) => entry.id === leadId, + ); + proposal = session?.blocks.find( + (block) => block.id === blockId, + )?.orchestration; + if (!session || session.busy || proposal?.status !== "ready") + throw new Error( + "Wait for the proposal to finish before confirming.", + ); + if ( + session.harness !== proposal.author.harness || + session.model !== proposal.author.model + ) + throw new Error( + "The lead model has changed. Switch back to the model shown on this card, or generate a new proposal.", + ); + const starting = updateOrchestrationCard(leadId, blockId, { + ...proposal, + status: "starting", + })!; + // Save the edited card before anything can execute. + await upsertSession(starting); + await orchestrator.startApproved(leadId, blockId, proposal); + updateOrchestrationCard(leadId, blockId, { + ...proposal, + status: "approved", + }); + } catch (error) { + if (proposal) + updateOrchestrationCard(leadId, blockId, { + ...proposal, + status: "ready", + }); + throw error; + } finally { + confirmingOrchestration.current.delete(leadId); + } + }, + retry: (leadId: string, blockId: string) => { + const session = sessionsRef.current.find( + (entry) => entry.id === leadId, + ); + const proposal = session?.blocks.find( + (block) => block.id === blockId, + )?.orchestration; + if (!session || session.busy || !proposal) return; + onSubmit(leadId, proposal.request, [], { intent: "orchestrate" }); + }, + }), + [onOpenApprovalSession, queueWorkerPanes, onSubmit, updateOrchestrationCard], + ); + const onSelectLiveAgent = useCallback( (sessionId: string) => { setSearchViewOpen(false); @@ -5297,15 +6006,21 @@ export default function App({ const sidebarHistory = useMemo( () => - historyWithLiveSessions(history, sessions, sidebarCwd, { - ...(projectBranches?.current - ? { branch: projectBranches.current } - : {}), - ...(sidebarCwd && sidebarCwd !== "~" - ? { repo: projectName(sidebarCwd) } - : {}), - }), - [history, projectBranches, sessions, sidebarCwd], + historyWithLiveSessions( + history, + sessions, + sidebarCwd, + { + ...(projectBranches?.current + ? { branch: projectBranches.current } + : {}), + ...(sidebarCwd && sidebarCwd !== "~" + ? { repo: projectName(sidebarCwd) } + : {}), + }, + orchestrationRuns, + ), + [history, projectBranches, sessions, sidebarCwd, orchestrationRuns], ); const { unseen: inboxUnseen, @@ -5347,7 +6062,9 @@ export default function App({ sessions .filter( (session) => - !session.inboxAsk && sameProjectPath(session.cwd, sidebarCwd), + !session.inboxAsk && + !session.orchestrationLeadId && + sameProjectPath(session.cwd, sidebarCwd), ) .map((session) => summaryFromSession(session, { @@ -5983,410 +6700,421 @@ export default function App({ }; return ( -
- - session.busy && session.cwd ? [session.cwd] : [], - )} - liveAgents={liveAgents} - onSelectAgent={onSelectLiveAgent} - onSelectProject={onSelectProject} - onOpenProject={pickProject} - onRemoveProject={onRemoveProject} - onNew={onNew} - openSessions={openProjectSessions} - onNewTerminal={onNewTerminal} - onSearch={onOpenSearch} - onOpenInbox={onOpenInbox} - onOpenInboxItem={onOpenLinkedWorkItem} - onOpenNotes={notesEnabled ? onOpenNotes : undefined} - onGoToFile={onGoToFile} - searchActive={searchViewOpen} - inboxActive={inboxViewOpen} - notesActive={notesViewOpen} - notesEnabled={notesEnabled} - projectRailOpen={projectRailOpen} - onToggleProjectRail={onToggleProjectRail} - unseenFinishedIds={unseenFinishedIds} - inboxUnseen={inboxUnseen} - linkedSessionUpdateIds={linkedSessionUpdateIds} - settingsOpen={settingsOpen} - settingsSection={settingsSection} - onOpenSettings={onOpenSettings} - onSelectSettingsSection={onSelectSettingsSection} - onCloseSettings={onCloseSettings} - updateNotice={updateNotice} - onOpenWhatsNew={onOpenWhatsNew} - onDismissUpdate={() => setUpdateNotice(null)} - /> - -
+ +
- {!IS_MAC ? ( - onCloseTab(activeTabId) : undefined - } - onCloseOtherTabs={onCloseOtherTabs} - onCloseAllTabs={onCloseAllTabs} - onPickProject={pickProject} - onFindInProject={onFindInProject} - onSearch={onOpenSearch} - onOpenInbox={onOpenInbox} - onOpenNotes={notesEnabled ? onOpenNotes : undefined} - onZoomIn={() => { - const next = saveUiScale(zoomInUiScale(loadUiScale())); - void applyUiScale(next); - }} - onZoomOut={() => { - const next = saveUiScale(zoomOutUiScale(loadUiScale())); - void applyUiScale(next); - }} - onZoomReset={() => { - saveUiScale(UI_SCALE_DEFAULT); - void applyUiScale(UI_SCALE_DEFAULT); - }} - /> - ) : null} - + session.busy && session.cwd ? [session.cwd] : [], + )} + liveAgents={liveAgents} + onSelectAgent={onSelectLiveAgent} + onSelectProject={onSelectProject} + onOpenProject={pickProject} + onRemoveProject={onRemoveProject} onNew={onNew} + openSessions={openProjectSessions} onNewTerminal={onNewTerminal} - onShowTerminal={onShowProjectTerminal} - projectTerminalActive={ - !!currentProjectDock && currentProjectDock.pane.files.length > 0 - } - onOpenSettings={onOpenSettings} + onSearch={onOpenSearch} onOpenInbox={onOpenInbox} + onOpenInboxItem={onOpenLinkedWorkItem} onOpenNotes={notesEnabled ? onOpenNotes : undefined} - onClose={onCloseTitleTab} - onCloseMany={onCloseTabs} - onReorder={onReorderTabs} onGoToFile={onGoToFile} - recents={recents} - onSelectProject={onSelectProject} + searchActive={searchViewOpen} + inboxActive={inboxViewOpen} + notesActive={notesViewOpen} + notesEnabled={notesEnabled} + projectRailOpen={projectRailOpen} + onToggleProjectRail={onToggleProjectRail} + unseenFinishedIds={unseenFinishedIds} + inboxUnseen={inboxUnseen} + linkedSessionUpdateIds={linkedSessionUpdateIds} + settingsOpen={settingsOpen} + settingsSection={settingsSection} + onOpenSettings={onOpenSettings} + onSelectSettingsSection={onSelectSettingsSection} + onCloseSettings={onCloseSettings} + updateNotice={updateNotice} + onOpenWhatsNew={onOpenWhatsNew} + onDismissUpdate={() => setUpdateNotice(null)} /> -
+
- {projectTerminals.map((dock) => { - const show = - dock.open && sameProjectPath(dock.projectPath, projectCwd); - return ( -
- - onOpenTerminal(active?.cwd ?? projectCwd) - } - onSelectTerminal={onSelectProjectTerminal} - onCloseTerminal={onCloseProjectTerminal} - onCloseOtherTerminals={onCloseOtherProjectTerminals} - onReorderTerminals={onReorderProjectTerminals} - onTerminalMetaChange={onTerminalMetaChange} - /> -
- ); - })} -
-
- {tabs.map((tab) => ( -
-
- - onRatio(tab.id, splitId, index, ratio) + {!IS_MAC ? ( + onCloseTab(activeTabId) : undefined + } + onCloseOtherTabs={onCloseOtherTabs} + onCloseAllTabs={onCloseAllTabs} + onPickProject={pickProject} + onFindInProject={onFindInProject} + onSearch={onOpenSearch} + onOpenInbox={onOpenInbox} + onOpenNotes={notesEnabled ? onOpenNotes : undefined} + onZoomIn={() => { + const next = saveUiScale(zoomInUiScale(loadUiScale())); + void applyUiScale(next); + }} + onZoomOut={() => { + const next = saveUiScale(zoomOutUiScale(loadUiScale())); + void applyUiScale(next); + }} + onZoomReset={() => { + saveUiScale(UI_SCALE_DEFAULT); + void applyUiScale(UI_SCALE_DEFAULT); + }} + /> + ) : null} + 0 + } + onOpenSettings={onOpenSettings} + onOpenInbox={onOpenInbox} + onOpenNotes={notesEnabled ? onOpenNotes : undefined} + onClose={onCloseTitleTab} + onCloseMany={onCloseTabs} + onReorder={onReorderTabs} + onGoToFile={onGoToFile} + recents={recents} + onSelectProject={onSelectProject} + /> + +
+
+ {projectTerminals.map((dock) => { + const show = + dock.open && + sameProjectPath(dock.projectPath, projectCwd); + return ( +
+ + onOpenTerminal(active?.cwd ?? projectCwd) } - editorNavigation={editorNavigation} - onUpdatePlan={onUpdatePlan} - onMovePane={onMovePane} + onSelectTerminal={onSelectProjectTerminal} + onCloseTerminal={onCloseProjectTerminal} + onCloseOtherTerminals={onCloseOtherProjectTerminals} + onReorderTerminals={onReorderProjectTerminals} onTerminalMetaChange={onTerminalMetaChange} />
+ ); + })} +
+
+ {tabs.map((tab) => ( +
+
+ + onRatio(tab.id, splitId, index, ratio) + } + editorNavigation={editorNavigation} + onUpdatePlan={onUpdatePlan} + onMovePane={onMovePane} + onTerminalMetaChange={onTerminalMetaChange} + /> +
+
+ ))}
- ))} +
-
+
- -
- {searchViewOpen ? ( - !session.inboxAsk)} - focusToken={searchViewFocusToken} - besideRail={projectRailOpen} - onClose={onLeaveSearch} - onToggleSidebar={onToggleSidebar} - onOpenFile={onOpenFile} - onOpenSession={onSelectHistorySession} - onOpenProject={onSelectProject} - /> - ) : null} -
- {sessions - .filter((session) => session.inboxAsk) - .map((session) => { - const visible = - inboxViewOpen && inboxAskPortal?.sessionId === session.id; - return ( - - - - ); - })} -
- {inboxViewOpen ? ( - - ) : null} - {notesViewOpen ? ( - - ) : null} - {settingsOpen ? ( - - onRemoveProject(path, { purgeData: true }) + {searchViewOpen ? ( + !session.inboxAsk)} + focusToken={searchViewFocusToken} + besideRail={projectRailOpen} + onClose={onLeaveSearch} + onToggleSidebar={onToggleSidebar} + onOpenFile={onOpenFile} + onOpenSession={onSelectHistorySession} + onOpenProject={onSelectProject} + /> + ) : null} +
+ {sessions + .filter((session) => session.inboxAsk) + .map((session) => { + const visible = + inboxViewOpen && inboxAskPortal?.sessionId === session.id; + return ( + + + + ); + })} +
+ {inboxViewOpen ? ( + + ) : null} + {notesViewOpen ? ( + + ) : null} + {settingsOpen ? ( + + onRemoveProject(path, { purgeData: true }) + } + onOpenWhatsNew={onOpenWhatsNew} + /> + ) : null} + {searchViewOpen || + inboxViewOpen || + notesViewOpen || + settingsOpen ? null : ( + + )} +
+ + {filePickerOpen ? ( + setFilePickerOpen(false)} + /> + ) : null} + + - ) : null} - {searchViewOpen || - inboxViewOpen || - notesViewOpen || - settingsOpen ? null : ( - openSettings()} + onHeightChange={setReminderNoticesHeight} /> - )} - - - {filePickerOpen ? ( - setFilePickerOpen(false)} - /> - ) : null} - - - openSettings()} - onHeightChange={setReminderNoticesHeight} - /> - {whatsNewVersion ? ( - setWhatsNewVersion(null)} - /> - ) : null} - + {whatsNewVersion ? ( + setWhatsNewVersion(null)} + /> + ) : null} + + + ); } - function conversationTitle(session: Session): string { const title = sessionDisplayTitle(session.title, session.harness); return title === "New session" ? "" : title; diff --git a/src/chrome/Composer.tsx b/src/chrome/Composer.tsx index d2a0283a..4c811e75 100644 --- a/src/chrome/Composer.tsx +++ b/src/chrome/Composer.tsx @@ -9,6 +9,7 @@ import { Pencil, Play, Plus, + Share, Square, StickyNote, Trash2, @@ -66,7 +67,7 @@ import type { MessageQueueStatus, QueuedMessage, RuntimeMode, - TurnIntent, + ComposerTurnOptions, } from "../lib/session"; import { HARNESS_TITLE, harnessSupportsAttachments } from "../lib/session"; import type { @@ -181,7 +182,7 @@ type Props = { onSubmit: ( text: string, attachments: Attachment[], - options?: { intent?: TurnIntent }, + options?: ComposerTurnOptions, ) => void; onStop?: () => void; onCompactContext?: () => boolean; @@ -467,6 +468,7 @@ export function Composer({ const [fileDrag, setFileDrag] = useState(false); const [plusOpen, setPlusOpen] = useState(false); const [planSelected, setPlanSelected] = useState(false); + const [orchestrationSelected, setOrchestrationSelected] = useState(false); const [slash, setSlash] = useState(null); const [skillActive, setSkillActive] = useState(0); const [creatingSkill, setCreatingSkill] = useState(false); @@ -830,7 +832,10 @@ export function Composer({ syncHasValue(next, attachmentsRef.current); setSlash(null); setCreatingSkill(false); - if (planCommand) setPlanSelected(true); + if (planCommand) { + setPlanSelected(true); + setOrchestrationSelected(false); + } el.focus(); }, [onPlaceInFolder, openSessionFolderPicker, syncHasValue], @@ -1005,7 +1010,12 @@ export function Composer({ const files = attachments; if (!text && files.length === 0 && !noteCard && !handoffCard) return; onSubmit(text, files, { - intent: planSelected || command.planning ? "plan" : "default", + intent: + planSelected || command.planning + ? "plan" + : orchestrationSelected + ? "orchestrate" + : "default", }); if (!ref.current) return; ref.current.value = ""; @@ -1014,6 +1024,7 @@ export function Composer({ onDraftChange?.(""); setAttachments([]); setPlanSelected(false); + setOrchestrationSelected(false); setSessionFolderSelected(false); setSessionFolderOpen(false); setPlusOpen(false); @@ -1443,9 +1454,9 @@ export function Composer({ Upload file - + {attachmentsSupported - ? "Attach files or images to this message" + ? "Attach files or images" : `${HARNESS_TITLE[harness]} does not support attachments`} @@ -1456,6 +1467,7 @@ export function Composer({ onMouseDown={(e) => e.preventDefault()} onClick={() => { setPlanSelected((selected) => !selected); + setOrchestrationSelected(false); setPlusOpen(false); ref.current?.focus(); }} @@ -1464,17 +1476,64 @@ export function Composer({ Plan mode - - Create a plan to review before building + + Review a plan before building {planSelected ? ( ) : null} + {!hideTopBar && ( + + )} ) : null} + {orchestrationSelected && ( + + )} {planSelected ? ( + {open && ( + setOpen(false)} + data-model-picker + className="flex flex-col overflow-hidden" + > + +
+ {matches.map((choice, index) => ( + + ))} + {!matches.length && ( +

+ No matching models +

+ )} +
+
+ )} + + ); +} + +function WorkerHelp() { + const [hovered, setHovered] = useState(false); + const [open, setOpen] = useState(false); + const anchor = useRef(null); + return ( + <> + + {(hovered || open) && ( + setOpen(false)} + className={`px-2.5 py-2 ${open ? "" : "pointer-events-none"}`} + > +
+ How many workers run at once +
+
+ The rest of the tasks wait their turn, and a task that depends on + another waits for it either way. Every worker edits this same + project folder, so a lower number means fewer changes landing in it + at the same time. +
+
+ )} + + ); +} + +function InstructionsField({ + label, + value, + className, + onChange, +}: { + label: string; + value: string; + className: string; + onChange(value: string): void; +}) { + const ref = useRef(null); + const lockOverscroll = useLockOverscroll(); + // The same growth the composer uses: fit the text, stop at `max-h-40` and + // scroll from there. The value is controlled, so one layout effect covers + // typing and edits that arrive from the lead alike. + useLayoutEffect(() => { + if (ref.current) resizeComposer(ref.current); + }, [value]); + return ( +