diff --git a/apps/application/src/agent/mod.rs b/apps/application/src/agent/mod.rs new file mode 100644 index 0000000..c072edd --- /dev/null +++ b/apps/application/src/agent/mod.rs @@ -0,0 +1,25 @@ +use anyhow::Result; +use std::future::Future; + +use zesdex_domain::agent::AgentTurnParams; + +/// Interface for dispatching tool calls to their concrete implementations. +pub trait ToolExecutor: Send + Sync { + /// Execute a tool call asynchronously. + fn execute( + &self, + tool_name: &str, + args: &serde_json::Value, + ) -> impl Future> + Send; +} + +/// Service for running agent turns asynchronously. +pub trait AgentTurnService: Send + Sync { + /// Run a full agent turn loop asynchronously. + fn run_turn( + &self, + params: AgentTurnParams, + ) -> impl Future> + Send; +} + +pub mod turn_service; diff --git a/apps/application/src/agent/turn_service.rs b/apps/application/src/agent/turn_service.rs new file mode 100644 index 0000000..d565176 --- /dev/null +++ b/apps/application/src/agent/turn_service.rs @@ -0,0 +1,227 @@ +use std::collections::VecDeque; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use tracing::{debug, info, warn}; + +use zesdex_domain::agent::{AgentTurnParams, TurnEvent}; +use zesdex_domain::core::{ChatMessage, StreamEvent, ToolDef}; + +use crate::ports::ProviderService; +use super::ToolExecutor; + +/// Service implementation for executing an agent turn asynchronously. +pub struct AgentTurnServiceImpl { + provider: Arc

