From 75c609117e2f63cf4d9b9f9c43cfdcc9d7b300d4 Mon Sep 17 00:00:00 2001 From: Nick <60738984+hardbeat920@users.noreply.github.com> Date: Sat, 12 Sep 2026 11:50:00 +0100 Subject: [PATCH 01/13] Add local agent orchestration workflow - Add orchestrator UI, planning, delegation, and worker tracking - Add authenticated local control CLI and loopback transport - Persist orchestration history and group worker sessions in the sidebar --- Cargo.lock | 1 + src-tauri/Cargo.toml | 1 + src-tauri/src/control.rs | 467 +++++++++ src-tauri/src/control_cli.rs | 170 ++++ src-tauri/src/harness.rs | 2 + src-tauri/src/lib.rs | 13 + src-tauri/src/main.rs | 5 + src-tauri/src/session_store.rs | 229 ++++- src/App.tsx | 442 ++++++++- src/chrome/Composer.tsx | 62 +- src/chrome/OrchestrationActions.ts | 22 + src/chrome/OrchestrationFlow.test.ts | 469 +++++++++ src/chrome/OrchestrationPanel.tsx | 336 +++++++ src/chrome/OrchestrationPreview.tsx | 465 +++++++++ src/chrome/OrchestrationSidebarAgents.tsx | 77 ++ src/chrome/Sidebar.tsx | 32 +- src/chrome/SidebarRename.test.ts | 52 + src/lib/approvalToast.test.ts | 38 +- src/lib/approvalToast.ts | 2 +- src/lib/harness/apply.ts | 17 +- src/lib/harness/registry.ts | 9 + src/lib/liveAgents.test.ts | 5 + src/lib/liveAgents.ts | 2 +- src/lib/orchestration.test.ts | 496 ++++++++++ src/lib/orchestration.ts | 1064 +++++++++++++++++++++ src/lib/orchestrationCatalog.test.ts | 68 ++ src/lib/orchestrationCatalog.ts | 27 + src/lib/orchestrationPlan.test.ts | 137 +++ src/lib/orchestrationPlan.ts | 267 ++++++ src/lib/orchestrationSummary.ts | 57 ++ src/lib/orchestrationWorkspace.test.ts | 75 ++ src/lib/orchestrationWorkspace.ts | 56 ++ src/lib/session.ts | 9 +- src/lib/sessionHistory.test.ts | 124 ++- src/lib/sessionHistory.ts | 46 +- src/lib/sessionStore.test.ts | 17 + src/lib/sessionStore.ts | 29 +- src/surfaces/AgentTranscript.test.ts | 51 + src/surfaces/AgentTranscript.tsx | 21 +- src/surfaces/PaneTree.tsx | 4 +- src/surfaces/SessionPane.tsx | 217 +++-- 41 files changed, 5510 insertions(+), 173 deletions(-) create mode 100644 src-tauri/src/control.rs create mode 100644 src-tauri/src/control_cli.rs create mode 100644 src/chrome/OrchestrationActions.ts create mode 100644 src/chrome/OrchestrationFlow.test.ts create mode 100644 src/chrome/OrchestrationPanel.tsx create mode 100644 src/chrome/OrchestrationPreview.tsx create mode 100644 src/chrome/OrchestrationSidebarAgents.tsx create mode 100644 src/lib/orchestration.test.ts create mode 100644 src/lib/orchestration.ts create mode 100644 src/lib/orchestrationCatalog.test.ts create mode 100644 src/lib/orchestrationCatalog.ts create mode 100644 src/lib/orchestrationPlan.test.ts create mode 100644 src/lib/orchestrationPlan.ts create mode 100644 src/lib/orchestrationSummary.ts create mode 100644 src/lib/orchestrationWorkspace.test.ts create mode 100644 src/lib/orchestrationWorkspace.ts diff --git a/Cargo.lock b/Cargo.lock index 63b2101d..04e133c5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2237,6 +2237,7 @@ dependencies = [ "tauri-plugin-updater", "tauri-plugin-window-state", "ureq", + "uuid", "windows-sys 0.61.2", ] diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index d023c040..015eb637 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..dfcc5b6e --- /dev/null +++ b/src-tauri/src/control.rs @@ -0,0 +1,467 @@ +//! 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 Inner { + grants: HashMap, + pending: HashMap, + workers: HashMap, + active: HashMap, +} +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 { + grants: HashMap::new(), + pending: HashMap::new(), + workers: HashMap::new(), + active: HashMap::new(), + })); + 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, path)| id != &session_id && paths_overlap(path, &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, 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 }; + let leads: Vec = inner + .grants + .values() + .filter(|grant| grant.window == label) + .map(|grant| grant.session.clone()) + .collect(); + let mut ids = leads.clone(); + ids.extend( + inner + .workers + .iter() + .filter(|(_, lead)| leads.contains(lead)) + .map(|(id, _)| id.clone()), + ); + ids + }; + for id in &ids { + let _ = crate::harness::harness_kill(app.state(), id.clone()); + } + if let Ok(mut inner) = host.inner.lock() { + inner.grants.retain(|id, _| !ids.contains(id)); + inner.workers.retain(|id, _| !ids.contains(id)); + inner.active.retain(|id, _| !ids.contains(id)); + inner.pending.retain(|_, pending| { + if pending.window != label { + return true; + } + let _ = pending + .reply + .send(json!({"ok":false,"error":"MonoCode window closed"})); + false + }); + }; +} + +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 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..eaa63849 --- /dev/null +++ b/src-tauri/src/control_cli.rs @@ -0,0 +1,170 @@ +//! 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}; + +pub const HELP: &str = r#"MonoCode local control + +Usage: monocode control ACTION [--json JSON | --input FILE|-] [--request-id ID] + +Actions: + list List the run, workers and available harness/model choices. + delegate {"title":"Fix UI","harness":"codex","prompt":"...","files":["src/ui"],"dependsOn":[]} + get {"taskId":"..."} + message {"taskId":"...","text":"..."} Send a follow-up worker turn. + cancel {"taskId":"..."} Cancel a worker, including queued work. + wait {"timeoutSeconds":20} Wait for worker state changes (maximum 25s). + review {"taskId":"..."} Accept a completed worker result. + finish Complete the run after all remaining results are accepted. + +All responses are JSON. --input - reads JSON from stdin. Use --request-id to +retry a mutation without duplicating it. A worker task belongs to the running +MonoCode app, not this CLI process. Closing the CLI does not cancel the task. + +Choose Orchestrator from MonoCode's composer + menu and confirm its proposal. +MonoCode then supplies +MONOCODE_CONTROL_ENDPOINT and MONOCODE_CONTROL_TOKEN to that harness only. +"#; + +pub fn run(args: Vec) -> i32 { + if args.is_empty() || matches!(args[0].as_str(), "help" | "--help" | "-h") { + println!("{HELP}"); + return 0; + } + match execute(args) { + Ok(value) => { + println!("{value}"); + if value.get("ok").and_then(Value::as_bool) == Some(true) { + 0 + } else { + 1 + } + } + Err(error) => { + println!("{}", json!({"ok":false,"error":error})); + 1 + } + } +} + +fn execute(args: Vec) -> Result { + let (action, input, request_id) = parse_args(&args)?; + let endpoint = std::env::var("MONOCODE_CONTROL_ENDPOINT").map_err(|_| { + "No MonoCode connection. Confirm the Orchestrator proposal in MonoCode first." + })?; + let token = std::env::var("MONOCODE_CONTROL_TOKEN") + .map_err(|_| "No MonoCode session credential. Start the lead from MonoCode.")?; + let address: SocketAddr = endpoint.parse().map_err(|_| "Invalid MonoCode endpoint")?; + if !address.ip().is_loopback() { + return Err("MonoCode control only connects to localhost".into()); + } + let mut stream = TcpStream::connect_timeout(&address, Duration::from_secs(3)) + .map_err(|_| "MonoCode is not running or this connection has expired.")?; + stream + .set_read_timeout(Some(Duration::from_secs(40))) + .map_err(|e| e.to_string())?; + stream + .set_write_timeout(Some(Duration::from_secs(5))) + .map_err(|e| e.to_string())?; + writeln!( + stream, + "{}", + json!({"token":token,"action":action,"input":input,"requestId":request_id}) + ) + .map_err(|e| e.to_string())?; + let mut line = String::new(); + BufReader::new(stream) + .take(2_000_001) + .read_line(&mut line) + .map_err(|e| e.to_string())?; + if line.len() > 2_000_000 { + return Err("MonoCode response is too large".into()); + } + serde_json::from_str(&line).map_err(|_| "MonoCode returned an invalid response".into()) +} + +fn parse_args(args: &[String]) -> Result<(String, Value, String), String> { + let action = args.first().ok_or("Missing action")?.clone(); + if ![ + "list", "delegate", "get", "message", "cancel", "wait", "review", "finish", + ] + .contains(&action.as_str()) + { + return Err(format!("Unknown action: {action}. Run control --help.")); + } + 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]; + let value = args + .get(index + 1) + .ok_or_else(|| format!("Missing value for {flag}"))?; + 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" { + value.clone() + } else if value == "-" { + let mut raw = String::new(); + std::io::stdin() + .take(262_145) + .read_to_string(&mut raw) + .map_err(|e| e.to_string())?; + raw + } else { + let file = std::fs::File::open(value).map_err(|e| e.to_string())?; + let mut raw = String::new(); + file.take(262_145) + .read_to_string(&mut raw) + .map_err(|e| e.to_string())?; + raw + }; + if raw.len() > 262_144 { + return Err("Input exceeds 256 KiB".into()); + } + let parsed: Value = + serde_json::from_str(&raw).map_err(|e| format!("Invalid JSON: {e}"))?; + if !parsed.is_object() { + return Err("Input must be a JSON object".into()); + } + input = Some(parsed); + } + _ => return Err(format!("Unknown option: {flag}")), + } + index += 2; + } + if request_id.is_empty() || request_id.len() > 128 { + return Err("Invalid request ID".into()); + } + Ok((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() + } + #[test] + fn validates_inputs_without_invoking_a_shell() { + let (_, input, id) = parse_args(&args(&[ + "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!(parse_args(&args(&["delegate", "--json", "[]"])).is_err()); + assert!(parse_args(&args(&["delegate", "--json", "{}", "--json", "{}"])).is_err()); + assert!(parse_args(&args(&["unknown"])).is_err()); + } +} 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 7b933d11..905bd61d 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; @@ -175,6 +177,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())?; @@ -198,6 +201,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, @@ -376,6 +388,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..2c24d914 100644 --- a/src-tauri/src/session_store.rs +++ b/src-tauri/src/session_store.rs @@ -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)?, @@ -1123,6 +1259,7 @@ fn get_session(conn: &Connection, session_id: &str) -> rusqlite::Result(null); const openingInboxSessions = useRef(new Map>()); const [notesViewOpen, setNotesViewOpen] = useState(false); + const [inspectedWorkerId, setInspectedWorkerId] = useState(null); + const orchestrationRuns = useSyncExternalStore(orchestrator.subscribe, orchestrator.snapshot, orchestrator.snapshot); const notesEnabled = useSyncExternalStore( subscribeNotesEnabled, loadNotesEnabled, @@ -801,6 +819,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, ); @@ -964,7 +983,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]); @@ -1010,7 +1032,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]); @@ -1357,6 +1382,13 @@ 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); + } const keepUnseen = liveAgentsEnabled; const idleDetached = sessions.filter( (session) => @@ -1381,7 +1413,7 @@ 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); @@ -2832,9 +2864,15 @@ export default function App({ const onSelectHistorySession = useCallback( async (sessionId: string) => { - if (focusOpenSession(sessionId)) return; - const session = await ensureOpenSession(sessionId); + let session = await ensureOpenSession(sessionId); if (!session || session.inboxAsk) return; + const parentId = session.orchestrationLeadId ?? orchestrator.forSession(sessionId)?.leadId; + if (parentId && parentId !== sessionId) { + setInspectedWorkerId(sessionId); + session = await ensureOpenSession(parentId); + if (!session) return; + } + if (focusOpenSession(session.id)) return; if (replaceBlankPaneWithSession(session)) return; const tab = newTab(session.id); appendTab(tab, session.cwd); @@ -3793,8 +3831,23 @@ 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; @@ -3802,6 +3855,16 @@ 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) => @@ -3846,7 +3909,7 @@ export default function App({ if (current.busy && !pendingSwitch) { const followUpBehavior = - intent === "plan" + intent === "plan" || intent === "orchestrate" ? "queue" : (options?.followUpBehavior ?? loadFollowUpBehavior()); if (followUpBehavior === "queue") { @@ -3944,6 +4007,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, @@ -4096,10 +4179,46 @@ 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, @@ -4151,6 +4270,11 @@ 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; @@ -4162,7 +4286,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; @@ -4178,7 +4302,7 @@ export default function App({ cwd: workCwd, }); const turnPrompt = - intent === "plan" && !rawCommand ? planTurnPrompt(prompt) : prompt; + proposalDraft ? orchestrationPlanningPrompt(prompt, proposalDraft.settings) : intent === "plan" && !rawCommand ? planTurnPrompt(prompt) : prompt; const earlier = queuedHandoff ? userMessagesAfterHandoff(current) : []; @@ -4189,8 +4313,8 @@ export default function App({ model: current.model, modelSettings: current.modelSettings, runtimeMode: current.runtimeMode, - intent, - text: inboxAskPrompt( + intent: intent === "orchestrate" ? "plan" : intent, + text: orchestrator.prompt(sessionId, inboxAskPrompt( rawCommand ? undefined : current.inboxAsk, wrap && !rawCommand ? wrapHandoffPrompt( @@ -4200,10 +4324,14 @@ export default function App({ 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" || @@ -4212,7 +4340,7 @@ 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); }, @@ -4238,6 +4366,7 @@ export default function App({ error instanceof Error ? error.message : `${current.harness} adapter failed`; + controlOutcome.error = message; if (!providerFailureSeen) { enqueueHarnessEvent(sessionId, { type: "session.error", @@ -4248,6 +4377,11 @@ 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. @@ -4264,8 +4398,9 @@ export default function App({ const providerFailed = providerFailureSeen || isProviderFailureText(lastAssistantTextInTurn(stopped)); - const finalized = - intent === "plan" && !nativePlanSeen && !providerFailed + const finalized = 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" @@ -4297,7 +4432,18 @@ 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); + }); }, [enqueueHarnessEvent, flushHarnessEvents], ); @@ -4348,6 +4494,7 @@ export default function App({ !session || session.busy || block?.role !== "plan" || + !!block.orchestration || !block.text.trim() || block.plan?.status === "streaming" || block.plan?.status === "building" || @@ -4479,14 +4626,20 @@ 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( @@ -4729,7 +4882,11 @@ 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(); @@ -4835,13 +4992,252 @@ export default function App({ 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); + 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."); + return; + } + const lead = sessionsRef.current.find( + (session) => session.id === run.leadId, + ); + if (!lead) throw new Error("Lead session is unavailable"); + 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 } + : { + ...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 })); + }, + 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 orchestrationWorkers = useMemo(() => ({ + sessions, selectedId: inspectedWorkerId, inspect: setInspectedWorkerId, + }), [sessions, inspectedWorkerId]); + 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, + 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, onSubmit, updateOrchestrationCard], + ); + const onSelectLiveAgent = useCallback( (sessionId: string) => { setSearchViewOpen(false); @@ -4880,8 +5276,8 @@ export default function App({ ...(sidebarCwd && sidebarCwd !== "~" ? { repo: projectName(sidebarCwd) } : {}), - }), - [history, projectBranches, sessions, sidebarCwd], + }, orchestrationRuns), + [history, projectBranches, sessions, sidebarCwd, orchestrationRuns], ); const inboxRelatedSessions = useMemo(() => { const byId = new Map(); @@ -4917,7 +5313,7 @@ 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, { @@ -5543,6 +5939,8 @@ export default function App({ }; return ( + +
) : null}
+ + ); } diff --git a/src/chrome/Composer.tsx b/src/chrome/Composer.tsx index dc0d4107..b39cfc5b 100644 --- a/src/chrome/Composer.tsx +++ b/src/chrome/Composer.tsx @@ -1,6 +1,7 @@ import { ArrowUp, AiIdea, + MessageMultiple, Check, CornerDownRight, FilePlus, @@ -66,7 +67,7 @@ import type { MessageQueueStatus, QueuedMessage, RuntimeMode, - TurnIntent, + ComposerTurnOptions, } from "../lib/session"; import { HARNESS_TITLE, harnessSupportsAttachments } from "../lib/session"; import type { @@ -179,7 +180,7 @@ type Props = { onSubmit: ( text: string, attachments: Attachment[], - options?: { intent?: TurnIntent }, + options?: ComposerTurnOptions, ) => void; onStop?: () => void; onCompactContext?: () => boolean; @@ -465,6 +466,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); @@ -823,7 +825,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], @@ -998,7 +1003,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 = ""; @@ -1007,6 +1017,7 @@ export function Composer({ onDraftChange?.(""); setAttachments([]); setPlanSelected(false); + setOrchestrationSelected(false); setSessionFolderSelected(false); setSessionFolderOpen(false); setPlusOpen(false); @@ -1449,6 +1460,7 @@ export function Composer({ onMouseDown={(e) => e.preventDefault()} onClick={() => { setPlanSelected((selected) => !selected); + setOrchestrationSelected(false); setPlusOpen(false); ref.current?.focus(); }} @@ -1465,9 +1477,51 @@ export function Composer({ ) : null} + {!hideTopBar && ( + + )} ) : null} + {orchestrationSelected && ( + + )} {planSelected ? ( + +

+ {working} working · {finished}/{run.tasks.length} accepted +

+ {!collapsed && ( +
    + {run.tasks.map((task) => { + const worker = workers.sessions.find( + (entry) => entry.id === task.sessionId, + ); + const approval = worker && pendingApprovalForSession(worker); + const approvalDetail = + approval?.block?.tool?.detail?.trim() || approval?.block?.text; + const open = workers.selectedId === task.sessionId || !!approval; + const latest = latestWorkerBlock(worker); + const activity = + approval?.label ?? + orchestrator.waitingFor(run, task) ?? + (latest?.tool + ? toolCallLabel(latest, run.cwd) + : latest?.text) ?? + task.result; + const model = + run.allowedModels?.find( + (choice) => + choice.harness === task.harness && + choice.model === task.model, + )?.name ?? task.model; + return ( +
  • + + {!open && activity && ( +

    + {activity.slice(0, 180)} +

    + )} + {open && ( +
    + {activity && ( +

    + {activity.slice(-2000)} +

    + )} +

    + {task.files.join(", ")} +

    + {task.error && ( +

    {task.error}

    + )} + {approval?.kind === "approval" && ( + <> + {approvalDetail && ( + 
    +                              {approvalDetail}
+
    + )} +
    + + +
    + + )} + {worker?.pendingQuestion && ( + + onQuestionReply(task.sessionId, requestId, reply) + } + onInteraction={(requestId) => + onQuestionInteraction?.(task.sessionId, requestId) + } + /> + )} + {["queued", "running"].includes(task.status) && ( + + )} +
    + )} +
    • + ); + })} +
+ )} + {(error || run.error) && ( +

+ {error ?? run.error} +

+ )} +
+

+ Direct the agents through your lead in the main conversation. +

+
+ + {run.maxWorkers} parallel · {run.status} + + {run.status === "paused" && ( + + )} + {["active", "paused"].includes(run.status) && ( + + )} +
+
+ + + ); +} diff --git a/src/chrome/OrchestrationPreview.tsx b/src/chrome/OrchestrationPreview.tsx new file mode 100644 index 00000000..40c53ef8 --- /dev/null +++ b/src/chrome/OrchestrationPreview.tsx @@ -0,0 +1,465 @@ +import { + useContext, + useEffect, + useRef, + useState, + useSyncExternalStore, +} from "react"; +import { HARNESS_TITLE, type Block } from "../lib/session"; +import type { + OrchestrationChoice, + ProposedTask, +} from "../lib/orchestrationPlan"; +import { orchestrator } from "../lib/orchestration"; +import { OrchestrationActions } from "./OrchestrationActions"; +import { HarnessIcon } from "./HarnessIcon"; +import { Popover } from "./Popover"; +import { + Check, + ChevronDown, + ChevronRight, + CircleDashed, + MessageMultiple, + Play, + Search, +} from "./icons"; + +function AssignmentModel({ + task, + choices, + onChange, +}: { + task: ProposedTask; + choices: OrchestrationChoice[]; + onChange(choice: OrchestrationChoice): void; +}) { + const [open, setOpen] = useState(false); + const [query, setQuery] = useState(""); + const anchor = useRef(null); + const selected = choices.find( + (choice) => choice.harness === task.harness && choice.model === task.model, + ); + return ( +
+ + {open && ( + setOpen(false)} + className="p-1.5" + > +

+ Available models +

+ +
+ {choices + .filter((choice) => + `${choice.name} ${choice.model} ${HARNESS_TITLE[choice.harness]}` + .toLowerCase() + .includes(query.toLowerCase()), + ) + .map((choice) => ( + + ))} + {!choices.some((choice) => + `${choice.name} ${choice.model} ${HARNESS_TITLE[choice.harness]}` + .toLowerCase() + .includes(query.toLowerCase()), + ) && ( +

+ No matching models +

+ )} +
+ + )} +
+ ); +} + +export function OrchestrationPreview({ + block, + busy, +}: { + block: Block; + busy?: boolean; +}) { + const actions = useContext(OrchestrationActions); + const runs = useSyncExternalStore( + orchestrator.subscribe, + orchestrator.snapshot, + orchestrator.snapshot, + ); + const proposal = block.orchestration!; + const run = runs.find( + (entry) => + entry.leadId === proposal.leadId && entry.proposalId === block.id, + ); + const [pending, setPending] = useState(false); + const [error, setError] = useState(); + const [expanded, setExpanded] = useState([]); + const [showAll, setShowAll] = useState(false); + useEffect(() => { + if (actions) + void orchestrator + .hydrate(proposal.leadId) + .catch((reason: unknown) => setError(String(reason))); + }, [actions, proposal.leadId]); + const editable = + proposal.status === "ready" && !run && !pending && !busy && !!actions; + const planning = proposal.status === "planning"; + const starting = pending || proposal.status === "starting"; + const visible = showAll ? proposal.tasks : proposal.tasks.slice(0, 3); + const perform = async (fn: () => Promise) => { + setPending(true); + setError(undefined); + try { + await fn(); + } catch (reason) { + setError(reason instanceof Error ? reason.message : String(reason)); + } finally { + setPending(false); + } + }; + const change = (id: string, patch: Partial) => { + setError(undefined); + actions?.update(proposal.leadId, block.id, { + ...proposal, + tasks: proposal.tasks.map((task) => + task.id === id ? { ...task, ...patch } : task, + ), + }); + }; + const secondary = + "flex h-7 items-center gap-1.5 rounded-md px-2.5 text-[11px] text-content/50 hover:bg-content/8 hover:text-content disabled:opacity-35"; + const field = + "mt-1 w-full rounded-md border border-content/10 bg-background-base/40 px-2 py-1.5 text-[12px] text-content/85 outline-none focus:border-content/30"; + return ( +
+
+ + {planning || proposal.status === "starting" ? ( + + ) : ( + + )} + +
+
+ {planning ? "Planning assignments…" : proposal.title} +
+
+ + + Lead · {proposal.author.name} + + {!!proposal.tasks.length && ( + + · {proposal.tasks.length}{" "} + {proposal.tasks.length === 1 ? "task" : "tasks"} + + )} +
+
+
+ {!run && proposal.status === "invalid" && ( + + )} + {!run && ["ready", "starting"].includes(proposal.status) && ( + + )} + {run && ( + + )} +
+
+ {(planning || proposal.summary) && ( +

+ {planning + ? proposal.settings.choices.length + ? "Your lead is choosing tasks and worker models. Review the assignments here before starting." + : "Checking available harnesses and models…" + : proposal.summary} +

+ )} + {!!proposal.tasks.length && ( +
    + {visible.map((task) => { + const index = proposal.tasks.indexOf(task); + const open = expanded.includes(task.id); + return ( +
  • +
    + +
    + {editable ? ( + + change(task.id, { + harness: choice.harness, + model: choice.model, + }) + } + /> + ) : ( + + + + {proposal.settings.choices.find( + (choice) => + choice.harness === task.harness && + choice.model === task.model, + )?.name ?? task.model} + + + )} +
    +
    + {open && ( +
    + {editable ? ( + <> + +