From 5e98ef95879fdc117e983fb86973eba794520bc9 Mon Sep 17 00:00:00 2001 From: rick12345 Date: Fri, 13 Nov 2026 17:18:46 +0800 Subject: [PATCH 1/2] fix(model): reduce stack usage during provider request encoding Generated-by: OpenAI Codex --- plugins/agent/model/src/adapters/wire/frozen.rs | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/plugins/agent/model/src/adapters/wire/frozen.rs b/plugins/agent/model/src/adapters/wire/frozen.rs index 2639bea3d7..72a461eb9f 100644 --- a/plugins/agent/model/src/adapters/wire/frozen.rs +++ b/plugins/agent/model/src/adapters/wire/frozen.rs @@ -561,13 +561,13 @@ impl WirePlan { counts: Counts::default(), expected: self.counts, }); - let Sink::Backed(build) = emit( + let Sink::Backed(build) = Box::pin(emit( &self.encoding, self.source.as_ref(), work.budget(), sink, self.projection, - ) + )) .await? else { unreachable!("backed encoding pass") @@ -605,13 +605,14 @@ pub(super) async fn count( ) -> Result { let work = resources.acquire_work().await.map_err(error)?; let budget = work.budget(); - let Sink::Count(counts) = emit( + // Keep the codec's large async state out of the enclosing request futures. + let Sink::Count(counts) = Box::pin(emit( &encoding, source.as_ref(), budget, Sink::Count(Counts::default()), projection, - ) + )) .await? else { unreachable!("count pass") @@ -664,13 +665,13 @@ pub(super) async fn build( counts: Counts::default(), expected: count, }; - let Sink::Build(mut build) = emit( + let Sink::Build(mut build) = Box::pin(emit( &encoding, source.as_ref(), budget, Sink::Build(build), projection, - ) + )) .await? else { unreachable!("build pass") From 21f8f7862ab21b9a89543ae0d260141af99da0f1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CM4n5ter=E2=80=9D?= Date: Fri, 13 Nov 2026 20:55:14 +0800 Subject: [PATCH 2/2] refactor(model): centralize wire encoder future allocation Generated-by: OpenAI Codex --- .../agent/model/src/adapters/wire/frozen.rs | 228 +++++++++--------- 1 file changed, 115 insertions(+), 113 deletions(-) diff --git a/plugins/agent/model/src/adapters/wire/frozen.rs b/plugins/agent/model/src/adapters/wire/frozen.rs index 72a461eb9f..9943d98fbf 100644 --- a/plugins/agent/model/src/adapters/wire/frozen.rs +++ b/plugins/agent/model/src/adapters/wire/frozen.rs @@ -387,131 +387,134 @@ impl Sink<'_> { Ok(()) } } -async fn emit<'a>( +fn emit<'a>( encoding: &WireEncoding, source: &dyn MessageSource, budget: Arc, mut sink: Sink<'a>, projection: chat_source::Projection, -) -> Result, ModelError> { - let mut recipes = source.open(0); - let mut metadata = source.open_metadata(0); - let anthropic_compiler = match encoding.options.as_ref() { - WireOptions::Anthropic(options) => { - let compiler = anthropic::Compiler::new(options.clone(), encoding.language); - Some(if matches!(&sink, Sink::Count(_)) { - compiler.directory() - } else { - compiler - }) - } - _ => None, - }; - let mut chat_compiler = match encoding.options.as_ref() { - WireOptions::Chat(options) => Some(match projection { - chat_source::Projection::Generated => { - chat_source::Compiler::new(options, encoding.language) - } - chat_source::Projection::Captured => { - chat_source::Compiler::captured(options, encoding.language) +) -> impl Future, ModelError>> + Send { + // Keep the codec's large async state out of enclosing request futures. + Box::pin(async move { + let mut recipes = source.open(0); + let mut metadata = source.open_metadata(0); + let anthropic_compiler = match encoding.options.as_ref() { + WireOptions::Anthropic(options) => { + let compiler = anthropic::Compiler::new(options.clone(), encoding.language); + Some(if matches!(&sink, Sink::Count(_)) { + compiler.directory() + } else { + compiler + }) } - }), - _ => None, - }; - let mut next_recipe = recipes.next().await.map_err(error)?; - for message_index in 0..source.len() { - tokio::task::yield_now().await; - let before = sink.shape(); - sink.begin(); - let recipe = next_recipe - .take() - .ok_or_else(|| ModelError::Adapter("truncated message recipe directory".into()))?; - next_recipe = recipes.next().await.map_err(error)?; - let last_tool = !matches!( - &next_recipe, - Some(maka_agent_facts::model::prompt_source::source::MessageRecipe::Tool { .. }) - ); - let message_metadata = - metadata.next().await.map_err(error)?.ok_or_else(|| { + _ => None, + }; + let mut chat_compiler = match encoding.options.as_ref() { + WireOptions::Chat(options) => Some(match projection { + chat_source::Projection::Generated => { + chat_source::Compiler::new(options, encoding.language) + } + chat_source::Projection::Captured => { + chat_source::Compiler::captured(options, encoding.language) + } + }), + _ => None, + }; + let mut next_recipe = recipes.next().await.map_err(error)?; + for message_index in 0..source.len() { + tokio::task::yield_now().await; + let before = sink.shape(); + sink.begin(); + let recipe = next_recipe + .take() + .ok_or_else(|| ModelError::Adapter("truncated message recipe directory".into()))?; + next_recipe = recipes.next().await.map_err(error)?; + let last_tool = !matches!( + &next_recipe, + Some(maka_agent_facts::model::prompt_source::source::MessageRecipe::Tool { .. }) + ); + let message_metadata = metadata.next().await.map_err(error)?.ok_or_else(|| { ModelError::Adapter("truncated message metadata directory".into()) })?; - let role = message_metadata.role; - match encoding.options.as_ref() { - WireOptions::Chat(_) => { - let mut cursor = chat_compiler.as_mut().expect("Chat compiler").open( - recipe, - last_tool, - budget.clone(), - ); - let mut first = true; - while let Some(mut item) = cursor.next().await? { + let role = message_metadata.role; + match encoding.options.as_ref() { + WireOptions::Chat(_) => { + let mut cursor = chat_compiler.as_mut().expect("Chat compiler").open( + recipe, + last_tool, + budget.clone(), + ); + let mut first = true; + while let Some(mut item) = cursor.next().await? { + if first { + item.metadata = + Some(wire::WireItemMetadata::Message(message_metadata.clone())); + first = false; + } + sink.item(message_index, item, budget.clone()).await?; + } if first { - item.metadata = - Some(wire::WireItemMetadata::Message(message_metadata.clone())); - first = false; + sink.item( + message_index, + SourceItem { + fragments: Arc::from([]), + metadata: Some(wire::WireItemMetadata::Message(message_metadata)), + citation: None, + owned_bytes: 0, + }, + budget.clone(), + ) + .await?; } - sink.item(message_index, item, budget.clone()).await?; } - if first { - sink.item( - message_index, - SourceItem { - fragments: Arc::from([]), - metadata: Some(wire::WireItemMetadata::Message(message_metadata)), - citation: None, - owned_bytes: 0, - }, + WireOptions::Responses { plaintext, options } => { + let mut cursor = responses::open( + recipe, + encoding.language, + *plaintext, + options, budget.clone(), - ) - .await?; - } - } - WireOptions::Responses { plaintext, options } => { - let mut cursor = responses::open( - recipe, - encoding.language, - *plaintext, - options, - budget.clone(), - ); - let mut first = true; - while let Some(mut item) = cursor.next().await? { + ); + let mut first = true; + while let Some(mut item) = cursor.next().await? { + if first { + item.metadata = + Some(wire::WireItemMetadata::Message(message_metadata.clone())); + first = false; + } + sink.item(message_index, item, budget.clone()).await?; + } if first { - item.metadata = - Some(wire::WireItemMetadata::Message(message_metadata.clone())); - first = false; + sink.item( + message_index, + SourceItem { + fragments: Arc::from([]), + metadata: Some(wire::WireItemMetadata::Message(message_metadata)), + citation: None, + owned_bytes: 0, + }, + budget.clone(), + ) + .await?; } - sink.item(message_index, item, budget.clone()).await?; - } - if first { - sink.item( - message_index, - SourceItem { - fragments: Arc::from([]), - metadata: Some(wire::WireItemMetadata::Message(message_metadata)), - citation: None, - owned_bytes: 0, - }, - budget.clone(), - ) - .await?; } - } - WireOptions::Anthropic(_) => { - let mut cursor = anthropic_compiler - .as_ref() - .expect("Anthropic compiler") - .open(recipe, budget.clone()) - .await?; - while let Some(item) = cursor.next().await? { - sink.item(message_index, item, budget.clone()).await?; + WireOptions::Anthropic(_) => { + let mut cursor = anthropic_compiler + .as_ref() + .expect("Anthropic compiler") + .open(recipe, budget.clone()) + .await?; + while let Some(item) = cursor.next().await? { + sink.item(message_index, item, budget.clone()).await?; + } } } + sink.end(message_index, role, before).await?; } - sink.end(message_index, role, before).await?; - } - Ok(sink) + Ok(sink) + }) } + pub struct WirePlan { pub(super) encoding: WireEncoding, source: Arc, @@ -561,13 +564,13 @@ impl WirePlan { counts: Counts::default(), expected: self.counts, }); - let Sink::Backed(build) = Box::pin(emit( + let Sink::Backed(build) = emit( &self.encoding, self.source.as_ref(), work.budget(), sink, self.projection, - )) + ) .await? else { unreachable!("backed encoding pass") @@ -605,14 +608,13 @@ pub(super) async fn count( ) -> Result { let work = resources.acquire_work().await.map_err(error)?; let budget = work.budget(); - // Keep the codec's large async state out of the enclosing request futures. - let Sink::Count(counts) = Box::pin(emit( + let Sink::Count(counts) = emit( &encoding, source.as_ref(), budget, Sink::Count(Counts::default()), projection, - )) + ) .await? else { unreachable!("count pass") @@ -665,13 +667,13 @@ pub(super) async fn build( counts: Counts::default(), expected: count, }; - let Sink::Build(mut build) = Box::pin(emit( + let Sink::Build(mut build) = emit( &encoding, source.as_ref(), budget, Sink::Build(build), projection, - )) + ) .await? else { unreachable!("build pass")