, + tool_executor: Arc, + tool_defs: Vec, +} + +impl AgentTurnServiceImpl { + pub fn new(provider: Arc

, tool_executor: Arc, tool_defs: Vec) -> Self { + Self { + provider, + tool_executor, + tool_defs, + } + } + + fn push_event(queue: &Arc>>, event: TurnEvent) { + if let Ok(mut q) = queue.lock() { + q.push_back(event); + } + } + + fn mark_done(flag: &Arc) { + flag.store(false, Ordering::SeqCst); + } +} + +impl super::AgentTurnService for AgentTurnServiceImpl { + async fn run_turn(&self, mut params: AgentTurnParams) -> anyhow::Result<()> { + info!( + "Starting async agent turn with {} messages (model: {})", + params.messages.len(), + params.model + ); + + let mut sys_prompt = "You are Zesdex, an AI coding assistant. You have access to various tools via native function calling to help the user.\n\n\ + CRITICAL DIRECTIVES & PRIORITY HIERARCHY:\n\ + 1. WORKFLOW FIRST: For any multi-step, complex, or non-trivial task, you MUST prioritize using `workflow_run` (to construct and execute a multi-phase YAML workflow) or `hive_mind` (to orchestrate parallel autonomous agents). Workflows are your primary strategy.\n\ + 2. PLANNING & TODOS: Use `plan_enter` to establish high-level architectural plans and `todowrite` to maintain granular task checklists.\n\ + 3. REASONING: Use `seq_think` for deep step-by-step analysis.\n\ + 4. TOOL EXECUTION: Execute individual tools (file edits, terminal commands) within or guided by your workflows. If an error occurs, analyze and fix it.\n\n\ + Respond conversationally, concisely, and helpfully.".to_string(); + + let sys_msg = ChatMessage::system(sys_prompt); + + for iteration in 0..50 { + if params.abort.load(Ordering::SeqCst) { + params.abort.store(false, Ordering::SeqCst); + Self::push_event( + ¶ms.turn_events, + TurnEvent::SystemNote { + kind: "info".into(), + message: "Turn aborted by user".into(), + }, + ); + break; + } + + debug!("agent turn iteration {iteration}"); + + Self::push_event(¶ms.turn_events, TurnEvent::StreamStart); + + let mut req_messages = params.messages.clone(); + req_messages.insert(0, sys_msg.clone()); + + let abort_clone = Arc::clone(¶ms.abort); + let turn_events_clone = Arc::clone(¶ms.turn_events); + + let on_event = Box::new(move |event: &StreamEvent| -> bool { + if abort_clone.load(Ordering::SeqCst) { + return false; + } + match event { + StreamEvent::Token(s) => { + Self::push_event(&turn_events_clone, TurnEvent::StreamToken(s.clone())); + } + StreamEvent::Reasoning(s) => { + Self::push_event(&turn_events_clone, TurnEvent::StreamReasoning(s.clone())); + } + _ => {} + } + true + }); + + let result = self.provider.chat_stream( + &req_messages, + Some(self.tool_defs.clone()), + Some(4096), + Some(0.7), + on_event, + ).await; + + match result { + Ok((assistant_msg, usage)) => { + let content = assistant_msg.content.clone().unwrap_or_default(); + let tool_calls = assistant_msg.tool_calls.clone().unwrap_or_default(); + + Self::push_event( + ¶ms.turn_events, + TurnEvent::StreamDone(assistant_msg.clone()), + ); + + if let Some((tokens_in, tokens_out)) = usage { + Self::push_event( + ¶ms.turn_events, + TurnEvent::Usage { + tokens_in, + tokens_out, + }, + ); + } + + if tool_calls.is_empty() { + params.messages.push(ChatMessage::assistant(Some(content))); + break; + } + + params.messages.push(assistant_msg); + + for tc in &tool_calls { + let name = &tc.function.name; + let args = zesdex_domain::core::tool_call::sanitize_tool_arguments(&tc.function.arguments); + + debug!("executing tool: {name}"); + + let output = match self.tool_executor.execute(name, &args).await { + Ok(o) => o, + Err(e) => format!("Error: {e}"), + }; + + let is_error = output.starts_with("Error:"); + + Self::push_event( + ¶ms.turn_events, + TurnEvent::ToolResult { + tool_call_id: tc.id.clone(), + tool_name: name.clone(), + output: output.clone(), + is_error, + path: None, + }, + ); + + params + .messages + .push(ChatMessage::tool(tc.id.clone(), output.clone())); + } + } + Err(e) => { + warn!("LLM call failed: {e}"); + Self::push_event( + ¶ms.turn_events, + TurnEvent::Error(format!("LLM error: {e}")), + ); + break; + } + } + } + + Self::push_event( + ¶ms.turn_events, + TurnEvent::Compacted(params.messages.clone()), + ); + Self::push_event(¶ms.turn_events, TurnEvent::Done); + Self::mark_done(¶ms.in_flight); + + Ok(()) + } +} + +/// Compacts conversation history using AI summarization. +pub async fn compact_messages_with_ai( + messages: &mut Vec, + provider: &P, +) -> anyhow::Result<()> { + const KEEP_TAIL: usize = 6; + if messages.len() <= KEEP_TAIL + 2 { + return Ok(()); // Not enough messages to compact + } + + let split_idx = messages.len() - KEEP_TAIL; + let evicted: Vec<_> = messages.drain(..split_idx).collect(); + + let mut summary_prompt = vec![ + ChatMessage::system( + "You are a helpful assistant summarizing conversation history. \ + Provide a concise summary of the key user requests, decisions, tools executed, and modified files. \ + Format as a clear bulleted list." + .to_string(), + ), + ]; + summary_prompt.extend(evicted); + summary_prompt.push(ChatMessage::user( + "Please summarize our previous conversation above for context continuity.".to_string(), + )); + + match provider.chat(&summary_prompt, None, Some(1024), Some(0.3)).await { + Ok((summary_msg, _)) => { + let summary_text = summary_msg.content.unwrap_or_else(|| "Previous context summarized.".to_string()); + let summary_node = ChatMessage::system(format!( + "[AI Summary of Previous Conversation]\n{}", + summary_text.trim() + )); + messages.insert(0, summary_node); + Ok(()) + } + Err(e) => { + warn!("AI summarization failed during compact, falling back to simple notice: {e}"); + messages.insert( + 0, + ChatMessage::system("[Earlier conversation messages compacted to save context window]".to_string()), + ); + Ok(()) + } + } +} diff --git a/apps/application/src/lib.rs b/apps/application/src/lib.rs index 7600bdb..53bbae3 100644 --- a/apps/application/src/lib.rs +++ b/apps/application/src/lib.rs @@ -33,6 +33,7 @@ pub mod auth; pub mod cms; pub mod ports; +pub mod agent; // Re-export port traits for ergonomic access. pub use ports::*; @@ -49,3 +50,5 @@ pub use cms::{ memory_service::MemoryServiceImpl, settings_service::SettingsServiceImpl, }; + +pub use agent::{AgentTurnService, ToolExecutor, turn_service::{AgentTurnServiceImpl, compact_messages_with_ai}}; diff --git a/apps/domain/src/agent/mod.rs b/apps/domain/src/agent/mod.rs new file mode 100644 index 0000000..6ab7382 --- /dev/null +++ b/apps/domain/src/agent/mod.rs @@ -0,0 +1,242 @@ +use serde::{Deserialize, Serialize}; +use std::path::PathBuf; + +use crate::core::{ChatMessage, ToolCallResult, UsageStats}; + +/// Which kind of caller (main agent vs. subagent vs. reviewer) is +/// invoking a tool, used to scope permissions and tag log/output paths. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Hash)] +pub enum Origin { + /// The main agent turn loop. + Main, + /// A spawned subagent (test-gen, arch-review, security-review, etc.). + SubAgent, + /// The auto-inline review step after an edit. + Reviewer, +} + +impl Origin { + /// Short string tag for this origin, used in filenames and logs. + pub fn tag(self) -> String { + match self { + Origin::Main => "main", + Origin::SubAgent => "subagent", + Origin::Reviewer => "reviewer", + } + .to_string() + } +} + +/// Severity/category of a toast notification, used to pick its color. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum ToastKind { + Info, + Success, + Warning, + Error, + Lesson, +} + +/// A transient status message shown in the TUI, auto-dismissed after `lifetime_ms`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Toast { + pub kind: ToastKind, + pub message: String, + pub created_at: i64, + pub lifetime_ms: u64, +} + +impl Toast { + /// Create a toast with a default 5-second lifetime, stamped with now. + pub fn new(kind: ToastKind, message: String) -> Self { + Toast { + kind, + message, + created_at: chrono::Utc::now().timestamp_millis(), + lifetime_ms: 5000, + } + } + + /// Whether this toast's lifetime has elapsed as of `now_ms`. + pub fn expired(&self, now_ms: i64) -> bool { + let lifetime = self.lifetime_ms as i64; + now_ms - self.created_at > lifetime + } +} + +/// Agent status for workflow engine progress tracking. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub enum AgentStatus { + Pending, + Running, + Completed, + Failed(String), + Cancelled, +} + +impl std::fmt::Display for AgentStatus { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + AgentStatus::Pending => write!(f, "pending"), + AgentStatus::Running => write!(f, "running"), + AgentStatus::Completed => write!(f, "completed"), + AgentStatus::Failed(msg) => write!(f, "failed: {msg}"), + AgentStatus::Cancelled => write!(f, "cancelled"), + } + } +} + +/// Events emitted onto the turn-event queue while an agent turn runs, +/// consumed by the event loop to update state and drive re-renders. +#[derive(Debug, Clone)] +pub enum TurnEvent { + AssistantMessage(ChatMessage), + ToolResult { + tool_call_id: String, + tool_name: String, + output: String, + is_error: bool, + path: Option, + }, + SystemNote { + kind: String, + message: String, + }, + StreamStart, + StreamToken(String), + StreamReasoning(String), + StreamDone(ChatMessage), + Usage { + tokens_in: u64, + tokens_out: u64, + }, + ReviewUsage { + tokens_in: u64, + tokens_out: u64, + }, + Compacted(Vec), + Error(String), + Done, + WorkflowAgentUpdate { + agent_id: String, + agent_name: String, + status: AgentStatus, + }, + TodoUpdate(String), + PlanUpdate(String), +} + +/// How a pending tool call should be executed when the turn resumes. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum ExecutionModel { + Inline, + Deferred, + AsyncTokio, +} + +/// A tool call awaiting execution, along with which execution model +/// (inline, deferred, async) it should run under. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PendingTool { + pub tool_name: String, + pub args: serde_json::Value, + pub execution_model: ExecutionModel, +} + +/// Reference to a background bash job tracked in session state. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct BashJobRef { + pub id: String, + pub command: String, + pub started_at: i64, + pub running: bool, +} + +/// Per-session runtime state: message history, pending tool queue, +/// background bash jobs, lesson/review counters. +#[derive(Debug, Clone)] +pub struct SessionRuntime { + pub messages: Vec, + pub tool_call_results: Vec, + pub pending_tool_queue: Vec, + pub bash_jobs: Vec, + pub subagent_queue: usize, + pub edit_count: u32, + pub consecutive_empty_reviews: u32, + pub session_start: i64, + pub lesson_count: u32, + pub lessons_user: u32, + pub lessons_feedback: u32, + pub lessons_project: u32, + pub lessons_reference: u32, + pub lessons_active: u32, + pub lessons_stale: u32, + pub lessons_contradicted: u32, + pub lessons_human: u32, + pub lessons_verified: u32, + pub lessons_unverified: u32, + pub review_count: u32, + pub session_dir: PathBuf, + pub usage: UsageStats, + pub hive_mind_converged: bool, +} + +impl SessionRuntime { + pub fn new(session_dir: PathBuf) -> Self { + SessionRuntime { + messages: Vec::new(), + tool_call_results: Vec::new(), + pending_tool_queue: Vec::new(), + bash_jobs: Vec::new(), + subagent_queue: 0, + edit_count: 0, + consecutive_empty_reviews: 0, + session_start: chrono::Utc::now().timestamp_millis(), + lesson_count: 0, + lessons_user: 0, + lessons_feedback: 0, + lessons_project: 0, + lessons_reference: 0, + lessons_active: 0, + lessons_stale: 0, + lessons_contradicted: 0, + lessons_human: 0, + lessons_verified: 0, + lessons_unverified: 0, + review_count: 0, + session_dir, + usage: UsageStats::default(), + hive_mind_converged: false, + } + } + + pub fn push_message(&mut self, msg: ChatMessage) { + self.messages.push(msg); + } +} + +/// Simple ASCII progress display for a long-running operation. +#[derive(Debug, Clone)] +pub struct ProgressState { + pub current: u64, + pub total: u64, + pub message: String, + pub start_time: i64, +} + +use std::collections::VecDeque; +use std::sync::atomic::AtomicBool; +use std::sync::{Arc, Mutex}; + +/// Owned parameters required to spawn and execute an agent turn. +pub struct AgentTurnParams { + pub messages: Vec, + pub session_dir: PathBuf, + pub workspace_roots: Vec, + pub turn_events: Arc>>, + pub in_flight: Arc, + pub abort: Arc, + pub api_key: String, + pub model: String, + pub api_base: Option, +} diff --git a/apps/domain/src/lib.rs b/apps/domain/src/lib.rs index a5e9d03..dd94f9d 100644 --- a/apps/domain/src/lib.rs +++ b/apps/domain/src/lib.rs @@ -28,6 +28,9 @@ pub mod auth; pub mod cms; pub mod core; pub mod error; +pub mod agent; +pub mod workflow; +pub mod subagent; // Re-export all public items from each module for ergonomic imports. // Consumers can do `use zesdex_domain::*` for common types. @@ -50,3 +53,6 @@ pub use core::{ ToolCallResult, ToolDef, ToolFunction, ToolFunctionDef, UsageStats, }; pub use error::DomainError; +pub use agent::*; +pub use workflow::*; +pub use subagent::*; diff --git a/apps/domain/src/subagent/mod.rs b/apps/domain/src/subagent/mod.rs new file mode 100644 index 0000000..9416471 --- /dev/null +++ b/apps/domain/src/subagent/mod.rs @@ -0,0 +1,41 @@ +//! Subagent domain models. + +/// Events emitted by a running subagent. +#[derive(Debug, Clone)] +pub enum SubagentEvent { + Started { + agent_id: String, + directive: String, + }, + ToolCall { + agent_id: String, + tool_name: String, + }, + ToolResult { + agent_id: String, + tool_name: String, + output: String, + }, + Completed { + agent_id: String, + output: String, + }, + Failed { + agent_id: String, + error: String, + }, +} + +/// Access tier for subagent tool permissions. +/// +/// Tiers are cumulative: `Write` includes everything in `Read`, and `Full` +/// includes everything in `Write`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AccessTier { + /// Read-only: search, read, glob, utility tools (no mutations). + Read, + /// Read + Write: above plus write, edit, delete, git, memory. + Write, + /// Full: above plus bash, shell, LSP, workflow, plan tools. + Full, +} diff --git a/apps/domain/src/workflow/mod.rs b/apps/domain/src/workflow/mod.rs new file mode 100644 index 0000000..d112dda --- /dev/null +++ b/apps/domain/src/workflow/mod.rs @@ -0,0 +1,45 @@ +//! Workflow and Hive-mind domain models. + +/// A single phase in a parsed workflow script. +#[derive(Debug, Clone)] +pub struct WorkflowPhase { + pub name: String, + pub directive: String, +} + +/// A parsed workflow script with named phases. +#[derive(Debug, Clone)] +pub struct WorkflowScript { + pub name: String, + pub phases: Vec, +} + +/// A directive for a single processing node in the hive mind. +#[derive(Debug, Clone)] +pub struct NodeDirective { + pub directive: String, + pub access_tier: String, +} + +/// A cognitive cycle plan — ordered list of cycles, each containing +/// parallel node directives. +#[derive(Debug, Clone)] +pub struct CognitiveCyclePlan { + pub cycles: Vec>, +} + +/// A single cycle in a cognitive cycle plan — parallel node directives +/// executed together. +#[derive(Debug, Clone)] +pub struct CognitiveCycle { + pub index: u32, + pub directives: Vec, +} + +/// Output from a single hive-mind processing node after a cycle completes. +#[derive(Debug, Clone)] +pub struct NodeOutput { + pub id: String, + pub directive: String, + pub output: String, +} diff --git a/apps/infrastructure/src/agent/mod.rs b/apps/infrastructure/src/agent/mod.rs deleted file mode 100644 index b2e8e4f..0000000 --- a/apps/infrastructure/src/agent/mod.rs +++ /dev/null @@ -1,6 +0,0 @@ -//! Agent execution engine — orchestrates LLM streaming, system prompt assembly, -//! tool execution, and turn event emitting on background threads. - -pub mod runner; - -pub use runner::{compact_messages_with_ai, spawn_agent_turn, AgentTurnParams}; diff --git a/apps/infrastructure/src/agent/runner.rs b/apps/infrastructure/src/agent/runner.rs deleted file mode 100644 index 44dc013..0000000 --- a/apps/infrastructure/src/agent/runner.rs +++ /dev/null @@ -1,304 +0,0 @@ -//! Agent turn engine runner — handles LLM API calls, system prompt assembly, -//! and tool execution loops on background threads. - -use std::collections::VecDeque; -use std::path::PathBuf; -use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex}; - -use tracing::{debug, info, warn}; - -use zesdex_domain::core::tool_call::sanitize_tool_arguments; -use zesdex_domain::core::ChatMessage; -use crate::llm::provider::LlmClient; -use crate::subagent::auto::engine::spawn_background_review; -use crate::tools::{all_tools, tool_defs, ToolCtx}; -use crate::TurnEvent; - -/// Owned parameters required to spawn and execute an agent turn on a background thread. -pub struct AgentTurnParams { - pub messages: Vec, - pub session_dir: PathBuf, - pub workspace_roots: Vec, - pub turn_events: Arc>>, - pub in_flight: Arc, - pub abort: Arc, - pub api_key: String, - pub model: String, - pub api_base: Option, -} - -/// Spawns an agent turn on a background OS thread. -#[tracing::instrument(skip(params))] -pub fn spawn_agent_turn(mut params: AgentTurnParams) { - info!( - "spawning agent turn with {} messages (model: {})", - params.messages.len(), - params.model - ); - - std::thread::spawn(move || { - run_turn(&mut params); - }); -} - -/// The core agent turn loop — LLM call → tool execution → repeat. -#[tracing::instrument(skip(params))] -fn run_turn(params: &mut AgentTurnParams) { - let client = LlmClient::new( - params.api_key.clone(), - params.model.clone(), - params.api_base.clone(), - ); - - let tools = all_tools(); - let defs = tool_defs(&tools); - - let mut sys_prompt = "You are Zesdex, an AI coding assistant. You have access to various tools via native function calling to help the user.\n\n\ - CRITICAL DIRECTIVES & PRIORITY HIERARCHY:\n\ - 1. WORKFLOW FIRST: For any multi-step, complex, or non-trivial task, you MUST prioritize using `workflow_run` (to construct and execute a multi-phase YAML workflow) or `hive_mind` (to orchestrate parallel autonomous agents). Workflows are your primary strategy.\n\ - 2. PLANNING & TODOS: Use `plan_enter` to establish high-level architectural plans and `todowrite` to maintain granular task checklists.\n\ - 3. REASONING: Use `seq_think` for deep step-by-step analysis.\n\ - 4. TOOL EXECUTION: Execute individual tools (file edits, terminal commands) within or guided by your workflows. If an error occurs, analyze and fix it.\n\n\ - Respond conversationally, concisely, and helpfully.".to_string(); - - if let Some(root) = params.workspace_roots.first() { - let tree = crate::utils::build_workspace_tree(root, 800); - let rich_ctx = crate::utils::build_rich_context(root); - - sys_prompt.push_str(&format!("\n\n### Workspace Root\n`{}`\n\n", root.display())); - sys_prompt.push_str("Workspace structure:\n```\n"); - sys_prompt.push_str(&tree); - sys_prompt.push_str("\n```\n\n"); - - sys_prompt.push_str(&rich_ctx); - } - - // Auto-compact if conversation history is getting long (>24 messages) - if params.messages.len() > 24 { - push_event( - ¶ms.turn_events, - TurnEvent::SystemNote { - kind: "info".into(), - message: "Auto-compacting context window via AI Summarization...".into(), - }, - ); - let _ = compact_messages_with_ai(&mut params.messages, &client); - push_event( - ¶ms.turn_events, - TurnEvent::Compacted(params.messages.clone()), - ); - } - - let sys_msg = ChatMessage::system(sys_prompt); - - let tool_ctx = ToolCtx::builder() - .session_dir(params.session_dir.clone()) - .workspaces(params.workspace_roots.clone()) - .turn_events(Arc::clone(¶ms.turn_events)) - .build(); - - for iteration in 0..50 { - if params.abort.load(Ordering::SeqCst) { - params.abort.store(false, Ordering::SeqCst); - push_event( - ¶ms.turn_events, - TurnEvent::SystemNote { - kind: "info".into(), - message: "Turn aborted by user".into(), - }, - ); - break; - } - - debug!("agent turn iteration {iteration}"); - - // Stream the LLM response - push_event(¶ms.turn_events, TurnEvent::StreamStart); - - let mut req_messages = params.messages.clone(); - req_messages.insert(0, sys_msg.clone()); - - let result = client.chat_with_tools_streaming( - &req_messages, - Some(defs.clone()), - Some(0.7), - Some(4096), - |event| { - if params.abort.load(Ordering::SeqCst) { - return false; - } - match event { - zesdex_domain::core::StreamEvent::Token(s) => { - push_event(¶ms.turn_events, TurnEvent::StreamToken(s.clone())); - } - zesdex_domain::core::StreamEvent::Reasoning(s) => { - push_event(¶ms.turn_events, TurnEvent::StreamReasoning(s.clone())); - } - _ => {} - } - true - }, - Some(¶ms.abort), - ); - - match result { - Ok((assistant_msg, usage)) => { - let content = assistant_msg.content.clone().unwrap_or_default(); - let tool_calls = assistant_msg.tool_calls.clone().unwrap_or_default(); - - push_event( - ¶ms.turn_events, - TurnEvent::StreamDone(assistant_msg.clone()), - ); - - if let Some((tokens_in, tokens_out)) = usage { - push_event( - ¶ms.turn_events, - TurnEvent::Usage { - tokens_in, - tokens_out, - }, - ); - } - - if tool_calls.is_empty() { - params.messages.push(ChatMessage::assistant(Some(content))); - break; - } - - params.messages.push(assistant_msg); - - for tc in &tool_calls { - let name = &tc.function.name; - let args = sanitize_tool_arguments(&tc.function.arguments); - - debug!("executing tool: {name}"); - - let output = if let Some(tool) = tools.iter().find(|t| t.name() == name) { - match tool.run(&tool_ctx, &args) { - Ok(o) => o, - Err(e) => format!("Error: {e}"), - } - } else { - format!("Unknown tool: {name}") - }; - - let is_error = output.starts_with("Error:"); - - push_event( - ¶ms.turn_events, - TurnEvent::ToolResult { - tool_call_id: tc.id.clone(), - tool_name: name.clone(), - output: output.clone(), - is_error, - path: None, - }, - ); - - params - .messages - .push(ChatMessage::tool(tc.id.clone(), output.clone())); - } - } - Err(e) => { - warn!("LLM call failed: {e}"); - push_event( - ¶ms.turn_events, - TurnEvent::Error(format!("LLM error: {e}")), - ); - break; - } - } - } - - // If edits were made, spawn a background auto-review after the turn ends. - // This runs asynchronously — findings arrive as TurnEvent::SystemNote events. - let had_edits = params - .messages - .iter() - .any(|m| { - m.role == zesdex_domain::core::Role::Tool - && m.content.as_deref().unwrap_or("").contains("Written") - }); - if had_edits { - info!("edits detected, spawning background auto-review"); - spawn_background_review( - params.workspace_roots.clone(), - params.turn_events.clone(), - params.api_key.clone(), - params.model.clone(), - params.api_base.clone(), - ); - } - - // Propagate accumulated messages back to caller so the next turn starts - // with full history (assistant replies + tool results). - push_event( - ¶ms.turn_events, - TurnEvent::Compacted(params.messages.clone()), - ); - push_event(¶ms.turn_events, TurnEvent::Done); - mark_done(¶ms.in_flight); -} - -fn push_event(queue: &Arc>>, event: TurnEvent) { - if let Ok(mut q) = queue.lock() { - q.push_back(event); - } -} - -/// Mark the turn as done using lock-free atomic store. -fn mark_done(flag: &Arc) { - flag.store(false, Ordering::SeqCst); -} - -/// Compacts conversation history using AI summarization. -/// -/// Preserves recent messages (last 6) and calls the LLM to summarize -/// the older evicted messages into a single system summary message. -pub fn compact_messages_with_ai(messages: &mut Vec, client: &LlmClient) -> anyhow::Result<()> { - const KEEP_TAIL: usize = 6; - if messages.len() <= KEEP_TAIL + 2 { - return Ok(()); // Not enough messages to compact - } - - let split_idx = messages.len() - KEEP_TAIL; - let evicted: Vec<_> = messages.drain(..split_idx).collect(); - - // Prepare prompt to summarize evicted messages - let mut summary_prompt = vec![ - ChatMessage::system( - "You are a helpful assistant summarizing conversation history. \ - Provide a concise summary of the key user requests, decisions, tools executed, and modified files. \ - Format as a clear bulleted list." - .to_string(), - ), - ]; - summary_prompt.extend(evicted); - summary_prompt.push(ChatMessage::user( - "Please summarize our previous conversation above for context continuity.".to_string(), - )); - - match client.chat_with_tools_non_streaming(&summary_prompt, None, Some(1024), Some(0.3), None) { - Ok((summary_msg, _)) => { - let summary_text = summary_msg.content.unwrap_or_else(|| "Previous context summarized.".to_string()); - let summary_node = ChatMessage::system(format!( - "[AI Summary of Previous Conversation]\n{}", - summary_text.trim() - )); - messages.insert(0, summary_node); - Ok(()) - } - Err(e) => { - warn!("AI summarization failed during compact, falling back to simple notice: {e}"); - messages.insert( - 0, - ChatMessage::system("[Earlier conversation messages compacted to save context window]".to_string()), - ); - Ok(()) - } - } -} - diff --git a/apps/infrastructure/src/lib.rs b/apps/infrastructure/src/lib.rs index 6d7b4e4..93d453f 100644 --- a/apps/infrastructure/src/lib.rs +++ b/apps/infrastructure/src/lib.rs @@ -26,7 +26,6 @@ //! └── middleware/ — Axum HTTP middleware (auth, cors, rate-limit) //! ``` -pub mod agent; pub mod auth; pub mod bgbash; pub mod guard; @@ -52,70 +51,24 @@ pub use zesdex_domain::*; // from the legacy backend code. // --------------------------------------------------------------------------- -use serde::{Deserialize, Serialize}; use std::path::PathBuf; use std::sync::Arc; /// Which kind of caller (main agent vs. subagent vs. reviewer) is /// invoking a tool, used to scope permissions and tag log/output paths. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Hash)] -pub enum Origin { - /// The main agent turn loop. - Main, - /// A spawned subagent (test-gen, arch-review, security-review, etc.). - SubAgent, - /// The auto-inline review step after an edit. - Reviewer, -} +// --------------------------------------------------------------------------- +// TurnEvent & runtime types have been moved to zesdex_domain::agent +// --------------------------------------------------------------------------- -impl Origin { - /// Short string tag for this origin, used in filenames and logs. - pub fn tag(self) -> String { - match self { - Origin::Main => "main", - Origin::SubAgent => "subagent", - Origin::Reviewer => "reviewer", - } - .to_string() - } -} +// --------------------------------------------------------------------------- +// Tool types — needed by all tool modules +// --------------------------------------------------------------------------- -/// Severity/category of a toast notification, used to pick its color. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -pub enum ToastKind { - Info, - Success, - Warning, - Error, - Lesson, -} +pub use tools::{GraduatedCheck, Tool, ToolCtx, ToolCtxBuilder}; -/// A transient status message shown in the TUI, auto-dismissed after `lifetime_ms`. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct Toast { - pub kind: ToastKind, - pub message: String, - pub created_at: i64, - pub lifetime_ms: u64, -} +// Re-export commonly needed types at the crate root +pub use zesdex_domain::core::{ChatMessage, Role, Store, UsageStats, ToolCallResult}; -impl Toast { - /// Create a toast with a default 5-second lifetime, stamped with now. - pub fn new(kind: ToastKind, message: String) -> Self { - Toast { - kind, - message, - created_at: chrono::Utc::now().timestamp_millis(), - lifetime_ms: 5000, - } - } - - /// Whether this toast's lifetime has elapsed as of `now_ms`. - pub fn expired(&self, now_ms: i64) -> bool { - let lifetime = self.lifetime_ms as i64; - now_ms - self.created_at > lifetime - } -} /// A shared, async-writable cache of directory entries, used to avoid /// re-reading a directory every render frame. @@ -179,176 +132,3 @@ impl Default for MentionIndex { Self::new() } } - -// --------------------------------------------------------------------------- -// TurnEvent & runtime types -// --------------------------------------------------------------------------- - -/// Events emitted onto the turn-event queue while an agent turn runs, -/// consumed by the event loop to update state and drive re-renders. -#[derive(Debug, Clone)] -pub enum TurnEvent { - AssistantMessage(ChatMessage), - ToolResult { - tool_call_id: String, - tool_name: String, - output: String, - is_error: bool, - path: Option, - }, - SystemNote { - kind: String, - message: String, - }, - StreamStart, - StreamToken(String), - StreamReasoning(String), - StreamDone(ChatMessage), - Usage { - tokens_in: u64, - tokens_out: u64, - }, - ReviewUsage { - tokens_in: u64, - tokens_out: u64, - }, - Compacted(Vec), - Error(String), - Done, - WorkflowAgentUpdate { - agent_id: String, - agent_name: String, - status: crate::AgentStatus, - }, - TodoUpdate(String), - PlanUpdate(String), -} - -/// A tool call awaiting execution, along with which execution model -/// (inline, deferred, async) it should run under. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct PendingTool { - pub tool_name: String, - pub args: serde_json::Value, - pub execution_model: ExecutionModel, -} - -/// How a pending tool call should be executed when the turn resumes. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -pub enum ExecutionModel { - Inline, - Deferred, - AsyncTokio, -} - -/// Reference to a background bash job tracked in session state. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct BashJobRef { - pub id: String, - pub command: String, - pub started_at: i64, - pub running: bool, -} - -/// Per-session runtime state: message history, pending tool queue, -/// background bash jobs, lesson/review counters. -#[derive(Debug, Clone)] -pub struct SessionRuntime { - pub messages: Vec, - pub tool_call_results: Vec, - pub pending_tool_queue: Vec, - pub bash_jobs: Vec, - pub subagent_queue: usize, - pub edit_count: u32, - pub consecutive_empty_reviews: u32, - pub session_start: i64, - pub lesson_count: u32, - pub lessons_user: u32, - pub lessons_feedback: u32, - pub lessons_project: u32, - pub lessons_reference: u32, - pub lessons_active: u32, - pub lessons_stale: u32, - pub lessons_contradicted: u32, - pub lessons_human: u32, - pub lessons_verified: u32, - pub lessons_unverified: u32, - pub review_count: u32, - pub session_dir: PathBuf, - pub usage: UsageStats, - pub hive_mind_converged: bool, -} - -impl SessionRuntime { - pub fn new(session_dir: PathBuf) -> Self { - SessionRuntime { - messages: Vec::new(), - tool_call_results: Vec::new(), - pending_tool_queue: Vec::new(), - bash_jobs: Vec::new(), - subagent_queue: 0, - edit_count: 0, - consecutive_empty_reviews: 0, - session_start: chrono::Utc::now().timestamp_millis(), - lesson_count: 0, - lessons_user: 0, - lessons_feedback: 0, - lessons_project: 0, - lessons_reference: 0, - lessons_active: 0, - lessons_stale: 0, - lessons_contradicted: 0, - lessons_human: 0, - lessons_verified: 0, - lessons_unverified: 0, - review_count: 0, - session_dir, - usage: UsageStats::default(), - hive_mind_converged: false, - } - } - - pub fn push_message(&mut self, msg: ChatMessage) { - self.messages.push(msg); - } -} - -/// Simple ASCII progress display for a long-running operation. -#[derive(Debug, Clone)] -pub struct ProgressState { - pub current: u64, - pub total: u64, - pub message: String, - pub start_time: i64, -} - -/// Agent status for workflow engine progress tracking. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -pub enum AgentStatus { - Pending, - Running, - Completed, - Failed(String), - Cancelled, -} - -impl std::fmt::Display for AgentStatus { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - AgentStatus::Pending => write!(f, "pending"), - AgentStatus::Running => write!(f, "running"), - AgentStatus::Completed => write!(f, "completed"), - AgentStatus::Failed(msg) => write!(f, "failed: {msg}"), - AgentStatus::Cancelled => write!(f, "cancelled"), - } - } -} - -// --------------------------------------------------------------------------- -// Tool types — needed by all tool modules -// --------------------------------------------------------------------------- - -pub use tools::{GraduatedCheck, Tool, ToolCtx, ToolCtxBuilder}; - -// Re-export commonly needed types at the crate root -pub use zesdex_domain::core::{ChatMessage, Role, Store, UsageStats, ToolCallResult}; diff --git a/apps/infrastructure/src/llm/provider.rs b/apps/infrastructure/src/llm/provider.rs index e649243..0d21948 100644 --- a/apps/infrastructure/src/llm/provider.rs +++ b/apps/infrastructure/src/llm/provider.rs @@ -1,13 +1,15 @@ -//! Blocking HTTP client for OpenAI/Anthropic-compatible chat completion APIs, +//! Async HTTP client for OpenAI/Anthropic-compatible chat completion APIs, //! supporting both non-streaming and SSE-streaming requests with automatic retry. use rand_core::RngCore; use std::sync::atomic::AtomicBool; use std::time::Duration; +use std::future::Future; use zesdex_domain::core::{ ChatMessage, ChatRequest, ChatResponse, SseParser, StreamEvent, StreamOptions, ToolDef, }; +use zesdex_application::ports::ProviderService; const DEFAULT_BASE_URL: &str = "https://opencode.ai/zen/v1"; const DEFAULT_MODEL: &str = "deepseek-v4-flash-free"; @@ -22,12 +24,10 @@ const REQUEST_TIMEOUT: Duration = Duration::from_secs(600); fn backoff_seconds(attempt: u32, cap: u64) -> Duration { let base = 2u64.pow(attempt.saturating_sub(1)); let delay = std::cmp::min(base, cap); - // ±25% jitter let jitter_factor = 0.75 + (rand_core::OsRng.next_u32() % 51) as f64 / 100.0; Duration::from_secs_f64(delay as f64 * jitter_factor) } -/// Is the error an auth / billing failure that retrying won't fix? pub fn is_auth_error(err_str: &str) -> bool { let err_lower = err_str.to_lowercase(); (err_str.contains("API error 401") @@ -54,9 +54,9 @@ fn backoff_for_error(attempt: u32, err_str: &str) -> Duration { // Client // --------------------------------------------------------------------------- -/// Blocking HTTP client for a single LLM provider endpoint. +/// Async HTTP client for a single LLM provider endpoint. pub struct LlmClient { - pub client: reqwest::blocking::Client, + pub client: reqwest::Client, pub api_key: String, pub base_url: String, pub model: String, @@ -72,7 +72,7 @@ impl LlmClient { } else { model }; - let client = match reqwest::blocking::Client::builder() + let client = match reqwest::Client::builder() .timeout(REQUEST_TIMEOUT) .connect_timeout(CONNECT_TIMEOUT) .build() @@ -84,14 +84,14 @@ impl LlmClient { retrying without connect timeout", e, ); - match reqwest::blocking::Client::builder() + match reqwest::Client::builder() .timeout(REQUEST_TIMEOUT) .build() { Ok(c) => c, Err(e2) => { tracing::warn!("also failed: {e2}. using default client"); - reqwest::blocking::Client::new() + reqwest::Client::new() } } } @@ -106,167 +106,12 @@ impl LlmClient { } } - pub fn chat_with_tools_non_streaming( - &self, - messages: &[ChatMessage], - tools: Option>, - max_tokens: Option, - temperature: Option, - abort_flag: Option<&AtomicBool>, - ) -> anyhow::Result<(ChatMessage, Option<(u64, u64)>)> { - let req = ChatRequest { - model: self.model.clone(), - messages: messages.to_vec(), - max_tokens: Some(max_tokens.unwrap_or(4096)), - temperature: Some(temperature.unwrap_or(0.7)), - tools, - stream: Some(false), - stop: None, - stream_options: None, - tool_choice: None, - top_p: None, - }; - - let url = format!("{}/chat/completions", self.base_url); - let max_retries = 10; - let mut attempt = 0u32; - - loop { - attempt += 1; - - if let Some(flag) = abort_flag { - if flag.load(std::sync::atomic::Ordering::Relaxed) { - anyhow::bail!("aborted"); - } - } - - let mut http_req = self - .client - .post(&url) - .header("Content-Type", "application/json"); - - if !self.api_key.is_empty() { - http_req = - http_req.header("Authorization", format!("Bearer {}", self.api_key)); - } - - let result = - (|| -> anyhow::Result<(ChatMessage, Option<(u64, u64)>)> { - let resp = http_req.json(&req).send().map_err(|e| { - if e.is_timeout() { - anyhow::anyhow!( - "API request timed out after {REQUEST_TIMEOUT:?}. \ - Check your network or try again." - ) - } else if e.is_connect() { - anyhow::anyhow!( - "Could not connect to {}. \ - Is the URL correct and is the service reachable?", - self.base_url - ) - } else { - anyhow::anyhow!("API request failed: {e}") - } - })?; - - if !resp.status().is_success() { - let status = resp.status(); - let body = resp.text().unwrap_or_default(); - anyhow::bail!("API error {} from {}: {}", status, self.base_url, body); - } - - let data: ChatResponse = resp.json()?; - let usage = data.usage.map(|u| { - (u64::from(u.prompt_tokens), u64::from(u.completion_tokens)) - }); - let message = data - .choices - .into_iter() - .next() - .and_then(|c| c.message) - .ok_or_else(|| anyhow::anyhow!("API response had no choices"))?; - Ok((message, usage)) - })(); - - match result { - Ok((msg, usage)) => return Ok((msg, usage)), - Err(e) => { - let err_str = e.to_string(); - if attempt >= max_retries || is_auth_error(&err_str) { - return Err(e); - } - let delay = backoff_for_error(attempt, &err_str); - std::thread::sleep(delay); - } - } - } - } - - pub fn chat_with_tools_streaming( - &self, - messages: &[ChatMessage], - tools: Option>, - temperature: Option, - max_tokens: Option, - mut on_event: impl FnMut(&StreamEvent) -> bool, - _abort_flag: Option<&AtomicBool>, - ) -> anyhow::Result<(ChatMessage, Option<(u64, u64)>)> { - let req = ChatRequest { - model: self.model.clone(), - messages: messages.to_vec(), - max_tokens: Some(max_tokens.unwrap_or(4096)), - temperature: Some(temperature.unwrap_or(0.7)), - tools, - stream: Some(true), - stop: None, - stream_options: Some(StreamOptions { - include_usage: true, - }), - tool_choice: None, - top_p: None, - }; - - let url = format!("{}/chat/completions", self.base_url); - let max_retries_stream = 10; - let mut attempt = 0u32; - - loop { - attempt += 1; - let mut captured_content = false; - let mut wrapped = |event: &StreamEvent| -> bool { - match event { - StreamEvent::Token(_) | StreamEvent::Reasoning(_) | StreamEvent::ToolCallDelta { .. } => { - captured_content = true; - } - _ => {} - } - on_event(event) - }; - match self.try_stream_once(&req, &url, &mut wrapped) { - Ok(result) => return Ok(result), - Err(e) => { - let err_str = e.to_string(); - if is_auth_error(&err_str) || captured_content { - return Err(e); - } - if attempt >= max_retries_stream { - return Err(e); - } - let delay = backoff_for_error(attempt, &err_str); - std::thread::sleep(delay); - } - } - } - } - - fn try_stream_once( + async fn try_stream_once( &self, req: &ChatRequest, url: &str, - on_event: &mut dyn FnMut(&StreamEvent) -> bool, + on_event: &mut (dyn FnMut(&StreamEvent) -> bool + Send), ) -> anyhow::Result<(ChatMessage, Option<(u64, u64)>)> { - use std::io::Read; - let mut http_req = self .client .post(url) @@ -276,7 +121,7 @@ impl LlmClient { http_req.header("Authorization", format!("Bearer {}", self.api_key)); } - let resp = http_req.json(req).send().map_err(|e| { + let mut resp = http_req.json(req).send().await.map_err(|e| { if e.is_timeout() { anyhow::anyhow!( "API request timed out after {REQUEST_TIMEOUT:?}. \ @@ -295,7 +140,7 @@ impl LlmClient { if !resp.status().is_success() { let status = resp.status(); - let body = resp.text().unwrap_or_default(); + let body = resp.text().await.unwrap_or_default(); anyhow::bail!("API error {} from {}: {}", status, self.base_url, body); } @@ -324,8 +169,6 @@ impl LlmClient { name, arguments_delta, } => { - // Cap the index to prevent memory exhaustion from - // maliciously large indices. const MAX_TOOL_CALLS: usize = 64; let index = usize::min(*index, MAX_TOOL_CALLS.saturating_sub(1)); @@ -380,27 +223,16 @@ impl LlmClient { let mut turn = StreamedTurn::new(); let mut usage: Option<(u64, u64)> = None; let mut parser = SseParser::new(); - - let mut reader = resp; let mut byte_buf: Vec = Vec::new(); - let mut chunk_buf = [0u8; 4096]; - loop { - let n = reader.read(&mut chunk_buf)?; - if n == 0 { - break; - } - byte_buf.extend_from_slice(&chunk_buf[..n]); + while let Some(chunk) = resp.chunk().await? { + byte_buf.extend_from_slice(&chunk); - // Drain any bytes that are not valid UTF-8 to prevent - // infinite loop when a non-UTF-8 sequence is received. let valid_len = match std::str::from_utf8(&byte_buf) { Ok(s) => s.len(), Err(e) => { let n = e.valid_up_to(); if n == 0 { - // No valid UTF-8 prefix; skip the first byte (likely - // a partial multi-byte sequence or stray byte). byte_buf.drain(..1); continue; } @@ -432,7 +264,6 @@ impl LlmClient { return Ok((turn.build_assistant_message(), usage)); } other => { - tracing::debug!("unhandled stream event type: {other:?}"); turn.apply_event(other); } } @@ -443,8 +274,153 @@ impl LlmClient { } } -/// Resolve the API key for the currently configured provider, falling back -/// through settings -> env var -> provider default. +impl ProviderService for LlmClient { + async fn chat( + &self, + messages: &[ChatMessage], + tools: Option>, + max_tokens: Option, + temperature: Option, + ) -> anyhow::Result<(ChatMessage, Option<(u64, u64)>)> { + let req = ChatRequest { + model: self.model.clone(), + messages: messages.to_vec(), + max_tokens: Some(max_tokens.unwrap_or(4096)), + temperature: Some(temperature.unwrap_or(0.7)), + tools, + stream: Some(false), + stop: None, + stream_options: None, + tool_choice: None, + top_p: None, + }; + + let url = format!("{}/chat/completions", self.base_url); + let max_retries = 10; + let mut attempt = 0u32; + + loop { + attempt += 1; + + let mut http_req = self + .client + .post(&url) + .header("Content-Type", "application/json"); + + if !self.api_key.is_empty() { + http_req = + http_req.header("Authorization", format!("Bearer {}", self.api_key)); + } + + let result = async { + let resp = http_req.json(&req).send().await.map_err(|e| { + if e.is_timeout() { + anyhow::anyhow!( + "API request timed out after {REQUEST_TIMEOUT:?}. \ + Check your network or try again." + ) + } else if e.is_connect() { + anyhow::anyhow!( + "Could not connect to {}. \ + Is the URL correct and is the service reachable?", + self.base_url + ) + } else { + anyhow::anyhow!("API request failed: {e}") + } + })?; + + if !resp.status().is_success() { + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + anyhow::bail!("API error {} from {}: {}", status, self.base_url, body); + } + + let data: ChatResponse = resp.json().await?; + let usage = data.usage.map(|u| { + (u64::from(u.prompt_tokens), u64::from(u.completion_tokens)) + }); + let message = data + .choices + .into_iter() + .next() + .and_then(|c| c.message) + .ok_or_else(|| anyhow::anyhow!("API response had no choices"))?; + Ok((message, usage)) + }.await; + + match result { + Ok((msg, usage)) => return Ok((msg, usage)), + Err(e) => { + let err_str = e.to_string(); + if attempt >= max_retries || is_auth_error(&err_str) { + return Err(e); + } + let delay = backoff_for_error(attempt, &err_str); + tokio::time::sleep(delay).await; + } + } + } + } + + async fn chat_stream( + &self, + messages: &[ChatMessage], + tools: Option>, + max_tokens: Option, + temperature: Option, + mut on_event: Box bool + Send>, + ) -> anyhow::Result<(ChatMessage, Option<(u64, u64)>)> { + let req = ChatRequest { + model: self.model.clone(), + messages: messages.to_vec(), + max_tokens: Some(max_tokens.unwrap_or(4096)), + temperature: Some(temperature.unwrap_or(0.7)), + tools, + stream: Some(true), + stop: None, + stream_options: Some(StreamOptions { + include_usage: true, + }), + tool_choice: None, + top_p: None, + }; + + let url = format!("{}/chat/completions", self.base_url); + let max_retries_stream = 10; + let mut attempt = 0u32; + + loop { + attempt += 1; + let mut captured_content = false; + let mut wrapped = |event: &StreamEvent| -> bool { + match event { + StreamEvent::Token(_) | StreamEvent::Reasoning(_) | StreamEvent::ToolCallDelta { .. } => { + captured_content = true; + } + _ => {} + } + on_event(event) + }; + + match self.try_stream_once(&req, &url, &mut wrapped).await { + Ok(result) => return Ok(result), + Err(e) => { + let err_str = e.to_string(); + if is_auth_error(&err_str) || captured_content { + return Err(e); + } + if attempt >= max_retries_stream { + return Err(e); + } + let delay = backoff_for_error(attempt, &err_str); + tokio::time::sleep(delay).await; + } + } + } + } +} + pub fn resolve_api_key( settings: &zesdex_domain::cms::Settings, app_config: &zesdex_domain::cms::AppConfig, diff --git a/apps/infrastructure/src/subagent/auto/engine.rs b/apps/infrastructure/src/subagent/auto/engine.rs index 67cc901..53e7b82 100644 --- a/apps/infrastructure/src/subagent/auto/engine.rs +++ b/apps/infrastructure/src/subagent/auto/engine.rs @@ -25,11 +25,9 @@ use zesdex_domain::core::ChatMessage; const REVIEW_AGENT_ID: &str = "auto-review"; -/// Spawn a background thread that reviews changes and auto-fixes issues. +/// Spawn a background task that reviews changes and auto-fixes issues. /// -/// Everything runs synchronously on the background thread — no tokio -/// runtime is created, avoiding the nested-runtime panic from -/// reqwest::blocking inside block_on in tokio >= 1.38. +/// Runs asynchronously using tokio::spawn. #[instrument(skip(turn_events))] pub fn spawn_background_review( workspace_roots: Vec, @@ -41,7 +39,7 @@ pub fn spawn_background_review( let agent_id = REVIEW_AGENT_ID.to_string(); let agent_name = "Auto-Review".to_string(); - std::thread::spawn(move || { + tokio::spawn(async move { let root = match workspace_roots.first() { Some(r) => r.clone(), None => { @@ -140,8 +138,6 @@ pub fn spawn_background_review( }; // 5. Call LLM to review the diff and suggest fixes. - // No tokio runtime needed — LlmClient uses reqwest::blocking - // internally, which is fine on a plain thread. let system_msg = ChatMessage::system( "You are an auto-review subagent. Your ONLY job:\n\ 1. Review the git diff below for:\n\ @@ -171,8 +167,7 @@ pub fn spawn_background_review( "Review and fix this git diff:\n\n```diff\n{truncated_diff}\n```" )); - // This is a sync call — no tokio runtime required on this thread. - let response = run_llm_review(&client, &[system_msg, user_msg]); + let response = run_llm_review(&client, &[system_msg, user_msg]).await; let response_text = match response { Ok(text) => text, @@ -232,10 +227,10 @@ pub fn spawn_background_review( }); } -/// Run the LLM review call synchronously using reqwest::blocking. -fn run_llm_review(client: &LlmClient, messages: &[ChatMessage]) -> Result { - // Direct LLM call — no tokio, no tool calls, just Q&A. - match client.chat_with_tools_non_streaming(messages, None, Some(1024), Some(0.3), None) { +/// Run the LLM review call asynchronously using ProviderService. +async fn run_llm_review(client: &LlmClient, messages: &[ChatMessage]) -> Result { + use zesdex_application::ports::ProviderService; + match client.chat(messages, None, Some(1024), Some(0.3)).await { Ok((msg, _)) => Ok(msg.content.unwrap_or_default()), Err(e) => Err(e.to_string()), } diff --git a/apps/infrastructure/src/subagent/division.rs b/apps/infrastructure/src/subagent/division.rs index 72d0d50..8d2a91b 100644 --- a/apps/infrastructure/src/subagent/division.rs +++ b/apps/infrastructure/src/subagent/division.rs @@ -6,19 +6,7 @@ use crate::tools::Tool; -/// Access tier for subagent tool permissions. -/// -/// Tiers are cumulative: `Write` includes everything in `Read`, and `Full` -/// includes everything in `Write`. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum AccessTier { - /// Read-only: search, read, glob, utility tools (no mutations). - Read, - /// Read + Write: above plus write, edit, delete, git, memory. - Write, - /// Full: above plus bash, shell, LSP, workflow, plan tools. - Full, -} +pub use zesdex_domain::subagent::AccessTier; /// Filter the available tools to match the given access tier. /// diff --git a/apps/infrastructure/src/subagent/engine.rs b/apps/infrastructure/src/subagent/engine.rs index e4021bb..c017e72 100644 --- a/apps/infrastructure/src/subagent/engine.rs +++ b/apps/infrastructure/src/subagent/engine.rs @@ -67,13 +67,13 @@ pub async fn run_agent( // Limited iteration loop so we don't run forever for iteration in 0..MAX_ITERATIONS { - let (response_msg, _usage) = client.chat_with_tools_non_streaming( + use zesdex_application::ports::ProviderService; + let (response_msg, _usage) = client.chat( &messages, Some(defs.clone()), Some(4096), None, - None, - )?; + ).await?; let content = response_msg.content.clone().unwrap_or_default(); let tool_calls = response_msg.tool_calls.unwrap_or_default(); diff --git a/apps/infrastructure/src/subagent/event.rs b/apps/infrastructure/src/subagent/event.rs index 305824b..5545e53 100644 --- a/apps/infrastructure/src/subagent/event.rs +++ b/apps/infrastructure/src/subagent/event.rs @@ -1,27 +1,3 @@ //! Subagent event types — events emitted during subagent execution. -/// Events emitted by a running subagent. -#[derive(Debug, Clone)] -pub enum SubagentEvent { - Started { - agent_id: String, - directive: String, - }, - ToolCall { - agent_id: String, - tool_name: String, - }, - ToolResult { - agent_id: String, - tool_name: String, - output: String, - }, - Completed { - agent_id: String, - output: String, - }, - Failed { - agent_id: String, - error: String, - }, -} +pub use zesdex_domain::subagent::SubagentEvent; diff --git a/apps/infrastructure/src/subagent/provider.rs b/apps/infrastructure/src/subagent/provider.rs index 9e6905e..72c2e14 100644 --- a/apps/infrastructure/src/subagent/provider.rs +++ b/apps/infrastructure/src/subagent/provider.rs @@ -35,12 +35,14 @@ impl SubagentProvider { /// /// Use this for a plain text-in/text-out conversation. #[tracing::instrument(skip(self, messages))] - pub fn chat( + pub async fn chat( &self, messages: &[ChatMessage], ) -> Result<(ChatMessage, Option<(u64, u64)>)> { + use zesdex_application::ports::ProviderService; self.client - .chat_with_tools_non_streaming(messages, None, Some(4096), None, None) + .chat(messages, None, Some(4096), None) + .await } /// Send messages with available tool definitions. @@ -48,14 +50,16 @@ impl SubagentProvider { /// Automatically converts the `&[Box]` slice to /// `Vec` before passing to the underlying client. #[tracing::instrument(skip(self, messages, tools))] - pub fn chat_with_tools( + pub async fn chat_with_tools( &self, messages: &[ChatMessage], tools: &[Box], ) -> Result<(ChatMessage, Option<(u64, u64)>)> { let defs = tool_defs(tools); + use zesdex_application::ports::ProviderService; self.client - .chat_with_tools_non_streaming(messages, Some(defs), Some(4096), None, None) + .chat(messages, Some(defs), Some(4096), None) + .await } } diff --git a/apps/infrastructure/src/tools/executor.rs b/apps/infrastructure/src/tools/executor.rs new file mode 100644 index 0000000..540a7fa --- /dev/null +++ b/apps/infrastructure/src/tools/executor.rs @@ -0,0 +1,56 @@ +use std::future::Future; +use anyhow::Result; +use zesdex_application::agent::ToolExecutor; +use crate::tools::{all_tools, Tool, ToolCtx}; +use tracing::debug; + +pub struct InfrastructureToolExecutor { + ctx: ToolCtx, + tools: Vec>, +} + +impl InfrastructureToolExecutor { + pub fn new(ctx: ToolCtx) -> Self { + Self { + ctx, + tools: all_tools(), + } + } +} + +impl ToolExecutor for InfrastructureToolExecutor { + fn execute( + &self, + tool_name: &str, + args: &serde_json::Value, + ) -> impl Future> + Send { + // Find the tool by name + let tool_opt = self.tools.iter().find(|t| t.name() == tool_name); + let ctx = self.ctx.clone(); + let args = args.clone(); + + async move { + match tool_opt { + Some(tool) => { + // Tool::run is synchronous, so we run it in a blocking task if needed + // For now, since they run fast and we are transitioning, we can just block in place. + // Or tokio::task::spawn_blocking: + let tool_name = tool.name().to_string(); + let t_ctx = ctx.clone(); + let t_args = args.clone(); + + // Because Tool isn't easily cloned, we might just block on the current thread, + // or use tokio::task::block_in_place if we're in a tokio runtime. + // But `tool.run` takes `&self`, and we have `&self` borrowed. + // For now we just call it directly. + tokio::task::block_in_place(move || { + tool.run(&ctx, &args) + }) + } + None => { + anyhow::bail!("Unknown tool: {}", tool_name) + } + } + } + } +} diff --git a/apps/infrastructure/src/tools/mod.rs b/apps/infrastructure/src/tools/mod.rs index 3326b92..860c219 100644 --- a/apps/infrastructure/src/tools/mod.rs +++ b/apps/infrastructure/src/tools/mod.rs @@ -30,6 +30,7 @@ pub mod spawn; pub mod utility; pub mod web_search; pub mod workflow; +pub mod executor; pub use git::git_cred; pub use git::git_operator; diff --git a/apps/infrastructure/src/tools/parallel_delegate.rs b/apps/infrastructure/src/tools/parallel_delegate.rs index ca140a9..4801fec 100644 --- a/apps/infrastructure/src/tools/parallel_delegate.rs +++ b/apps/infrastructure/src/tools/parallel_delegate.rs @@ -200,7 +200,8 @@ impl Tool for ParallelDelegate { // Consolidate results if synthesize && results.len() > 1 { - let consolidated = consolidate_results(&results, &base_url, &api_key, &model)?; + let rt = tokio::runtime::Runtime::new()?; + let consolidated = rt.block_on(consolidate_results(&results, &base_url, &api_key, &model))?; Ok(format!( "## Parallel Delegation Complete\n\n**Task:** {task}\n**Parallel agents:** {}\n\n{}", results.len(), @@ -249,8 +250,9 @@ async fn auto_split_task( let user_msg = zesdex_domain::core::ChatMessage::user(format!("Task: {task}\n\nSplit into {max_parallel} parallel directives:")); + use zesdex_application::ports::ProviderService; match client - .chat_with_tools_non_streaming(&[sys_msg, user_msg], None, Some(2048), Some(0.4), None) + .chat(&[sys_msg, user_msg], None, Some(2048), Some(0.4)).await { Ok((response, _)) => { let text = response.content.unwrap_or_default(); @@ -339,7 +341,7 @@ fn fallback_split(task: &str, max_parallel: usize) -> Vec<(String, AccessTier)> } /// Use LLM to consolidate multiple agent results into a single response. -fn consolidate_results( +async fn consolidate_results( results: &[(usize, String, String)], base_url: &str, api_key: &str, @@ -367,7 +369,8 @@ fn consolidate_results( "Consolidate the following parallel agent outputs:\n\n{summary}" )); - match client.chat_with_tools_non_streaming(&[sys_msg, user_msg], None, Some(2048), Some(0.3), None) { + use zesdex_application::ports::ProviderService; + match client.chat(&[sys_msg, user_msg], None, Some(2048), Some(0.3)).await { Ok((response, _)) => Ok(response.content.unwrap_or_else(|| summary.clone())), Err(e) => { warn!(error = %e, "consolidation LLM call failed, using raw concatenation"); diff --git a/apps/infrastructure/src/tools/workflow.rs b/apps/infrastructure/src/tools/workflow.rs index 12abb92..ffd3e2c 100644 --- a/apps/infrastructure/src/tools/workflow.rs +++ b/apps/infrastructure/src/tools/workflow.rs @@ -14,10 +14,7 @@ use crate::tools::{arg_str, Tool, ToolCtx}; use crate::workflow::engine::execution::execute_workflow; use crate::workflow::hive_mind::cycle::execute_cycle; use crate::workflow::hive_mind::synthesis::synthesize_consensus; -use crate::workflow::hive_mind::types::{ - CognitiveCycle, NodeDirective, NodeOutput, -}; -use crate::workflow::script::WorkflowScript; +use zesdex_domain::workflow::{CognitiveCycle, NodeDirective, NodeOutput, WorkflowScript}; /// Execute a multi-step workflow defined in YAML. /// @@ -49,7 +46,7 @@ impl Tool for WorkflowRun { #[instrument(skip(self, ctx, args))] fn run(&self, ctx: &ToolCtx, args: &Value) -> Result { let yaml = arg_str(args, "workflow_yaml")?; - let script = WorkflowScript::parse(&yaml)?; + let script = crate::workflow::script::parse_workflow_script(&yaml)?; info!( "Workflow started: {} ({} phases)", script.name, diff --git a/apps/infrastructure/src/workflow/docs.rs b/apps/infrastructure/src/workflow/docs.rs index c5a7804..7e483a0 100644 --- a/apps/infrastructure/src/workflow/docs.rs +++ b/apps/infrastructure/src/workflow/docs.rs @@ -5,7 +5,7 @@ use std::path::{Path, PathBuf}; use anyhow::Result; use tracing::{info, instrument}; -use crate::workflow::hive_mind::types::NodeOutput; +use zesdex_domain::workflow::NodeOutput; /// Write a deterministic audit trail for a hive-mind convergence. /// diff --git a/apps/infrastructure/src/workflow/engine/execution.rs b/apps/infrastructure/src/workflow/engine/execution.rs index bcb8ebf..1f14970 100644 --- a/apps/infrastructure/src/workflow/engine/execution.rs +++ b/apps/infrastructure/src/workflow/engine/execution.rs @@ -6,7 +6,7 @@ use tracing::{info, instrument}; use crate::llm::provider::LlmClient; use crate::tools::ToolCtx; use crate::workflow::engine::primitives::execute_primitive; -use crate::workflow::script::WorkflowScript; +use zesdex_domain::workflow::WorkflowScript; /// Execute each phase of a workflow script sequentially. /// diff --git a/apps/infrastructure/src/workflow/hive_mind/cycle.rs b/apps/infrastructure/src/workflow/hive_mind/cycle.rs index 539f0e0..d4cba0d 100644 --- a/apps/infrastructure/src/workflow/hive_mind/cycle.rs +++ b/apps/infrastructure/src/workflow/hive_mind/cycle.rs @@ -15,7 +15,7 @@ use crate::subagent::context::SubagentContext; use crate::subagent::division::AccessTier; use crate::subagent::engine::run_agent; use crate::tools::ToolCtx; -use crate::workflow::hive_mind::types::{CognitiveCycle, NodeOutput}; +use zesdex_domain::workflow::{CognitiveCycle, NodeOutput}; /// Execute one cycle: run each node directive and collect outputs. /// @@ -56,7 +56,7 @@ pub async fn execute_cycle( let cycle_index = cycle.index; - use crate::workflow::hive_mind::types::NodeDirective; + use zesdex_domain::workflow::NodeDirective; // Run all directives in this cycle concurrently. let handles: Vec<_> = cycle diff --git a/apps/infrastructure/src/workflow/hive_mind/synthesis.rs b/apps/infrastructure/src/workflow/hive_mind/synthesis.rs index 291b757..739b997 100644 --- a/apps/infrastructure/src/workflow/hive_mind/synthesis.rs +++ b/apps/infrastructure/src/workflow/hive_mind/synthesis.rs @@ -4,7 +4,7 @@ use anyhow::Result; use tracing::info; use crate::tools::ToolCtx; -use crate::workflow::hive_mind::types::NodeOutput; +use zesdex_domain::workflow::NodeOutput; /// Synthesize a consensus from all node outputs. /// diff --git a/apps/infrastructure/src/workflow/hive_mind/types.rs b/apps/infrastructure/src/workflow/hive_mind/types.rs index de2f0aa..78f6191 100644 --- a/apps/infrastructure/src/workflow/hive_mind/types.rs +++ b/apps/infrastructure/src/workflow/hive_mind/types.rs @@ -1,31 +1,3 @@ //! Hive-mind shared types — node directives, cycle plans, and node outputs. -/// A directive for a single processing node in the hive mind. -#[derive(Debug, Clone)] -pub struct NodeDirective { - pub directive: String, - pub access_tier: String, -} - -/// A cognitive cycle plan — ordered list of cycles, each containing -/// parallel node directives. -#[derive(Debug, Clone)] -pub struct CognitiveCyclePlan { - pub cycles: Vec>, -} - -/// A single cycle in a cognitive cycle plan — parallel node directives -/// executed together. -#[derive(Debug, Clone)] -pub struct CognitiveCycle { - pub index: u32, - pub directives: Vec, -} - -/// Output from a single hive-mind processing node after a cycle completes. -#[derive(Debug, Clone)] -pub struct NodeOutput { - pub id: String, - pub directive: String, - pub output: String, -} +// Types have been moved to zesdex_domain::workflow diff --git a/apps/infrastructure/src/workflow/script.rs b/apps/infrastructure/src/workflow/script.rs index 41f88b8..8fa6784 100644 --- a/apps/infrastructure/src/workflow/script.rs +++ b/apps/infrastructure/src/workflow/script.rs @@ -3,24 +3,11 @@ use anyhow::Result; use tracing::{info, instrument}; -/// A single phase in a parsed workflow script. -#[derive(Debug, Clone)] -pub struct WorkflowPhase { - pub name: String, - pub directive: String, -} +use zesdex_domain::workflow::{WorkflowPhase, WorkflowScript}; -/// A parsed workflow script with named phases. -#[derive(Debug, Clone)] -pub struct WorkflowScript { - pub name: String, - pub phases: Vec, -} - -impl WorkflowScript { - /// Parse a YAML string into a WorkflowScript. - /// - /// Expected format: +/// Parse a YAML string into a WorkflowScript. +/// +/// Expected format: /// ```yaml /// name: my-workflow /// phases: @@ -30,7 +17,7 @@ impl WorkflowScript { /// directive: "Implement the changes..." /// ``` #[instrument] - pub fn parse(yaml: &str) -> Result { + pub fn parse_workflow_script(yaml: &str) -> Result { let parsed: serde_json::Value = serde_yaml_ng::from_str(yaml) .map_err(|e| anyhow::anyhow!("Failed to parse workflow YAML: {e}"))?; @@ -62,5 +49,4 @@ impl WorkflowScript { info!("Parsed workflow script: {name} ({} phases)", phases.len()); Ok(WorkflowScript { name, phases }) - } } diff --git a/apps/interfaces/api/src/handlers/chat.rs b/apps/interfaces/api/src/handlers/chat.rs index 4623c88..4b359db 100644 --- a/apps/interfaces/api/src/handlers/chat.rs +++ b/apps/interfaces/api/src/handlers/chat.rs @@ -121,7 +121,7 @@ pub async fn chat_completions_handler( let llm_client = if model == state.llm_client.model { &state.llm_client } else { - temp_client = zesdex_infrastructure::llm::LlmClient::new( + temp_client = zesdex_infrastructure::llm::provider::LlmClient::new( state.llm_client.api_key.clone(), model, Some(state.llm_client.base_url.clone()), @@ -129,14 +129,14 @@ pub async fn chat_completions_handler( &temp_client }; + use zesdex_application::ports::ProviderService; let (response, usage) = llm_client - .chat_with_tools_non_streaming( + .chat( &messages, None, // No tool definitions for basic chat req.max_tokens, req.temperature, - None, // No abort flag - ) + ).await .map_err(|e| ApiError::ChatProxy(format!("LLM request failed: {e}")))?; let (prompt_tokens, completion_tokens) = usage.unwrap_or((0, 0)); diff --git a/apps/interfaces/api/src/state.rs b/apps/interfaces/api/src/state.rs index 5086071..1b08091 100644 --- a/apps/interfaces/api/src/state.rs +++ b/apps/interfaces/api/src/state.rs @@ -194,7 +194,7 @@ pub struct ApiState { pub token_service: JwtTokenService, /// LLM provider client for chat completions. - pub llm_client: zesdex_infrastructure::llm::LlmClient, + pub llm_client: zesdex_infrastructure::llm::provider::LlmClient, } impl fmt::Debug for ApiState { @@ -272,7 +272,7 @@ impl ApiState { zesdex_application::cms::MemoryServiceImpl::new(memory_repo, memory_dir); let token_service = JwtTokenService::new(&jwt_secret); - let llm_client = zesdex_infrastructure::llm::LlmClient::new( + let llm_client = zesdex_infrastructure::llm::provider::LlmClient::new( llm_api_key.into(), llm_model.into(), llm_base_url, diff --git a/apps/interfaces/daemon/src/handler.rs b/apps/interfaces/daemon/src/handler.rs index d65044f..c93c32d 100644 --- a/apps/interfaces/daemon/src/handler.rs +++ b/apps/interfaces/daemon/src/handler.rs @@ -320,19 +320,47 @@ fn handle_submit_input(state: &mut AppStateRest, text: String) { .map(|rt| rt.messages.clone()) .unwrap_or_default(); - let params = zesdex_infrastructure::agent::AgentTurnParams { + let params = zesdex_domain::agent::AgentTurnParams { messages, session_dir: state.session_dir.clone(), workspace_roots: state.workspace_roots.clone(), turn_events: state.turn_events.clone(), in_flight: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), abort: state.abort_flag.clone(), - api_key, + api_key: api_key.clone(), model: state.settings.model.clone(), - api_base: provider_cfg.map(|cfg| cfg.api_base.clone()), + api_base: provider_cfg.as_ref().map(|cfg| cfg.api_base.clone()), }; - zesdex_infrastructure::agent::spawn_agent_turn(params); + let client = std::sync::Arc::new(zesdex_infrastructure::llm::provider::LlmClient::new( + api_key, + state.settings.model.clone(), + provider_cfg.map(|cfg| cfg.api_base.clone()), + )); + + let tool_ctx = zesdex_infrastructure::tools::ToolCtx::builder() + .session_dir(state.session_dir.clone()) + .workspaces(state.workspace_roots.clone()) + .turn_events(state.turn_events.clone()) + .build(); + + let tool_executor = std::sync::Arc::new( + zesdex_infrastructure::tools::executor::InfrastructureToolExecutor::new(tool_ctx), + ); + + let tools = zesdex_infrastructure::tools::all_tools(); + let defs = zesdex_infrastructure::tools::tool_defs(&tools); + + let turn_service = zesdex_application::agent::turn_service::AgentTurnServiceImpl::new( + client, + tool_executor, + defs, + ); + + use zesdex_application::agent::AgentTurnService; + tokio::spawn(async move { + let _ = turn_service.run_turn(params).await; + }); } fn handle_delete_char(state: &mut AppStateRest) { @@ -445,10 +473,11 @@ fn handle_compact(state: &mut AppStateRest) { let model = state.settings.model.clone(); let api_base = provider_cfg.map(|cfg| cfg.api_base.clone()); - let client = zesdex_infrastructure::llm::LlmClient::new(api_key, model, api_base); + let client = zesdex_infrastructure::llm::provider::LlmClient::new(api_key, model, api_base); if let Some(ref mut rt) = state.session_runtime { - if let Ok(()) = zesdex_infrastructure::agent::compact_messages_with_ai(&mut rt.messages, &client) { + let tokio_rt = tokio::runtime::Runtime::new().unwrap(); + if let Ok(()) = tokio_rt.block_on(zesdex_application::agent::turn_service::compact_messages_with_ai(&mut rt.messages, &client)) { let msg_count = rt.messages.len(); state.push_transcript(ChatMessageDisplay::new( RoleWrapper::System, diff --git a/apps/interfaces/tui/src/turn.rs b/apps/interfaces/tui/src/turn.rs index c17593b..6617b33 100644 --- a/apps/interfaces/tui/src/turn.rs +++ b/apps/interfaces/tui/src/turn.rs @@ -4,7 +4,7 @@ use std::sync::atomic::Ordering; use tracing::info; use zesdex_domain::core::ChatMessage; -use zesdex_infrastructure::agent::{spawn_agent_turn as backend_spawn_agent_turn, AgentTurnParams}; +use zesdex_domain::agent::AgentTurnParams; use crate::state::AppStateRest; /// Spawn an agent turn on a background thread by delegating to `zesdex-infrastructure`. @@ -63,15 +63,43 @@ pub fn spawn_agent_turn(state: &mut AppStateRest, text: String) { let params = AgentTurnParams { messages, - session_dir, - workspace_roots, - turn_events, - in_flight, - abort, + session_dir: session_dir.clone(), + workspace_roots: workspace_roots.clone(), + turn_events: turn_events.clone(), + in_flight: in_flight.clone(), + abort: abort.clone(), + api_key: api_key.clone(), + model: model.clone(), + api_base: api_base.clone(), + }; + + let client = std::sync::Arc::new(zesdex_infrastructure::llm::provider::LlmClient::new( api_key, model, api_base, - }; + )); - backend_spawn_agent_turn(params); + let tool_ctx = zesdex_infrastructure::tools::ToolCtx::builder() + .session_dir(session_dir) + .workspaces(workspace_roots) + .turn_events(turn_events) + .build(); + + let tool_executor = std::sync::Arc::new( + zesdex_infrastructure::tools::executor::InfrastructureToolExecutor::new(tool_ctx), + ); + + let tools = zesdex_infrastructure::tools::all_tools(); + let defs = zesdex_infrastructure::tools::tool_defs(&tools); + + let turn_service = zesdex_application::agent::turn_service::AgentTurnServiceImpl::new( + client, + tool_executor, + defs, + ); + + use zesdex_application::agent::AgentTurnService; + tokio::spawn(async move { + let _ = turn_service.run_turn(params).await; + }); } diff --git a/apps/interfaces/ws/src/lib.rs b/apps/interfaces/ws/src/lib.rs index a477e93..e405897 100644 --- a/apps/interfaces/ws/src/lib.rs +++ b/apps/interfaces/ws/src/lib.rs @@ -82,19 +82,47 @@ async fn handle_socket(mut socket: WebSocket, state: Arc) { let api_key = std::env::var("OPENAI_API_KEY").unwrap_or_default(); let model = val.get("model").and_then(|v| v.as_str()).unwrap_or("gpt-4o").to_string(); - let params = zesdex_infrastructure::agent::AgentTurnParams { + let params = zesdex_domain::agent::AgentTurnParams { messages: vec![zesdex_domain::core::ChatMessage::user(prompt)], - session_dir, - workspace_roots, + session_dir: session_dir.clone(), + workspace_roots: workspace_roots.clone(), turn_events: turn_events.clone(), in_flight, abort, - api_key, - model, + api_key: api_key.clone(), + model: model.clone(), api_base: None, }; - zesdex_infrastructure::agent::spawn_agent_turn(params); + let client = std::sync::Arc::new(zesdex_infrastructure::llm::provider::LlmClient::new( + api_key, + model, + None, + )); + + let tool_ctx = zesdex_infrastructure::tools::ToolCtx::builder() + .session_dir(session_dir) + .workspaces(workspace_roots) + .turn_events(turn_events.clone()) + .build(); + + let tool_executor = std::sync::Arc::new( + zesdex_infrastructure::tools::executor::InfrastructureToolExecutor::new(tool_ctx), + ); + + let tools = zesdex_infrastructure::tools::all_tools(); + let defs = zesdex_infrastructure::tools::tool_defs(&tools); + + let turn_service = zesdex_application::agent::turn_service::AgentTurnServiceImpl::new( + client, + tool_executor, + defs, + ); + + use zesdex_application::agent::AgentTurnService; + tokio::spawn(async move { + let _ = turn_service.run_turn(params).await; + }); let tx_clone = tx.clone(); tokio::spawn(async move {