1320 lines
60 KiB
Rust
1320 lines
60 KiB
Rust
//! The agent loop: stream a model response, run tool calls (with permission
|
||
//! checks), feed results back, repeat until the model stops.
|
||
|
||
use crate::app::{App, PermissionReply};
|
||
use crate::permission::{Decision, Evaluator, bash_prefix, split_command};
|
||
use crate::prompts::{self, Agent};
|
||
use crate::provider::{ChatRequest, Provider, StreamEvent};
|
||
use crate::store::UsageRecord;
|
||
use crate::tools::{self, Tool, ToolCtx};
|
||
use crate::types::{Event, Message, Part, Role, Session, Usage, new_id, now_ms};
|
||
use anyhow::{Result, anyhow, bail};
|
||
use futures::future::join_all;
|
||
use serde_json::Value;
|
||
use std::path::{Path, PathBuf};
|
||
use std::sync::Arc;
|
||
use std::time::Instant;
|
||
use tokio::sync::mpsc;
|
||
use tokio_util::sync::CancellationToken;
|
||
|
||
const MAX_STEPS: usize = 250;
|
||
const MAX_SUBAGENT_DEPTH: usize = 3;
|
||
|
||
/// What happened to a prompt.
|
||
pub enum Submitted {
|
||
Started,
|
||
Queued(crate::store::QueueItem),
|
||
}
|
||
|
||
fn emit_queue(app: &App, session_id: &str) {
|
||
let items = app.store.queue_list(session_id).unwrap_or_default();
|
||
app.emit(Event::QueueUpdated { session_id: session_id.into(), items: serde_json::json!(items) });
|
||
}
|
||
|
||
/// Take all queued messages (one user message's worth of parts).
|
||
fn take_queue(app: &App, session_id: &str) -> Vec<Part> {
|
||
let items = app.store.queue_take(session_id).unwrap_or_default();
|
||
if !items.is_empty() {
|
||
emit_queue(app, session_id);
|
||
}
|
||
items.into_iter().flat_map(|i| i.parts).collect()
|
||
}
|
||
|
||
/// Start processing a user prompt. If the session is busy, the prompt (with
|
||
/// attachments) is queued and delivered at the agent's next step.
|
||
pub fn start_prompt(app: Arc<App>, session_id: String, parts: Vec<Part>) -> Result<()> {
|
||
submit(app, session_id, parts, "user").map(|_| ())
|
||
}
|
||
|
||
pub fn submit(app: Arc<App>, session_id: String, parts: Vec<Part>, source: &str) -> Result<Submitted> {
|
||
// Queueing and the end-of-run check run under the same runtime lock, so a
|
||
// message is never lost or delivered twice.
|
||
let cancel = CancellationToken::new();
|
||
let queued = app.with_runtime(&session_id, |r| -> Result<Option<crate::store::QueueItem>> {
|
||
if r.cancel.is_some() {
|
||
let item = crate::store::QueueItem { id: new_id(), parts: parts.clone(), source: source.into(), created_at: now_ms() };
|
||
app.store.queue_add(&session_id, &item)?;
|
||
Ok(Some(item))
|
||
} else {
|
||
r.cancel = Some(cancel.clone());
|
||
Ok(None)
|
||
}
|
||
})?;
|
||
if let Some(item) = queued {
|
||
emit_queue(&app, &session_id);
|
||
return Ok(Submitted::Queued(item));
|
||
}
|
||
let text_only: String = parts
|
||
.iter()
|
||
.filter_map(|p| if let Part::Text { text } = p { Some(text.as_str()) } else { None })
|
||
.collect::<Vec<_>>()
|
||
.join("\n");
|
||
app.emit(Event::SessionStatus { session_id: session_id.clone(), busy: true });
|
||
let source = source.to_string();
|
||
|
||
tokio::spawn(async move {
|
||
let res = async {
|
||
let mut parts = parts;
|
||
if source == "user" {
|
||
let session = app.session(&session_id)?;
|
||
let sandbox = crate::sandbox::spec_for(&app, session.owner_id.as_deref());
|
||
let payload = serde_json::json!({"session_id": session_id, "cwd": session.cwd, "prompt": text_only});
|
||
let o = crate::hooks::run(&app, "user_prompt", None, payload, Path::new(&session.cwd), sandbox.as_ref()).await;
|
||
if let Some(reason) = o.blocked {
|
||
anyhow::bail!("Prompt blocked by a hook: {}", reason.trim());
|
||
}
|
||
if !o.context.trim().is_empty() {
|
||
parts.push(Part::Text { text: format!("<hook-context event=\"user_prompt\">\n{}\n</hook-context>", o.context.trim()) });
|
||
}
|
||
}
|
||
if source == "user" {
|
||
parts.extend(crate::mentions::expand(&app, &session_id, &text_only).await);
|
||
}
|
||
// Empty parts: continue from the existing history (regenerate).
|
||
if !parts.is_empty() {
|
||
let msg = Message::new(Role::User, parts);
|
||
app.store.add_message(&session_id, &msg)?;
|
||
app.emit(Event::MessageAdded { session_id: session_id.clone(), message: msg });
|
||
maybe_generate_title(app.clone(), &session_id, &text_only);
|
||
}
|
||
run_session(app.clone(), &session_id, cancel.clone(), 0).await
|
||
}
|
||
.await;
|
||
if let Err(e) = res {
|
||
tracing::warn!("session {session_id}: {e:#}");
|
||
app.emit(Event::Error { session_id: session_id.clone(), message: format!("{e:#}") });
|
||
}
|
||
// Messages queued after the last step start a new turn – unless the user
|
||
// stopped the agent: then they stay queued until sent explicitly.
|
||
let leftover = app.with_runtime(&session_id, |r| {
|
||
r.cancel = None;
|
||
r.effort = None;
|
||
if cancel.is_cancelled() { Vec::new() } else { take_queue(&app, &session_id) }
|
||
});
|
||
app.emit(Event::SessionStatus { session_id: session_id.clone(), busy: false });
|
||
if !leftover.is_empty() {
|
||
let _ = start_prompt(app, session_id, leftover);
|
||
}
|
||
});
|
||
Ok(Submitted::Started)
|
||
}
|
||
|
||
/// Send queued messages now (after a stop or a server restart).
|
||
pub fn flush_queue(app: Arc<App>, session_id: &str) -> Result<bool> {
|
||
if app.is_busy(session_id) {
|
||
return Ok(false);
|
||
}
|
||
let parts = take_queue(&app, session_id);
|
||
if parts.is_empty() {
|
||
return Ok(false);
|
||
}
|
||
start_prompt(app, session_id.to_string(), parts)?;
|
||
Ok(true)
|
||
}
|
||
|
||
pub fn abort(app: &App, session_id: &str) {
|
||
app.with_runtime(session_id, |r| {
|
||
if let Some(c) = &r.cancel {
|
||
c.cancel();
|
||
}
|
||
});
|
||
}
|
||
|
||
/// Messages since the last compaction summary. Tool calls left without a
|
||
/// result (run interrupted, e.g. by a server restart) get a synthetic result:
|
||
/// providers reject histories with unanswered tool calls.
|
||
fn active_history(all: Vec<Message>) -> Vec<Message> {
|
||
let start = all
|
||
.iter()
|
||
.rposition(|m| m.parts.iter().any(|p| matches!(p, Part::Summary { .. })))
|
||
.unwrap_or(0);
|
||
let msgs: Vec<Message> = all.into_iter().skip(start).collect();
|
||
let mut out: Vec<Message> = Vec::with_capacity(msgs.len());
|
||
for (i, m) in msgs.iter().enumerate() {
|
||
out.push(m.clone());
|
||
if m.role != Role::Assistant {
|
||
continue;
|
||
}
|
||
let answered: std::collections::HashSet<&str> = msgs
|
||
.get(i + 1)
|
||
.filter(|n| n.role == Role::User)
|
||
.map(|n| {
|
||
n.parts
|
||
.iter()
|
||
.filter_map(|p| if let Part::ToolResult { call_id, .. } = p { Some(call_id.as_str()) } else { None })
|
||
.collect()
|
||
})
|
||
.unwrap_or_default();
|
||
let missing: Vec<Part> = m
|
||
.parts
|
||
.iter()
|
||
.filter_map(|p| match p {
|
||
Part::ToolCall { id, name, .. } if !answered.contains(id.as_str()) => {
|
||
Some(tool_result(id, name, "Not executed: the run was interrupted (e.g. server restart).".into(), true, None))
|
||
}
|
||
_ => None,
|
||
})
|
||
.collect();
|
||
if missing.is_empty() {
|
||
continue;
|
||
}
|
||
// Results must directly follow the call: merge into the next user message or insert one.
|
||
match msgs.get(i + 1) {
|
||
Some(n) if n.role == Role::User => {
|
||
// The next message is pushed in the next iteration; prepend there.
|
||
let mut next = n.clone();
|
||
let mut parts = missing;
|
||
parts.extend(next.parts);
|
||
next.parts = parts;
|
||
// Replace by pushing now and skipping the original.
|
||
out.push(next);
|
||
}
|
||
_ => out.push(Message::new(Role::User, missing)),
|
||
}
|
||
}
|
||
// Remove duplicates created by the merge above (same id pushed twice: keep the merged one).
|
||
let mut dedup: Vec<Message> = Vec::with_capacity(out.len());
|
||
for m in out {
|
||
if let Some(last) = dedup.last() {
|
||
if last.id == m.id {
|
||
continue;
|
||
}
|
||
}
|
||
dedup.push(m);
|
||
}
|
||
dedup
|
||
}
|
||
|
||
fn normalize_tool_name(s: &str) -> String {
|
||
s.chars().filter(|c| c.is_ascii_alphanumeric()).collect::<String>().to_lowercase()
|
||
}
|
||
|
||
pub fn available_tools(app: &App, agent: &Agent, owner: Option<&crate::store::User>) -> Vec<Arc<dyn Tool>> {
|
||
let cfg = app.config();
|
||
let mut all = tools::builtin_tools();
|
||
if !cfg.images.enabled {
|
||
all.retain(|t| t.name() != "generate_image");
|
||
}
|
||
let (direct, deferred) = app.mcp.tools(&cfg, owner);
|
||
all.extend(direct);
|
||
if !deferred.is_empty() {
|
||
all.push(Arc::new(crate::mcp::McpSearch));
|
||
all.push(Arc::new(crate::mcp::McpCall));
|
||
}
|
||
all.into_iter()
|
||
.filter(|t| !agent.read_only || t.read_only())
|
||
// Craft tools only where they are useful.
|
||
.filter(|t| agent.name == "craft" || !matches!(t.name(), "browser_snapshot" | "image_compare" | "binary_info" | "decompile" | "craft_phase") || agent.tools.is_some())
|
||
.filter(|t| match &agent.tools {
|
||
None => true,
|
||
Some(list) => {
|
||
let n = normalize_tool_name(t.name());
|
||
list.iter().any(|a| {
|
||
let a = normalize_tool_name(a);
|
||
a == n || (t.name().starts_with("mcp") && a == "mcp")
|
||
})
|
||
}
|
||
})
|
||
.collect()
|
||
}
|
||
|
||
struct Step {
|
||
message: Message,
|
||
stop_reason: String,
|
||
}
|
||
|
||
/// Boxed with an explicit `Send` bound: subagents recurse into the agent loop,
|
||
/// and the declared type breaks the cycle for the compiler.
|
||
fn run_session_boxed(
|
||
app: Arc<App>,
|
||
session_id: String,
|
||
cancel: CancellationToken,
|
||
depth: usize,
|
||
) -> futures::future::BoxFuture<'static, Result<String>> {
|
||
Box::pin(async move { run_session(app, &session_id, cancel, depth).await })
|
||
}
|
||
|
||
/// Run the loop for a session until the model stops. Returns the final text.
|
||
pub async fn run_session(app: Arc<App>, session_id: &str, cancel: CancellationToken, depth: usize) -> Result<String> {
|
||
let mut final_text = String::new();
|
||
let mut stop_hooks = 0;
|
||
let mut refine_done = 0u32;
|
||
let turn_start = now_ms();
|
||
for _ in 0..MAX_STEPS {
|
||
if cancel.is_cancelled() {
|
||
break;
|
||
}
|
||
let session = app.session(session_id)?;
|
||
let cwd = PathBuf::from(&session.cwd);
|
||
let roots = app.user_roots(session.owner_id.as_deref());
|
||
let root = app.root_for(session.owner_id.as_deref(), &cwd);
|
||
let mut agent = prompts::find_agent_for(&app, &root, &cwd, &session.agent, session.owner_id.as_deref());
|
||
let clean_room = session.clean_room(app.config().craft.clean_room);
|
||
// The craft prompt follows the session's clean room / deep switch.
|
||
if agent.name == "craft" && agent.source == "builtin" {
|
||
agent.prompt = prompts::craft_prompt(clean_room);
|
||
}
|
||
let model_spec = agent.model.clone().unwrap_or_else(|| session.model.clone());
|
||
let (provider, mut model) = app.registry().resolve(&model_spec)?;
|
||
if let Some(effort) = app.with_runtime(session_id, |r| r.effort.clone()) {
|
||
model.effort = Some(effort);
|
||
}
|
||
|
||
// Auto-compaction when the last request used most of the context window.
|
||
if let (Some(u), Some(ctx_len)) = (&session.last_usage, model.context) {
|
||
let threshold = app.config().server.auto_compact_threshold;
|
||
if threshold > 0.0 && u.context_tokens() as f64 > ctx_len as f64 * threshold {
|
||
compact(app.clone(), session_id, cancel.clone(), Some((u.context_tokens(), ctx_len))).await?;
|
||
continue;
|
||
}
|
||
}
|
||
|
||
let owner = session.owner_id.as_deref().and_then(|id| app.store.user(id).ok().flatten());
|
||
let tools = available_tools(&app, &agent, owner.as_ref());
|
||
let history = active_history(app.store.messages(session_id)?);
|
||
let req = ChatRequest {
|
||
system: prompts::with_chat_instructions(prompts::system_prompt(&app, &agent, &root, &cwd, &model_spec, &session.permission_mode, owner.as_ref()), &session.instructions),
|
||
messages: history,
|
||
tools: tools.iter().map(|t| t.def()).collect(),
|
||
max_tokens: model.max_output.unwrap_or(16_000),
|
||
reasoning: true,
|
||
session_id: Some(session_id.to_string()),
|
||
model: model.clone(),
|
||
};
|
||
let step = match stream_step(&app, &session, &agent.name, &model_spec, provider, req.clone(), &cancel).await {
|
||
Ok(step) => step,
|
||
// Host down or budget used up: switch to the model's fallback for this step.
|
||
Err(e) if model.fallback.is_some() && !cancel.is_cancelled() && needs_fallback(&e) => {
|
||
let fb = model.fallback.clone().unwrap_or_default();
|
||
let (fb_provider, fb_model) = app.registry().resolve(&fb)?;
|
||
app.emit(Event::Error { session_id: session_id.into(), message: format!("{model_spec} is not available ({}) – using the fallback model {fb}", short_error(&e)) });
|
||
let req = ChatRequest { model: fb_model.clone(), max_tokens: fb_model.max_output.unwrap_or(16_000), ..req };
|
||
stream_step(&app, &session, &agent.name, &fb, fb_provider, req, &cancel).await?
|
||
}
|
||
Err(e) => return Err(e),
|
||
};
|
||
final_text = step
|
||
.message
|
||
.parts
|
||
.iter()
|
||
.filter_map(|p| if let Part::Text { text } = p { Some(text.as_str()) } else { None })
|
||
.collect::<Vec<_>>()
|
||
.join("");
|
||
|
||
let calls: Vec<(String, String, Value)> = step
|
||
.message
|
||
.parts
|
||
.iter()
|
||
.filter_map(|p| match p {
|
||
Part::ToolCall { id, name, input, .. } => Some((id.clone(), name.clone(), input.clone())),
|
||
_ => None,
|
||
})
|
||
.collect();
|
||
|
||
if cancel.is_cancelled() {
|
||
// Every tool call needs a result, otherwise the history is invalid.
|
||
if !calls.is_empty() {
|
||
let results = calls
|
||
.iter()
|
||
.map(|(id, name, _)| tool_result(id, name, "Aborted by user".into(), true, None))
|
||
.collect();
|
||
save_user_parts(&app, session_id, results)?;
|
||
}
|
||
break;
|
||
}
|
||
if calls.is_empty() && depth == 0 && stop_hooks < 3 && matches!(step.stop_reason.as_str(), "stop" | "end_turn" | "") {
|
||
let payload = serde_json::json!({"session_id": session_id, "cwd": session.cwd, "last_message": final_text});
|
||
let sandbox = crate::sandbox::spec_for(&app, session.owner_id.as_deref());
|
||
let o = crate::hooks::run(&app, "stop", None, payload, &cwd, sandbox.as_ref()).await;
|
||
if let Some(reason) = o.blocked {
|
||
stop_hooks += 1;
|
||
save_user_parts(&app, session_id, vec![Part::Text { text: format!("<hook-feedback event=\"stop\">\n{}\n</hook-feedback>", reason.trim()) }])?;
|
||
continue;
|
||
}
|
||
}
|
||
let normal_end = matches!(step.stop_reason.as_str(), "stop" | "end_turn" | "");
|
||
if calls.is_empty() && depth == 0 && normal_end && !cancel.is_cancelled() {
|
||
// Refinement loop: a reviewer checks the turn; findings send the agent back to work.
|
||
let rounds = agent.refine.as_ref().map(|r| r.rounds).unwrap_or(0);
|
||
if refine_done < rounds {
|
||
refine_done += 1;
|
||
match review_turn(&app, &session, &agent, &model_spec, turn_start, refine_done, rounds, &cancel).await {
|
||
Ok(Some(feedback)) => {
|
||
save_user_parts(&app, session_id, vec![Part::Text { text: feedback }])?;
|
||
continue;
|
||
}
|
||
Ok(None) => {}
|
||
Err(e) => tracing::warn!("review of {session_id} failed: {e:#}"),
|
||
}
|
||
}
|
||
// Craft: a finished check phase means the result is ready.
|
||
if agent.name == "craft" {
|
||
advance_phase(&app, session_id, "ready", Some("check"));
|
||
}
|
||
}
|
||
if calls.is_empty() {
|
||
match step.stop_reason.as_str() {
|
||
"max_tokens" | "length" => app.emit(Event::Error {
|
||
session_id: session_id.into(),
|
||
message: "The response hit the output token limit (max_output).".into(),
|
||
}),
|
||
"refusal" | "content_filter" => app.emit(Event::Error {
|
||
session_id: session_id.into(),
|
||
message: "The model declined to respond (refusal).".into(),
|
||
}),
|
||
_ => {}
|
||
}
|
||
break;
|
||
}
|
||
|
||
let base = ToolCtx {
|
||
app: app.clone(),
|
||
session_id: session_id.to_string(),
|
||
call_id: String::new(),
|
||
cwd: cwd.clone(),
|
||
cancel: cancel.clone(),
|
||
agent: agent.name.clone(),
|
||
roots: roots.clone(),
|
||
sandbox: crate::sandbox::spec_for(&app, session.owner_id.as_deref()),
|
||
clean_room,
|
||
owner_id: None,
|
||
};
|
||
let by_name = |n: &str| tools.iter().find(|t| t.name() == n).cloned();
|
||
let all_read_only = calls.iter().all(|(_, n, _)| by_name(n).is_some_and(|t| t.read_only()));
|
||
let mut results = Vec::new();
|
||
if all_read_only {
|
||
let futs = calls.iter().map(|(id, name, input)| {
|
||
execute_call(&app, &session, &base, by_name(name), id, name, input.clone(), depth)
|
||
});
|
||
results = join_all(futs).await;
|
||
} else {
|
||
for (id, name, input) in &calls {
|
||
results.push(execute_call(&app, &session, &base, by_name(name), id, name, input.clone(), depth).await);
|
||
}
|
||
}
|
||
// Deliver messages the user queued while we were working (with attachments).
|
||
results.extend(take_queue(&app, session_id));
|
||
save_user_parts(&app, session_id, results)?;
|
||
}
|
||
Ok(final_text)
|
||
}
|
||
|
||
fn tool_result(call_id: &str, name: &str, content: String, is_error: bool, meta: Option<Value>) -> Part {
|
||
Part::ToolResult { call_id: call_id.into(), name: name.into(), content, images: vec![], is_error, meta }
|
||
}
|
||
|
||
fn save_user_parts(app: &App, session_id: &str, parts: Vec<Part>) -> Result<()> {
|
||
let msg = Message::new(Role::User, parts);
|
||
app.store.add_message(session_id, &msg)?;
|
||
app.emit(Event::MessageAdded { session_id: session_id.into(), message: msg });
|
||
Ok(())
|
||
}
|
||
|
||
#[allow(clippy::too_many_arguments)]
|
||
async fn execute_call(
|
||
app: &Arc<App>,
|
||
session: &Session,
|
||
base: &ToolCtx,
|
||
tool: Option<Arc<dyn Tool>>,
|
||
call_id: &str,
|
||
name: &str,
|
||
input: Value,
|
||
depth: usize,
|
||
) -> Part {
|
||
let Some(tool) = tool else {
|
||
return tool_result(call_id, name, format!("Unknown or unavailable tool '{name}' in this mode."), true, None);
|
||
};
|
||
if let Some(raw) = input.get("__invalid_json") {
|
||
return tool_result(
|
||
call_id,
|
||
name,
|
||
format!("Tool input was not valid JSON (possibly truncated). Retry with a smaller input.\n{}", raw.as_str().unwrap_or("").chars().take(300).collect::<String>()),
|
||
true,
|
||
None,
|
||
);
|
||
}
|
||
if name == "task" && depth >= MAX_SUBAGENT_DEPTH {
|
||
return tool_result(call_id, name, "Maximum subagent depth reached.".into(), true, None);
|
||
}
|
||
|
||
let ctx = ToolCtx {
|
||
app: app.clone(),
|
||
session_id: base.session_id.clone(),
|
||
call_id: call_id.into(),
|
||
cwd: base.cwd.clone(),
|
||
cancel: base.cancel.clone(),
|
||
agent: base.agent.clone(),
|
||
roots: base.roots.clone(),
|
||
sandbox: base.sandbox.clone(),
|
||
clean_room: base.clean_room,
|
||
owner_id: None,
|
||
};
|
||
// mcp_call is judged (and approved) as the MCP tool it calls.
|
||
let call_name = name;
|
||
let (tool, input, name): (Arc<dyn Tool>, Value, String) = if name == "mcp_call" {
|
||
match crate::mcp::resolve_call_target(&ctx, &input) {
|
||
Ok(target) => {
|
||
let args = input.get("arguments").cloned().unwrap_or(serde_json::json!({}));
|
||
let n = target.full_name.clone();
|
||
(target, args, n)
|
||
}
|
||
Err(e) => return tool_result(call_id, call_name, format!("Error: {e:#}"), true, None),
|
||
}
|
||
} else {
|
||
(tool, input, name.to_string())
|
||
};
|
||
let name = name.as_str();
|
||
|
||
// Hunks of a file edit the user accepted in the review (None = whole change).
|
||
let mut partial: Option<Vec<usize>> = None;
|
||
// Permission check. Approvals are tracked on the top-level session.
|
||
let root = app.root_session_id(&session.id);
|
||
let cfg = app.config();
|
||
let decision = app.with_runtime(&root, |r| {
|
||
Evaluator { cfg: &cfg.permissions, mode: &session.permission_mode, approved: &r.approved }.evaluate(tool.as_ref(), &input)
|
||
});
|
||
match decision {
|
||
Decision::Deny => {
|
||
return tool_result(call_id, call_name, "Denied by the permission rules in the configuration.".into(), true, None);
|
||
}
|
||
Decision::Ask => {
|
||
let diff = tool.preview(&input, &ctx).await;
|
||
let (reply, message, hunks) = app
|
||
.request_permission(&session.id, name, tool.title(&input), input.clone(), diff, base.cancel.clone())
|
||
.await;
|
||
match reply {
|
||
PermissionReply::Reject => {
|
||
let mut text = "The user rejected this tool call.".to_string();
|
||
if let Some(m) = message.filter(|m| !m.trim().is_empty()) {
|
||
text.push_str(&format!(" User feedback: {m}"));
|
||
}
|
||
return tool_result(call_id, call_name, text, true, None);
|
||
}
|
||
PermissionReply::Always => app.with_runtime(&root, |r| {
|
||
if name == "bash" {
|
||
let cmd = input["command"].as_str().unwrap_or("");
|
||
for part in split_command(cmd).unwrap_or_default() {
|
||
r.approved.insert(format!("bash:{}", bash_prefix(&part)));
|
||
}
|
||
} else {
|
||
r.approved.insert(name.to_string());
|
||
}
|
||
}),
|
||
PermissionReply::Once => {}
|
||
}
|
||
if let Some(sel) = hunks {
|
||
if sel.is_empty() {
|
||
return tool_result(call_id, call_name, "The user rejected all parts of this change.".into(), true, None);
|
||
}
|
||
partial = Some(sel);
|
||
}
|
||
}
|
||
Decision::Allow => {}
|
||
}
|
||
|
||
// Hooks see the tool input (cloned only when hooks are configured).
|
||
let hook_input = if app.config().hooks.is_empty() { Value::Null } else { input.clone() };
|
||
let hook_payload = |extra: Option<(&str, Value)>| {
|
||
let mut v = serde_json::json!({"session_id": session.id, "cwd": session.cwd, "tool": name, "input": hook_input});
|
||
if let Some((k, x)) = extra {
|
||
v[k] = x;
|
||
}
|
||
v
|
||
};
|
||
let pre = crate::hooks::run(app, "pre_tool", Some(name), hook_payload(None), &ctx.cwd, ctx.sandbox.as_ref()).await;
|
||
if let Some(reason) = pre.blocked {
|
||
return tool_result(call_id, call_name, format!("Blocked by a hook: {}", reason.trim()), true, None);
|
||
}
|
||
|
||
// Checkpoint: remember files before the agent changes them (rewind can restore them).
|
||
if tool.is_edit() {
|
||
if let Some(path) = input["path"].as_str().and_then(|p| ctx.resolve(p).ok()) {
|
||
let content = std::fs::read(&path).ok().filter(|c| c.len() <= 5 * 1024 * 1024);
|
||
if path.exists() == content.is_some() {
|
||
let _ = app.store.add_checkpoint(&session.id, &path.display().to_string(), content.as_deref());
|
||
}
|
||
}
|
||
}
|
||
let input_path = input["path"].as_str().map(String::from);
|
||
let input_for_phase = if base.agent == "craft" { input.clone() } else { Value::Null };
|
||
let res = if let (Some(sel), true) = (&partial, matches!(name, "write" | "edit")) {
|
||
apply_partial(&ctx, name, &input, sel).await
|
||
} else if name == "task" {
|
||
// Pass depth through the agent name channel: subagents run with depth+1.
|
||
run_task(app, &ctx, input, depth).await
|
||
} else {
|
||
tool.run(input, &ctx).await
|
||
};
|
||
// Diagnostics: report new errors of the changed file right away.
|
||
let res = match res {
|
||
Ok(mut out) if tool.is_edit() && !out.is_error => {
|
||
if let Some(path) = input_path.as_deref().and_then(|p| ctx.resolve(p).ok()) {
|
||
let wait = std::time::Duration::from_secs(4);
|
||
if let Some(Ok(d)) = crate::lsp::diagnostics(app, &path, &ctx.cwd, ctx.sandbox.as_ref(), wait).await {
|
||
out.content.push_str(&crate::lsp::format(input_path.as_deref().unwrap_or(""), &d, false));
|
||
}
|
||
}
|
||
Ok(out)
|
||
}
|
||
other => other,
|
||
};
|
||
let res = match res {
|
||
Ok(mut out) => {
|
||
let result = serde_json::json!({"content": out.content.chars().take(20_000).collect::<String>(), "is_error": out.is_error});
|
||
let post = crate::hooks::run(app, "post_tool", Some(name), hook_payload(Some(("result", result))), &ctx.cwd, ctx.sandbox.as_ref()).await;
|
||
if let Some(reason) = post.blocked {
|
||
out.content.push_str(&format!("\n\n<hook-feedback event=\"post_tool\">\n{}\n</hook-feedback>", reason.trim()));
|
||
} else if !post.context.trim().is_empty() {
|
||
out.content.push_str(&format!("\n\n<hook-context event=\"post_tool\">\n{}\n</hook-context>", post.context.trim()));
|
||
}
|
||
Ok(out)
|
||
}
|
||
e => e,
|
||
};
|
||
if matches!(&res, Ok(o) if !o.is_error) && base.agent == "craft" && name != "craft_phase" {
|
||
if let Some(phase) = infer_phase(name, &input_for_phase, session.clean_room == Some(false)) {
|
||
advance_phase(app, &session.id, phase, None);
|
||
}
|
||
}
|
||
match res {
|
||
Ok(out) => Part::ToolResult {
|
||
call_id: call_id.into(),
|
||
name: call_name.into(),
|
||
content: out.content,
|
||
images: out.images,
|
||
is_error: out.is_error,
|
||
meta: out.meta,
|
||
},
|
||
Err(e) => tool_result(call_id, call_name, format!("Error: {e:#}"), true, None),
|
||
}
|
||
}
|
||
|
||
async fn run_task(app: &Arc<App>, ctx: &ToolCtx, input: Value, depth: usize) -> Result<tools::ToolOutput> {
|
||
let prompt = tools::arg_str(&input, "prompt")?.to_string();
|
||
let description = input["description"].as_str().unwrap_or("subagent").to_string();
|
||
let agent_name = input["subagent_type"].as_str().unwrap_or("general").to_string();
|
||
let background = input["background"].as_bool().unwrap_or(false);
|
||
let (child, agent) = create_child(app, ctx, &agent_name, &description, prompt)?;
|
||
let meta = serde_json::json!({
|
||
"session_id": child.id, "agent": agent.name, "model": child.model, "background": background
|
||
});
|
||
|
||
if background {
|
||
let app = app.clone();
|
||
let parent_id = ctx.session_id.clone();
|
||
let child_id = child.id.clone();
|
||
let agent_name = agent.name.clone();
|
||
let desc = description.clone();
|
||
// Background tasks outlive the current turn; they are stopped via their own session.
|
||
let cancel = CancellationToken::new();
|
||
tokio::spawn(async move {
|
||
let res = run_child(app.clone(), &child, &agent, cancel, depth + 1).await;
|
||
let body = match res {
|
||
Ok(text) => text,
|
||
Err(e) => format!("The background task failed: {e:#}"),
|
||
};
|
||
let text = format!(
|
||
"<background-task-result agent=\"{agent_name}\" session=\"{child_id}\" description=\"{desc}\">\n{body}\n</background-task-result>"
|
||
);
|
||
// Injected into the running turn, or starts a new turn when the parent is idle.
|
||
if app.store.get_session(&parent_id).ok().flatten().is_some() {
|
||
if let Err(e) = submit(app.clone(), parent_id.clone(), vec![Part::Text { text }], "background") {
|
||
tracing::warn!("could not deliver background result: {e:#}");
|
||
}
|
||
}
|
||
});
|
||
return Ok(tools::ToolOutput::text(format!(
|
||
"Started background task '{description}' with subagent '{}' (model {}). Its result will arrive \
|
||
automatically as a message when it finishes – continue with other work and do not poll.",
|
||
meta["agent"].as_str().unwrap_or(""),
|
||
meta["model"].as_str().unwrap_or("")
|
||
))
|
||
.with_meta(meta));
|
||
}
|
||
|
||
let text = run_child(app.clone(), &child, &agent, ctx.cancel.child_token(), depth + 1).await?;
|
||
Ok(tools::ToolOutput::text(text).with_meta(meta))
|
||
}
|
||
|
||
/// Entry point used by the task tool when called outside the runner.
|
||
pub async fn run_subagent(app: Arc<App>, ctx: &ToolCtx, agent: &str, description: &str, prompt: String) -> Result<(String, String)> {
|
||
let (child, agent) = create_child(&app, ctx, agent, description, prompt)?;
|
||
let text = run_child(app, &child, &agent, ctx.cancel.child_token(), 1).await?;
|
||
Ok((child.id, text))
|
||
}
|
||
|
||
/// Create the subagent session with its first message.
|
||
fn create_child(app: &Arc<App>, ctx: &ToolCtx, agent_name: &str, description: &str, prompt: String) -> Result<(Session, Agent)> {
|
||
let parent = app.session(&ctx.session_id)?;
|
||
let root = ctx.root();
|
||
let parent_agent = prompts::find_agent_for(app, &root, &ctx.cwd, &parent.agent, parent.owner_id.as_deref());
|
||
let agents = prompts::agents(app, &root, &ctx.cwd);
|
||
let mut agent = agents
|
||
.iter()
|
||
.find(|a| a.name == agent_name && a.mode != "primary")
|
||
.cloned()
|
||
.ok_or_else(|| {
|
||
let names: Vec<_> = agents.iter().filter(|a| a.mode != "primary").map(|a| a.name.as_str()).collect();
|
||
anyhow!("unknown subagent '{agent_name}'. Available: {}", names.join(", "))
|
||
})?;
|
||
// Read-only parents may only spawn read-only subagents.
|
||
if parent_agent.read_only && !agent.read_only {
|
||
agent = agents.into_iter().find(|a| a.name == "explore").unwrap_or(agent);
|
||
}
|
||
if let Some(m) = &agent.model {
|
||
app.registry().resolve(m).map_err(|e| anyhow!("subagent '{}' has an unusable model: {e:#}", agent.name))?;
|
||
}
|
||
let now = now_ms();
|
||
let child = Session {
|
||
id: new_id(),
|
||
title: description.to_string(),
|
||
cwd: parent.cwd.clone(),
|
||
model: agent.model.clone().unwrap_or_else(|| parent.model.clone()),
|
||
agent: agent.name.clone(),
|
||
permission_mode: parent.permission_mode.clone(),
|
||
parent_id: Some(parent.id.clone()),
|
||
owner_id: parent.owner_id.clone(),
|
||
created_at: now,
|
||
updated_at: now,
|
||
todos: Value::Null,
|
||
last_usage: None,
|
||
clean_room: parent.clean_room,
|
||
phase: None,
|
||
worktree: None,
|
||
pinned: false,
|
||
archived: false,
|
||
folder: None,
|
||
instructions: String::new(),
|
||
};
|
||
let child = app.save_session(child)?;
|
||
let msg = Message::new(Role::User, vec![Part::Text { text: prompt }]);
|
||
app.store.add_message(&child.id, &msg)?;
|
||
app.emit(Event::MessageAdded { session_id: child.id.clone(), message: msg });
|
||
Ok((child, agent))
|
||
}
|
||
|
||
/// Run a subagent session to completion, honouring the subagent limits.
|
||
async fn run_child(app: Arc<App>, child: &Session, agent: &Agent, cancel: CancellationToken, depth: usize) -> Result<String> {
|
||
// Registered like a normal run so the user can stop it from its own session.
|
||
app.with_runtime(&child.id, |r| r.cancel = Some(cancel.clone()));
|
||
app.emit(Event::SessionStatus { session_id: child.id.clone(), busy: true });
|
||
// Only first-level subagents count against the limits: nested subagents would
|
||
// otherwise deadlock waiting for a slot their own parent holds.
|
||
let _slot = if depth <= 1 {
|
||
let (slot, waited) = app.subagent_slot(&agent.name, agent.max_parallel, &child.id, &cancel).await;
|
||
if waited {
|
||
app.emit(Event::Waiting { session_id: child.id.clone(), reason: None });
|
||
}
|
||
slot
|
||
} else {
|
||
None
|
||
};
|
||
let res = if cancel.is_cancelled() {
|
||
Err(anyhow!("aborted"))
|
||
} else {
|
||
run_session_boxed(app.clone(), child.id.clone(), cancel, depth).await
|
||
};
|
||
app.with_runtime(&child.id, |r| r.cancel = None);
|
||
app.emit(Event::SessionStatus { session_id: child.id.clone(), busy: false });
|
||
let text = res?;
|
||
Ok(if text.trim().is_empty() { "(subagent returned no text)".into() } else { text })
|
||
}
|
||
|
||
/// Rough prompt size (≈ 3.8 characters per token, ~1,000 tokens per image).
|
||
fn estimate_prompt_tokens(req: &ChatRequest) -> u64 {
|
||
let mut chars = req.system.len();
|
||
let mut images = 0u64;
|
||
for m in &req.messages {
|
||
for p in &m.parts {
|
||
match p {
|
||
Part::Text { text } | Part::Reasoning { text, .. } | Part::Summary { text, .. } => chars += text.len(),
|
||
Part::ToolCall { input, .. } => chars += input.to_string().len(),
|
||
Part::ToolResult { content, images: imgs, .. } => {
|
||
chars += content.len();
|
||
images += imgs.len() as u64;
|
||
}
|
||
Part::Image { .. } => images += 1,
|
||
}
|
||
}
|
||
}
|
||
for t in &req.tools {
|
||
chars += t.description.len() + t.input_schema.to_string().len() + t.name.len();
|
||
}
|
||
(chars as f64 / 3.8) as u64 + images * 1000
|
||
}
|
||
|
||
/// Errors after which the fallback model takes over: the host is unreachable
|
||
/// (connection, timeout, gateway errors) or the provider's budget is used up.
|
||
fn needs_fallback(e: &anyhow::Error) -> bool {
|
||
let s = format!("{e:#}").to_lowercase();
|
||
["connect", "timed out", "timeout", "http 502", "http 503", "http 504", "dns", "budget"].iter().any(|k| s.contains(k))
|
||
}
|
||
|
||
fn short_error(e: &anyhow::Error) -> String {
|
||
format!("{e:#}").lines().next().unwrap_or("").chars().take(120).collect()
|
||
}
|
||
|
||
fn is_retryable(e: &anyhow::Error) -> bool {
|
||
let s = format!("{e:#}").to_lowercase();
|
||
["http 429", "http 5", "overloaded", "connect", "timed out", "timeout"].iter().any(|k| s.contains(k))
|
||
}
|
||
|
||
/// One model request with streaming to the UI, usage recording and retries.
|
||
async fn stream_step(
|
||
app: &Arc<App>,
|
||
session: &Session,
|
||
agent: &str,
|
||
model_spec: &str,
|
||
provider: Arc<dyn Provider>,
|
||
req: ChatRequest,
|
||
cancel: &CancellationToken,
|
||
) -> Result<Step> {
|
||
let provider_name = model_spec.split_once('/').map(|(p, _)| p).unwrap_or(model_spec);
|
||
crate::budget::check(app, provider_name, session.owner_id.as_deref())?;
|
||
let mut attempt = 0;
|
||
loop {
|
||
attempt += 1;
|
||
let msg_id = new_id();
|
||
let (tx, mut rx) = mpsc::channel::<StreamEvent>(256);
|
||
let p = provider.clone();
|
||
let r = req.clone();
|
||
let c = cancel.clone();
|
||
// Wait for a free slot on the model/host; not counted as time to first token.
|
||
let registry = app.registry();
|
||
if registry.is_saturated(model_spec) {
|
||
app.emit(Event::Waiting { session_id: session.id.clone(), reason: Some(format!("model:{model_spec}")) });
|
||
}
|
||
let slot = tokio::select! {
|
||
s = registry.acquire(model_spec) => s,
|
||
_ = cancel.cancelled() => bail!("aborted"),
|
||
};
|
||
app.emit(Event::Waiting { session_id: session.id.clone(), reason: None });
|
||
app.emit(Event::StepStarted {
|
||
session_id: session.id.clone(),
|
||
message_id: msg_id.clone(),
|
||
started_at: now_ms(),
|
||
prompt_tokens: estimate_prompt_tokens(&req),
|
||
estimated: true,
|
||
});
|
||
let handle = tokio::spawn(async move {
|
||
let _slot = slot;
|
||
p.stream(r, tx, c).await
|
||
});
|
||
|
||
let started = Instant::now();
|
||
let mut ttft: Option<u128> = None;
|
||
let mut parts: Vec<Part> = Vec::new();
|
||
let mut usage = Usage::default();
|
||
let mut stop_reason = String::new();
|
||
while let Some(ev) = rx.recv().await {
|
||
if ttft.is_none() && !matches!(ev, StreamEvent::Usage(_) | StreamEvent::Done { .. }) {
|
||
ttft = Some(started.elapsed().as_millis());
|
||
}
|
||
match ev {
|
||
StreamEvent::TextDelta(t) => {
|
||
match parts.last_mut() {
|
||
Some(Part::Text { text }) => text.push_str(&t),
|
||
_ => parts.push(Part::Text { text: t.clone() }),
|
||
}
|
||
app.emit(Event::TextDelta { session_id: session.id.clone(), message_id: msg_id.clone(), text: t });
|
||
}
|
||
StreamEvent::ReasoningDelta(t) => {
|
||
match parts.last_mut() {
|
||
Some(Part::Reasoning { text, signature: None }) => text.push_str(&t),
|
||
_ => parts.push(Part::Reasoning { text: t.clone(), signature: None }),
|
||
}
|
||
app.emit(Event::ReasoningDelta { session_id: session.id.clone(), message_id: msg_id.clone(), text: t });
|
||
}
|
||
StreamEvent::ReasoningSignature(sig) => match parts.last_mut() {
|
||
Some(Part::Reasoning { signature, .. }) if signature.is_none() => *signature = Some(sig),
|
||
// Thinking without visible text (display omitted / redacted).
|
||
_ => parts.push(Part::Reasoning { text: String::new(), signature: Some(sig) }),
|
||
},
|
||
StreamEvent::ToolCall { id, name, input, extra } => {
|
||
app.emit(Event::ToolStarted {
|
||
session_id: session.id.clone(),
|
||
message_id: msg_id.clone(),
|
||
call_id: id.clone(),
|
||
name: name.clone(),
|
||
input: input.clone(),
|
||
});
|
||
parts.push(Part::ToolCall { id, name, input, extra });
|
||
}
|
||
StreamEvent::Usage(u) => usage = u,
|
||
StreamEvent::Done { stop_reason: s } => stop_reason = s,
|
||
}
|
||
}
|
||
let result = handle.await.map_err(|e| anyhow!("provider task failed: {e}"))?;
|
||
let duration = started.elapsed().as_millis() as u64;
|
||
let (provider_name, model_id) = model_spec.split_once('/').unwrap_or(("", model_spec));
|
||
let record = |error: bool, usage: &Usage| UsageRecord {
|
||
ts: now_ms(),
|
||
session_id: session.id.clone(),
|
||
root_session_id: app.root_session_id(&session.id),
|
||
provider: provider_name.into(),
|
||
model: model_id.into(),
|
||
agent: agent.into(),
|
||
input_tokens: usage.input_tokens,
|
||
output_tokens: usage.output_tokens,
|
||
cache_read_tokens: usage.cache_read_tokens,
|
||
cache_write_tokens: usage.cache_write_tokens,
|
||
ttft_ms: ttft.unwrap_or(duration as u128) as u64,
|
||
duration_ms: duration,
|
||
cost: req.model.cost(usage),
|
||
error,
|
||
user_id: None,
|
||
allowance: usage.allowance,
|
||
};
|
||
|
||
if let Err(e) = result {
|
||
let _ = app.store.add_usage(&record(true, &usage));
|
||
if parts.is_empty() && attempt < 4 && is_retryable(&e) && !cancel.is_cancelled() {
|
||
let wait = std::time::Duration::from_secs(2u64.pow(attempt));
|
||
app.emit(Event::Error {
|
||
session_id: session.id.clone(),
|
||
message: format!("{e:#} – retrying in {}s", wait.as_secs()),
|
||
});
|
||
tokio::time::sleep(wait).await;
|
||
continue;
|
||
}
|
||
return Err(e);
|
||
}
|
||
if cancel.is_cancelled() {
|
||
// Keep what was streamed, but drop unfinished tool calls and unsigned thinking.
|
||
parts.retain(|p| matches!(p, Part::Text { .. }));
|
||
}
|
||
if usage.input_tokens + usage.output_tokens > 0 || !cancel.is_cancelled() {
|
||
let _ = app.store.add_usage(&record(false, &usage));
|
||
}
|
||
let mut message = Message::new(Role::Assistant, parts);
|
||
message.id = msg_id;
|
||
message.model = Some(model_spec.to_string());
|
||
message.usage = Some(usage.clone());
|
||
if !message.parts.is_empty() {
|
||
app.store.add_message(&session.id, &message)?;
|
||
app.emit(Event::MessageAdded { session_id: session.id.clone(), message: message.clone() });
|
||
}
|
||
app.update_session(&session.id, |s| s.last_usage = Some(usage))?;
|
||
return Ok(Step { message, stop_reason });
|
||
}
|
||
}
|
||
|
||
/// Collect a complete, tool-free answer (titles, summaries).
|
||
async fn complete_text(app: &App, spec: &str, provider: Arc<dyn Provider>, req: ChatRequest, cancel: CancellationToken) -> Result<(String, Usage)> {
|
||
let slot = app.registry().acquire(spec).await;
|
||
let (tx, mut rx) = mpsc::channel(256);
|
||
let handle = tokio::spawn(async move {
|
||
let _slot = slot;
|
||
provider.stream(req, tx, cancel).await
|
||
});
|
||
let mut text = String::new();
|
||
let mut usage = Usage::default();
|
||
while let Some(ev) = rx.recv().await {
|
||
match ev {
|
||
StreamEvent::TextDelta(t) => text.push_str(&t),
|
||
StreamEvent::Usage(u) => usage = u,
|
||
_ => {}
|
||
}
|
||
}
|
||
handle.await??;
|
||
Ok((text, usage))
|
||
}
|
||
|
||
fn small_model(app: &App, session: &Session) -> String {
|
||
let cfg = app.config();
|
||
if !cfg.server.small_model.is_empty() && app.registry().resolve(&cfg.server.small_model).is_ok() {
|
||
cfg.server.small_model.clone()
|
||
} else {
|
||
session.model.clone()
|
||
}
|
||
}
|
||
|
||
fn record_side_usage(app: &App, session: &Session, spec: &str, model: &crate::config::ModelConfig, usage: &Usage, agent: &str) {
|
||
let (p, m) = spec.split_once('/').unwrap_or(("", spec));
|
||
let _ = app.store.add_usage(&UsageRecord {
|
||
ts: now_ms(),
|
||
session_id: session.id.clone(),
|
||
root_session_id: app.root_session_id(&session.id),
|
||
provider: p.into(),
|
||
model: m.into(),
|
||
agent: agent.into(),
|
||
input_tokens: usage.input_tokens,
|
||
output_tokens: usage.output_tokens,
|
||
cache_read_tokens: usage.cache_read_tokens,
|
||
cache_write_tokens: usage.cache_write_tokens,
|
||
ttft_ms: 0,
|
||
duration_ms: 0,
|
||
cost: model.cost(usage),
|
||
error: false,
|
||
user_id: None,
|
||
allowance: usage.allowance,
|
||
});
|
||
}
|
||
|
||
fn maybe_generate_title(app: Arc<App>, session_id: &str, text: &str) {
|
||
let Ok(session) = app.session(session_id) else { return };
|
||
if session.parent_id.is_some() || !session.title.is_empty() {
|
||
return;
|
||
}
|
||
let text: String = text.chars().take(2000).collect();
|
||
tokio::spawn(async move {
|
||
let spec = small_model(&app, &session);
|
||
let Ok((provider, mut model)) = app.registry().resolve(&spec) else { return };
|
||
model.effort = Some("low".into());
|
||
let req = ChatRequest {
|
||
system: "Write a short title (max 6 words) for a coding session that starts with the user's message below. Use the user's language. Reply with the title only, no quotes or punctuation at the end.".into(),
|
||
messages: vec![Message::new(Role::User, vec![Part::Text { text }])],
|
||
tools: vec![],
|
||
max_tokens: 2000,
|
||
reasoning: false,
|
||
session_id: Some(session.id.clone()),
|
||
model: model.clone(),
|
||
};
|
||
match complete_text(&app, &spec, provider, req, CancellationToken::new()).await {
|
||
Ok((title, usage)) => {
|
||
record_side_usage(&app, &session, &spec, &model, &usage, "title");
|
||
let title = title.lines().find(|l| !l.trim().is_empty()).unwrap_or("").trim().trim_matches('"').to_string();
|
||
if !title.is_empty() {
|
||
let _ = app.update_session(&session.id, |s| {
|
||
if s.title.is_empty() {
|
||
s.title = title.chars().take(80).collect();
|
||
}
|
||
});
|
||
}
|
||
}
|
||
Err(e) => tracing::debug!("title generation failed: {e:#}"),
|
||
}
|
||
});
|
||
}
|
||
|
||
const COMPACT_PROMPT: &str = "Summarize this coding session so far so that the work can continue seamlessly from the summary alone. Include: the user's goals and explicit requests (verbatim where important), decisions made, files created/changed (with paths) and their current state, commands/tests run and results, open problems, and the exact next steps. Be thorough but concise; use headings and bullet points.";
|
||
|
||
/// Suggest a title for a session from its conversation, with the given model
|
||
/// (or the small model). The title is returned, not saved.
|
||
pub async fn suggest_title(app: Arc<App>, session_id: &str, model_spec: Option<String>) -> Result<String> {
|
||
let session = app.session(session_id)?;
|
||
let spec = model_spec.filter(|m| !m.is_empty()).unwrap_or_else(|| small_model(&app, &session));
|
||
let (provider, mut model) = app.registry().resolve(&spec)?;
|
||
model.effort = Some("low".into());
|
||
let history = active_history(app.store.messages(session_id)?);
|
||
let mut transcript = transcript_of(&history);
|
||
if transcript.len() > 12_000 {
|
||
// Beginning (the goal) and end (where it went).
|
||
let head = clip(&transcript, 6_000);
|
||
let mut start = transcript.len() - 5_000;
|
||
while !transcript.is_char_boundary(start) {
|
||
start += 1;
|
||
}
|
||
transcript = format!("{head}\n…\n{}", &transcript[start..]);
|
||
}
|
||
if transcript.trim().is_empty() {
|
||
bail!("the session has no messages yet");
|
||
}
|
||
let req = ChatRequest {
|
||
system: "Write a short, specific title (max 6 words) for the coding session below. Use the language of the user's messages. Reply with the title only, no quotes, no punctuation at the end.".into(),
|
||
messages: vec![Message::new(Role::User, vec![Part::Text { text: format!("<session>\n{transcript}\n</session>") }])],
|
||
tools: vec![],
|
||
max_tokens: 2000,
|
||
reasoning: false,
|
||
session_id: Some(session.id.clone()),
|
||
model: model.clone(),
|
||
};
|
||
let (text, usage) = complete_text(&app, &spec, provider, req, CancellationToken::new()).await?;
|
||
record_side_usage(&app, &session, &spec, &model, &usage, "title");
|
||
let text = strip_think(&text);
|
||
let title = text.lines().find(|l| !l.trim().is_empty()).unwrap_or("").trim().trim_matches(['"', '\'', '*', '#', ' ']).to_string();
|
||
if title.is_empty() {
|
||
bail!("the model returned no title");
|
||
}
|
||
Ok(title.chars().take(80).collect())
|
||
}
|
||
|
||
/// Write only the hunks of a proposed write/edit that the user accepted.
|
||
async fn apply_partial(ctx: &ToolCtx, name: &str, input: &Value, selected: &[usize]) -> Result<tools::ToolOutput> {
|
||
let path = ctx.resolve(tools::arg_str(input, "path")?)?;
|
||
let old = tokio::fs::read_to_string(&path).await.unwrap_or_default();
|
||
let proposed = if name == "write" { tools::arg_str(input, "content")?.to_string() } else { tools::fs::apply_edit(&old, input)? };
|
||
let (result, total) = tools::fs::apply_hunks(&old, &proposed, selected);
|
||
let accepted = selected.iter().filter(|i| **i < total).count();
|
||
if let Some(dir) = path.parent() {
|
||
tokio::fs::create_dir_all(dir).await?;
|
||
}
|
||
tokio::fs::write(&path, &result).await?;
|
||
ctx.app.mark_read(&ctx.session_id, &path);
|
||
let rel = path.strip_prefix(&ctx.cwd).map(|p| p.display().to_string()).unwrap_or_else(|_| path.display().to_string());
|
||
let applied = tools::fs::unified_diff(&rel, &old, &result);
|
||
let rejected = tools::fs::unified_diff(&rel, &result, &proposed);
|
||
Ok(tools::ToolOutput::text(format!(
|
||
"The user reviewed your change to {rel} and applied only {accepted} of {total} hunks. \
|
||
The file now contains the accepted parts; the rejected parts were NOT applied – do not re-apply them unless the user asks.\n\n\
|
||
Rejected (what is still missing compared to your proposal):\n{}",
|
||
clip(&rejected, 8000)
|
||
))
|
||
.with_meta(serde_json::json!({"diff": applied, "path": path, "partial": {"accepted": accepted, "total": total}})))
|
||
}
|
||
|
||
const PHASES: &[&str] = &["study", "specify", "build", "check", "ready"];
|
||
|
||
/// Craft phase implied by a tool call (fallback when the agent does not report it).
|
||
fn infer_phase(tool: &str, input: &Value, deep: bool) -> Option<&'static str> {
|
||
let path = input["path"].as_str().unwrap_or("");
|
||
let label = input["label"].as_str().unwrap_or("");
|
||
let cmd = input["command"].as_str().unwrap_or("");
|
||
Some(match tool {
|
||
"write" | "edit" => {
|
||
let doc = path.contains(".codecraft/spec") || path.starts_with("spec/") || path.contains("/spec/") || path.starts_with("docs/") || path.contains("proposal/");
|
||
if doc && deep { "specify" } else if doc { return None } else { "build" }
|
||
}
|
||
"image_compare" | "openscad" => "check",
|
||
"browser_snapshot" if label.starts_with("build") => "check",
|
||
"bash" if ["test", "pytest", "cargo check", "cargo build", "npm run build", "http.server", "serve", "vite"].iter().any(|k| cmd.contains(k)) => "check",
|
||
"read" | "glob" | "grep" | "list" | "browser_snapshot" | "web_fetch" | "web_search" | "binary_info" | "decompile" => "study",
|
||
_ => return None,
|
||
})
|
||
}
|
||
|
||
/// Move the craft phase forward (never backwards); `only_from` restricts the transition.
|
||
fn advance_phase(app: &App, session_id: &str, phase: &str, only_from: Option<&str>) {
|
||
let root = app.root_session_id(session_id);
|
||
let Ok(s) = app.session(&root) else { return };
|
||
let cur = s.phase.as_deref().and_then(|p| PHASES.iter().position(|x| *x == p));
|
||
if let Some(from) = only_from {
|
||
if s.phase.as_deref() != Some(from) {
|
||
return;
|
||
}
|
||
}
|
||
let Some(next) = PHASES.iter().position(|x| *x == phase) else { return };
|
||
// Studying again after building is normal – no step back automatically.
|
||
if cur.is_some_and(|c| c >= next) {
|
||
return;
|
||
}
|
||
// Clean room has no specify phase.
|
||
let phase = if phase == "specify" && s.clean_room != Some(false) { "study" } else { phase };
|
||
if s.phase.as_deref() == Some(phase) {
|
||
return;
|
||
}
|
||
let _ = app.update_session(&root, |s| s.phase = Some(phase.to_string()));
|
||
}
|
||
|
||
/// Review the last turn against the user's request. Returns feedback for the
|
||
/// agent, or None when the reviewer approves.
|
||
#[allow(clippy::too_many_arguments)]
|
||
async fn review_turn(app: &Arc<App>, session: &Session, agent: &prompts::Agent, model_spec: &str, turn_start: i64, round: u32, rounds: u32, cancel: &CancellationToken) -> Result<Option<String>> {
|
||
let refine = agent.refine.clone().unwrap_or_default();
|
||
let spec = refine.model.clone().filter(|m| !m.is_empty()).unwrap_or_else(|| model_spec.to_string());
|
||
let (provider, mut model) = app.registry().resolve(&spec)?;
|
||
model.effort = Some("medium".into());
|
||
app.emit(Event::Waiting { session_id: session.id.clone(), reason: Some(format!("review:{round}/{rounds}")) });
|
||
let result = async {
|
||
let history = active_history(app.store.messages(&session.id)?);
|
||
// The user's request: the last user message with text that is not feedback/tool output.
|
||
let req_idx = history
|
||
.iter()
|
||
.rposition(|m| matches!(m.role, Role::User) && m.parts.iter().any(|p| matches!(p, Part::Text { text } if !text.starts_with("<refine-feedback") && !text.starts_with("<hook-"))))
|
||
.unwrap_or(0);
|
||
let request: String = history.get(req_idx).map(|m| m.parts.iter().filter_map(|p| if let Part::Text { text } = p { Some(text.as_str()) } else { None }).collect::<Vec<_>>().join("\n")).unwrap_or_default();
|
||
let mut work = transcript_of(&history[(req_idx + 1).min(history.len())..]);
|
||
if work.len() > 30_000 {
|
||
let mut start = work.len() - 30_000;
|
||
while !work.is_char_boundary(start) {
|
||
start += 1;
|
||
}
|
||
work = format!("…{}", &work[start..]);
|
||
}
|
||
// Actual file changes of this turn (checkpoint vs. current content).
|
||
let mut changes = String::new();
|
||
let mut seen = std::collections::HashSet::new();
|
||
for (path, before) in app.store.checkpoints_since(&session.id, turn_start)? {
|
||
if !seen.insert(path.clone()) || changes.len() > 30_000 {
|
||
continue;
|
||
}
|
||
let old = before.map(|b| String::from_utf8_lossy(&b).into_owned()).unwrap_or_default();
|
||
let new = std::fs::read_to_string(&path).unwrap_or_default();
|
||
let diff = similar::TextDiff::from_lines(&old, &new).unified_diff().context_radius(3).header(&path, &path).to_string();
|
||
changes.push_str(&clip(&diff, 12_000));
|
||
changes.push('\n');
|
||
}
|
||
let criteria = refine.criteria.clone().filter(|c| !c.trim().is_empty()).unwrap_or_else(|| prompts::refine_criteria(&agent.name).to_string());
|
||
let system = "You review the work of an AI coding agent for the user. Be strict about what matters for the user's request and ignore style nitpicks. \
|
||
Reply in exactly this format: the first line is APPROVED or CHANGES_NEEDED. For CHANGES_NEEDED, follow with at most 8 concrete, actionable findings as a bullet list \
|
||
(with file:line where possible). If the important work is done and correct, reply APPROVED.";
|
||
let body = format!(
|
||
"<request>\n{request}\n</request>\n\n<criteria>\n{criteria}\n</criteria>\n\n<agent-work mode=\"{}\">\n{work}\n</agent-work>\n\n<file-changes>\n{}\n</file-changes>",
|
||
agent.name,
|
||
if changes.is_empty() { "(no files changed in this turn)" } else { changes.as_str() }
|
||
);
|
||
let req = ChatRequest { system: system.into(), messages: vec![Message::new(Role::User, vec![Part::Text { text: body }])], tools: vec![], max_tokens: 4000, reasoning: false, model: model.clone(), session_id: Some(session.id.clone()) };
|
||
let (text, usage) = complete_text(app, &spec, provider, req, cancel.clone()).await?;
|
||
record_side_usage(app, session, &spec, &model, &usage, "review");
|
||
let text = strip_think(&text);
|
||
let mut lines = text.lines().skip_while(|l| l.trim().is_empty());
|
||
let verdict = lines.next().unwrap_or("").trim().to_uppercase();
|
||
if verdict.starts_with("APPROVED") || text.trim().is_empty() {
|
||
return Ok(None);
|
||
}
|
||
let findings: String = lines.collect::<Vec<_>>().join("\n");
|
||
let findings = if findings.trim().is_empty() { text.clone() } else { findings };
|
||
Ok(Some(format!(
|
||
"<refine-feedback round=\"{round}/{rounds}\" reviewer=\"{spec}\">\n{}\n</refine-feedback>\nAddress these review findings (or explain briefly why one does not apply), verify, then finish.",
|
||
findings.trim()
|
||
)))
|
||
}
|
||
.await;
|
||
app.emit(Event::Waiting { session_id: session.id.clone(), reason: None });
|
||
result
|
||
}
|
||
|
||
/// Remove a leading `<think>…</think>` block some models put into the text.
|
||
fn strip_think(text: &str) -> String {
|
||
let t = text.trim_start();
|
||
if t.starts_with("<think>") {
|
||
if let Some(end) = t.find("</think>") {
|
||
return t[end + 8..].trim().to_string();
|
||
}
|
||
return String::new();
|
||
}
|
||
text.trim().to_string()
|
||
}
|
||
|
||
fn clip(s: &str, max: usize) -> String {
|
||
if s.len() <= max {
|
||
return s.to_string();
|
||
}
|
||
let mut end = max;
|
||
while !s.is_char_boundary(end) {
|
||
end -= 1;
|
||
}
|
||
format!("{} […{} more bytes]", &s[..end], s.len() - end)
|
||
}
|
||
|
||
/// The history as readable text for summarizing (tool output shortened).
|
||
pub fn transcript_of(history: &[Message]) -> String {
|
||
let mut out = String::new();
|
||
for m in history {
|
||
let role = if matches!(m.role, Role::User) { "USER" } else { "ASSISTANT" };
|
||
let mut body = String::new();
|
||
for p in &m.parts {
|
||
match p {
|
||
Part::Text { text } => body.push_str(&format!("{text}\n")),
|
||
Part::Summary { text, .. } => body.push_str(&format!("[Summary of the earlier conversation]\n{text}\n")),
|
||
Part::ToolCall { name, input, .. } => body.push_str(&format!("[tool call {name}: {}]\n", clip(&input.to_string(), 1500))),
|
||
Part::ToolResult { name, content, is_error, .. } => {
|
||
body.push_str(&format!("[{} of {name}]\n{}\n", if *is_error { "error" } else { "result" }, clip(content, 3000)))
|
||
}
|
||
Part::Image { .. } => body.push_str("[image]\n"),
|
||
Part::Reasoning { .. } => {}
|
||
}
|
||
}
|
||
if !body.trim().is_empty() {
|
||
out.push_str(&format!("### {role}\n{}\n", body.trim_end()));
|
||
}
|
||
}
|
||
out
|
||
}
|
||
|
||
/// Replace the active history with a summary (the old messages stay visible in the UI).
|
||
/// Summarize the history. `auto` = (context tokens, context limit) when triggered by a
|
||
/// nearly full context; the chat shows the progress and the reason.
|
||
pub async fn compact(app: Arc<App>, session_id: &str, cancel: CancellationToken, auto: Option<(u64, u64)>) -> Result<()> {
|
||
let event = |state: &str| Event::Compaction {
|
||
session_id: session_id.to_string(),
|
||
state: state.into(),
|
||
auto: auto.is_some(),
|
||
tokens: auto.map(|a| a.0),
|
||
limit: auto.map(|a| a.1),
|
||
};
|
||
app.emit(event("started"));
|
||
let res = compact_inner(app.clone(), session_id, cancel, auto).await;
|
||
app.emit(event(if res.is_ok() { "done" } else { "failed" }));
|
||
res
|
||
}
|
||
|
||
async fn compact_inner(app: Arc<App>, session_id: &str, cancel: CancellationToken, auto: Option<(u64, u64)>) -> Result<()> {
|
||
let session = app.session(session_id)?;
|
||
let history = active_history(app.store.messages(session_id)?);
|
||
if history.is_empty() {
|
||
return Ok(());
|
||
}
|
||
let spec = session.model.clone();
|
||
let (provider, model) = app.registry().resolve(&spec)?;
|
||
// The history goes in as a plain transcript: with real tool calls in the
|
||
// history (and no tools offered) some models answer with a tool call or
|
||
// only think, leaving the summary empty.
|
||
let transcript = transcript_of(&history);
|
||
let mut summary = String::new();
|
||
for attempt in 0..2 {
|
||
let prompt = if attempt == 0 {
|
||
format!("<transcript>\n{transcript}\n</transcript>\n\n{COMPACT_PROMPT}")
|
||
} else {
|
||
format!("<transcript>\n{transcript}\n</transcript>\n\n{COMPACT_PROMPT}\n\nAnswer directly with the summary as plain Markdown text. Do not call tools.")
|
||
};
|
||
let req = ChatRequest {
|
||
system: "You are a precise summarizer of software engineering sessions. You never call tools; you only write the requested summary.".into(),
|
||
messages: vec![Message::new(Role::User, vec![Part::Text { text: prompt }])],
|
||
tools: vec![],
|
||
max_tokens: 16_000,
|
||
reasoning: false,
|
||
session_id: Some(session.id.clone()),
|
||
model: model.clone(),
|
||
};
|
||
let (text, usage) = complete_text(&app, &spec, provider.clone(), req, cancel.clone()).await?;
|
||
record_side_usage(&app, &session, &spec, &model, &usage, "compaction");
|
||
summary = strip_think(&text);
|
||
if !summary.trim().is_empty() {
|
||
break;
|
||
}
|
||
tracing::warn!("compaction of {session_id}: empty summary (attempt {})", attempt + 1);
|
||
}
|
||
if summary.trim().is_empty() {
|
||
bail!("compaction produced an empty summary");
|
||
}
|
||
let tokens = auto.map(|a| a.0).or_else(|| session.last_usage.as_ref().map(|u| u.context_tokens()));
|
||
save_user_parts(&app, session_id, vec![Part::Summary { text: summary, auto: auto.is_some(), tokens }])?;
|
||
app.update_session(session_id, |s| s.last_usage = None)?;
|
||
// Files must be re-read after compaction before editing.
|
||
app.with_runtime(session_id, |r| r.read_files.clear());
|
||
Ok(())
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
fn call(id: &str) -> Part {
|
||
Part::ToolCall { id: id.into(), name: "bash".into(), input: serde_json::json!({}), extra: None }
|
||
}
|
||
|
||
#[test]
|
||
fn dangling_tool_calls_get_results() {
|
||
let a = Message::new(Role::Assistant, vec![call("1"), call("2")]);
|
||
let r = Message::new(Role::User, vec![tool_result("1", "bash", "ok".into(), false, None)]);
|
||
let a2 = Message::new(Role::Assistant, vec![call("3")]);
|
||
let h = active_history(vec![Message::new(Role::User, vec![Part::Text { text: "hi".into() }]), a, r, a2]);
|
||
// user, assistant(1,2), user(result 2 + result 1), assistant(3), user(result 3)
|
||
assert_eq!(h.len(), 5);
|
||
let ids = |m: &Message| m.parts.iter().filter_map(|p| if let Part::ToolResult { call_id, .. } = p { Some(call_id.clone()) } else { None }).collect::<Vec<_>>();
|
||
assert_eq!(ids(&h[2]), vec!["2", "1"]);
|
||
assert_eq!(ids(&h[4]), vec!["3"]);
|
||
}
|
||
}
|