feat: enhance explore phase with TUI workflow event handling

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