MythEclipse/zesdex
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

perf(agent): stabilkan async & parallel — satu runtime, bounded concurrency, isolasi error

Browse Source 
Seperti Claude Code: satu runtime shared, concurrency dibatasi, error
subagent terisolasi (satu node gagal tidak menggagalkan cycle).

- feat(runtime): global tokio runtime via OnceLock — ganti 9+ titik
  Runtime::new() per tool call (spawn, parallel_delegate, workflow,
  explore, dir_cache, daemon handler). Hemat resource, hilangkan panic
  path Runtime::new().expect() di daemon compaction.
- fix(workflow): execute_cycle ganti try_join_all (fail-fast) →
  buffer_unordered(8) + isolasi error per node; node gagal di-log dan
  diganti [ERROR], hasil node lain tetap dipakai (Claude Code-style).
- fix(parallel_delegate): spawn subagent dibatasi per batch max_parallel
  (tidak unbounded threads).
- perf(subagent): run_agent adaptif max_tokens (800/1600/4096), temp 0.2,
  truncate tool output 12k, error-recovery note utk tool error berulang.
- test: runtime singleton + block_on (2 test).
This commit is contained in:
asepharyana
2026-08-27 23:25:44 +07:00
parent 3847c0e6fd
commit 6a98d52d54
11 changed files with 225 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/infrastructure/src/tools/parallel_delegate.rs
+41 -37
View File
@@ -123,7 +123,7 @@ impl Tool for ParallelDelegate {
.collect()
} else {
// Auto-split using LLM
let rt = tokio::runtime::Runtime::new()?;
let rt = crate::runtime::runtime();
let directives = rt.block_on(auto_split_task(
&task,
max_parallel,
@@ -146,49 +146,53 @@ impl Tool for ParallelDelegate {
"parallel delegation: starting subagents"
);
// Spawn agents in parallel
let mut handles = Vec::new();
for (i, (directive, access)) in directives.iter().enumerate() {
let subagent_ctx = SubagentContext::new(
directive.clone(),
ctx.clone(),
format!("{access:?}"),
base_url.clone(),
api_key.clone(),
model.clone(),
);
debug!(agent_index = i, access = ?access, "spawning parallel agent");
let handle = spawn_subagent(subagent_ctx, directive.clone(), *access, ctx.clone());
handles.push((i, handle));
}
// Join all results
// Spawn agents in parallel — bounded: never more than `max_parallel`
// subagent threads in flight at once (Claude Code-style isolation).
let mut results: Vec<(usize, String, String)> = Vec::new();
for (i, handle) in handles {
match handle.join() {
Ok(Ok(output)) => {
info!(agent_index = i, "parallel agent completed");
results.push((i, directives[i].0.clone(), output));
}
Ok(Err(e)) => {
warn!(agent_index = i, error = %e, "parallel agent failed");
results.push((i, directives[i].0.clone(), format!("[ERROR] {e}")));
}
Err(e) => {
warn!(agent_index = i, error = ?e, "parallel agent panicked");
results.push((
i,
directives[i].0.clone(),
"[ERROR] Agent panicked".to_string(),
));
for batch in directives.chunks(max_parallel) {
let mut handles = Vec::with_capacity(batch.len());
for (i, (directive, access)) in batch.iter().enumerate() {
let global_idx = results.len() + i;
let subagent_ctx = SubagentContext::new(
directive.clone(),
ctx.clone(),
format!("{access:?}"),
base_url.clone(),
api_key.clone(),
model.clone(),
);
debug!(agent_index = global_idx, access = ?access, "spawning parallel agent");
let handle = spawn_subagent(subagent_ctx, directive.clone(), *access, ctx.clone());
handles.push((global_idx, handle));
}
// Join this batch before spawning the next.
for (i, handle) in handles {
match handle.join() {
Ok(Ok(output)) => {
info!(agent_index = i, "parallel agent completed");
results.push((i, directives[i].0.clone(), output));
}
Ok(Err(e)) => {
warn!(agent_index = i, error = %e, "parallel agent failed");
results.push((i, directives[i].0.clone(), format!("[ERROR] {e}")));
}
Err(e) => {
warn!(agent_index = i, error = ?e, "parallel agent panicked");
results.push((
i,
directives[i].0.clone(),
"[ERROR] Agent panicked".to_string(),
));
}
}
}
}
// Consolidate results
if synthesize && results.len() > 1 {
let rt = tokio::runtime::Runtime::new()?;
let rt = crate::runtime::runtime();
let consolidated =
rt.block_on(consolidate_results(&results, &base_url, &api_key, &model))?;
Ok(format!(