Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/apps/desktop/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1976,7 +1976,7 @@ async fn init_agentic_system() -> anyhow::Result<(
let persistence_manager = Arc::new(persistence::PersistenceManager::new(path_manager.clone())?);

let context_store = Arc::new(session::SessionContextStore::new());
let context_compressor = Arc::new(session::ContextCompressor::new(Default::default()));
let context_compressor = Arc::new(session::ContextCompressor::new());

let session_manager = Arc::new(session::SessionManager::new(
context_store,
Expand Down
70 changes: 34 additions & 36 deletions src/crates/adapters/ai-adapters/src/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -22,18 +22,13 @@ use crate::types::*;
use anyhow::Result;
use format::ApiFormat;
use log::warn;
use openbitfun_core_types::errors::{AiProviderError, ErrorCategory};
use openbitfun_core_types::errors::AiProviderError;
use reqwest::Client;
use std::time::Duration;
use tokio::sync::mpsc;

const SEND_MESSAGE_STREAM_ATTEMPTS: usize = 10;
const SEND_MESSAGE_STREAM_ATTEMPTS: usize = openbitfun_agent_stream::retry::MAX_MODEL_ATTEMPTS;
const TEST_CONNECTION_STREAM_ATTEMPTS: usize = 5;
const SEND_MESSAGE_RETRY_BASE_DELAY_MS: u64 = 500;
const SEND_MESSAGE_RATE_LIMIT_RETRY_BASE_DELAY_MS: u64 = 2_000;
const SEND_MESSAGE_MAX_EXPONENTIAL_DELAY_MS: u64 = 30_000;
const SEND_MESSAGE_MAX_RATE_LIMIT_DELAY_MS: u64 = 60_000;
const SEND_MESSAGE_MAX_RETRY_EXPONENT_SHIFT: u32 = 6;

/// Streamed response result with the parsed stream and optional raw SSE receiver.
pub struct StreamResponse {
Expand Down Expand Up @@ -360,6 +355,37 @@ impl AIClient {
.await
}

/// Aggregate exactly one provider attempt; the caller owns retries and validation.
pub async fn send_message_once_with_trace_and_request_context(
&self,
messages: Vec<Message>,
tools: Option<Vec<ToolDefinition>>,
request_context: Option<ModelRequestContext>,
trace: Option<ModelExchangeTraceConfig>,
) -> Result<GeminiResponse> {
let response = self
.send_message_stream_once_with_request_context(
messages,
tools,
request_context,
trace.clone(),
)
.await?;
let handle = response.trace_handle.clone();
match response_aggregator::aggregate_stream_response_preserving_tool_presence(response)
.await
{
Ok(response) => {
complete_aggregated_trace(trace.as_ref(), handle.as_ref(), &response).await;
Ok(response)
}
Err(error) => {
fail_aggregated_trace(trace.as_ref(), handle.as_ref(), &error.to_string()).await;
Err(error)
}
}
}

pub async fn send_message_with_extra_body_and_trace(
&self,
messages: Vec<Message>,
Expand Down Expand Up @@ -578,35 +604,7 @@ fn send_message_retry_delay_ms_with_provider(
error_message: &str,
provider_error: Option<&AiProviderError>,
) -> u64 {
let shift = u32::try_from(attempt_index)
.unwrap_or(u32::MAX)
.min(SEND_MESSAGE_MAX_RETRY_EXPONENT_SHIFT);
let msg = error_message.to_lowercase();
let is_rate_limit = provider_error
.is_some_and(|error| error.category == ErrorCategory::RateLimit)
|| msg.contains("429")
|| msg.contains("rate limit")
|| msg.contains("too many requests");

let fallback = if is_rate_limit {
SEND_MESSAGE_RATE_LIMIT_RETRY_BASE_DELAY_MS
.saturating_mul(1u64 << shift)
.min(SEND_MESSAGE_MAX_RATE_LIMIT_DELAY_MS)
} else {
SEND_MESSAGE_RETRY_BASE_DELAY_MS
.saturating_mul(1u64 << shift)
.min(SEND_MESSAGE_MAX_EXPONENTIAL_DELAY_MS)
};

match provider_error.and_then(|error| error.retry_after_ms) {
Some(retry_after_ms) if is_rate_limit => retry_after_ms
.max(fallback)
.min(SEND_MESSAGE_MAX_RATE_LIMIT_DELAY_MS),
Some(retry_after_ms) if retry_after_ms > 0 => {
retry_after_ms.min(SEND_MESSAGE_MAX_RATE_LIMIT_DELAY_MS)
}
Some(_) | None => fallback,
}
openbitfun_agent_stream::retry::delay_ms(attempt_index, error_message, provider_error)
}

async fn complete_aggregated_trace(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,22 @@ use super::StreamResponse;
pub(crate) async fn aggregate_stream_response(
stream_response: StreamResponse,
) -> Result<GeminiResponse> {
aggregate_stream_response_inner(stream_response, false).await
}

/// Preserve even malformed tool-call presence so summary callers cannot accept
/// text accompanied by tool deltas that the aggregator could not reconstruct.
pub(crate) async fn aggregate_stream_response_preserving_tool_presence(
stream_response: StreamResponse,
) -> Result<GeminiResponse> {
aggregate_stream_response_inner(stream_response, true).await
}

async fn aggregate_stream_response_inner(
stream_response: StreamResponse,
preserve_tool_presence: bool,
) -> Result<GeminiResponse> {
let mut saw_tool_call = false;
let mut stream = stream_response.stream;

let mut full_text = String::new();
Expand Down Expand Up @@ -46,6 +62,7 @@ pub(crate) async fn aggregate_stream_response(
}

if let Some(tool_call) = tool_call {
saw_tool_call = true;
let crate::stream::UnifiedToolCall {
tool_call_index,
id,
Expand Down Expand Up @@ -153,7 +170,8 @@ pub(crate) async fn aggregate_stream_response(
Ok(GeminiResponse {
text: full_text,
reasoning_content: (!full_reasoning.is_empty()).then_some(full_reasoning),
tool_calls: (!tool_calls.is_empty()).then_some(tool_calls),
tool_calls: (!tool_calls.is_empty() || (preserve_tool_presence && saw_tool_call))
.then_some(tool_calls),
usage,
finish_reason,
provider_metadata,
Expand Down
2 changes: 1 addition & 1 deletion src/crates/assembly/core/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -468,7 +468,7 @@ ssh-remote = [
tokio-tungstenite = { workspace = true }
openbitfun-runtime-services = { path = "../../execution/runtime-services", features = ["test-support"] }
tempfile = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
tokio = { workspace = true, features = ["macros", "rt-multi-thread", "test-util"] }

[[test]]
name = "git_contracts"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15095,9 +15095,9 @@ mod tests {
use crate::agentic::goal_mode::thread_goal_patch;
use crate::agentic::persistence::PersistenceManager;
use crate::agentic::session::{
compression::{CompressionConfig, ContextCompressor},
PromptCachePolicy, SessionContextStore, SessionManager, SessionManagerConfig,
SystemPromptCacheIdentity, UserContextCacheIdentity, TEST_MODEL_RESOLUTION_AI_CONFIG,
compression::ContextCompressor, PromptCachePolicy, SessionContextStore, SessionManager,
SessionManagerConfig, SystemPromptCacheIdentity, UserContextCacheIdentity,
TEST_MODEL_RESOLUTION_AI_CONFIG,
};
use crate::agentic::skill_agent_snapshot::SkillSnapshotEntry;
use crate::agentic::tools::framework::{
Expand Down Expand Up @@ -16722,7 +16722,7 @@ mod tests {
)),
event_queue.clone(),
session_manager.clone(),
Arc::new(ContextCompressor::new(CompressionConfig::default())),
Arc::new(ContextCompressor::new()),
ExecutionEngineConfig::default(),
));
let coordinator = ConversationCoordinator::new_with_coordination_database_file(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3148,7 +3148,7 @@ mod tests {
};
use crate::agentic::persistence::PersistenceManager;
use crate::agentic::session::{
compression::{CompressionConfig, ContextCompressor},
compression::ContextCompressor,
revert::{SessionRevertPhase, SessionRevertState, SESSION_REVERT_SCHEMA_VERSION},
PromptCachePolicy, SessionContextStore, SessionManagerConfig,
TEST_MODEL_RESOLUTION_AI_CONFIG,
Expand Down Expand Up @@ -3259,7 +3259,7 @@ mod tests {
)),
event_queue.clone(),
session_manager.clone(),
Arc::new(ContextCompressor::new(CompressionConfig::default())),
Arc::new(ContextCompressor::new()),
ExecutionEngineConfig::default(),
));
let coordinator = Arc::new(ConversationCoordinator::new(
Expand Down
9 changes: 9 additions & 0 deletions src/crates/assembly/core/src/agentic/execution/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,15 @@ to name a trusted Bash 4+ executable and uses only isolated synthetic commands.

For automatic/manual context compaction cancellation, preparation, and commit races, use:

Each fixed summary plan has ten total model attempts, shared across transport,
stream aggregation, empty summaries, and tool-call responses. Context overflow
immediately starts a smaller plan with a fresh budget (at most four plans).
Both summary and ordinary model requests use the provider-neutral policy in
`openbitfun-agent-stream::retry` and single-attempt adapter entry points.
Failed compression terminates the turn without replacing context; there is no
locally reconstructed summary fallback. Historical compression payloads remain
readable.

```bash
cargo test --locked -p openbitfun-core --no-default-features --features agent-runtime,git --lib compression
cargo test --locked -p openbitfun-core --no-default-features --features agent-runtime,git --lib compaction
Expand Down
Loading
Loading