diff --git a/apps/application/src/auth/oauth_service.rs b/apps/application/src/auth/oauth_service.rs index ba514d6..454bd04 100644 --- a/apps/application/src/auth/oauth_service.rs +++ b/apps/application/src/auth/oauth_service.rs @@ -104,7 +104,7 @@ fn generate_pkce_pair() -> (String, String) { bytes[..16].copy_from_slice(uuid::Uuid::new_v4().as_bytes()); bytes[16..].copy_from_slice(uuid::Uuid::new_v4().as_bytes()); - let verifier = URL_SAFE_NO_PAD.encode(&bytes); + let verifier = URL_SAFE_NO_PAD.encode(bytes); let challenge = { let mut hasher = Sha256::new(); hasher.update(verifier.as_bytes()); diff --git a/apps/application/src/auth/session_service.rs b/apps/application/src/auth/session_service.rs index e5ecf0a..a41b487 100644 --- a/apps/application/src/auth/session_service.rs +++ b/apps/application/src/auth/session_service.rs @@ -51,7 +51,7 @@ impl { fn create_session(&self, title: &str) -> Result { let id = SessionId::new(&Uuid::new_v4().to_string()) - .map_err(|e| ServiceError::Other(e))?; + .map_err(ServiceError::Other)?; let title_owned = if title.is_empty() { "New Session".to_string() } else { diff --git a/apps/gateway/src/bin/migrate.rs b/apps/gateway/src/bin/migrate.rs index 92ef7ac..0633330 100644 --- a/apps/gateway/src/bin/migrate.rs +++ b/apps/gateway/src/bin/migrate.rs @@ -4,7 +4,6 @@ //! schema for each one. Standalone CLI tool invoked as `cargo run --bin migrate`. use std::path::Path; -use tracing; fn main() -> anyhow::Result<()> { let store = zesdex_domain::core::Store::new(); diff --git a/apps/gateway/src/bin/seed.rs b/apps/gateway/src/bin/seed.rs index a243b4a..fd4636b 100644 --- a/apps/gateway/src/bin/seed.rs +++ b/apps/gateway/src/bin/seed.rs @@ -4,7 +4,6 @@ //! configuration files plus a seed session for development/testing. //! Invoked as `cargo run --bin seed`. -use tracing; fn main() -> anyhow::Result<()> { let store = zesdex_domain::core::Store::new(); diff --git a/apps/infrastructure/src/bgbash/control.rs b/apps/infrastructure/src/bgbash/control.rs index def04ad..e53c9bc 100644 --- a/apps/infrastructure/src/bgbash/control.rs +++ b/apps/infrastructure/src/bgbash/control.rs @@ -18,7 +18,15 @@ impl BashControl { jobs: Mutex::new(HashMap::new()), } } +} +impl Default for BashControl { + fn default() -> Self { + Self::new() + } +} + +impl BashControl { /// Register a new background job. pub fn register(&self, job: Arc) { if let Ok(mut guard) = self.jobs.lock() { diff --git a/apps/infrastructure/src/bgbash/job.rs b/apps/infrastructure/src/bgbash/job.rs index b30b30e..a77aac0 100644 --- a/apps/infrastructure/src/bgbash/job.rs +++ b/apps/infrastructure/src/bgbash/job.rs @@ -33,18 +33,38 @@ pub fn spawn_bash_job(cmd: String) -> Arc { cancelled: AtomicBool::new(false), }); - // Spawn a monitor thread (in production this would use an async task) + // Spawn a monitor thread using try_wait() polling so the lock is never + // held across a blocking wait, allowing cancel() to acquire the lock. let job_clone = Arc::clone(&job); std::thread::spawn(move || { - let mut guard = match job_clone.process.lock() { - Ok(g) => g, - Err(poisoned) => { - error!("bgbash job mutex poisoned, recovering"); - poisoned.into_inner() + loop { + let mut exited = false; + { + let mut guard = match job_clone.process.lock() { + Ok(g) => g, + Err(poisoned) => { + error!("bgbash job mutex poisoned, recovering"); + poisoned.into_inner() + } + }; + if let Some(ref mut child) = *guard { + match child.try_wait() { + Ok(Some(_)) => exited = true, + Ok(None) => {} // still running + Err(e) => { + error!("bgbash wait error: {e}"); + exited = true; + } + } + } else { + exited = true; // no child process + } + } // lock is dropped here — cancel() can now acquire it + + if exited || job_clone.cancelled.load(Ordering::SeqCst) { + break; } - }; - if let Some(ref mut child) = *guard { - let _ = child.wait(); + std::thread::sleep(std::time::Duration::from_millis(50)); } }); @@ -69,8 +89,6 @@ impl BashJob { let Ok(mut guard) = self.process.lock() else { return false; }; - guard.as_mut().map_or(false, |c| { - matches!(c.try_wait(), Ok(None)) - }) + guard.as_mut().is_some_and(|c| matches!(c.try_wait(), Ok(None))) } } diff --git a/apps/infrastructure/src/llm/provider.rs b/apps/infrastructure/src/llm/provider.rs index 01cb294..8430f8a 100644 --- a/apps/infrastructure/src/llm/provider.rs +++ b/apps/infrastructure/src/llm/provider.rs @@ -134,7 +134,7 @@ impl LlmClient { loop { attempt += 1; - if let Some(ref flag) = abort_flag { + if let Some(flag) = abort_flag { if flag.load(std::sync::atomic::Ordering::Relaxed) { anyhow::bail!("aborted"); } @@ -267,7 +267,7 @@ impl LlmClient { } if meaningful_content { - if let Some(ref flag) = abort_flag { + if let Some(flag) = abort_flag { if flag.load(std::sync::atomic::Ordering::Relaxed) { return Err(anyhow::anyhow!("aborted")); } diff --git a/apps/infrastructure/src/lsp/manager.rs b/apps/infrastructure/src/lsp/manager.rs index 2fdf3ad..40a8d56 100644 --- a/apps/infrastructure/src/lsp/manager.rs +++ b/apps/infrastructure/src/lsp/manager.rs @@ -19,6 +19,15 @@ impl LspManager { clients: HashMap::new(), } } +} + +impl Default for LspManager { + fn default() -> Self { + Self::new() + } +} + +impl LspManager { pub fn start(&mut self, language: &str, command: &str, args: &[String]) -> anyhow::Result<()> { let client = LspClient::start(command, args)?; @@ -31,7 +40,7 @@ impl LspManager { } pub fn shutdown_all(&mut self) { - for (_lang, client) in &self.clients { + for client in self.clients.values() { let _ = client.shutdown(); } self.clients.clear(); diff --git a/apps/infrastructure/src/mcp/manager.rs b/apps/infrastructure/src/mcp/manager.rs index 8d2689f..6f52fa2 100644 --- a/apps/infrastructure/src/mcp/manager.rs +++ b/apps/infrastructure/src/mcp/manager.rs @@ -22,6 +22,15 @@ impl McpManager { servers: HashMap::new(), } } +} + +impl Default for McpManager { + fn default() -> Self { + Self::new() + } +} + +impl McpManager { pub fn register(&mut self, name: &str, transport: &str) { self.servers.insert( diff --git a/apps/infrastructure/src/subagent/engine.rs b/apps/infrastructure/src/subagent/engine.rs index 11de5e9..8bc7e01 100644 --- a/apps/infrastructure/src/subagent/engine.rs +++ b/apps/infrastructure/src/subagent/engine.rs @@ -26,7 +26,7 @@ const MAX_ITERATIONS: u32 = 25; /// a. Call the LLM (non-streaming) with accumulated messages + tool defs. /// b. If the response has no tool calls → return the text content. /// c. Otherwise execute each tool call and append the result as a -/// tool-role message. +/// tool-role message. /// d. If the response also contained text, append an assistant message. /// 4. If the loop exits naturally, return the iteration-limit message. pub async fn run_agent( diff --git a/apps/infrastructure/src/subagent/provider.rs b/apps/infrastructure/src/subagent/provider.rs index 0b411f7..ba006d0 100644 --- a/apps/infrastructure/src/subagent/provider.rs +++ b/apps/infrastructure/src/subagent/provider.rs @@ -3,7 +3,7 @@ //! //! Flow: `resolve_subagent_provider` is called at startup to pick a provider //! + model → `SubagentProvider` wraps that pair around an `LlmClient` for use -//! inside the subagent engine loop. +//! inside the subagent engine loop. use anyhow::Result; diff --git a/apps/infrastructure/src/subagent/workspace.rs b/apps/infrastructure/src/subagent/workspace.rs index 6862272..9835300 100644 --- a/apps/infrastructure/src/subagent/workspace.rs +++ b/apps/infrastructure/src/subagent/workspace.rs @@ -1,9 +1,9 @@ //! Subagent workspace management — create isolated workspaces for subagents. -use std::path::PathBuf; +use std::path::{Path, PathBuf}; /// Create an isolated workspace directory for a subagent. -pub fn create_subagent_workspace(base_dir: &PathBuf, agent_id: &str) -> anyhow::Result { +pub fn create_subagent_workspace(base_dir: &Path, agent_id: &str) -> anyhow::Result { let ws = base_dir.join("subagent-workspaces").join(agent_id); std::fs::create_dir_all(&ws)?; Ok(ws) diff --git a/apps/infrastructure/src/workflow/hive_mind/live.rs b/apps/infrastructure/src/workflow/hive_mind/live.rs index 9260833..eff3e5c 100644 --- a/apps/infrastructure/src/workflow/hive_mind/live.rs +++ b/apps/infrastructure/src/workflow/hive_mind/live.rs @@ -14,6 +14,15 @@ impl LiveHiveMind { nodes: Mutex::new(HashMap::new()), } } +} + +impl Default for LiveHiveMind { + fn default() -> Self { + Self::new() + } +} + +impl LiveHiveMind { pub fn set_status(&self, agent_id: &str, status: &str) { if let Ok(mut guard) = self.nodes.lock() { diff --git a/apps/interfaces/daemon/src/client.rs b/apps/interfaces/daemon/src/client.rs index 753487a..3f30d38 100644 --- a/apps/interfaces/daemon/src/client.rs +++ b/apps/interfaces/daemon/src/client.rs @@ -217,7 +217,7 @@ fn draw(frame: &mut ratatui::Frame, state: &AppStateRest) { // Show toasts at the top if present. for toast in &state.misc.toasts { - content_lines.push(format!("[{}] {}", format!("{:?}", toast.kind), toast.message)); + content_lines.push(format!("[{:?}] {}", toast.kind, toast.message)); } // Show active overlay name. diff --git a/apps/interfaces/daemon/src/handler.rs b/apps/interfaces/daemon/src/handler.rs index b61c18f..74b7490 100644 --- a/apps/interfaces/daemon/src/handler.rs +++ b/apps/interfaces/daemon/src/handler.rs @@ -196,7 +196,10 @@ fn handle_tick(state: &mut AppStateRest) { .turn_events .lock() .map(|mut events| events.drain(..).collect()) - .unwrap_or_default(); + .unwrap_or_else(|e| { + tracing::error!("turn_events mutex poisoned, recovering"); + e.into_inner().drain(..).collect() + }); for event in drained { use zesdex_infrastructure::TurnEvent; @@ -209,11 +212,9 @@ fn handle_tick(state: &mut AppStateRest) { } handle_system_note(state, message); } - TurnEvent::AssistantMessage(msg) => { - state.push_transcript(ChatMessageDisplay::new( - RoleWrapper::Assistant, - msg.content.unwrap_or_default(), - )); + TurnEvent::AssistantMessage(_msg) => { + // AssistantMessage is handled via StreamDone to avoid duplicates. + state.dirty = true; } TurnEvent::StreamToken(_token) => { state.dirty = true; @@ -248,6 +249,11 @@ fn handle_tick(state: &mut AppStateRest) { } state.dirty = true; } + TurnEvent::ToolResult { .. } => { + // Tool result events are logged but the content is already in + // the session runtime messages — no transcript push needed here. + state.dirty = true; + } _ => { state.dirty = true; } @@ -392,11 +398,14 @@ fn handle_abort_turn(state: &mut AppStateRest) { fn handle_compact(state: &mut AppStateRest) { tracing::info!("compacting conversation"); - const KEEP_COUNT: usize = 10; + const KEEP_HEAD: usize = 2; // system prompt + tool definitions + const KEEP_TAIL: usize = 10; // recent conversation messages if let Some(ref mut rt) = state.session_runtime { - if rt.messages.len() > KEEP_COUNT { - let keep = rt.messages.split_off(rt.messages.len() - KEEP_COUNT); - rt.messages = keep; + if rt.messages.len() > KEEP_HEAD + KEEP_TAIL { + let tail = rt.messages.split_off(rt.messages.len() - KEEP_TAIL); + let head: Vec<_> = rt.messages.drain(..KEEP_HEAD.min(rt.messages.len())).collect(); + rt.messages = head; + rt.messages.extend(tail); let msg_count = rt.messages.len(); state.push_transcript(ChatMessageDisplay::new( RoleWrapper::System, @@ -418,14 +427,26 @@ fn handle_open_editor(state: &mut AppStateRest, path: String) { fn handle_mcp_add(state: &mut AppStateRest, name: String, command: String) { tracing::info!("adding MCP server: {name}"); - state.toast_info(format!("MCP server '{name}' registered with command: {command}")); + state.mcp_manager.register(&name, &command); + state.toast_success(format!("MCP server '{name}' added with command: {command}")); state.dirty = true; } fn handle_start_oauth(state: &mut AppStateRest, provider: String) { tracing::info!("starting OAuth for provider: {provider}"); state.toast_info(format!("OAuth flow started for {provider}...")); - if let Err(e) = webbrowser::open(&format!("https://{provider}.com/auth")) { + // Construct provider-specific OAuth authorization URL. + let auth_url_owned; + let url_to_open = match provider.as_str() { + "github" => "https://github.com/login/oauth/authorize", + "google" => "https://accounts.google.com/o/oauth2/v2/auth", + "anthropic" => "https://anthropic.com/api/oauth/authorize", + other => { + auth_url_owned = format!("https://{other}.com/auth"); + &auth_url_owned + } + }; + if let Err(e) = webbrowser::open(url_to_open) { tracing::warn!("Failed to open browser for OAuth: {e}"); state.toast_error(format!("Failed to open browser: {e}")); } @@ -749,10 +770,14 @@ pub fn handle_daemon_client( if let Some(text) = state.misc.pending_clipboard_copy.take() { conn.send(&DaemonFrame::ClipboardCopy(text))?; } - send_daemon_update(&mut conn, state)?; + // Only send state update for valid requests, not on EOF/disconnect. + if running { + send_daemon_update(&mut conn, state)?; + } } None => { running = false; + // Don't call send_daemon_update — connection is dead. } } } diff --git a/apps/interfaces/tui/src/state.rs b/apps/interfaces/tui/src/state.rs index d9df03d..b630621 100644 --- a/apps/interfaces/tui/src/state.rs +++ b/apps/interfaces/tui/src/state.rs @@ -788,7 +788,7 @@ pub fn count_tokens(text: &str) -> usize { return bpe.encode_with_special_tokens(text).len(); } // Fallback: ~4 chars per token - (text.len() + 3) / 4 + text.len().div_ceil(4) } // --------------------------------------------------------------------------- diff --git a/apps/interfaces/tui/src/turn.rs b/apps/interfaces/tui/src/turn.rs index 9e4ff02..b65e142 100644 --- a/apps/interfaces/tui/src/turn.rs +++ b/apps/interfaces/tui/src/turn.rs @@ -3,7 +3,7 @@ //! Flow: push user message → spawn OS thread → loop: call blocking LLM //! client → execute tool calls → push TurnEvents → repeat until done. -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::collections::VecDeque; @@ -53,7 +53,7 @@ pub fn spawn_agent_turn(state: &mut AppStateRest, text: String) { /// The core agent turn — LLM call → tool execution → repeat. fn run_turn( messages: &mut Vec, - session_dir: &PathBuf, + session_dir: &Path, workspace_roots: &[PathBuf], turn_events: &Arc>>, in_flight: &Arc>, @@ -89,7 +89,7 @@ fn run_turn( messages.insert(0, sys_msg); let tool_ctx = ToolCtx::builder() - .session_dir(session_dir.clone()) + .session_dir(session_dir.to_path_buf()) .workspaces(workspace_roots.to_vec()) .build();