asepharyana
9a67137954
Crate renames: - zesdex-entities::seaorm → domain (misleading name, no SeaORM used) - zesdex-dto → merged into zesdex-entities (100% re-exports) - zesdex-libs → zesdex-infra (vague name) Module renames: - app/harness → guard (misleading: safety gatekeeper, not test harness) - runtime/commands → action_dispatch (name clashed with controller/command) - resources → prompts (embedded prompt text, not general resources) - tool/seqthink → sequential_think (unreadable abbreviation) - msglog/query → insert (module only inserts, never queries) Dead code removal: - app/mode/help.rs (orphaned — not declared in mod.rs) - app/mode/loading.rs (orphaned — not declared in mod.rs) File splitting (71 new files, avg ~115 lines/file): - app/runtime/actions/: 1→8 files (was 2030 lines) - view/overlays/: 1→16 files (was 1167 lines) - tool/lsp/: 1→8 per-tool files (was 909 lines) - main.rs: 1→5 files (session, daemon, attach, event_loop) - workflow/engine + hive_mind: 2→10 files - subagent/engine + auto: 2→9 files - lsp/provisioner: 1→5 files - review/: 1→6 files - guard/: 1→2 files (extracted patterns) - state/misc: 1→3 files (input, scroll) - mcp/: 1→3 files (transport, adapter) - stream/json_repair extracted from turn.rs DRY: - Pattern constants (STUB_PATTERNS etc) in guard/patterns shared with subagent - 3 near-identical background spawners → 1 generic + thin wrappers - Shared spawn_subagent_with_drain() extracted - Shared create_session() in main - write_osc52 deduplicated Bug fixes: - archive_message(): sess.db → db (wrong variable name) - execute_one_tool(): wrong parameter name - check_credential_read() function was missing (restored from test expectations)
427 lines
20 KiB
Rust
427 lines
20 KiB
Rust
//! Subagent execution loop: drive an LLM conversation, run tools, and stream
|
|
//! progress events to the parent via an mpsc channel.
|
|
//!
|
|
//! Tool gating and pattern-constant definitions live in sibling modules
|
|
//! (`gating`, `provider`, `tools`, `workspace`) rather than here, so each
|
|
//! concern is independently testable and maintainable.
|
|
|
|
use super::context::SubagentContext;
|
|
use super::event::SubagentEvent;
|
|
use super::gating::gate_subagent_tool_call;
|
|
use super::provider::{require_api_key, resolve_provider_config};
|
|
use super::tools::build_subagent_tools;
|
|
use super::workspace::generate_workspace_tree;
|
|
use crate::dto::chat::message::ChatMessage;
|
|
use crate::dto::provider::request::ToolDef;
|
|
use crate::tool::tool_is_risky;
|
|
use sha2::Digest;
|
|
use tokio::sync::mpsc;
|
|
use zesdex_cms::domain::repository::EditLogRepository;
|
|
|
|
fn format_subagent_progress(prefix: &str, text: &str) -> String {
|
|
let lines: Vec<&str> = text.lines().filter(|l| !l.trim().is_empty()).collect();
|
|
if lines.is_empty() {
|
|
format!("{prefix}...")
|
|
} else if lines.len() == 1 {
|
|
format!("{prefix}: {}", lines[0])
|
|
} else {
|
|
lines[lines.len() - 2..].join("\n")
|
|
}
|
|
}
|
|
|
|
/// Synchronous subagent entry point: run up to `ctx.max_steps` iterations
|
|
/// of the LLM tool loop.
|
|
///
|
|
/// Flow: inject system prompt (with workspace tree if available) → for each
|
|
/// step: resolve provider config, build an LLM client, call
|
|
/// `chat_with_tools_streaming` (with abort check per SSE event), process
|
|
/// tool calls (gated against both the allowlist and Harness-style content
|
|
/// safety checks) or collect text output → send `SubagentEvent`s on `tx` →
|
|
/// break on first text-only (non-empty) response.
|
|
///
|
|
/// Why: runs synchronously on a dedicated thread so the main async event
|
|
/// loop is not blocked. Tool gating prevents restricted, risky, or
|
|
/// malicious/poor-quality tool calls from executing.
|
|
///
|
|
/// Return: the concatenated text output, or an `anyhow::Error` if the LLM
|
|
/// call fails at any step.
|
|
pub fn run_subagent(
|
|
ctx: &SubagentContext,
|
|
tx: &mpsc::Sender<SubagentEvent>,
|
|
) -> anyhow::Result<String> {
|
|
let mut output = String::new();
|
|
let mut messages: Vec<ChatMessage> = Vec::new();
|
|
|
|
// Build system prompt with workspace tree context if we have workspaces,
|
|
// giving subagents the same project-awareness as the main agent.
|
|
let system_with_context = if ctx.workspaces.is_empty() {
|
|
ctx.system_prompt.clone()
|
|
} else {
|
|
let tree_info = generate_workspace_tree(&ctx.workspaces);
|
|
format!("{}\n\n{}", ctx.system_prompt, tree_info)
|
|
};
|
|
messages.push(ChatMessage::system(system_with_context));
|
|
|
|
let tool_ctx = crate::tool::ToolCtx::builder()
|
|
.session_dir(ctx.session_dir.clone())
|
|
.workspaces(ctx.workspaces.clone())
|
|
.origin(crate::app::state::types::Origin::SubAgent)
|
|
.workflow_findings(ctx.workflow_findings.clone())
|
|
.build();
|
|
|
|
// Build tool list once before the loop
|
|
let (tools, tdefs) = build_subagent_tools(&ctx.allowed_tools);
|
|
let tdefs_opt: Option<Vec<ToolDef>> = if tdefs.is_empty() { None } else { Some(tdefs) };
|
|
|
|
// Cache provider config once before the loop instead of re-resolving
|
|
// from disk on every step (Settings::load + AppConfig::load each parse
|
|
// JSON files, and the config cannot change between steps).
|
|
let (api_key, model, base_url, provider) = resolve_provider_config();
|
|
|
|
// Fail fast on a missing key instead of sending a doomed request: an
|
|
// empty api_key still reaches the network (base_url falls back to a
|
|
// default endpoint), so without this check every step burns a full
|
|
// 10-retry timeout/backoff cycle against a server that was never going
|
|
// to authenticate, and the real cause (no key configured) never
|
|
// surfaces past a buried WARN log.
|
|
if let Err(error) = require_api_key(&api_key, &provider) {
|
|
let error = error.to_string();
|
|
let _ = tx.blocking_send(SubagentEvent::StepFailed {
|
|
step: 0,
|
|
error: error.clone(),
|
|
});
|
|
anyhow::bail!(error);
|
|
}
|
|
|
|
let client = crate::service::provider::LlmClient::new(api_key, model, base_url);
|
|
|
|
for step in 0..ctx.max_steps {
|
|
// Check abort flag before each LLM call so a stuck subagent can
|
|
// be cancelled from the parent (mirrors main agent behaviour).
|
|
if ctx
|
|
.abort_flag
|
|
.as_ref()
|
|
.is_some_and(|f| f.load(std::sync::atomic::Ordering::SeqCst))
|
|
{
|
|
let _ = tx.blocking_send(SubagentEvent::StepFailed {
|
|
step,
|
|
error: "subagent aborted by parent".to_string(),
|
|
});
|
|
anyhow::bail!("subagent aborted by parent at step {step}");
|
|
}
|
|
|
|
let tx_clone = tx.clone();
|
|
let mut current_thinking = String::new();
|
|
let mut current_token = String::new();
|
|
let mut step_usage: Option<(u64, u64)> = None;
|
|
|
|
// Use streaming API so the abort flag is checked per SSE event,
|
|
// making the subagent responsive to cancellation even during an
|
|
// LLM call (non-streaming would block for 10-30s unchecked).
|
|
let stream_result = client.chat_with_tools_streaming(
|
|
&messages,
|
|
tdefs_opt.clone(),
|
|
Some(0.7),
|
|
Some(4096),
|
|
|event| -> bool {
|
|
// Check abort on every SSE event for responsive cancellation.
|
|
if ctx
|
|
.abort_flag
|
|
.as_ref()
|
|
.is_some_and(|f| f.load(std::sync::atomic::Ordering::SeqCst))
|
|
{
|
|
return false; // signals provider to abort
|
|
}
|
|
match event {
|
|
crate::app::runtime::stream::StreamEvent::Reasoning(text) => {
|
|
current_thinking.push_str(text);
|
|
let prog = format_subagent_progress("thinking", ¤t_thinking);
|
|
let _ = tx_clone.blocking_send(SubagentEvent::Progress(prog));
|
|
}
|
|
crate::app::runtime::stream::StreamEvent::Token(text) => {
|
|
current_token.push_str(text);
|
|
let prog = format_subagent_progress("replying", ¤t_token);
|
|
let _ = tx_clone.blocking_send(SubagentEvent::Progress(prog));
|
|
}
|
|
crate::app::runtime::stream::StreamEvent::Usage {
|
|
prompt_tokens,
|
|
completion_tokens,
|
|
..
|
|
} => {
|
|
// Capture usage so the drain thread can route it
|
|
// to the parent's `UsageStats::review_tokens`.
|
|
// Last writer wins — providers send exactly one
|
|
// Usage event per streaming call.
|
|
step_usage = Some((*prompt_tokens, *completion_tokens));
|
|
}
|
|
_ => {}
|
|
}
|
|
true
|
|
},
|
|
);
|
|
|
|
let (response, returned_usage) = match stream_result {
|
|
Ok(result) => result,
|
|
Err(e) => {
|
|
let is_abort = ctx
|
|
.abort_flag
|
|
.as_ref()
|
|
.is_some_and(|f| f.load(std::sync::atomic::Ordering::SeqCst))
|
|
|| e.to_string().contains("aborted");
|
|
let _ = tx.blocking_send(SubagentEvent::StepFailed {
|
|
step,
|
|
error: if is_abort {
|
|
"subagent aborted by user".to_string()
|
|
} else {
|
|
e.to_string()
|
|
},
|
|
});
|
|
if is_abort {
|
|
anyhow::bail!("subagent aborted by parent at step {step}");
|
|
}
|
|
// No non-streaming fallback — API must support streaming.
|
|
// Non-streaming calls block for up to 1 min without checking
|
|
// abort_flag, making cancellation unresponsive.
|
|
anyhow::bail!("subagent call failed at step {step}: {e}");
|
|
}
|
|
};
|
|
|
|
// Emit the token usage from this streaming call so the parent's
|
|
// drain thread can accumulate it and update the Usage panel.
|
|
// Without this, the Usage panel always shows zeros because the
|
|
// subagent never tells the parent about the tokens consumed.
|
|
let (mut tok_in, mut tok_out) = returned_usage.unwrap_or((0, 0));
|
|
if tok_in == 0 {
|
|
let prompt_chars: usize = messages
|
|
.iter()
|
|
.filter_map(|m| m.content.as_deref())
|
|
.map(str::len)
|
|
.sum();
|
|
tok_in = (prompt_chars / 4).max(1) as u64;
|
|
}
|
|
if tok_out == 0 {
|
|
let response_chars = response.content.as_deref().map_or(0, str::len);
|
|
tok_out = (response_chars / 4).max(1) as u64;
|
|
}
|
|
let _ = tx.blocking_send(SubagentEvent::Usage {
|
|
tokens_in: tok_in,
|
|
tokens_out: tok_out,
|
|
});
|
|
|
|
let has_tool_calls = response.tool_calls.is_some()
|
|
&& response
|
|
.tool_calls
|
|
.as_ref()
|
|
.is_some_and(|tc| !tc.is_empty());
|
|
|
|
let content = response.content.clone().unwrap_or_default();
|
|
|
|
// Emit thinking/reasoning text as StepCompleted so the parent's
|
|
// drain thread can show it as progress instead of just the tool name.
|
|
if !content.is_empty() {
|
|
let _ = tx.blocking_send(SubagentEvent::StepCompleted {
|
|
output: content.clone(),
|
|
});
|
|
}
|
|
|
|
if has_tool_calls {
|
|
let tool_calls = response.tool_calls.clone().unwrap_or_default();
|
|
// Push the assistant message with tool_calls into the conversation
|
|
messages.push(response);
|
|
|
|
let mut results_vec = Vec::new();
|
|
std::thread::scope(|s| {
|
|
let mut handles = Vec::new();
|
|
let tools_ref = &tools;
|
|
let tool_ctx_ref = &tool_ctx;
|
|
for tool_call in &tool_calls {
|
|
let handle = s.spawn(move || {
|
|
// Check abort flag before each tool execution
|
|
if ctx.abort_flag.as_ref().is_some_and(|f| f.load(std::sync::atomic::Ordering::SeqCst)) {
|
|
return (tool_call, Err(anyhow::anyhow!("subagent aborted by parent during tool execution")));
|
|
}
|
|
|
|
let tool_name = &tool_call.function.name;
|
|
let args = crate::dto::chat::tool::sanitize_tool_arguments(&tool_call.function.arguments);
|
|
let explicitly_allowed = ctx.allowed_tools.contains(tool_name);
|
|
let generally_allowed = ctx.allowed_tools.is_empty() || explicitly_allowed;
|
|
|
|
// Level 1: allowlist check — is this tool even permitted?
|
|
if !generally_allowed {
|
|
return (tool_call, Ok(format!("tool '{tool_name}' not allowed for this subagent")));
|
|
}
|
|
|
|
// Level 2: risky tool check — risky tools require explicit permission
|
|
if tool_is_risky(tool_name) && !explicitly_allowed {
|
|
return (tool_call, Ok(format!("risky tool '{tool_name}' requires explicit permission; not allowed for this subagent")));
|
|
}
|
|
|
|
// Level 3: Harness-style content safety gating
|
|
if let Some(block_reason) = gate_subagent_tool_call(tool_name, &args) {
|
|
return (tool_call, Ok(format!("Blocked by subagent gate: {block_reason}")));
|
|
}
|
|
|
|
let result = match tools_ref.iter().find(|t| t.name() == tool_name.as_str()) {
|
|
Some(tool) => {
|
|
let is_edit = tool_name == "write" || tool_name == "edit";
|
|
if is_edit && !tool_call.id.is_empty() {
|
|
if let Ok(conn) = crate::model::msglog::open_or_create(&ctx.session_dir) {
|
|
let path = args.get("path").and_then(|v| v.as_str()).unwrap_or("");
|
|
if let Ok(abs_path) = crate::tool::resolve_path(&tool_ctx_ref.workspaces, path) {
|
|
if let Ok(bytes) = std::fs::read(&abs_path) {
|
|
let session_id = ctx.session_dir
|
|
.file_name()
|
|
.and_then(|n| n.to_str())
|
|
.unwrap_or("unknown");
|
|
let _ = crate::model::msglog::store_blob(
|
|
&conn, session_id, &tool_call.id, &bytes, None,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
let run_res = tool.run(tool_ctx_ref, &args);
|
|
|
|
if is_edit && run_res.is_ok() {
|
|
let reason = args
|
|
.get("reason")
|
|
.and_then(|v| v.as_str())
|
|
.unwrap_or("unnamed");
|
|
let path = args
|
|
.get("path")
|
|
.and_then(|v| v.as_str())
|
|
.unwrap_or("unknown");
|
|
let content_sha256 = {
|
|
let content = args.get("content").or_else(|| args.get("new"));
|
|
let hash = sha2::Sha256::digest(
|
|
content.and_then(|v| v.as_str()).unwrap_or("").as_bytes(),
|
|
);
|
|
hex::encode(hash)
|
|
};
|
|
let bytes_delta = if tool_name == "write" {
|
|
args.get("content")
|
|
.and_then(|v| v.as_str())
|
|
.map_or(0, |s| s.len() as i64)
|
|
} else {
|
|
let old = args.get("old").and_then(|v| v.as_str()).unwrap_or("");
|
|
let new = args.get("new").and_then(|v| v.as_str()).unwrap_or("");
|
|
(new.len() as i64 - old.len() as i64).abs()
|
|
};
|
|
let session_id = ctx.session_dir
|
|
.file_name()
|
|
.and_then(|n| n.to_str())
|
|
.unwrap_or("unknown")
|
|
.to_string();
|
|
let entry = zesdex_cms::domain::edit_log::EditLogEntry {
|
|
ts: chrono::Utc::now().timestamp_millis(),
|
|
tool: tool_name.clone(),
|
|
path: path.to_string(),
|
|
reason: reason.to_string(),
|
|
content_sha256,
|
|
bytes_delta,
|
|
origin: tool_ctx_ref.origin.tag(),
|
|
session_id,
|
|
};
|
|
let repo = zesdex_cms::infrastructure::persistence::edit_log_repo::JsonlEditLogRepository::new();
|
|
if let Ok(mut el) = repo.open(&ctx.session_dir) {
|
|
let _ = repo.append(&ctx.session_dir, &mut el, entry);
|
|
}
|
|
}
|
|
run_res
|
|
}
|
|
None => Err(anyhow::anyhow!("tool '{tool_name}' not found")),
|
|
};
|
|
(tool_call, result)
|
|
});
|
|
handles.push(handle);
|
|
}
|
|
for h in handles {
|
|
if let Ok(res) = h.join() {
|
|
results_vec.push(res);
|
|
}
|
|
}
|
|
});
|
|
|
|
for (tool_call, result) in results_vec {
|
|
let tool_name = &tool_call.function.name;
|
|
let args =
|
|
crate::dto::chat::tool::sanitize_tool_arguments(&tool_call.function.arguments);
|
|
|
|
let _ = tx.blocking_send(SubagentEvent::ToolCall {
|
|
tool: tool_name.clone(),
|
|
args: args.clone(),
|
|
});
|
|
|
|
match result {
|
|
Ok(output_text) => {
|
|
messages.push(ChatMessage::tool_result(
|
|
tool_call.id.clone(),
|
|
output_text.clone(),
|
|
));
|
|
let _ = tx.blocking_send(SubagentEvent::ToolResult {
|
|
tool: tool_name.clone(),
|
|
args: args.clone(),
|
|
});
|
|
|
|
let is_readonly = tool_name == "read"
|
|
|| tool_name == "view_file"
|
|
|| tool_name == "grep"
|
|
|| tool_name == "grep_search"
|
|
|| tool_name == "glob"
|
|
|| tool_name == "dir_list"
|
|
|| tool_name == "list_dir";
|
|
|
|
if is_readonly {
|
|
if let Some(ref findings) = ctx.workflow_findings {
|
|
if let Ok(mut f) = findings.lock() {
|
|
let args_json =
|
|
serde_json::to_string(&args).unwrap_or_default();
|
|
let mut shared_text = output_text;
|
|
if shared_text.len() > 50_000 {
|
|
shared_text.truncate(50_000);
|
|
shared_text.push_str("\n...[truncated]");
|
|
}
|
|
f.push(format!("[Auto-Shared] Sibling drone executed '{tool_name}' with args {args_json}:\n{shared_text}"));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
Err(e) => {
|
|
let err_str = e.to_string();
|
|
if err_str.contains("subagent aborted by parent") {
|
|
let _ = tx.blocking_send(SubagentEvent::StepFailed {
|
|
step,
|
|
error: err_str.clone(),
|
|
});
|
|
anyhow::bail!("{err_str}");
|
|
}
|
|
let msg = format!("tool '{tool_name}' failed: {e}");
|
|
messages.push(ChatMessage::tool_result(tool_call.id.clone(), msg.clone()));
|
|
let _ = tx.blocking_send(SubagentEvent::ToolResult {
|
|
tool: tool_name.clone(),
|
|
args: args.clone(),
|
|
});
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
// Text-only response — accumulate and finish
|
|
if !content.is_empty() {
|
|
output.push_str(&content);
|
|
output.push('\n');
|
|
}
|
|
let _ = tx.blocking_send(SubagentEvent::StepCompleted {
|
|
output: content.clone(),
|
|
});
|
|
// Break only when we got real content; empty means something went wrong
|
|
if !content.is_empty() {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
let _ = tx.blocking_send(SubagentEvent::Completed);
|
|
Ok(output)
|
|
}
|