feat: implement mandatory explore phase with parallel subagents
- Added `ExploreService` trait and `ExploreServiceImpl` struct to handle the exploration of codebase context before agent turns. - Implemented three parallel subagents: Code Structure, Symbol Index, and Semantic Context, each with specific directives. - Integrated the explore phase into the agent turn process, ensuring that each turn starts with a consolidated context message. - Enhanced `spawn_agent_turn` function to include explore service wiring and context preparation.
This commit is contained in:
asepharyana
@@ -0,0 +1,52 @@
|
||||
//! Mandatory explore phase — spawns parallel subagents to discover context
|
||||
//! before the main agent begins its turn.
|
||||
//!
|
||||
//! # Flow
|
||||
//!
|
||||
//! Before the main agent's LLM loop, [`ExploreService::explore`] dispatches
|
||||
//! at least 3 subagents in parallel (code-structure scan, symbol-index query,
|
||||
//! semantic-context search). Their findings are consolidated into a single
|
||||
//! system message that is prepended to the conversation.
|
||||
//!
|
||||
//! # Why mandatory
|
||||
//!
|
||||
//! Without structured exploration the main agent works from an empty context
|
||||
//! window. The explore phase guarantees that every turn starts with a compact
|
||||
//! snapshot of what the codebase contains and where relevant code lives.
|
||||
|
||||
use anyhow::Result;
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
|
||||
/// The consolidated output of an explore phase — a set of system-level
|
||||
/// context messages injected before the main agent prompt.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ExploreOutput {
|
||||
/// One or more system messages summarising what the explore subagents
|
||||
/// discovered. Prepended to the conversation by the turn service.
|
||||
pub context_messages: Vec<String>,
|
||||
/// Short human-readable summary of what was explored.
|
||||
pub summary: String,
|
||||
}
|
||||
|
||||
/// Service trait for the mandatory pre-turn exploration phase.
|
||||
///
|
||||
/// Implementors spawn ≥3 parallel subagents, each analysing a different
|
||||
/// aspect of the workspace, and return a consolidated summary.
|
||||
///
|
||||
/// # Object safety
|
||||
///
|
||||
/// This trait is `dyn`-safe — it returns `Pin<Box<dyn Future>>` so it can
|
||||
/// be stored as `Arc<dyn ExploreService>`.
|
||||
pub trait ExploreService: Send + Sync {
|
||||
/// Run the explore phase.
|
||||
///
|
||||
/// `query` — the user's current input phrase.
|
||||
/// `workspace_root` — absolute path to the workspace root.
|
||||
/// Returns structured context messages and a summary blob.
|
||||
fn explore<'a>(
|
||||
&'a self,
|
||||
query: &'a str,
|
||||
workspace_root: &'a str,
|
||||
) -> Pin<Box<dyn Future<Output = Result<ExploreOutput>> + Send + 'a>>;
|
||||
}
|
||||
@@ -22,4 +22,8 @@ pub trait AgentTurnService: Send + Sync {
|
||||
) -> impl Future<Output = Result<()>> + Send;
|
||||
}
|
||||
|
||||
pub mod explore;
|
||||
pub mod turn_service;
|
||||
|
||||
pub use explore::{ExploreOutput, ExploreService};
|
||||
pub use turn_service::{compact_messages_with_ai, AgentTurnServiceImpl};
|
||||
|
||||
@@ -8,7 +8,7 @@ use zesdex_domain::core::{ChatMessage, StreamEvent, ToolDef};
|
||||
use zesdex_domain::main_agent_prompt;
|
||||
|
||||
use crate::ports::ProviderService;
|
||||
use super::ToolExecutor;
|
||||
use super::{ExploreService, ToolExecutor};
|
||||
|
||||
/// Maximum tool-call iterations per agent turn before forcing termination.
|
||||
const MAX_TURN_ITERATIONS: u32 = 50;
|
||||
@@ -107,21 +107,45 @@ fn emit_usage(turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>, usage: Option<(u64,
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Service implementation for executing an agent turn asynchronously.
|
||||
///
|
||||
/// # Explore phase
|
||||
///
|
||||
/// Before the main LLM loop begins, [`AgentTurnServiceImpl`] runs a mandatory
|
||||
/// explore phase that spawns ≥3 parallel subagents (code structure, symbol
|
||||
/// index, semantic context) and injects their consolidated findings as a
|
||||
/// system message. See [`ExploreService`] for the trait contract.
|
||||
pub struct AgentTurnServiceImpl<P: ProviderService, T: ToolExecutor> {
|
||||
provider: Arc<P>,
|
||||
tool_executor: Arc<T>,
|
||||
tool_defs: Vec<ToolDef>,
|
||||
/// Optional explore-phase service. When `Some`, the explore phase runs
|
||||
/// before every turn; when `None` it is skipped (tests, daemon mode).
|
||||
explore_service: Option<Arc<dyn ExploreService>>,
|
||||
}
|
||||
|
||||
impl<P: ProviderService, T: ToolExecutor> AgentTurnServiceImpl<P, T> {
|
||||
pub fn new(provider: Arc<P>, tool_executor: Arc<T>, tool_defs: Vec<ToolDef>) -> Self {
|
||||
pub fn new(
|
||||
provider: Arc<P>,
|
||||
tool_executor: Arc<T>,
|
||||
tool_defs: Vec<ToolDef>,
|
||||
) -> Self {
|
||||
Self {
|
||||
provider,
|
||||
tool_executor,
|
||||
tool_defs,
|
||||
explore_service: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Attach an optional explore-phase service.
|
||||
///
|
||||
/// When set, every call to `run_turn` will first run the explore phase
|
||||
/// and inject the consolidated context as a system message.
|
||||
pub fn with_explore(mut self, service: Arc<dyn ExploreService>) -> Self {
|
||||
self.explore_service = Some(service);
|
||||
self
|
||||
}
|
||||
|
||||
/// Execute a single LLM call with the current message list, handling
|
||||
/// streaming events and error reporting.
|
||||
async fn call_llm(
|
||||
@@ -153,6 +177,61 @@ impl<P: ProviderService, T: ToolExecutor> super::AgentTurnService for AgentTurnS
|
||||
params.model
|
||||
);
|
||||
|
||||
// ── Phase 0: Mandatory explore ──────────────────────────────────
|
||||
// Spawn ≥3 parallel subagents to discover code structure, symbols,
|
||||
// and semantic context. The consolidated summary is injected as a
|
||||
// system message before the main agent prompt.
|
||||
if let Some(ref explorer) = self.explore_service {
|
||||
// Determine workspace root from the first message's context or
|
||||
// the first workspace root in params.
|
||||
let user_query = params
|
||||
.messages
|
||||
.last()
|
||||
.map(|m| m.content.clone().unwrap_or_default())
|
||||
.unwrap_or_default();
|
||||
let workspace_root = params
|
||||
.workspace_roots
|
||||
.first()
|
||||
.map(|p| p.to_string_lossy().to_string())
|
||||
.unwrap_or_else(|| ".".to_string());
|
||||
|
||||
push_event(
|
||||
¶ms.turn_events,
|
||||
TurnEvent::SystemNote {
|
||||
kind: "info".into(),
|
||||
message: "🔍 Exploring codebase structure...".into(),
|
||||
},
|
||||
);
|
||||
|
||||
match explorer.explore(&user_query, &workspace_root).await {
|
||||
Ok(output) => {
|
||||
// Insert each context message as a system message.
|
||||
// They go at index 0 and are removed after the turn
|
||||
// like the main agent prompt.
|
||||
for ctx_msg in &output.context_messages {
|
||||
params
|
||||
.messages
|
||||
.insert(0, ChatMessage::system(ctx_msg.clone()));
|
||||
}
|
||||
info!(
|
||||
"Explore phase complete: {} context messages, {}",
|
||||
output.context_messages.len(),
|
||||
output.summary
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Explore phase failed (non-fatal): {e}");
|
||||
push_event(
|
||||
¶ms.turn_events,
|
||||
TurnEvent::SystemNote {
|
||||
kind: "warn".into(),
|
||||
message: format!("Explore phase failed: {e}"),
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Insert system prompt at position 0 once and keep it there for the
|
||||
// entire turn, avoiding per-iteration clones of the full message list.
|
||||
// It is removed before emitting the Compacted event so persistence
|
||||
|
||||
@@ -51,4 +51,7 @@ pub use cms::{
|
||||
settings_service::SettingsServiceImpl,
|
||||
};
|
||||
|
||||
pub use agent::{AgentTurnService, ToolExecutor, turn_service::{AgentTurnServiceImpl, compact_messages_with_ai}};
|
||||
pub use agent::{
|
||||
AgentTurnService, ExploreOutput, ExploreService, ToolExecutor,
|
||||
turn_service::{AgentTurnServiceImpl, compact_messages_with_ai},
|
||||
};
|
||||
|
||||
@@ -0,0 +1,229 @@
|
||||
//! Mandatory explore phase — spawns ≥3 parallel subagents to discover
|
||||
//! codebase context before every agent turn.
|
||||
//!
|
||||
//! # 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.
|
||||
|
||||
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::future::Future;
|
||||
use std::pin::Pin;
|
||||
use std::thread;
|
||||
use tracing::{info, warn};
|
||||
use zesdex_application::agent::{ExploreOutput, ExploreService};
|
||||
|
||||
/// Number of parallel explore subagents.
|
||||
const EXPLORE_AGENT_COUNT: usize = 3;
|
||||
|
||||
/// Labels for each explore agent.
|
||||
const EXPLORE_LABELS: [&str; 3] = [
|
||||
"📁 Code Structure",
|
||||
"🔣 Symbol Index",
|
||||
"🔍 Semantic Context",
|
||||
];
|
||||
|
||||
/// Directives for each explore subagent.
|
||||
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\
|
||||
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\
|
||||
Use the ls_dir, read, grep, and glob tools. Be concise.",
|
||||
|
||||
// Agent 1: Symbol Index
|
||||
"You are a symbol index explorer.\n\
|
||||
1. Call the 'rebuild_index' tool to rebuild the symbol index.\n\
|
||||
2. Call the 'list_symbols' tool with max_results: 100.\n\
|
||||
3. Identify public APIs, entry points, and key types.\n\
|
||||
4. Group symbols by language and kind.\n\
|
||||
Be concise. Report what symbols exist and where they live.",
|
||||
|
||||
// Agent 2: Semantic Context
|
||||
"You are a semantic context explorer.\n\
|
||||
1. Call the 'rebuild_index' tool to ensure the index is fresh.\n\
|
||||
2. Search for symbols related to the user's query using semantic_search.\n\
|
||||
3. Search for config files, env variables, and settings.\n\
|
||||
4. Search for test files and test patterns.\n\
|
||||
Be concise. Report relevant code areas for the task.\n\
|
||||
Use the semantic_search, grep, glob, and read tools.",
|
||||
];
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Credentials
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// LLM credentials for explore subagents.
|
||||
pub struct Credentials {
|
||||
pub base_url: String,
|
||||
pub api_key: String,
|
||||
pub model: String,
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// ExploreServiceImpl — implements the application-layer trait
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// 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.
|
||||
pub struct ExploreServiceImpl {
|
||||
tool_ctx: ToolCtx,
|
||||
credentials: Credentials,
|
||||
}
|
||||
|
||||
impl ExploreServiceImpl {
|
||||
pub fn new(tool_ctx: ToolCtx, credentials: Credentials) -> Self {
|
||||
ExploreServiceImpl {
|
||||
tool_ctx,
|
||||
credentials,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl ExploreService for ExploreServiceImpl {
|
||||
fn explore<'a>(
|
||||
&'a self,
|
||||
query: &'a str,
|
||||
workspace_root: &'a str,
|
||||
) -> Pin<Box<dyn Future<Output = Result<ExploreOutput>> + Send + 'a>> {
|
||||
Box::pin(async move {
|
||||
let context = run_explore_phase(
|
||||
query,
|
||||
workspace_root,
|
||||
&self.tool_ctx,
|
||||
&self.credentials,
|
||||
)
|
||||
.await?;
|
||||
|
||||
Ok(ExploreOutput {
|
||||
context_messages: vec![context],
|
||||
summary: format!("{EXPLORE_AGENT_COUNT} explore agents dispatched"),
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Core orchestration
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Spawn EXPLORE_AGENT_COUNT subagents in parallel, join, consolidate.
|
||||
async fn run_explore_phase(
|
||||
query: &str,
|
||||
workspace_root: &str,
|
||||
tool_ctx: &ToolCtx,
|
||||
credentials: &Credentials,
|
||||
) -> Result<String> {
|
||||
// ── Prepare directives ──────────────────────────────────────────────
|
||||
let mut directives: Vec<String> = 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}\""));
|
||||
}
|
||||
d.push_str(&format!("\n\nWorkspace root: {workspace_root}"));
|
||||
directives.push(d);
|
||||
}
|
||||
|
||||
// ── Spawn all agents on threads with their own tokio runtime ────────
|
||||
let mut handles: Vec<(usize, thread::JoinHandle<Result<String>>)> =
|
||||
Vec::with_capacity(EXPLORE_AGENT_COUNT);
|
||||
|
||||
for i in 0..EXPLORE_AGENT_COUNT {
|
||||
let ctx = SubagentContext::new(
|
||||
directives[i].clone(),
|
||||
tool_ctx.clone(),
|
||||
"read".to_string(),
|
||||
credentials.base_url.clone(),
|
||||
credentials.api_key.clone(),
|
||||
credentials.model.clone(),
|
||||
);
|
||||
|
||||
let directive = directives[i].clone();
|
||||
let tc = tool_ctx.clone();
|
||||
|
||||
let handle = thread::spawn(move || {
|
||||
let rt = tokio::runtime::Runtime::new()
|
||||
.context("create explore subagent tokio runtime")?;
|
||||
rt.block_on(run_agent(ctx, &directive, AccessTier::Read, tc))
|
||||
});
|
||||
|
||||
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.
|
||||
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");
|
||||
(i, output, true)
|
||||
}
|
||||
Ok(Err(e)) => {
|
||||
warn!(agent = i, error = %e, "explore subagent failed");
|
||||
(i, format!("Error: {e}"), false)
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(agent = i, error = ?e, "explore subagent panicked");
|
||||
(i, format!("Thread panic: {e:?}"), false)
|
||||
}
|
||||
};
|
||||
out.push(entry);
|
||||
}
|
||||
out
|
||||
})
|
||||
.await
|
||||
.context("explore join task panicked")?;
|
||||
|
||||
// ── Build consolidated context ──────────────────────────────────────
|
||||
Ok(build_explore_context(&results))
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Consolidation
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Format explore results as a system-level context message.
|
||||
fn build_explore_context(results: &[(usize, String, bool)]) -> String {
|
||||
let success_count = results.iter().filter(|r| r.2).count();
|
||||
let total = results.len();
|
||||
|
||||
let mut msg = format!(
|
||||
"[Explore Phase — {success_count}/{total} agents succeeded]\n\n"
|
||||
);
|
||||
|
||||
for (i, output, success) in results {
|
||||
let label = EXPLORE_LABELS.get(*i).unwrap_or(&"❓ Unknown");
|
||||
if *success {
|
||||
msg.push_str(&format!("=== {label} ===\n{output}\n\n"));
|
||||
} else {
|
||||
msg.push_str(&format!("=== {label} (FAILED) ===\n{output}\n\n"));
|
||||
}
|
||||
}
|
||||
|
||||
msg
|
||||
}
|
||||
@@ -22,6 +22,7 @@
|
||||
pub mod arch_audit;
|
||||
pub mod code_quality;
|
||||
pub mod commit;
|
||||
pub mod explore;
|
||||
pub mod skills;
|
||||
|
||||
use anyhow::Result;
|
||||
|
||||
@@ -44,6 +44,7 @@ pub fn all_tools() -> Vec<Box<dyn super::Tool>> {
|
||||
Box::new(super::web_search::WebSearch),
|
||||
Box::new(super::semantic_search::SemanticSearch),
|
||||
Box::new(super::semantic_search::RebuildIndex),
|
||||
Box::new(super::semantic_search::ListSymbols),
|
||||
Box::new(super::parallel_delegate::ParallelDelegate),
|
||||
// ── Best-practice tools (built-in) ─────────────────────────
|
||||
Box::new(super::best_practice::BestPractice),
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -110,6 +110,11 @@ pub fn spawn_agent_turn(state: &mut AppStateRest, text: String) {
|
||||
api_base: api_base.clone(),
|
||||
};
|
||||
|
||||
// Clone credentials before moving into LlmClient.
|
||||
let explore_api_key = api_key.clone();
|
||||
let explore_model = model.clone();
|
||||
let explore_base_url = api_base.clone().unwrap_or_else(|| "https://api.openai.com/v1".to_string());
|
||||
|
||||
let client = std::sync::Arc::new(LlmClient::new(api_key, model, api_base));
|
||||
|
||||
let tool_ctx = ToolCtx::builder()
|
||||
@@ -118,13 +123,30 @@ pub fn spawn_agent_turn(state: &mut AppStateRest, text: String) {
|
||||
.turn_events(turn_events)
|
||||
.build();
|
||||
|
||||
// Clone ToolCtx for the explore service (before moving into executor).
|
||||
let explore_ctx = tool_ctx.clone();
|
||||
|
||||
let tool_executor =
|
||||
std::sync::Arc::new(InfrastructureToolExecutor::new(tool_ctx));
|
||||
|
||||
let tools = all_tools();
|
||||
let defs = tool_defs(&tools);
|
||||
|
||||
let turn_service = AgentTurnServiceImpl::new(client, tool_executor, defs);
|
||||
// Wire the mandatory explore phase (3+ parallel subagents).
|
||||
let explore_creds = zesdex_infrastructure::best_practice::explore::Credentials {
|
||||
base_url: explore_base_url,
|
||||
api_key: explore_api_key,
|
||||
model: explore_model,
|
||||
};
|
||||
let explore_service = std::sync::Arc::new(
|
||||
zesdex_infrastructure::best_practice::explore::ExploreServiceImpl::new(
|
||||
explore_ctx,
|
||||
explore_creds,
|
||||
),
|
||||
);
|
||||
|
||||
let turn_service = AgentTurnServiceImpl::new(client, tool_executor, defs)
|
||||
.with_explore(explore_service);
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _ = turn_service.run_turn(params).await;
|
||||
|
||||
Reference in New Issue
Block a user