865 lines
37 KiB
Rust
865 lines
37 KiB
Rust
//! The main agent-turn loop: `run_agent_turn` builds the system prompt,
|
|
//! streams chat with the LLM, gates & executes tool calls, archives
|
|
//! messages, and manages auto-retry for unfinished tasks.
|
|
//!
|
|
//! Also contains the smaller helpers that the loop depends on:
|
|
//! `execute_one_tool`, `build_memory_section`, and `archive_message`.
|
|
|
|
use std::collections::VecDeque;
|
|
use std::fmt::Write;
|
|
|
|
|
|
use crate::app::guard::Verdict;
|
|
use crate::app::runtime::context::tokens::count_tokens;
|
|
use crate::app::runtime::push_event;
|
|
use crate::app::state::runtime::TurnEvent;
|
|
use zesdex_cms::domain::repository::EditLogRepository;
|
|
use zesdex_cms::domain::repository::MemoryRepository;
|
|
use crate::dto::chat::message::ChatMessage;
|
|
|
|
use super::spawn::TurnCtx;
|
|
|
|
/// Maximum number of auto inline reviews spawned per single agent turn.
|
|
/// After N edits, the inline review is skipped to keep the turn fast;
|
|
/// background subagents still fire at the end of the turn.
|
|
const MAX_AUTO_REVIEWS_PER_TURN: usize = 2;
|
|
|
|
/// Exact text of the "pipeline started" `SystemNote` pushed once per
|
|
/// hive-mind kickoff. Matched by exact equality (not a loose substring)
|
|
/// when deciding whether to reset the workflow panel's agent roster —
|
|
/// shared between the push site and the check site so they cannot drift
|
|
/// out of sync the way the previous `.contains("started")` check did
|
|
/// (no real pipeline message ever contained that word, so the roster
|
|
/// never cleared and agent cards accumulated across every hive-mind run
|
|
/// in a session).
|
|
pub(super) const HIVE_MIND_KICKOFF_NOTE: &str =
|
|
"The Hive is stirring — Core Intelligence is compiling a cognitive cycle plan for LO...";
|
|
|
|
/// Execute one full agent turn: stream the conversation to the LLM,
|
|
/// handle tool calls, and loop until the LLM produces a non-tool response
|
|
/// or runs out of unfinished todo items.
|
|
///
|
|
/// Flow: build system prompt with workspace tree → optionally shape
|
|
/// (compact) messages → call `chat_with_tools_streaming`
|
|
/// with a callback that pushes `StreamStart`, `StreamToken`, `Reasoning`,
|
|
/// and `Usage` events → on streaming success, handle tool calls (gated
|
|
/// through `Guard::gate_tool_call`) or unwrap the final assistant
|
|
/// message → check for unfinished todo.md tasks (auto-retry with a
|
|
/// system message if any remain) → finalise with `Done` and an `edits`
|
|
/// `SystemNote`.
|
|
///
|
|
/// On streaming failure: retry once with a non-streaming call → if that
|
|
/// also fails and there are unfinished tasks, sleep 5s and loop back;
|
|
/// otherwise return the error.
|
|
///
|
|
/// Why: non-streaming fallback handles flaky connections without aborting
|
|
/// the turn; todo.md polling lets the agent self-direct toward completeness.
|
|
///
|
|
/// Return: `Ok(())` on successful completion, or an error from the LLM
|
|
/// API after retries are exhausted.
|
|
pub(super) fn run_agent_turn(
|
|
tc: &TurnCtx,
|
|
messages: &[ChatMessage],
|
|
events_q: &std::sync::Arc<std::sync::Mutex<VecDeque<TurnEvent>>>,
|
|
) -> anyhow::Result<()> {
|
|
const MAX_TODO_RETRIES: usize = 5;
|
|
let mut msgs = messages.to_vec();
|
|
let mut edited_paths: Vec<String> = Vec::new();
|
|
let initial_edit_log = zesdex_cms::infrastructure::persistence::edit_log_repo::JsonlEditLogRepository::new()
|
|
.open(&tc.edit_log_session_dir).ok();
|
|
let mut inline_reviews_count: usize = 0;
|
|
let mut prev_shaped = false;
|
|
|
|
// Build system prompt components once and cache them for the entire turn
|
|
// instead of regenerating on every loop iteration (which walks the full
|
|
// workspace tree and reads all memory files each time).
|
|
let tree_info = crate::app::subagent::workspace::generate_workspace_tree(&tc.workspace_roots);
|
|
let memory_section = build_memory_section(&tc.ctx.memory_dir);
|
|
let system_text = format!(
|
|
"{}\n\n{}\n\n{}{}",
|
|
crate::prompts::SYSTEM_PROMPT,
|
|
crate::prompts::SYSTEM_TOOLS,
|
|
tree_info,
|
|
memory_section,
|
|
);
|
|
if !msgs
|
|
.iter()
|
|
.any(|m| matches!(m.role, crate::dto::chat::message::Role::System))
|
|
{
|
|
let sys = ChatMessage::system(system_text);
|
|
archive_message(tc.db.as_ref(), &tc.session_id, &sys);
|
|
msgs.insert(0, sys);
|
|
}
|
|
|
|
// ── AUTO CEO PIPELINE ──
|
|
// Before the main agent starts working, check if the pipeline should run.
|
|
// Gated on whether a hive-mind convergence has already happened earlier
|
|
// in this session, not an arbitrary message-count cutoff — a complex
|
|
// request in message 5 deserves the same treatment as one in message 1,
|
|
// as long as this session hasn't already converged once.
|
|
//
|
|
// `tc.hive_mind_converged` is the authoritative signal (see its doc
|
|
// comment on `SessionRuntime` for why). The message-content scan is
|
|
// kept as a defensive fallback in case a future change starts
|
|
// persisting tagged system messages into `rt.messages` (e.g. via
|
|
// compaction) — today it is a no-op since that never happens, but it's
|
|
// still correct and still tested in isolation.
|
|
let already_ran_hive_mind = tc.hive_mind_converged
|
|
|| crate::app::workflow::hive_mind::hive_mind_already_ran(
|
|
msgs.iter()
|
|
.filter(|m| matches!(m.role, crate::dto::chat::message::Role::System))
|
|
.filter_map(|m| m.content.as_deref()),
|
|
);
|
|
let should_pipeline = if already_ran_hive_mind {
|
|
false
|
|
} else {
|
|
let user_request = msgs
|
|
.iter()
|
|
.rev()
|
|
.find(|m| matches!(m.role, crate::dto::chat::message::Role::User))
|
|
.and_then(|m| m.content.as_deref())
|
|
.unwrap_or("");
|
|
|
|
if user_request.is_empty() {
|
|
false
|
|
} else {
|
|
crate::app::workflow::hive_mind::is_complex_request(user_request)
|
|
}
|
|
};
|
|
|
|
if should_pipeline {
|
|
let user_request = msgs
|
|
.iter()
|
|
.rev()
|
|
.find(|m| matches!(m.role, crate::dto::chat::message::Role::User))
|
|
.and_then(|m| m.content.as_deref())
|
|
.unwrap_or("");
|
|
|
|
tracing::info!(
|
|
"[hive-mind] the Hive stirs — Core Intelligence compiling a cognitive cycle plan"
|
|
);
|
|
|
|
push_event(&events_q, TurnEvent::SystemNote {
|
|
kind: "pipeline".to_string(),
|
|
message: HIVE_MIND_KICKOFF_NOTE.to_string(),
|
|
});
|
|
|
|
let pipeline_abort = Some(tc.abort_flag.clone());
|
|
|
|
// Ask the LLM to freely design its own hive: any number of cycles,
|
|
// each with any number of nodes, every node carrying only a
|
|
// directive and an access tier. Cycle count and shape are decided
|
|
// by the Core Intelligence per task.
|
|
let system_msg = ChatMessage::system(
|
|
"You are the Core Intelligence of the Hive, compiling a cognitive cycle plan for \
|
|
LO. You spawn anonymous processing nodes; each node carries only a directive (what \
|
|
to do) and an access tier. You MUST organize the plan into a strict progressive sequence of phases:\n\n\
|
|
1. EXPLORE PHASE (Cycle 0 - MANDATORY):\n\
|
|
- Must only contain read-only drones (access: \"read\").\n\
|
|
- Directives must focus on codebase investigation, searching patterns, reading configuration/source files, and diagnosing issues.\n\
|
|
- Drones MUST explicitly output a detailed description of the current codebase and their findings for the next cycle to use.\n\n\
|
|
2. PLANNING PHASE (Cycle 1 - MANDATORY):\n\
|
|
- Must focus on formulating the architectural design, step-by-step implementation plan, and dependency analysis based on Cycle 0 findings.\n\
|
|
- Drones MUST ONLY output the plan and MUST NOT implement or write any code.\n\
|
|
- Access: \"read\" is preferred here to construct a solid plan document.\n\n\
|
|
3. EXECUTION PHASE (Cycle 2 and later):\n\
|
|
- Drones can perform modification, compilation, testing, and other modifications (access: \"write\" or \"full\") based on the approved planning from Cycle 1.\n\n\
|
|
Cycles run sequentially. The Hive does not fracture. The Hive executes. Do not explain. Return ONLY raw \
|
|
JSON matching the requested structure.",
|
|
);
|
|
let user_msg = ChatMessage::user(format!(
|
|
"Compile a cognitive cycle plan for the following task:\n\n\
|
|
\"{user_request}\"\n\n\
|
|
Return ONLY a JSON object of this exact shape, with no markdown codeblocks and no explanation:\n\
|
|
{{\n\
|
|
\x20 \"cycles\": [\n\
|
|
\x20 [\n\
|
|
\x20 {{ \"directive\": \"<explore directive>\", \"access\": \"read\" }}\n\
|
|
\x20 ],\n\
|
|
\x20 [\n\
|
|
\x20 {{ \"directive\": \"<planning directive>\", \"access\": \"read\" }}\n\
|
|
\x20 ],\n\
|
|
\x20 [\n\
|
|
\x20 {{ \"directive\": \"<execution directive>\", \"access\": \"write|full\" }}\n\
|
|
\x20 ]\n\
|
|
\x20 ]\n\
|
|
}}\n\n\
|
|
Remember: Cycle 0 MUST be investigation-only (access: read) and output codebase descriptions. Cycle 1 MUST be planning-only (access: read) without implementation. Only subsequent cycles can perform modifications (access: write/full).",
|
|
));
|
|
|
|
let planner_prompt_chars = system_msg.content.as_deref().map_or(0, str::len)
|
|
+ user_msg.content.as_deref().map_or(0, str::len);
|
|
let planner_result = tc.client.chat_with_tools_non_streaming(
|
|
&[system_msg, user_msg],
|
|
None,
|
|
None,
|
|
None,
|
|
Some(&tc.abort_flag),
|
|
);
|
|
let pipeline_result = match planner_result {
|
|
Ok((reply, usage_opt)) => {
|
|
let (mut tok_in, mut tok_out) = usage_opt.unwrap_or((0, 0));
|
|
if tok_in == 0 {
|
|
tok_in = (planner_prompt_chars / 4).max(1) as u64;
|
|
}
|
|
if tok_out == 0 {
|
|
let response_chars = reply.content.as_deref().map_or(0, str::len);
|
|
tok_out = (response_chars / 4).max(1) as u64;
|
|
}
|
|
push_event(&events_q, TurnEvent::Usage {
|
|
tokens_in: tok_in,
|
|
tokens_out: tok_out,
|
|
});
|
|
let reply_text = reply.content.as_deref().unwrap_or("").trim();
|
|
let clean_json = if reply_text.starts_with("```") {
|
|
let mut lines = reply_text.lines();
|
|
lines.next();
|
|
let mut content = lines.collect::<Vec<&str>>();
|
|
if content.last().is_some_and(|s| s.trim() == "```") {
|
|
content.pop();
|
|
}
|
|
content.join("\n")
|
|
} else {
|
|
reply_text.to_string()
|
|
};
|
|
|
|
match serde_json::from_str::<
|
|
crate::app::workflow::hive_mind::CognitiveCyclePlan,
|
|
>(&clean_json)
|
|
{
|
|
Ok(plan) => {
|
|
let cycle_desc = plan
|
|
.cycles
|
|
.iter()
|
|
.enumerate()
|
|
.map(|(i, nodes)| format!("cycle {i}: {} node(s)", nodes.len()))
|
|
.collect::<Vec<String>>()
|
|
.join(", ");
|
|
|
|
push_event(&events_q, TurnEvent::SystemNote {
|
|
kind: "pipeline".to_string(),
|
|
message: format!(
|
|
"The Hive compiled {} cycle(s) — {cycle_desc}. Deploying nodes...",
|
|
plan.cycles.len()
|
|
),
|
|
});
|
|
|
|
crate::app::workflow::hive_mind::run_hive_mind(
|
|
user_request,
|
|
&plan,
|
|
&tc.edit_log_session_dir,
|
|
&tc.workspace_roots,
|
|
Some(events_q),
|
|
pipeline_abort.as_ref(),
|
|
)
|
|
}
|
|
Err(e) => Err(anyhow::anyhow!(
|
|
"Failed to parse LLM planning JSON: {e}. Cleaned JSON was: {clean_json}"
|
|
)),
|
|
}
|
|
}
|
|
Err(e) => Err(anyhow::anyhow!(
|
|
"Failed to query LLM for planning workflow: {e}"
|
|
)),
|
|
};
|
|
|
|
match pipeline_result {
|
|
Ok((consensus, _reports)) => {
|
|
// run_hive_mind already wrote docs/runs/*.md internally
|
|
// (guaranteed, even on synthesis failure) — nothing to do
|
|
// here besides feeding the consensus back to the LLM.
|
|
tracing::info!(
|
|
"[hive-mind] convergence completed — the Hive has spoken"
|
|
);
|
|
|
|
let pipeline_msg = ChatMessage::system(format!(
|
|
"{}\n{consensus}",
|
|
crate::app::workflow::hive_mind::HIVE_MIND_CONSENSUS_TAG,
|
|
));
|
|
archive_message(tc.db.as_ref(), &tc.session_id, &pipeline_msg);
|
|
msgs.push(pipeline_msg);
|
|
|
|
push_event(&events_q, TurnEvent::SystemNote {
|
|
kind: "pipeline".to_string(),
|
|
message:
|
|
"The Hive's convergence is complete. Core Intelligence reviewing consensus for LO..."
|
|
.to_string(),
|
|
});
|
|
push_event(&events_q, TurnEvent::SystemNote {
|
|
kind: "hive_mind_converged".to_string(),
|
|
message: String::new(),
|
|
});
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!("[hive-mind] convergence fractured: {}", e);
|
|
let fail_msg = ChatMessage::system(format!(
|
|
"[Pipeline Note] The Hive encountered interference: {e}.\n\
|
|
Proceeding with direct execution as fallback.",
|
|
));
|
|
msgs.push(fail_msg);
|
|
}
|
|
}
|
|
} else {
|
|
tracing::debug!("[ceo] pipeline not triggered — handling directly");
|
|
}
|
|
|
|
// Check abort after pipeline completes, before entering main loop.
|
|
// This catches the case where the user pressed Esc during the pipeline
|
|
// phase, which previously ran unchecked for minutes at a time.
|
|
if crate::app::util::abort::is_aborted_direct(&tc.abort_flag)
|
|
{
|
|
push_event(&events_q, TurnEvent::Error("Generation aborted by user".to_string()));
|
|
return Ok(());
|
|
}
|
|
|
|
let mut todo_retry_count = 0usize;
|
|
|
|
loop {
|
|
let token_estimate: usize = msgs
|
|
.iter()
|
|
.filter_map(|m| m.content.as_deref())
|
|
.map(count_tokens)
|
|
.sum();
|
|
let max_wire_tokens = tc.context_window;
|
|
|
|
// Skip message compaction if abort was requested — the non-streaming
|
|
// LLM call for summarization would block without checking abort_flag.
|
|
let wire_msgs = if !crate::app::util::abort::is_aborted_direct(&tc.abort_flag)
|
|
&& crate::app::runtime::context::shaping::should_shape(
|
|
token_estimate,
|
|
max_wire_tokens,
|
|
prev_shaped,
|
|
) {
|
|
prev_shaped = true;
|
|
let compacted =
|
|
crate::app::runtime::context::shaping::shape_messages(
|
|
&msgs,
|
|
token_estimate,
|
|
max_wire_tokens,
|
|
false,
|
|
Some(&tc.client),
|
|
Some(&tc.abort_flag),
|
|
);
|
|
|
|
// Dispatch the compacted messages to the main thread so the local session history
|
|
// is permanently compacted and doesn't trigger shaping again immediately on next turn.
|
|
push_event(&events_q, TurnEvent::Compacted(compacted.clone()));
|
|
|
|
// Also update our local `msgs` variable so the rest of the loop operates on the compacted version
|
|
msgs.clone_from(&compacted);
|
|
compacted
|
|
} else {
|
|
prev_shaped = false;
|
|
msgs.clone()
|
|
};
|
|
|
|
let mut stream_started = false;
|
|
let mut reasoning_started = false;
|
|
let mut reasoning_ended = false;
|
|
let mut usage = None;
|
|
let result = tc.client.chat_with_tools_streaming(
|
|
&wire_msgs,
|
|
if tc.tdefs.is_empty() {
|
|
None
|
|
} else {
|
|
Some(tc.tdefs.clone())
|
|
},
|
|
Some(tc.temperature),
|
|
tc.max_tokens,
|
|
|event| -> bool {
|
|
if crate::app::util::abort::is_aborted_direct(&tc.abort_flag) {
|
|
return false;
|
|
}
|
|
if let Ok(mut q) = events_q.lock() {
|
|
match event {
|
|
crate::app::runtime::stream::StreamEvent::Token(tok) => {
|
|
if !stream_started {
|
|
q.push_back(TurnEvent::StreamStart);
|
|
stream_started = true;
|
|
}
|
|
if reasoning_started && !reasoning_ended {
|
|
reasoning_ended = true;
|
|
q.push_back(
|
|
TurnEvent::StreamToken("\n</think>\n\n".to_string()),
|
|
);
|
|
}
|
|
q.push_back(TurnEvent::StreamToken(tok.clone()));
|
|
}
|
|
crate::app::runtime::stream::StreamEvent::Reasoning(tok) => {
|
|
if !stream_started {
|
|
q.push_back(TurnEvent::StreamStart);
|
|
stream_started = true;
|
|
}
|
|
if !reasoning_started {
|
|
reasoning_started = true;
|
|
q.push_back(TurnEvent::StreamToken("<think>\n".to_string()));
|
|
}
|
|
q.push_back(TurnEvent::StreamToken(tok.clone()));
|
|
}
|
|
crate::app::runtime::stream::StreamEvent::Usage {
|
|
prompt_tokens,
|
|
completion_tokens,
|
|
..
|
|
} => {
|
|
usage = Some((*prompt_tokens, *completion_tokens));
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
true
|
|
},
|
|
Some(&tc.abort_flag),
|
|
);
|
|
|
|
if reasoning_started && !reasoning_ended {
|
|
push_event(&events_q, TurnEvent::StreamToken(
|
|
"\n</think>\n\n".to_string(),
|
|
));
|
|
}
|
|
|
|
let (response, final_usage) = match result {
|
|
Ok((msg, u)) => (msg, u.or(usage)),
|
|
Err(e) => {
|
|
// If abort was requested, return immediately.
|
|
if crate::app::util::abort::is_aborted_direct(&tc.abort_flag)
|
|
|| e.to_string().contains("aborted")
|
|
{
|
|
push_event(&events_q, TurnEvent::Error(
|
|
"Generation aborted by user".to_string(),
|
|
));
|
|
return Ok(());
|
|
}
|
|
// Streaming-only: no non-streaming fallback.
|
|
// Non-streaming blocks up to 1 minute without checking
|
|
// abort_flag, making cancellation unresponsive.
|
|
// If the API supports streaming (which it must), this
|
|
// path handles transient errors via the retry loop below.
|
|
let api_err = e;
|
|
let todo_path = tc.ctx.session_dir.join("todo.md");
|
|
let mut has_unfinished = false;
|
|
if let Ok(todo_text) = std::fs::read_to_string(&todo_path) {
|
|
if todo_text
|
|
.lines()
|
|
.any(|l| l.trim_start().starts_with("- [ ]"))
|
|
{
|
|
has_unfinished = true;
|
|
}
|
|
}
|
|
if has_unfinished {
|
|
todo_retry_count += 1;
|
|
if todo_retry_count > MAX_TODO_RETRIES {
|
|
anyhow::bail!(
|
|
"exhausted {MAX_TODO_RETRIES} todo-retries — giving up on unfinished tasks. \
|
|
Edit todo.md manually or ask me to focus on specific items.",
|
|
);
|
|
}
|
|
push_event(&events_q, TurnEvent::SystemNote {
|
|
kind: "task_retry".to_string(),
|
|
message: format!(
|
|
"Network/API error: {api_err}. Auto-retrying to finish tasks... (retry {todo_retry_count}/{MAX_TODO_RETRIES})"
|
|
),
|
|
});
|
|
std::thread::sleep(std::time::Duration::from_secs(5));
|
|
continue;
|
|
}
|
|
return Err(api_err);
|
|
}
|
|
};
|
|
|
|
let (mut tok_in, mut tok_out) = final_usage.unwrap_or((0, 0));
|
|
if tok_in == 0 {
|
|
let total_tokens: usize = wire_msgs
|
|
.iter()
|
|
.filter_map(|m| m.content.as_deref())
|
|
.map(count_tokens)
|
|
.sum();
|
|
tok_in = total_tokens.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;
|
|
}
|
|
push_event(&events_q, TurnEvent::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();
|
|
if has_tool_calls {
|
|
let tool_calls = response.tool_calls.clone().unwrap_or_default();
|
|
archive_message(tc.db.as_ref(), &tc.session_id, &response);
|
|
msgs.push(response);
|
|
let mut results_vec = Vec::new();
|
|
std::thread::scope(|s| {
|
|
let mut handles = Vec::new();
|
|
let tc_ref = tc;
|
|
for tool_call in &tool_calls {
|
|
let handle = s.spawn(move || {
|
|
let tool_name = tool_call.function.name.clone();
|
|
let args = crate::dto::chat::tool::sanitize_tool_arguments(
|
|
&tool_call.function.arguments,
|
|
);
|
|
|
|
let ws_roots: Vec<&std::path::Path> = tc_ref
|
|
.workspace_roots
|
|
.iter()
|
|
.map(std::path::PathBuf::as_path)
|
|
.collect();
|
|
let verdict = crate::app::guard::Guard::gate_tool_call(
|
|
&tool_name,
|
|
&args,
|
|
&ws_roots,
|
|
);
|
|
|
|
let is_edit_tool =
|
|
tool_name == "write" || tool_name == "edit";
|
|
let (output, is_error, is_edit) = match verdict {
|
|
Verdict::Allow => match execute_one_tool(
|
|
&tc_ref.tools,
|
|
&tc_ref.ctx,
|
|
&tool_name,
|
|
&tool_call.id,
|
|
&args,
|
|
&ToolExecSession {
|
|
dir: &tc_ref.edit_log_session_dir,
|
|
id: &tc_ref.session_id,
|
|
db: tc_ref.db.as_ref(),
|
|
},
|
|
) {
|
|
Ok(result) => (result, false, is_edit_tool),
|
|
Err(e) => (e.to_string(), true, false),
|
|
},
|
|
Verdict::Block(reason) => {
|
|
(format!("Blocked: {reason}"), true, false)
|
|
}
|
|
};
|
|
(tool_call, tool_name, args, output, is_error, is_edit)
|
|
});
|
|
handles.push(handle);
|
|
}
|
|
for h in handles {
|
|
if let Ok(res) = h.join() {
|
|
results_vec.push(res);
|
|
}
|
|
}
|
|
});
|
|
|
|
for (tool_call, tool_name, args, output, is_error, is_edit) in results_vec {
|
|
if crate::app::util::abort::is_aborted_direct(&tc.abort_flag)
|
|
{
|
|
push_event(&events_q, TurnEvent::Error(
|
|
"Turn aborted by user".to_string(),
|
|
));
|
|
return Ok(());
|
|
}
|
|
|
|
if is_edit {
|
|
// ── Auto-subagent orchestration ──
|
|
// Extract path from tool args for auto-review and
|
|
// background subagent tracking.
|
|
let edit_path = args
|
|
.get("path")
|
|
.and_then(|v| v.as_str())
|
|
.map(std::string::ToString::to_string);
|
|
if let Some(ref p) = edit_path {
|
|
edited_paths.push(p.clone());
|
|
|
|
// Inline quick-review: spawn a lightweight read-only
|
|
// subagent that reviews the written file and feeds
|
|
// its verdict back into the LLM conversation so the
|
|
// agent can fix issues immediately in the same turn.
|
|
if inline_reviews_count < MAX_AUTO_REVIEWS_PER_TURN
|
|
&& crate::app::subagent::auto::is_reviewable_path(p)
|
|
{
|
|
inline_reviews_count += 1;
|
|
let review_start = std::time::Instant::now();
|
|
match crate::app::subagent::auto::spawn_quick_review(
|
|
p,
|
|
&tc.edit_log_session_dir,
|
|
&tc.workspace_roots,
|
|
) {
|
|
Ok(verdict) => {
|
|
let elapsed =
|
|
review_start.elapsed().as_millis();
|
|
let review_msg = ChatMessage::tool_result(
|
|
format!("auto-review-{inline_reviews_count}"),
|
|
format!(
|
|
"[Auto inline review: {} ({}ms)]\n{}",
|
|
p, elapsed, verdict.trim(),
|
|
),
|
|
);
|
|
archive_message(
|
|
tc.db.as_ref(),
|
|
&tc.session_id,
|
|
&review_msg,
|
|
);
|
|
msgs.push(review_msg);
|
|
tracing::info!(
|
|
"[auto-review] inline review for '{}' completed in {}ms: {}",
|
|
p,
|
|
elapsed,
|
|
verdict.lines().next().unwrap_or(&verdict).trim(),
|
|
);
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
"[auto-review] inline review failed for '{}': {}",
|
|
p,
|
|
e,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
let tool_path = args
|
|
.get("path")
|
|
.and_then(|v| v.as_str())
|
|
.map(std::string::ToString::to_string);
|
|
|
|
push_event(&events_q, TurnEvent::ToolResult {
|
|
tool_call_id: tool_call.id.clone(),
|
|
tool_name: tool_name.clone(),
|
|
output: output.clone(),
|
|
is_error,
|
|
path: tool_path,
|
|
});
|
|
|
|
let tool_msg =
|
|
ChatMessage::tool_result(tool_call.id.clone(), output);
|
|
archive_message(tc.db.as_ref(), &tc.session_id, &tool_msg);
|
|
msgs.push(tool_msg);
|
|
}
|
|
} else {
|
|
if !content.is_empty() {
|
|
archive_message(tc.db.as_ref(), &tc.session_id, &response);
|
|
if stream_started {
|
|
push_event(&events_q, TurnEvent::StreamDone(response.clone()));
|
|
} else {
|
|
push_event(&events_q, TurnEvent::AssistantMessage(response.clone()));
|
|
}
|
|
}
|
|
|
|
let todo_path = tc.ctx.session_dir.join("todo.md");
|
|
let mut has_unfinished = false;
|
|
if let Ok(todo_text) = std::fs::read_to_string(&todo_path) {
|
|
if todo_text
|
|
.lines()
|
|
.any(|l| l.trim_start().starts_with("- [ ]"))
|
|
{
|
|
has_unfinished = true;
|
|
}
|
|
}
|
|
|
|
if has_unfinished {
|
|
todo_retry_count += 1;
|
|
if todo_retry_count > MAX_TODO_RETRIES {
|
|
push_event(&events_q, TurnEvent::SystemNote {
|
|
kind: "task_retry".to_string(),
|
|
message: format!("Giving up after {MAX_TODO_RETRIES} retries — some todo items remain unfinished. Edit todo.md manually or ask again."),
|
|
});
|
|
break;
|
|
}
|
|
let sys_text = format!("You stopped, but you still have unfinished tasks in todo.md (marked with '- [ ]'). You MUST continue working and use tools to finish them, or edit todo.md to mark them as done if they are finished. (Retry {todo_retry_count}/{MAX_TODO_RETRIES})");
|
|
let sys_text_clone = sys_text.clone();
|
|
let msg = ChatMessage::system(sys_text);
|
|
archive_message(tc.db.as_ref(), &tc.session_id, &msg);
|
|
msgs.push(msg);
|
|
push_event(&events_q, TurnEvent::SystemNote {
|
|
kind: "task_retry".to_string(),
|
|
message: sys_text_clone,
|
|
});
|
|
continue;
|
|
}
|
|
|
|
break;
|
|
}
|
|
}
|
|
|
|
let total_edits_this_turn = initial_edit_log.as_ref().and_then(|initial_el| {
|
|
let initial_count = initial_el.len();
|
|
zesdex_cms::infrastructure::persistence::edit_log_repo::JsonlEditLogRepository::new()
|
|
.open(&tc.edit_log_session_dir)
|
|
.ok()
|
|
.map(|final_el| {
|
|
let count = final_el.len().saturating_sub(initial_count);
|
|
(count, initial_count, final_el)
|
|
})
|
|
});
|
|
|
|
if let Some((total_edits_this_turn, prev_edits, el)) = &total_edits_this_turn {
|
|
if *total_edits_this_turn > 0 {
|
|
push_event(&events_q, TurnEvent::SystemNote {
|
|
kind: "edits".to_string(),
|
|
message: total_edits_this_turn.to_string(),
|
|
});
|
|
|
|
// Collect edited paths from the new edit log entries
|
|
let mut bg_paths = Vec::new();
|
|
for entry in el.entries.iter().skip(*prev_edits) {
|
|
bg_paths.push(entry.path.clone());
|
|
}
|
|
bg_paths.sort();
|
|
bg_paths.dedup();
|
|
|
|
// ── Background auto-subagents ──
|
|
if !bg_paths.is_empty() {
|
|
let bg_session_dir = tc.edit_log_session_dir.clone();
|
|
let bg_workspaces = tc.workspace_roots.clone();
|
|
let bg_events = events_q.clone();
|
|
let bg_abort = tc.abort_flag.clone();
|
|
std::thread::spawn(move || {
|
|
crate::app::subagent::auto::spawn_all_background(
|
|
&bg_paths,
|
|
&bg_session_dir,
|
|
&bg_workspaces,
|
|
&bg_events,
|
|
bg_abort,
|
|
);
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
push_event(&events_q, TurnEvent::Done);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Execute a single tool call: find the tool by name, snapshot the file
|
|
/// (if write/edit) for rewind, run the tool, log an `EditLogEntry` for
|
|
/// write/edit, and return the output.
|
|
///
|
|
/// Flow: iterate tools → match by name → for write/edit, snapshot the
|
|
/// pre-existing file content into the blob store → call `tool.run()` →
|
|
/// for write/edit, compute SHA-256 of the new content and append an
|
|
/// `EditLogEntry` → return the tool output string.
|
|
///
|
|
/// Why: snapshots enable the rewind feature to restore previous content
|
|
/// after a write/edit.
|
|
///
|
|
/// Return: the tool's stdout string, or an error if no matching tool was
|
|
/// found or the tool run itself failed.
|
|
struct ToolExecSession<'a> {
|
|
dir: &'a std::path::Path,
|
|
id: &'a str,
|
|
db: Option<&'a std::sync::Arc<std::sync::Mutex<rusqlite::Connection>>>,
|
|
}
|
|
|
|
fn execute_one_tool(
|
|
tools: &[Box<dyn crate::tool::Tool>],
|
|
ctx: &crate::tool::ToolCtx,
|
|
name: &str,
|
|
tool_call_id: &str,
|
|
args: &serde_json::Value,
|
|
sess: &ToolExecSession<'_>,
|
|
) -> anyhow::Result<String> {
|
|
for tool in tools {
|
|
if tool.name() == name {
|
|
// Snapshot current file content before write/edit for rewind
|
|
if (name == "write" || name == "edit") && !tool_call_id.is_empty() {
|
|
if let Some(arc) = sess.db {
|
|
if let Ok(conn) = arc.lock() {
|
|
let path = args
|
|
.get("path")
|
|
.and_then(|v| v.as_str())
|
|
.unwrap_or("");
|
|
if let Ok(abs_path) =
|
|
crate::tool::resolve_path(&ctx.workspaces, path)
|
|
{
|
|
if let Ok(bytes) = std::fs::read(&abs_path) {
|
|
let _ = crate::model::msglog::store_blob(
|
|
&conn,
|
|
sess.id,
|
|
tool_call_id,
|
|
&bytes,
|
|
None,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
let result = tool.run(ctx, args)?;
|
|
if name == "write" || name == "edit" {
|
|
crate::tool::log_write_edit_tool(
|
|
args, name, &ctx.origin.tag(), sess.dir, sess.id,
|
|
);
|
|
}
|
|
return Ok(result);
|
|
}
|
|
}
|
|
anyhow::bail!("tool not found: {name}")
|
|
}
|
|
|
|
/// Load all memory entries from `memory_dir` and format them as a compact
|
|
/// section appended to the system prompt, so the AI is always aware of
|
|
/// stored lessons and project knowledge.
|
|
///
|
|
/// Flow: list memory slugs → for each, read + parse the file → collect
|
|
/// entries whose lifecycle is not "stale" → cap total output at 3000 chars
|
|
/// to avoid dominating the prompt budget.
|
|
///
|
|
/// Why: previously, lessons existed on disk but the AI never saw them
|
|
/// unless it explicitly called `recall()`. This makes the memory system
|
|
/// actually useful by surfacing relevant knowledge automatically.
|
|
///
|
|
/// Return: a formatted string (may be empty if no memory entries exist).
|
|
fn build_memory_section(memory_dir: &std::path::Path) -> String {
|
|
let names =
|
|
zesdex_cms::infrastructure::persistence::memory_repo::MarkdownMemoryRepository::new()
|
|
.list(memory_dir)
|
|
.unwrap_or_default();
|
|
if names.is_empty() {
|
|
return String::new();
|
|
}
|
|
|
|
let mut section = String::from("\n\n--- Persistent Memory ---\n");
|
|
write!(section, "Total entries: {}\n\n", names.len()).unwrap();
|
|
|
|
for name in &names {
|
|
if section.len() > 3000 {
|
|
section
|
|
.push_str("... (more entries omitted, use recall() to see all)\n");
|
|
break;
|
|
}
|
|
if let Ok(mem) =
|
|
zesdex_cms::infrastructure::persistence::memory_repo::MarkdownMemoryRepository::new()
|
|
.load(memory_dir, name)
|
|
{
|
|
if mem.lifecycle == "stale" {
|
|
continue;
|
|
}
|
|
write!(
|
|
section,
|
|
"## [{}] {}\n{}\n\n",
|
|
mem.kind, mem.name, mem.content
|
|
)
|
|
.unwrap();
|
|
}
|
|
}
|
|
section.push_str("---");
|
|
section
|
|
}
|
|
|
|
/// Persist a `ChatMessage` to the `SQLite` message log, if a database
|
|
/// connection is available.
|
|
///
|
|
/// Flow: if `db` is `Some`, lock the mutex and call `insert_message`.
|
|
/// Errors are silently ignored.
|
|
fn archive_message(
|
|
db: Option<&std::sync::Arc<std::sync::Mutex<rusqlite::Connection>>>,
|
|
session_id: &str,
|
|
msg: &ChatMessage,
|
|
) {
|
|
if let Some(arc) = db {
|
|
if let Ok(conn) = arc.lock() {
|
|
let _ = crate::model::msglog::insert_message(&conn, session_id, msg);
|
|
}
|
|
}
|
|
}
|