fix: drive tool rounds iteratively instead of recursively to prevent stack overflow on long agentic turns to fix latent stack overflow recursion bug
This commit is contained in:
2 files changed
+183
-184
No files matched your search
+28
-28
@@ -556,7 +556,6 @@ async fn run(
|
||||
}
|
||||
}
|
||||
|
||||
#[async_recursion::async_recursion]
|
||||
async fn start_directive(
|
||||
ctx: &mut RequestContext,
|
||||
input: Input,
|
||||
@@ -576,37 +575,36 @@ async fn start_directive(
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let client = input.create_client()?;
|
||||
let extract_code = !*IS_STDOUT_TERMINAL && code_mode;
|
||||
ctx.before_chat_completion(&input)?;
|
||||
let (output, tool_results) = if !input.stream() || extract_code {
|
||||
call_chat_completions(
|
||||
&input,
|
||||
true,
|
||||
extract_code,
|
||||
client.as_ref(),
|
||||
ctx,
|
||||
abort_signal.clone(),
|
||||
)
|
||||
.await?
|
||||
} else {
|
||||
call_chat_completions_streaming(&input, client.as_ref(), ctx, abort_signal.clone()).await?
|
||||
};
|
||||
ctx.after_chat_completion(app.as_ref(), &input, &output, &tool_results)?;
|
||||
let mut input = input;
|
||||
loop {
|
||||
let client = input.create_client()?;
|
||||
ctx.before_chat_completion(&input)?;
|
||||
let (output, tool_results) = if !input.stream() || extract_code {
|
||||
call_chat_completions(
|
||||
&input,
|
||||
true,
|
||||
extract_code,
|
||||
client.as_ref(),
|
||||
ctx,
|
||||
abort_signal.clone(),
|
||||
)
|
||||
.await?
|
||||
} else {
|
||||
call_chat_completions_streaming(&input, client.as_ref(), ctx, abort_signal.clone())
|
||||
.await?
|
||||
};
|
||||
ctx.after_chat_completion(app.as_ref(), &input, &output, &tool_results)?;
|
||||
|
||||
if !tool_results.is_empty() {
|
||||
input = input.merge_tool_results(output, tool_results);
|
||||
continue;
|
||||
}
|
||||
|
||||
if !tool_results.is_empty() {
|
||||
start_directive(
|
||||
ctx,
|
||||
input.merge_tool_results(output, tool_results),
|
||||
code_mode,
|
||||
abort_signal,
|
||||
)
|
||||
.await?;
|
||||
} else {
|
||||
match check_pending_tasks_guardrail(ctx) {
|
||||
GuardrailAction::Inject(prompt) => {
|
||||
let guardrail_input = Input::from_str(ctx, &prompt, None)?;
|
||||
return start_directive(ctx, guardrail_input, code_mode, abort_signal).await;
|
||||
input = Input::from_str(ctx, &prompt, None)?;
|
||||
continue;
|
||||
}
|
||||
GuardrailAction::ForceTerminate(ids) => {
|
||||
warn!(
|
||||
@@ -617,6 +615,8 @@ async fn start_directive(
|
||||
}
|
||||
GuardrailAction::NoAction => {}
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
|
||||
ctx.exit_session()?;
|
||||
|
||||
+155
-156
@@ -1424,74 +1424,73 @@ pub async fn run_repl_command(
|
||||
Ok(false)
|
||||
}
|
||||
|
||||
#[async_recursion::async_recursion]
|
||||
async fn ask(
|
||||
ctx: &mut RequestContext,
|
||||
abort_signal: AbortSignal,
|
||||
mut input: Input,
|
||||
input: Input,
|
||||
with_embeddings: bool,
|
||||
) -> Result<()> {
|
||||
if input.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
ctx.session_abort = Some(abort_signal.clone());
|
||||
if with_embeddings {
|
||||
input.use_embeddings(abort_signal.clone()).await?;
|
||||
}
|
||||
while ctx.is_compressing_session() {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
||||
}
|
||||
|
||||
let app = Arc::clone(&ctx.app.config);
|
||||
|
||||
if graph::active_agent_graph_name(ctx).is_some() {
|
||||
ctx.before_chat_completion(&input)?;
|
||||
let output =
|
||||
graph::run_active_agent_graph(ctx, &input.text(), abort_signal.clone()).await?;
|
||||
app.print_markdown(&output)?;
|
||||
ctx.after_chat_completion(app.as_ref(), &input, &output, &[])?;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let client = input.create_client()?;
|
||||
ctx.before_chat_completion(&input)?;
|
||||
let (output, tool_results) = {
|
||||
let result = if input.stream() {
|
||||
call_chat_completions_streaming(&input, client.as_ref(), ctx, abort_signal.clone())
|
||||
.await
|
||||
} else {
|
||||
call_chat_completions(
|
||||
&input,
|
||||
true,
|
||||
false,
|
||||
client.as_ref(),
|
||||
ctx,
|
||||
abort_signal.clone(),
|
||||
)
|
||||
.await
|
||||
};
|
||||
match result {
|
||||
Ok(v) => v,
|
||||
Err(err) => {
|
||||
ctx.on_chat_completion_error(app.as_ref(), &input);
|
||||
return Err(err);
|
||||
}
|
||||
let mut input = input;
|
||||
let mut with_embeddings = with_embeddings;
|
||||
loop {
|
||||
if input.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
ctx.after_chat_completion(app.as_ref(), &input, &output, &tool_results)?;
|
||||
if !tool_results.is_empty() {
|
||||
ask(
|
||||
ctx,
|
||||
abort_signal,
|
||||
input.merge_tool_results(output, tool_results),
|
||||
false,
|
||||
)
|
||||
.await
|
||||
} else {
|
||||
ctx.session_abort = Some(abort_signal.clone());
|
||||
if with_embeddings {
|
||||
input.use_embeddings(abort_signal.clone()).await?;
|
||||
with_embeddings = false;
|
||||
}
|
||||
while ctx.is_compressing_session() {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
||||
}
|
||||
|
||||
let app = Arc::clone(&ctx.app.config);
|
||||
|
||||
if graph::active_agent_graph_name(ctx).is_some() {
|
||||
ctx.before_chat_completion(&input)?;
|
||||
let output =
|
||||
graph::run_active_agent_graph(ctx, &input.text(), abort_signal.clone()).await?;
|
||||
app.print_markdown(&output)?;
|
||||
ctx.after_chat_completion(app.as_ref(), &input, &output, &[])?;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let client = input.create_client()?;
|
||||
ctx.before_chat_completion(&input)?;
|
||||
let (output, tool_results) = {
|
||||
let result = if input.stream() {
|
||||
call_chat_completions_streaming(&input, client.as_ref(), ctx, abort_signal.clone())
|
||||
.await
|
||||
} else {
|
||||
call_chat_completions(
|
||||
&input,
|
||||
true,
|
||||
false,
|
||||
client.as_ref(),
|
||||
ctx,
|
||||
abort_signal.clone(),
|
||||
)
|
||||
.await
|
||||
};
|
||||
match result {
|
||||
Ok(v) => v,
|
||||
Err(err) => {
|
||||
ctx.on_chat_completion_error(app.as_ref(), &input);
|
||||
return Err(err);
|
||||
}
|
||||
}
|
||||
};
|
||||
ctx.after_chat_completion(app.as_ref(), &input, &output, &tool_results)?;
|
||||
if !tool_results.is_empty() {
|
||||
input = input.merge_tool_results(output, tool_results);
|
||||
continue;
|
||||
}
|
||||
|
||||
match check_pending_tasks_guardrail(ctx) {
|
||||
GuardrailAction::Inject(prompt) => {
|
||||
let guardrail_input = Input::from_str(ctx, &prompt, None)?;
|
||||
return ask(ctx, abort_signal, guardrail_input, false).await;
|
||||
input = Input::from_str(ctx, &prompt, None)?;
|
||||
continue;
|
||||
}
|
||||
GuardrailAction::ForceTerminate(ids) => {
|
||||
warn!(
|
||||
@@ -1533,103 +1532,103 @@ async fn ask(
|
||||
|
||||
format!("{prompt}\n\n{todo_state}")
|
||||
};
|
||||
let continuation_input = Input::from_str(ctx, &full_prompt, None)?;
|
||||
ask(ctx, abort_signal, continuation_input, false).await
|
||||
} else {
|
||||
reset_continuation(ctx);
|
||||
if let Some(reason) = ctx.auto_continue_paused.clone()
|
||||
&& ctx.todo_list.has_incomplete()
|
||||
{
|
||||
let remaining = ctx.todo_list.incomplete_count();
|
||||
let color = if app.light_theme() {
|
||||
nu_ansi_term::Color::LightGray
|
||||
} else {
|
||||
nu_ansi_term::Color::DarkGray
|
||||
};
|
||||
eprintln!(
|
||||
"\n⏸ {}",
|
||||
color.italic().paint(format!(
|
||||
"Auto-continue paused, awaiting user ({remaining} incomplete todo(s) preserved): {reason}"
|
||||
))
|
||||
);
|
||||
}
|
||||
if ctx.maybe_autoname_session() {
|
||||
let color = if app.light_theme() {
|
||||
nu_ansi_term::Color::LightGray
|
||||
} else {
|
||||
nu_ansi_term::Color::DarkGray
|
||||
};
|
||||
eprintln!("\n📢 {}", color.italic().paint("Autonaming the session."),);
|
||||
if let Err(err) = ctx.autoname_session(app.as_ref()).await {
|
||||
warn!("Failed to autonaming the session: {err}");
|
||||
}
|
||||
if let Some(session) = ctx.session.as_mut() {
|
||||
session.set_autonaming(false);
|
||||
}
|
||||
}
|
||||
|
||||
let needs_compression = ctx
|
||||
.session
|
||||
.as_ref()
|
||||
.is_some_and(|s| s.needs_compression(app.compression_threshold));
|
||||
|
||||
if needs_compression {
|
||||
let agent_can_continue_after_compress =
|
||||
should_continue(ctx) && !abort_signal.aborted_ctrlc();
|
||||
|
||||
if let Some(session) = ctx.session.as_mut() {
|
||||
session.set_compressing(true);
|
||||
}
|
||||
|
||||
let color = if app.light_theme() {
|
||||
nu_ansi_term::Color::LightGray
|
||||
} else {
|
||||
nu_ansi_term::Color::DarkGray
|
||||
};
|
||||
eprintln!("\n📢 {}", color.italic().paint("Compressing the session."),);
|
||||
|
||||
if let Err(err) = ctx.compress_session().await {
|
||||
warn!("Failed to compress the session: {err}");
|
||||
}
|
||||
if let Some(session) = ctx.session.as_mut() {
|
||||
session.set_compressing(false);
|
||||
}
|
||||
|
||||
if agent_can_continue_after_compress {
|
||||
let full_prompt = {
|
||||
let config = ctx.auto_continue_config();
|
||||
let todo_state = ctx.todo_list.render_for_model();
|
||||
let remaining = ctx.todo_list.incomplete_count();
|
||||
ctx.increment_auto_continue_count();
|
||||
let count = ctx.auto_continue_count;
|
||||
let max = config.max_continues;
|
||||
|
||||
let prompt = config
|
||||
.continuation_prompt
|
||||
.as_deref()
|
||||
.unwrap_or(DEFAULT_CONTINUATION_PROMPT);
|
||||
|
||||
let color = if app.light_theme() {
|
||||
nu_ansi_term::Color::LightGray
|
||||
} else {
|
||||
nu_ansi_term::Color::DarkGray
|
||||
};
|
||||
eprintln!(
|
||||
"\n📋 {}",
|
||||
color.italic().paint(format!(
|
||||
"Auto-continuing after compression ({count}/{max}): {remaining} incomplete todo(s) remain"
|
||||
))
|
||||
);
|
||||
|
||||
format!("{prompt}\n\n{todo_state}")
|
||||
};
|
||||
let continuation_input = Input::from_str(ctx, &full_prompt, None)?;
|
||||
return ask(ctx, abort_signal, continuation_input, false).await;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
input = Input::from_str(ctx, &full_prompt, None)?;
|
||||
continue;
|
||||
}
|
||||
|
||||
reset_continuation(ctx);
|
||||
if let Some(reason) = ctx.auto_continue_paused.clone()
|
||||
&& ctx.todo_list.has_incomplete()
|
||||
{
|
||||
let remaining = ctx.todo_list.incomplete_count();
|
||||
let color = if app.light_theme() {
|
||||
nu_ansi_term::Color::LightGray
|
||||
} else {
|
||||
nu_ansi_term::Color::DarkGray
|
||||
};
|
||||
eprintln!(
|
||||
"\n⏸ {}",
|
||||
color.italic().paint(format!(
|
||||
"Auto-continue paused, awaiting user ({remaining} incomplete todo(s) preserved): {reason}"
|
||||
))
|
||||
);
|
||||
}
|
||||
if ctx.maybe_autoname_session() {
|
||||
let color = if app.light_theme() {
|
||||
nu_ansi_term::Color::LightGray
|
||||
} else {
|
||||
nu_ansi_term::Color::DarkGray
|
||||
};
|
||||
eprintln!("\n📢 {}", color.italic().paint("Autonaming the session."),);
|
||||
if let Err(err) = ctx.autoname_session(app.as_ref()).await {
|
||||
warn!("Failed to autonaming the session: {err}");
|
||||
}
|
||||
if let Some(session) = ctx.session.as_mut() {
|
||||
session.set_autonaming(false);
|
||||
}
|
||||
}
|
||||
|
||||
let needs_compression = ctx
|
||||
.session
|
||||
.as_ref()
|
||||
.is_some_and(|s| s.needs_compression(app.compression_threshold));
|
||||
|
||||
if needs_compression {
|
||||
let agent_can_continue_after_compress =
|
||||
should_continue(ctx) && !abort_signal.aborted_ctrlc();
|
||||
|
||||
if let Some(session) = ctx.session.as_mut() {
|
||||
session.set_compressing(true);
|
||||
}
|
||||
|
||||
let color = if app.light_theme() {
|
||||
nu_ansi_term::Color::LightGray
|
||||
} else {
|
||||
nu_ansi_term::Color::DarkGray
|
||||
};
|
||||
eprintln!("\n📢 {}", color.italic().paint("Compressing the session."),);
|
||||
|
||||
if let Err(err) = ctx.compress_session().await {
|
||||
warn!("Failed to compress the session: {err}");
|
||||
}
|
||||
if let Some(session) = ctx.session.as_mut() {
|
||||
session.set_compressing(false);
|
||||
}
|
||||
|
||||
if agent_can_continue_after_compress {
|
||||
let full_prompt = {
|
||||
let config = ctx.auto_continue_config();
|
||||
let todo_state = ctx.todo_list.render_for_model();
|
||||
let remaining = ctx.todo_list.incomplete_count();
|
||||
ctx.increment_auto_continue_count();
|
||||
let count = ctx.auto_continue_count;
|
||||
let max = config.max_continues;
|
||||
|
||||
let prompt = config
|
||||
.continuation_prompt
|
||||
.as_deref()
|
||||
.unwrap_or(DEFAULT_CONTINUATION_PROMPT);
|
||||
|
||||
let color = if app.light_theme() {
|
||||
nu_ansi_term::Color::LightGray
|
||||
} else {
|
||||
nu_ansi_term::Color::DarkGray
|
||||
};
|
||||
eprintln!(
|
||||
"\n📋 {}",
|
||||
color.italic().paint(format!(
|
||||
"Auto-continuing after compression ({count}/{max}): {remaining} incomplete todo(s) remain"
|
||||
))
|
||||
);
|
||||
|
||||
format!("{prompt}\n\n{todo_state}")
|
||||
};
|
||||
input = Input::from_str(ctx, &full_prompt, None)?;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user