From b3a4d131d13d94bd307720cb50c2d02dedf92873 Mon Sep 17 00:00:00 2001 From: asepharyana Date: Tue, 21 Jul 2026 07:47:35 +0700 Subject: [PATCH] feat: enhance explore phase with TUI workflow event handling --- apps/application/src/agent/explore.rs | 5 + apps/application/src/agent/turn_service.rs | 2 +- .../src/best_practice/explore.rs | 117 ++++++++++++++---- 3 files changed, 99 insertions(+), 25 deletions(-) diff --git a/apps/application/src/agent/explore.rs b/apps/application/src/agent/explore.rs index cccdced..e7b2567 100644 --- a/apps/application/src/agent/explore.rs +++ b/apps/application/src/agent/explore.rs @@ -15,8 +15,11 @@ //! snapshot of what the codebase contains and where relevant code lives. use anyhow::Result; +use std::collections::VecDeque; use std::future::Future; use std::pin::Pin; +use std::sync::{Arc, Mutex}; +use zesdex_domain::agent::TurnEvent; /// The consolidated output of an explore phase — a set of system-level /// context messages injected before the main agent prompt. @@ -43,10 +46,12 @@ pub trait ExploreService: Send + Sync { /// /// `query` — the user's current input phrase. /// `workspace_root` — absolute path to the workspace root. + /// `turn_events` — shared event queue for TUI updates. /// Returns structured context messages and a summary blob. fn explore<'a>( &'a self, query: &'a str, workspace_root: &'a str, + turn_events: &'a Arc>>, ) -> Pin> + Send + 'a>>; } diff --git a/apps/application/src/agent/turn_service.rs b/apps/application/src/agent/turn_service.rs index e93d75c..4c09e28 100644 --- a/apps/application/src/agent/turn_service.rs +++ b/apps/application/src/agent/turn_service.rs @@ -203,7 +203,7 @@ impl super::AgentTurnService for AgentTurnS }, ); - match explorer.explore(&user_query, &workspace_root).await { + match explorer.explore(&user_query, &workspace_root, ¶ms.turn_events).await { Ok(output) => { // Insert each context message as a system message. // They go at index 0 and are removed after the turn diff --git a/apps/infrastructure/src/best_practice/explore.rs b/apps/infrastructure/src/best_practice/explore.rs index 63f1574..2ddabea 100644 --- a/apps/infrastructure/src/best_practice/explore.rs +++ b/apps/infrastructure/src/best_practice/explore.rs @@ -1,36 +1,40 @@ //! Mandatory explore phase — spawns ≥3 parallel subagents to discover -//! codebase context before every agent turn. +//! codebase context before every agent turn, visible in the TUI workflow tab. //! //! # Flow //! //! `ExploreServiceImpl::explore()` → //! -//! 1. **Code Structure** subagent — walks the workspace tree, reads -//! `Cargo.toml` / `package.json`, lists top-level modules. -//! 2. **Symbol Index** subagent — rebuilds the multi-language symbol index, -//! then lists symbols. -//! 3. **Semantic Context** subagent — searches code for context related to -//! the user's query. -//! -//! All three run on **dedicated OS threads** in parallel with their own -//! tokio runtimes. Joins are offloaded to `spawn_blocking` so they do not -//! stall the async runtime. +//! 1. Push `WorkflowAgentUpdate { Pending }` for each agent onto the turn-event +//! queue so the TUI workflow tab shows all 3. +//! 2. Spawn **Code Structure** subagent (thread + tokio runtime). +//! 3. Spawn **Symbol Index** subagent (thread + tokio runtime). +//! 4. Spawn **Semantic Context** subagent (thread + tokio runtime). +//! 5. Join all handles via `spawn_blocking`. +//! 6. Push `Completed` / `Failed` events for each agent. +//! 7. Consolidate findings into a system message → return. use crate::subagent::context::SubagentContext; use crate::subagent::division::AccessTier; use crate::subagent::engine::run_agent; use crate::tools::ToolCtx; use anyhow::{Context, Result}; +use std::collections::VecDeque; use std::future::Future; use std::pin::Pin; +use std::sync::{Arc, Mutex}; use std::thread; use tracing::{info, warn}; use zesdex_application::agent::{ExploreOutput, ExploreService}; +use zesdex_domain::agent::{AgentStatus, TurnEvent}; /// Number of parallel explore subagents. const EXPLORE_AGENT_COUNT: usize = 3; -/// Labels for each explore agent. +/// IDs for each explore agent (shown in the workflow tab). +const EXPLORE_IDS: [&str; 3] = ["explore-structure", "explore-symbols", "explore-context"]; + +/// Display names for the TUI workflow tab. const EXPLORE_LABELS: [&str; 3] = [ "📁 Code Structure", "🔣 Symbol Index", @@ -41,8 +45,8 @@ const EXPLORE_LABELS: [&str; 3] = [ const EXPLORE_DIRECTIVES: [&str; 3] = [ // Agent 0: Code Structure "You are a codebase structure explorer.\n\ - 1. List top-level directories and files in the workspace root.\n\ - 2. Read Cargo.toml, package.json, or pyproject.toml at root.\n\ + 1. List all top-level directories and files in the workspace root.\n\ + 2. Read Cargo.toml, package.json, or pyproject.toml at the root.\n\ 3. List the apps/ or src/ directory contents.\n\ 4. Identify main entry points (main.rs, main.py, index.ts, etc.).\n\ 5. Count files by extension type.\n\ @@ -84,8 +88,7 @@ pub struct Credentials { /// Concrete [`ExploreService`] that the turn service calls. /// /// Owns a shared `ToolCtx` and LLM credentials. Each call to `explore()` -/// spawns 3 subagents in parallel, waits for all to finish, and returns -/// a consolidated context message. +/// spawns 3 subagents in parallel with TUI workflow-tab visibility. pub struct ExploreServiceImpl { tool_ctx: ToolCtx, credentials: Credentials, @@ -105,6 +108,7 @@ impl ExploreService for ExploreServiceImpl { &'a self, query: &'a str, workspace_root: &'a str, + turn_events: &'a Arc>>, ) -> Pin> + Send + 'a>> { Box::pin(async move { let context = run_explore_phase( @@ -112,6 +116,7 @@ impl ExploreService for ExploreServiceImpl { workspace_root, &self.tool_ctx, &self.credentials, + turn_events, ) .await?; @@ -123,23 +128,82 @@ impl ExploreService for ExploreServiceImpl { } } +// --------------------------------------------------------------------------- +// Helpers for pushing workflow events +// --------------------------------------------------------------------------- + +fn push_event(events: &Arc>>, event: TurnEvent) { + if let Ok(mut q) = events.lock() { + q.push_back(event); + } +} + +fn emit_pending(events: &Arc>>, agent_id: &str, display: &str) { + push_event( + events, + TurnEvent::WorkflowAgentUpdate { + agent_id: agent_id.to_string(), + agent_name: display.to_string(), + status: AgentStatus::Pending, + }, + ); +} + +fn emit_running(events: &Arc>>, agent_id: &str, display: &str) { + push_event( + events, + TurnEvent::WorkflowAgentUpdate { + agent_id: agent_id.to_string(), + agent_name: display.to_string(), + status: AgentStatus::Running, + }, + ); +} + +fn emit_completed(events: &Arc>>, agent_id: &str, display: &str) { + push_event( + events, + TurnEvent::WorkflowAgentUpdate { + agent_id: agent_id.to_string(), + agent_name: display.to_string(), + status: AgentStatus::Completed, + }, + ); +} + +fn emit_failed(events: &Arc>>, agent_id: &str, display: &str, msg: &str) { + push_event( + events, + TurnEvent::WorkflowAgentUpdate { + agent_id: agent_id.to_string(), + agent_name: display.to_string(), + status: AgentStatus::Failed(msg.to_string()), + }, + ); +} + // --------------------------------------------------------------------------- // Core orchestration // --------------------------------------------------------------------------- -/// Spawn EXPLORE_AGENT_COUNT subagents in parallel, join, consolidate. +/// Spawn `EXPLORE_AGENT_COUNT` subagents in parallel, emit workflow events +/// for the TUI tab, join, and consolidate. async fn run_explore_phase( query: &str, workspace_root: &str, tool_ctx: &ToolCtx, credentials: &Credentials, + turn_events: &Arc>>, ) -> Result { - // ── Prepare directives ────────────────────────────────────────────── + // ── 1. Emit Pending for all agents (appears instantly in workflow tab) ─ + for i in 0..EXPLORE_AGENT_COUNT { + emit_pending(turn_events, EXPLORE_IDS[i], EXPLORE_LABELS[i]); + } + + // ── 2. Prepare directives ─────────────────────────────────────────── let mut directives: Vec = Vec::with_capacity(EXPLORE_AGENT_COUNT); for i in 0..EXPLORE_AGENT_COUNT { let mut d = EXPLORE_DIRECTIVES[i].to_string(); - - // Agent 2 gets the user query for targeted search. if i == 2 { d.push_str(&format!("\n\nThe user's current query is: \"{query}\"")); } @@ -147,11 +211,13 @@ async fn run_explore_phase( directives.push(d); } - // ── Spawn all agents on threads with their own tokio runtime ──────── + // ── 3. Spawn all agents on threads ────────────────────────────────── let mut handles: Vec<(usize, thread::JoinHandle>)> = Vec::with_capacity(EXPLORE_AGENT_COUNT); for i in 0..EXPLORE_AGENT_COUNT { + emit_running(turn_events, EXPLORE_IDS[i], EXPLORE_LABELS[i]); + let ctx = SubagentContext::new( directives[i].clone(), tool_ctx.clone(), @@ -173,22 +239,25 @@ async fn run_explore_phase( handles.push((i, handle)); } - // ── Join handles via spawn_blocking (blocking join, NOT async) ────── - // We move handles into a blocking task so the async executor stays free. + // ── 4. Join handles via spawn_blocking ────────────────────────────── + let turn_events_clone = Arc::clone(turn_events); let results: Vec<(usize, String, bool)> = tokio::task::spawn_blocking(move || { let mut out = Vec::with_capacity(EXPLORE_AGENT_COUNT); for (i, handle) in handles { let entry = match handle.join() { Ok(Ok(output)) => { info!(agent = i, "explore subagent completed"); + emit_completed(&turn_events_clone, EXPLORE_IDS[i], EXPLORE_LABELS[i]); (i, output, true) } Ok(Err(e)) => { warn!(agent = i, error = %e, "explore subagent failed"); + emit_failed(&turn_events_clone, EXPLORE_IDS[i], EXPLORE_LABELS[i], &e.to_string()); (i, format!("Error: {e}"), false) } Err(e) => { warn!(agent = i, error = ?e, "explore subagent panicked"); + emit_failed(&turn_events_clone, EXPLORE_IDS[i], EXPLORE_LABELS[i], "thread panicked"); (i, format!("Thread panic: {e:?}"), false) } }; @@ -199,7 +268,7 @@ async fn run_explore_phase( .await .context("explore join task panicked")?; - // ── Build consolidated context ────────────────────────────────────── + // ── 5. Build consolidated context ─────────────────────────────────── Ok(build_explore_context(&results)) }