fmq/AstroResearch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 大文件模块化拆分、自进化Skill管线、前端设计系统统一

Browse Source 
后端：
  - runtime: 拆分 AgentConfig/events/duplicate_detector 为独立模块
  - error_recovery: 1499行单体拆为 classification(21种FailoverReason)/overflow/mod
  - executor: 提取 helpers.rs (PreparedCall/ToolExecutionResult/execute_single_tool)
  - skills: 新增 SkillCreator + SelfImprovePipeline（模式检测→自动生成SKILL.md→质量审查）
  - clients/llm: 拆分为 chat/embedding/types 三个子模块
  - services/download: 1548行拆为 mod/headers(反爬+SSRF)/strategies(多级回退)
  - services/batch/asset: 1264行拆为 mod/helpers/process

  前端：
  - 设计系统统一: sky/indigo → blueprint 色系, rounded-xl→lg, shadow-lg→sm
  - 删除 Vite 模板残留 App.css
  - GlobalDialog/PaperDetailModal/UncachedPaperModal 提取公共 BaseModal 组件
This commit is contained in:
fmq
2026-06-27 09:56:36 +08:00
parent 5db4cc5998
commit 5f2d2d83f6
62 changed files with 6584 additions and 6308 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agent/runtime/config.rs
+93
View File
@@ -0,0 +1,93 @@
// src/agent/runtime/config.rs
//
// Agent 运行时配置参数。
/// Agent 配置参数
#[derive(Debug, Clone)]
pub struct AgentConfig {
/// 最大 ReAct 迭代次数
pub max_steps: usize,
/// 同质调用检测阈值（连续相同调用次数）
pub duplicate_call_threshold: usize,
/// 工具执行超时时间（秒）
pub tool_timeout_secs: u64,
/// 工具输出最大字符数
pub max_tool_output_chars: usize,
/// Token 预算软限制 — 各压缩层统一触发阈值
pub token_soft_limit: usize,
/// Token 预算硬限制（触发强制动作）
pub token_hard_limit: usize,
/// 最大消息数（超过此阈值触发 snip_compact 层压缩）
pub max_messages: usize,
/// 是否启用 LLM 思考模式（前端可控，默认关闭）
pub enable_thinking: bool,
/// 权限拒绝规则（逗号分隔，格式: ToolName 或 ToolName(content_pattern)）
pub permission_deny_rules: Vec<String>,
/// 权限允许规则（逗号分隔）
pub permission_allow_rules: Vec<String>,
/// 权限询问规则（逗号分隔）
pub permission_ask_rules: Vec<String>,
/// 权限模式: "default" | "accept_edits" | "bypass" | "dont_ask"
pub permission_mode: String,
/// 拒绝追踪：连续拒绝上限（默认 3）
pub denial_max_consecutive: usize,
/// 拒绝追踪：总拒绝上限（默认 20）
pub denial_max_total: usize,
/// 附加允许目录（逗号分隔，扩展文件沙箱范围）
pub additional_allowed_dirs: Vec<String>,
/// 子代理工具白名单（逗号分隔，空=全部工具可用）
pub subagent_allowed_tools: Vec<String>,
/// Agent 运行模式 ID（"default" / "deep-research" / "literature-reader"）
pub mode: String,
}
impl AgentConfig {
/// 从环境变量加载配置，缺失时使用默认值。
pub fn from_env_optional() -> Self {
AgentConfig {
max_steps: 8,
duplicate_call_threshold: 3,
tool_timeout_secs: 120,
max_tool_output_chars: 4000,
token_soft_limit: std::env::var("AGENT_TOKEN_SOFT_LIMIT")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(80000),
token_hard_limit: std::env::var("AGENT_TOKEN_HARD_LIMIT")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(100000),
max_messages: 50,
enable_thinking: false,
permission_deny_rules: parse_comma_list("AGENT_PERMISSIONS_DENY"),
permission_allow_rules: parse_comma_list("AGENT_PERMISSIONS_ALLOW"),
permission_ask_rules: parse_comma_list("AGENT_PERMISSIONS_ASK"),
permission_mode: std::env::var("AGENT_PERMISSION_MODE")
.unwrap_or_else(|_| "default".to_string()),
denial_max_consecutive: 3,
denial_max_total: 20,
additional_allowed_dirs: parse_comma_list("AGENT_ADDITIONAL_DIRS"),
subagent_allowed_tools: parse_comma_list("AGENT_SUBAGENT_ALLOWED_TOOLS"),
mode: "default".to_string(),
}
}
}
/// 解析逗号分隔的环境变量为字符串列表
fn parse_comma_list(env_key: &str) -> Vec<String> {
std::env::var(env_key)
.ok()
.map(|v| {
v.split(',')
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect()
})
.unwrap_or_default()
}
impl Default for AgentConfig {
fn default() -> Self {
Self::from_env_optional()
}
}
src/agent/runtime/duplicate_detector.rs
+27
View File
@@ -0,0 +1,27 @@
// src/agent/runtime/duplicate_detector.rs
//
// 同质调用检测器：检测连续重复的工具调用，防止死循环。
/// 同质调用检测器
#[derive(Debug, Default)]
pub struct DuplicateDetector {
last_call: Option<(String, String)>, // (tool_name, arguments)
consecutive_count: usize,
}
impl DuplicateDetector {
/// 记录一次调用，返回是否检测到死循环
pub fn record(&mut self, tool_name: &str, arguments: &str, threshold: usize) -> bool {
let key = (tool_name.to_string(), arguments.to_string());
if self.last_call.as_ref() == Some(&key) {
self.consecutive_count += 1;
if self.consecutive_count >= threshold {
return true;
}
} else {
self.last_call = Some(key);
self.consecutive_count = 1;
}
false
}
}
src/agent/runtime/error_recovery.rs
-1499
View File
File diff suppressed because it is too large Load Diff
src/agent/runtime/error_recovery/classification.rs
+624
View File
@@ -0,0 +1,624 @@
// src/agent/runtime/error_recovery/classification.rs
//
// 故障转移原因分类引擎：FailoverReason / ErrorKind / classify_* 函数族。
use tracing::{error, info};
// ── FailoverReason：21 种错误分类 ──
/// 故障转移原因 — 参考 Hermes FailoverReason 枚举。
/// 每个 variant 携带是否可重试、是否应触发凭据切换等信息。
#[derive(Debug, Clone, PartialEq)]
pub enum FailoverReason {
// ── Auth 类 ──
/// 认证失败（401），可切换凭据重试
Auth,
/// 永久认证失败，不应重试
AuthPermanent,
// ── Billing 类 ──
/// 计费问题（402/403 billing），可切换账号
Billing,
/// 限流（429），应退避重试
RateLimited,
// ── Server 类 ──
/// 服务过载（503/529），应退避重试
Overloaded,
/// 服务器错误（500/502），可重试
ServerError,
// ── Transport 类 ──
/// 网络超时，可重试
Timeout,
// ── Context/Payload 类 ──
/// 上下文溢出（prompt too long / 413）
ContextOverflow,
/// 载荷过大
PayloadTooLarge,
/// 图片过大
ImageTooLarge,
// ── Model/Provider Policy 类 ──
/// 模型不存在或无权访问（404）
ModelNotFound,
/// Provider 策略阻止
ProviderPolicyBlocked,
/// 内容策略阻止（安全过滤）
ContentPolicyBlocked,
// ── Format 类 ──
/// 响应格式错误
FormatError,
/// 无效加密内容
InvalidEncryptedContent,
/// 多模态工具内容不支持
MultimodalToolContentUnsupported,
// ── Provider-specific 类 ──
/// Thinking 签名错误（Qwen/DashScope specific）
ThinkingSignature,
/// 长上下文 tier 门控
LongContextTier,
/// OAuth 长上下文 beta 禁止
OauthLongContextBetaForbidden,
// ── 网络/传输 ──
/// SSL/TLS 瞬态错误
SslTransient,
// ── Catch-all ──
/// 未知错误
Unknown(String),
}
impl FailoverReason {
/// 是否可重试（auth/rate-limit/server/transport 类）
pub fn is_retryable(&self) -> bool {
matches!(
self,
FailoverReason::Auth
| FailoverReason::RateLimited
| FailoverReason::Overloaded
| FailoverReason::ServerError
| FailoverReason::Timeout
| FailoverReason::SslTransient
)
}
/// 是否应触发上下文压缩
pub fn should_compress(&self) -> bool {
matches!(
self,
FailoverReason::ContextOverflow
| FailoverReason::PayloadTooLarge
| FailoverReason::LongContextTier
)
}
/// 是否应切换凭据/模型
pub fn should_failover(&self) -> bool {
matches!(
self,
FailoverReason::Auth
| FailoverReason::Billing
| FailoverReason::ModelNotFound
| FailoverReason::ProviderPolicyBlocked
)
}
/// 是否为永久性错误（不应重试）
pub fn is_permanent(&self) -> bool {
matches!(
self,
FailoverReason::AuthPermanent
| FailoverReason::ContentPolicyBlocked
| FailoverReason::InvalidEncryptedContent
)
}
/// 错误归因：用户侧还是服务侧
pub fn is_user_error(&self) -> bool {
matches!(
self,
FailoverReason::ContextOverflow
| FailoverReason::PayloadTooLarge
| FailoverReason::ImageTooLarge
| FailoverReason::FormatError
)
}
}
impl std::fmt::Display for FailoverReason {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let label = match self {
FailoverReason::Auth => "auth",
FailoverReason::AuthPermanent => "auth_permanent",
FailoverReason::Billing => "billing",
FailoverReason::RateLimited => "rate_limited",
FailoverReason::Overloaded => "overloaded",
FailoverReason::ServerError => "server_error",
FailoverReason::Timeout => "timeout",
FailoverReason::ContextOverflow => "context_overflow",
FailoverReason::PayloadTooLarge => "payload_too_large",
FailoverReason::ImageTooLarge => "image_too_large",
FailoverReason::ModelNotFound => "model_not_found",
FailoverReason::ProviderPolicyBlocked => "provider_policy_blocked",
FailoverReason::ContentPolicyBlocked => "content_policy_blocked",
FailoverReason::FormatError => "format_error",
FailoverReason::InvalidEncryptedContent => "invalid_encrypted_content",
FailoverReason::MultimodalToolContentUnsupported => {
"multimodal_tool_content_unsupported"
}
FailoverReason::ThinkingSignature => "thinking_signature",
FailoverReason::LongContextTier => "long_context_tier",
FailoverReason::OauthLongContextBetaForbidden => "oauth_long_context_beta_forbidden",
FailoverReason::SslTransient => "ssl_transient",
FailoverReason::Unknown(s) => return write!(f, "unknown({})", s),
};
write!(f, "{}", label)
}
}
// ── 保留向后兼容的 ErrorKind 别名 ──
/// 错误类型分类（向后兼容简化版，新代码应使用 FailoverReason）
#[derive(Debug, Clone, PartialEq)]
pub enum ErrorKind {
PromptTooLong,
TokenExhausted,
ModelError(String),
Timeout,
RateLimited,
Overloaded,
}
impl From<&FailoverReason> for ErrorKind {
fn from(reason: &FailoverReason) -> Self {
match reason {
FailoverReason::ContextOverflow | FailoverReason::PayloadTooLarge => {
ErrorKind::PromptTooLong
}
FailoverReason::LongContextTier => ErrorKind::TokenExhausted,
FailoverReason::RateLimited => ErrorKind::RateLimited,
FailoverReason::Overloaded | FailoverReason::ServerError => ErrorKind::Overloaded,
FailoverReason::Timeout | FailoverReason::SslTransient => ErrorKind::Timeout,
other => ErrorKind::ModelError(other.to_string()),
}
}
}
// ── 8 步分类管线 ──
/// Step 1: Provider-specific 模式匹配（最高优先级）
fn classify_provider_specific(error_str: &str) -> Option<FailoverReason> {
let lower = error_str.to_lowercase();
// Content policy blocks (非 provider 特定，但在 status 分类前检查)
if lower.contains("content_policy_violation")
|| lower.contains("safety filter")
|| lower.contains("content filter")
|| lower.contains("responsible_ai")
|| lower.contains("content management policy")
{
return Some(FailoverReason::ContentPolicyBlocked);
}
// Thinking signature error (Qwen/DashScope)
if lower.contains("thinking_signature") || lower.contains("thinking signature") {
return Some(FailoverReason::ThinkingSignature);
}
// Long context tier gate
if lower.contains("long_context_tier")
|| lower.contains("long context tier")
|| lower.contains("context_length_exceeded_for_tier")
{
return Some(FailoverReason::LongContextTier);
}
// OAuth 1M beta forbidden
if lower.contains("oauth_long_context") || lower.contains("1m_beta_forbidden") {
return Some(FailoverReason::OauthLongContextBetaForbidden);
}
// Llama.cpp grammar error
if lower.contains("grammar_pattern") || lower.contains("llama_cpp_grammar") {
return Some(FailoverReason::FormatError);
}
// Multimodal tool content
if lower.contains("multimodal_tool_content") || lower.contains("tool content not supported") {
return Some(FailoverReason::MultimodalToolContentUnsupported);
}
None
}
/// Step 2: HTTP Status Code 分类
fn classify_by_http_status(error_str: &str) -> Option<FailoverReason> {
let lower = error_str.to_lowercase();
// 401 → auth (temporary)
if lower.contains("401") || lower.contains("unauthorized") {
return Some(FailoverReason::Auth);
}
// 402 → billing vs rate-limit disambiguation
if lower.contains("402") {
if lower.contains("rate") || lower.contains("limit") || lower.contains("try again") {
return Some(FailoverReason::RateLimited);
}
return Some(FailoverReason::Billing);
}
// 403
if lower.contains("403") || lower.contains("forbidden") {
if lower.contains("billing") || lower.contains("quota") || lower.contains("insufficient") {
return Some(FailoverReason::Billing);
}
if lower.contains("policy") || lower.contains("blocked") {
return Some(FailoverReason::ProviderPolicyBlocked);
}
return Some(FailoverReason::AuthPermanent);
}
// 404
if lower.contains("404") || lower.contains("not found") {
if lower.contains("model") || lower.contains("deployment") {
return Some(FailoverReason::ModelNotFound);
}
if lower.contains("billing") || lower.contains("subscription") {
return Some(FailoverReason::Billing);
}
if lower.contains("policy") || lower.contains("blocked") {
return Some(FailoverReason::ProviderPolicyBlocked);
}
}
// 413 → context overflow (default), could also be payload_too_large
if lower.contains("413") {
if lower.contains("image") || lower.contains("media") {
return Some(FailoverReason::ImageTooLarge);
}
return Some(FailoverReason::ContextOverflow);
}
// 429 → rate limit (definitive)
if lower.contains("429") {
return Some(FailoverReason::RateLimited);
}
// 500/502 → server error
if lower.contains("\"500\"")
|| lower.contains(" 500 ")
|| lower.contains("502")
|| lower.contains("internal server error")
|| lower.contains("bad gateway")
{
// Check for request validation signals
if lower.contains("invalid_request")
|| lower.contains("unsupported_parameter")
|| lower.contains("unknown_parameter")
{
return Some(FailoverReason::FormatError);
}
return Some(FailoverReason::ServerError);
}
// 503/529 → overloaded
if lower.contains("503") || lower.contains("529") || lower.contains("service unavailable") {
return Some(FailoverReason::Overloaded);
}
None
}
/// Step 3: Structured Error Code 分类
fn classify_by_error_code(error_str: &str) -> Option<FailoverReason> {
let lower = error_str.to_lowercase();
// Structured error codes from API responses
if lower.contains("\"code\"") || lower.contains("\"error_type\"") {
if lower.contains("resource_exhausted") {
return Some(FailoverReason::RateLimited);
}
if lower.contains("insufficient_quota") {
return Some(FailoverReason::Billing);
}
if lower.contains("model_not_found") || lower.contains("deployment_not_found") {
return Some(FailoverReason::ModelNotFound);
}
if lower.contains("invalid_api_key") || lower.contains("token_expired") {
return Some(FailoverReason::Auth);
}
if lower.contains("context_length_exceeded") {
return Some(FailoverReason::ContextOverflow);
}
}
None
}
/// Step 4: Message Pattern 匹配（14 组模式集）
fn classify_by_message_pattern(error_str: &str) -> Option<FailoverReason> {
let lower = error_str.to_lowercase();
// 4a. Billing patterns
if lower.contains("insufficient_quota")
|| lower.contains("billing account")
|| lower.contains("payment required")
|| lower.contains("quota exceeded")
|| lower.contains("run out of")
|| lower.contains("no credit")
|| lower.contains("check your balance")
{
return Some(FailoverReason::Billing);
}
// 4b. Rate-limit patterns
if lower.contains("rate limit")
|| lower.contains("rate_limit")
|| lower.contains("too many requests")
|| lower.contains("requests too frequent")
|| lower.contains("try again in")
|| lower.contains("resets at")
|| lower.contains("please slow down")
{
return Some(FailoverReason::RateLimited);
}
// 4c. Context overflow patterns
if lower.contains("prompt_too_long")
|| lower.contains("prompt too long")
|| lower.contains("context length")
|| lower.contains("context_length_exceeded")
|| lower.contains("input length")
|| lower.contains("maximum context")
|| lower.contains("reduce the length")
|| lower.contains("too many tokens")
|| lower.contains("max_tokens_exceeded")
{
return Some(FailoverReason::ContextOverflow);
}
// 4d. Auth patterns
if lower.contains("invalid api key")
|| lower.contains("invalid_api_key")
|| lower.contains("incorrect api key")
|| lower.contains("authentication failed")
|| lower.contains("auth error")
|| lower.contains("not authenticated")
|| lower.contains("token is invalid")
{
return Some(FailoverReason::Auth);
}
// 4e. Payload size
if lower.contains("request too large")
|| lower.contains("payload too large")
|| lower.contains("request size exceeds")
|| lower.contains("request entity too large")
{
return Some(FailoverReason::PayloadTooLarge);
}
// 4f. Image size
if lower.contains("image too large")
|| lower.contains("image size exceeds")
|| lower.contains("max image size")
|| lower.contains("image_resolution_exceeded")
{
return Some(FailoverReason::ImageTooLarge);
}
// 4g. Multimodal tool content
if lower.contains("tool content not supported") || lower.contains("multimodal content in tool")
{
return Some(FailoverReason::MultimodalToolContentUnsupported);
}
// 4h. Timeout patterns
if lower.contains("request timed out")
|| lower.contains("timed out")
|| lower.contains("deadline exceeded")
|| lower.contains("idle timeout")
|| lower.contains("connection timeout")
{
return Some(FailoverReason::Timeout);
}
// 4i. Invalid encrypted content
if lower.contains("invalid encrypted content") || lower.contains("encrypted_content_error") {
return Some(FailoverReason::InvalidEncryptedContent);
}
// 4j. Format error
if lower.contains("invalid_request")
&& (lower.contains("parameter") || lower.contains("format") || lower.contains("schema"))
{
return Some(FailoverReason::FormatError);
}
None
}
/// Step 5: SSL/TLS 瞬态错误检测
fn classify_ssl_transient(error_str: &str) -> Option<FailoverReason> {
let lower = error_str.to_lowercase();
if lower.contains("ssl") || lower.contains("tls") || lower.contains("certificate") {
// Temporary TLS errors (connection reset, handshake failure)
if lower.contains("reset") || lower.contains("handshake") || lower.contains("timeout") {
return Some(FailoverReason::SslTransient);
}
// Permanent TLS errors (certificate expired, hostname mismatch)
if lower.contains("expired") || lower.contains("hostname") || lower.contains("verify") {
return Some(FailoverReason::AuthPermanent);
}
return Some(FailoverReason::SslTransient);
}
None
}
/// Step 6: Server disconnect + large session → context overflow
fn classify_disconnect_with_large_session(
error_str: &str,
estimated_tokens: Option<usize>,
context_length: Option<usize>,
) -> Option<FailoverReason> {
let lower = error_str.to_lowercase();
// Server disconnected without response
if lower.contains("connection reset")
|| lower.contains("eof")
|| lower.contains("incomplete response")
|| lower.contains("peer closed")
{
// If session is large (>60% of context), blame context overflow
if let (Some(tokens), Some(ctx_len)) = (estimated_tokens, context_length) {
let ratio = tokens as f64 / ctx_len as f64;
if ratio > 0.6 {
return Some(FailoverReason::ContextOverflow);
}
}
return Some(FailoverReason::Timeout);
}
None
}
/// Step 7: Transport error heuristics
fn classify_transport_error(error_str: &str) -> Option<FailoverReason> {
let lower = error_str.to_lowercase();
if lower.contains("connection")
|| lower.contains("network")
|| lower.contains("dns")
|| lower.contains("unreachable")
|| lower.contains("refused")
|| lower.contains("broken pipe")
|| lower.contains("transport")
|| lower.contains("eof")
|| lower.contains("reset by peer")
{
return Some(FailoverReason::Timeout);
}
None
}
/// Step 8: Unknown fallback
fn classify_unknown(error_str: &str) -> FailoverReason {
FailoverReason::Unknown(error_str.to_string())
}
// ── 主分类函数 ──
/// 8步分类管线：按优先级依次尝试各分类器。
///
/// 可选传入 `estimated_tokens` 和 `context_length` 用于
/// disconnect + large session 的上下文溢出检测。
pub fn classify_failover(
error_str: &str,
estimated_tokens: Option<usize>,
context_length: Option<usize>,
) -> FailoverReason {
// Step 1: Provider-specific (HIGHEST priority)
if let Some(reason) = classify_provider_specific(error_str) {
info!("[ErrorClassify] Step 1 (provider-specific): {:?}", reason);
return reason;
}
// Step 2: HTTP status code
if let Some(reason) = classify_by_http_status(error_str) {
info!("[ErrorClassify] Step 2 (HTTP status): {:?}", reason);
return reason;
}
// Step 3: Structured error code
if let Some(reason) = classify_by_error_code(error_str) {
info!("[ErrorClassify] Step 3 (error code): {:?}", reason);
return reason;
}
// Step 4: Message patterns (14 groups)
if let Some(reason) = classify_by_message_pattern(error_str) {
info!("[ErrorClassify] Step 4 (message pattern): {:?}", reason);
return reason;
}
// Step 5: SSL/TLS transient
if let Some(reason) = classify_ssl_transient(error_str) {
info!("[ErrorClassify] Step 5 (SSL/TLS): {:?}", reason);
return reason;
}
// Step 6: Server disconnect + large session
if let Some(reason) =
classify_disconnect_with_large_session(error_str, estimated_tokens, context_length)
{
info!(
"[ErrorClassify] Step 6 (disconnect + large session): {:?}",
reason
);
return reason;
}
// Step 7: Transport heuristics
if let Some(reason) = classify_transport_error(error_str) {
info!("[ErrorClassify] Step 7 (transport): {:?}", reason);
return reason;
}
// Step 8: Unknown fallback
error!(
"[ErrorClassify] Step 8 (unknown): unclassified error: {}",
error_str
);
classify_unknown(error_str)
}
// ── 向后兼容的 classify_error ──
/// 从错误字符串分类错误类型。
/// 内部调用 classify_failover 并转换为简化的 ErrorKind。
pub fn classify_error(error_str: &str) -> ErrorKind {
let reason = classify_failover(error_str, None, None);
ErrorKind::from(&reason)
}
// ── 恢复建议生成 ──
/// 根据 FailoverReason 生成面向用户的恢复建议。
pub fn recovery_suggestion(reason: &FailoverReason) -> &'static str {
match reason {
FailoverReason::Auth => "认证失败，尝试切换 API Key 或刷新凭据。",
FailoverReason::AuthPermanent => "认证凭据永久失效，请检查 API Key 配置。",
FailoverReason::Billing => "账户余额不足或配额用尽，请检查计费状态。",
FailoverReason::RateLimited => "请求频率超限，系统正在自动退避重试。",
FailoverReason::Overloaded => "模型服务过载，系统正在自动重试。",
FailoverReason::ServerError => "服务器内部错误，将自动重试。",
FailoverReason::Timeout => "请求超时，将自动重试。请检查网络连接。",
FailoverReason::ContextOverflow => "上下文超过限制，正在自动压缩对话历史。",
FailoverReason::PayloadTooLarge => "请求载荷过大，请尝试减少工具调用数量。",
FailoverReason::ImageTooLarge => "图片过大，请使用更小的图片或降低分辨率。",
FailoverReason::ModelNotFound => "模型不可用，请检查模型名称或尝试切换模型。",
FailoverReason::ProviderPolicyBlocked => "Provider 策略阻止了此请求。",
FailoverReason::ContentPolicyBlocked => {
"内容被安全策略过滤，请修改请求内容后重试。此错误不可自动恢复。"
}
FailoverReason::FormatError => "请求格式错误，请检查工具参数格式。",
FailoverReason::InvalidEncryptedContent => "加密内容无效，请重新生成。",
FailoverReason::MultimodalToolContentUnsupported => {
"当前模型不支持工具结果中的多模态内容，请改用纯文本。"
}
FailoverReason::ThinkingSignature => "思考模式签名错误，请禁用 thinking 模式或切换模型。",
FailoverReason::LongContextTier => "超出当前 tier 的上下文长度限制，正在压缩或升级 tier。",
FailoverReason::OauthLongContextBetaForbidden => "长上下文 Beta 功能未对当前凭据开放。",
FailoverReason::SslTransient => "SSL/TLS 连接暂时中断，将自动重试。",
FailoverReason::Unknown(_) => "未知错误，系统将尝试恢复。",
}
}
src/agent/runtime/error_recovery/mod.rs
+726
View File
@@ -0,0 +1,726 @@
// src/agent/runtime/error_recovery/mod.rs
//
// LLM 调用错误恢复引擎。
// 子模块：
// classification — FailoverReason / ErrorKind / classify_* 分类函数族
// overflow — 上下文溢出解析与安全 Token 计算
pub(crate) mod classification;
pub(crate) mod overflow;
use tracing::{info, warn};
use crate::agent::runtime::token_budget::TokenBudget;
// Re-export 公共类型，保持外部引用路径不变
pub use classification::{
classify_error, classify_failover, recovery_suggestion, ErrorKind, FailoverReason,
};
pub use overflow::{
calculate_safe_max_tokens, parse_context_overflow, parse_retry_after, ContextOverflowInfo,
};
#[derive(Debug, Clone, PartialEq)]
pub enum RecoveryStep {
/// 调整 max_tokens 以适配上下文容量（ContextOverflow 首选策略）。
/// 参考 Claude Code — 解析 "X + Y > Z" 错误后计算安全值。
AdjustMaxTokens { new_max_tokens: usize },
/// 尝试更激进的 micro_compact
AggressiveCompact,
/// 使用 LLM 摘要压缩
ReactiveCompact,
/// 提升 token 上限
EscalateTokens { new_hard_limit: usize },
/// 分轮恢复（注入 meta 消息）
MultiTurn,
/// 放弃，暴露错误
Surface,
/// 指数退避重试（用于 429/529 瞬态错误）
RetryWithBackoff { attempt: u32, delay_ms: u64 },
}
/// 恢复尝试追踪
#[derive(Debug, Clone)]
pub struct RecoveryAttempts {
pub aggressive_compact: bool,
pub reactive_compact: bool,
pub escalate_tokens: bool,
pub multi_turn: bool,
}
impl RecoveryAttempts {
pub fn new() -> Self {
RecoveryAttempts {
aggressive_compact: false,
reactive_compact: false,
escalate_tokens: false,
multi_turn: false,
}
}
/// 是否有未尝试的恢复步骤
pub fn has_remaining(&self) -> bool {
!self.aggressive_compact
|| !self.reactive_compact
|| !self.escalate_tokens
|| !self.multi_turn
}
/// 获取下一个应尝试的恢复步骤。
///
/// 对于 ContextOverflow 类错误，恢复优先级为：
/// 1. AdjustMaxTokens — 从错误消息解析容量，精确计算安全的 max_tokens（参考 Claude Code）
/// 2. AggressiveCompact — 激进微压缩
/// 3. ReactiveCompact — LLM 摘要压缩
/// 4. EscalateTokens — 提升 hard_limit
/// 5. MultiTurn — 分轮
pub fn next_step(
&mut self,
error_kind: &ErrorKind,
overflow_info: Option<&ContextOverflowInfo>,
) -> Option<RecoveryStep> {
if matches!(error_kind, ErrorKind::RateLimited | ErrorKind::Overloaded) {
return Some(RecoveryStep::RetryWithBackoff {
attempt: 0,
delay_ms: 500,
});
}
match error_kind {
ErrorKind::PromptTooLong | ErrorKind::TokenExhausted => {
// Step 0: 尝试精确调整 max_tokens（仅在有 Overflow info 时）
if let Some(info) = overflow_info {
if let Some(safe_tokens) = calculate_safe_max_tokens(info) {
// 仅在使用默认路径（尚未尝试过其他恢复方式）时触发
if !self.aggressive_compact
&& !self.reactive_compact
&& !self.escalate_tokens
{
return Some(RecoveryStep::AdjustMaxTokens {
new_max_tokens: safe_tokens,
});
}
}
}
// Step 1: 激进微压缩
if !self.aggressive_compact {
self.aggressive_compact = true;
return Some(RecoveryStep::AggressiveCompact);
}
// Step 2: LLM 摘要压缩
if !self.reactive_compact {
self.reactive_compact = true;
return Some(RecoveryStep::ReactiveCompact);
}
// Step 3: 提升硬限制
if !self.escalate_tokens {
self.escalate_tokens = true;
return Some(RecoveryStep::EscalateTokens {
new_hard_limit: 64_000,
});
}
// Step 4: 分轮
if !self.multi_turn {
self.multi_turn = true;
return Some(RecoveryStep::MultiTurn);
}
}
_ => {
if !self.multi_turn {
self.multi_turn = true;
return Some(RecoveryStep::Surface);
}
}
}
None
}
/// 向后兼容的无 overflow_info 版本。
pub fn next_step_compat(&mut self, error_kind: &ErrorKind) -> Option<RecoveryStep> {
self.next_step(error_kind, None)
}
}
impl Default for RecoveryAttempts {
fn default() -> Self {
Self::new()
}
}
/// 错误恢复器
pub struct ErrorRecovery {
pub attempts: RecoveryAttempts,
pub token_budget: TokenBudget,
}
impl ErrorRecovery {
pub fn new(token_budget: TokenBudget) -> Self {
ErrorRecovery {
attempts: RecoveryAttempts::new(),
token_budget,
}
}
pub fn try_recover(
&mut self,
error_kind: &ErrorKind,
overflow_info: Option<&ContextOverflowInfo>,
) -> Option<RecoveryStep> {
let step = self.attempts.next_step(error_kind, overflow_info);
match &step {
Some(RecoveryStep::AdjustMaxTokens { new_max_tokens }) => {
info!(
"[ErrorRecovery] 尝试步骤 0/5: AdjustMaxTokens → {} (从错误消息计算)",
new_max_tokens
);
// 将硬限制下调至安全值，使下一次 LLM 调用能在容量内完成
self.token_budget.hard_limit = *new_max_tokens;
}
Some(RecoveryStep::AggressiveCompact) => {
info!("[ErrorRecovery] 尝试步骤 1/5: AggressiveCompact");
}
Some(RecoveryStep::ReactiveCompact) => {
info!("[ErrorRecovery] 尝试步骤 2/5: ReactiveCompact");
}
Some(RecoveryStep::EscalateTokens { new_hard_limit }) => {
info!(
"[ErrorRecovery] 尝试步骤 3/5: EscalateTokens → {}",
new_hard_limit
);
self.token_budget.escalate_hard_limit(*new_hard_limit);
}
Some(RecoveryStep::RetryWithBackoff { attempt, delay_ms }) => {
info!(
"[ErrorRecovery] 退避重试: attempt={}, delay={}ms",
attempt, delay_ms
);
}
Some(RecoveryStep::MultiTurn) => {
info!("[ErrorRecovery] 尝试步骤 4/5: MultiTurn");
}
Some(RecoveryStep::Surface) => {
warn!("[ErrorRecovery] 无法恢复，暴露错误");
}
None => {
warn!("[ErrorRecovery] 所有恢复步骤已尝试完毕");
}
}
step
}
/// 向后兼容的无 overflow_info 版本。
pub fn try_recover_compat(&mut self, error_kind: &ErrorKind) -> Option<RecoveryStep> {
self.try_recover(error_kind, None)
}
pub fn multi_turn_message() -> String {
"由于 token 限制，当前回答被截断。请基于已收集的信息继续分析，\
重点关注尚未完成的部分。你可以：\n\
1. 总结已有发现\n\
2. 使用 compress_context 手动压缩上下文\n\
3. 分步完成剩余工作"
.to_string()
}
pub fn is_recoverable(error_kind: &ErrorKind) -> bool {
matches!(
error_kind,
ErrorKind::PromptTooLong
| ErrorKind::TokenExhausted
| ErrorKind::RateLimited
| ErrorKind::Overloaded
)
}
}
/// 计算指数退避延迟（毫秒）
pub fn backoff_delay(attempt: u32, retry_after_secs: Option<u64>) -> u64 {
if let Some(ra) = retry_after_secs {
return ra * 1000;
}
let base = 500u64 * 2u64.pow(attempt.min(6));
let base = base.min(32_000);
let jitter = (base / 4) * (attempt as u64 % 5) / 5;
base + jitter
}
#[cfg(test)]
mod tests {
use super::*;
// ── FailoverReason tests ──
#[test]
fn test_is_retryable() {
assert!(FailoverReason::Auth.is_retryable());
assert!(FailoverReason::RateLimited.is_retryable());
assert!(FailoverReason::Overloaded.is_retryable());
assert!(!FailoverReason::AuthPermanent.is_retryable());
assert!(!FailoverReason::ContentPolicyBlocked.is_retryable());
}
#[test]
fn test_should_compress() {
assert!(FailoverReason::ContextOverflow.should_compress());
assert!(FailoverReason::PayloadTooLarge.should_compress());
assert!(!FailoverReason::RateLimited.should_compress());
}
#[test]
fn test_should_failover() {
assert!(FailoverReason::Auth.should_failover());
assert!(FailoverReason::Billing.should_failover());
assert!(FailoverReason::ModelNotFound.should_failover());
assert!(!FailoverReason::Timeout.should_failover());
}
#[test]
fn test_is_permanent() {
assert!(FailoverReason::AuthPermanent.is_permanent());
assert!(FailoverReason::ContentPolicyBlocked.is_permanent());
assert!(!FailoverReason::Auth.is_permanent());
}
#[test]
fn test_is_user_error() {
assert!(FailoverReason::ContextOverflow.is_user_error());
assert!(FailoverReason::ImageTooLarge.is_user_error());
assert!(!FailoverReason::Timeout.is_user_error());
}
#[test]
fn test_failover_reason_display() {
assert_eq!(FailoverReason::Auth.to_string(), "auth");
assert_eq!(FailoverReason::RateLimited.to_string(), "rate_limited");
assert_eq!(
FailoverReason::Unknown("test".into()).to_string(),
"unknown(test)"
);
}
// ── Classification pipeline tests ──
#[test]
fn test_classify_rate_limited_429() {
let reason = classify_failover("HTTP 429: Too Many Requests", None, None);
assert_eq!(reason, FailoverReason::RateLimited);
}
#[test]
fn test_classify_overloaded_529() {
let reason = classify_failover("HTTP 529: Service Overloaded", None, None);
assert_eq!(reason, FailoverReason::Overloaded);
}
#[test]
fn test_classify_context_overflow() {
let reason = classify_failover(
"Error: prompt_too_long: context length exceeded",
None,
None,
);
assert_eq!(reason, FailoverReason::ContextOverflow);
}
#[test]
fn test_classify_auth_401() {
let reason = classify_failover("HTTP 401: Unauthorized - invalid api key", None, None);
assert_eq!(reason, FailoverReason::Auth);
}
#[test]
fn test_classify_billing_402() {
let reason = classify_failover("HTTP 402: insufficient_quota", None, None);
assert_eq!(reason, FailoverReason::Billing);
}
#[test]
fn test_classify_content_policy_blocked() {
let reason = classify_failover(
"content_policy_violation: responsible_ai filter triggered",
None,
None,
);
assert_eq!(reason, FailoverReason::ContentPolicyBlocked);
}
#[test]
fn test_classify_model_not_found() {
let reason = classify_failover(
"HTTP 404: model_not_found - deployment not found",
None,
None,
);
assert_eq!(reason, FailoverReason::ModelNotFound);
}
#[test]
fn test_classify_transport_timeout() {
let reason = classify_failover("Connection reset by peer", None, None);
// Connection reset with no session info → timeout
assert_eq!(reason, FailoverReason::Timeout);
}
#[test]
fn test_classify_disconnect_with_large_session() {
let reason = classify_failover(
"connection reset",
Some(8000), // estimated tokens
Some(10000), // context length (80% → >60%)
);
assert_eq!(reason, FailoverReason::ContextOverflow);
}
#[test]
fn test_classify_disconnect_with_small_session() {
let reason = classify_failover(
"connection reset",
Some(4000), // estimated tokens
Some(10000), // context length (40% → <60%)
);
assert_eq!(reason, FailoverReason::Timeout);
}
#[test]
fn test_classify_priorities_content_policy_before_status() {
// content_policy_violation should be detected before 400 status
let reason = classify_failover(
"HTTP 400: content_policy_violation - content management policy",
None,
None,
);
assert_eq!(reason, FailoverReason::ContentPolicyBlocked);
}
#[test]
fn test_classify_unknown_fallback() {
let reason = classify_failover("Some weird error that doesn't match anything", None, None);
assert!(matches!(reason, FailoverReason::Unknown(_)));
}
#[test]
fn test_classify_image_too_large() {
let reason = classify_failover("HTTP 413: image too large, max 20MB", None, None);
assert_eq!(reason, FailoverReason::ImageTooLarge);
}
#[test]
fn test_classify_provider_policy_blocked() {
let reason = classify_failover("HTTP 403: provider policy blocked this region", None, None);
assert_eq!(reason, FailoverReason::ProviderPolicyBlocked);
}
#[test]
fn test_classify_ssl_transient() {
let reason = classify_failover("SSL handshake timeout: connection reset", None, None);
assert_eq!(reason, FailoverReason::SslTransient);
}
// ── Backward compatibility tests ──
#[test]
fn test_classify_error_backward_compat() {
let kind = classify_error("HTTP 429: Too Many Requests");
assert_eq!(kind, ErrorKind::RateLimited);
}
#[test]
fn test_classify_error_529_backward_compat() {
let kind = classify_error("HTTP 529: Service Overloaded");
assert_eq!(kind, ErrorKind::Overloaded);
}
// ── Recovery step tests ──
#[test]
fn test_all_steps_sequence() {
let mut attempts = RecoveryAttempts::new();
assert_eq!(
attempts.next_step_compat(&ErrorKind::PromptTooLong),
Some(RecoveryStep::AggressiveCompact)
);
assert_eq!(
attempts.next_step_compat(&ErrorKind::PromptTooLong),
Some(RecoveryStep::ReactiveCompact)
);
assert_eq!(
attempts.next_step_compat(&ErrorKind::PromptTooLong),
Some(RecoveryStep::EscalateTokens {
new_hard_limit: 64_000
})
);
assert_eq!(
attempts.next_step_compat(&ErrorKind::PromptTooLong),
Some(RecoveryStep::MultiTurn)
);
assert_eq!(attempts.next_step_compat(&ErrorKind::PromptTooLong), None);
}
#[test]
fn test_model_error_goes_straight_to_surface() {
let mut attempts = RecoveryAttempts::new();
assert_eq!(
attempts.next_step_compat(&ErrorKind::ModelError("test".into())),
Some(RecoveryStep::Surface)
);
assert_eq!(
attempts.next_step_compat(&ErrorKind::ModelError("test".into())),
None
);
}
#[test]
fn test_has_remaining() {
let mut attempts = RecoveryAttempts::new();
assert!(attempts.has_remaining());
for _ in 0..4 {
attempts.next_step_compat(&ErrorKind::PromptTooLong);
}
assert!(!attempts.has_remaining());
}
#[test]
fn test_is_recoverable() {
assert!(ErrorRecovery::is_recoverable(&ErrorKind::PromptTooLong));
assert!(ErrorRecovery::is_recoverable(&ErrorKind::TokenExhausted));
assert!(ErrorRecovery::is_recoverable(&ErrorKind::RateLimited));
assert!(ErrorRecovery::is_recoverable(&ErrorKind::Overloaded));
assert!(!ErrorRecovery::is_recoverable(&ErrorKind::ModelError(
"test".into()
)));
assert!(!ErrorRecovery::is_recoverable(&ErrorKind::Timeout));
}
#[test]
fn test_parse_retry_after() {
let err = "HTTP 429: retry_after=Some(30)";
assert_eq!(parse_retry_after(err), Some(30));
}
#[test]
fn test_parse_retry_after_header_format() {
let err = "HTTP 429\nRetry-After: 60\n";
assert_eq!(parse_retry_after(err), Some(60));
}
#[test]
fn test_parse_retry_after_none() {
let err = "HTTP 500: Internal Server Error";
assert_eq!(parse_retry_after(err), None);
}
#[test]
fn test_backoff_delay() {
let d0 = backoff_delay(0, None);
assert!(d0 >= 500 && d0 <= 700);
let d3 = backoff_delay(3, None);
assert!(d3 >= 4000 && d3 <= 5000);
let d10 = backoff_delay(10, None);
assert!(d10 <= 40_000);
let d_ra = backoff_delay(0, Some(15));
assert_eq!(d_ra, 15000);
}
#[test]
fn test_rate_limited_goes_to_retry() {
let mut attempts = RecoveryAttempts::new();
let step = attempts.next_step_compat(&ErrorKind::RateLimited);
assert!(matches!(step, Some(RecoveryStep::RetryWithBackoff { .. })));
}
#[test]
fn test_token_budget_escalation() {
let budget = TokenBudget::new(32_000, 40_000);
let mut recovery = ErrorRecovery::new(budget);
assert_eq!(recovery.token_budget.hard_limit, 40_000);
recovery
.attempts
.next_step_compat(&ErrorKind::PromptTooLong); // aggressive
recovery
.attempts
.next_step_compat(&ErrorKind::PromptTooLong); // reactive
let step = recovery.try_recover_compat(&ErrorKind::PromptTooLong); // escalate
assert_eq!(
step,
Some(RecoveryStep::EscalateTokens {
new_hard_limit: 64_000
})
);
assert_eq!(recovery.token_budget.hard_limit, 64_000);
}
#[test]
fn test_recovery_suggestion_all_variants() {
// Every variant should return a non-empty suggestion
let variants = [
FailoverReason::Auth,
FailoverReason::AuthPermanent,
FailoverReason::Billing,
FailoverReason::RateLimited,
FailoverReason::Overloaded,
FailoverReason::ServerError,
FailoverReason::Timeout,
FailoverReason::ContextOverflow,
FailoverReason::PayloadTooLarge,
FailoverReason::ImageTooLarge,
FailoverReason::ModelNotFound,
FailoverReason::ProviderPolicyBlocked,
FailoverReason::ContentPolicyBlocked,
FailoverReason::FormatError,
FailoverReason::InvalidEncryptedContent,
FailoverReason::MultimodalToolContentUnsupported,
FailoverReason::ThinkingSignature,
FailoverReason::LongContextTier,
FailoverReason::OauthLongContextBetaForbidden,
FailoverReason::SslTransient,
FailoverReason::Unknown("test".into()),
];
for v in &variants {
let suggestion = recovery_suggestion(v);
assert!(!suggestion.is_empty(), "Empty suggestion for {:?}", v);
}
}
// ── Context Overflow 解析测试 ──
#[test]
fn test_parse_overflow_anthropic_format() {
let err = "input length and max_tokens exceed context limit: 180000 + 32000 > 200000. \
please reduce the length of the messages or the max_tokens.";
let info = parse_context_overflow(err).unwrap();
assert_eq!(info.input_length, 180000);
assert_eq!(info.requested_max_tokens, 32000);
assert_eq!(info.context_limit, 200000);
}
#[test]
fn test_parse_overflow_anthropic_format_different_numbers() {
let err = "Error: input length and max_tokens exceed context limit: 50000 + 16000 > 64000";
let info = parse_context_overflow(err).unwrap();
assert_eq!(info.input_length, 50000);
assert_eq!(info.requested_max_tokens, 16000);
assert_eq!(info.context_limit, 64000);
}
#[test]
fn test_parse_overflow_generic_format() {
let err = "context length exceeded: 130000 + 32000 > 128000";
let info = parse_context_overflow(err).unwrap();
assert_eq!(info.input_length, 130000);
assert_eq!(info.requested_max_tokens, 32000);
assert_eq!(info.context_limit, 128000);
}
#[test]
fn test_parse_overflow_no_match() {
let err = "Something went wrong but no overflow pattern here";
assert!(parse_context_overflow(err).is_none());
}
#[test]
fn test_calculate_safe_max_tokens_normal() {
let info = ContextOverflowInfo {
input_length: 180000,
requested_max_tokens: 32000,
context_limit: 200000,
};
// available = 200000 - 180000 = 20000
// adjusted = 20000 - 1000 = 19000
assert_eq!(calculate_safe_max_tokens(&info), Some(19000));
}
#[test]
fn test_calculate_safe_max_tokens_no_room() {
let info = ContextOverflowInfo {
input_length: 199000,
requested_max_tokens: 32000,
context_limit: 200000,
};
// available = 200000 - 199000 = 1000
// adjusted = 1000 - 1000 = 0 < MIN_ACCEPTABLE_TOKENS(500)
assert_eq!(calculate_safe_max_tokens(&info), None);
}
#[test]
fn test_calculate_safe_max_tokens_input_exceeds_context() {
let info = ContextOverflowInfo {
input_length: 210000,
requested_max_tokens: 1000,
context_limit: 200000,
};
// input > context → None (must compress)
assert_eq!(calculate_safe_max_tokens(&info), None);
}
#[test]
fn test_recovery_uses_adjust_max_tokens_first() {
let mut attempts = RecoveryAttempts::new();
let info = ContextOverflowInfo {
input_length: 180000,
requested_max_tokens: 32000,
context_limit: 200000,
};
let step = attempts.next_step(&ErrorKind::PromptTooLong, Some(&info));
assert_eq!(
step,
Some(RecoveryStep::AdjustMaxTokens {
new_max_tokens: 19000
})
);
}
#[test]
fn test_recovery_falls_back_to_compact_when_no_room() {
let mut attempts = RecoveryAttempts::new();
// Input almost fills context — no room to adjust max_tokens
let info = ContextOverflowInfo {
input_length: 199500,
requested_max_tokens: 32000,
context_limit: 200000,
};
let step = attempts.next_step(&ErrorKind::PromptTooLong, Some(&info));
// Should skip AdjustMaxTokens and go to AggressiveCompact
assert_eq!(step, Some(RecoveryStep::AggressiveCompact));
}
#[test]
fn test_recovery_adjust_after_compact_exhausted() {
let mut attempts = RecoveryAttempts::new();
// Mark aggressive_compact as tried
attempts.aggressive_compact = true;
let info = ContextOverflowInfo {
input_length: 180000,
requested_max_tokens: 32000,
context_limit: 200000,
};
// AdjustMaxTokens check is skipped when aggressive_compact is already true
let step = attempts.next_step(&ErrorKind::PromptTooLong, Some(&info));
assert_eq!(step, Some(RecoveryStep::ReactiveCompact));
}
#[test]
fn test_try_recover_adjusts_token_budget() {
let budget = TokenBudget::new(32_000, 40_000);
let mut recovery = ErrorRecovery::new(budget);
let info = ContextOverflowInfo {
input_length: 180000,
requested_max_tokens: 32000,
context_limit: 200000,
};
let step = recovery.try_recover(&ErrorKind::PromptTooLong, Some(&info));
assert_eq!(
step,
Some(RecoveryStep::AdjustMaxTokens {
new_max_tokens: 19000
})
);
// Token budget hard_limit should be adjusted down
assert_eq!(recovery.token_budget.hard_limit, 19000);
}
}
src/agent/runtime/error_recovery/overflow.rs
+157
View File
@@ -0,0 +1,157 @@
// src/agent/runtime/error_recovery/overflow.rs
//
// 上下文溢出解析与安全 Token 计算。
#[derive(Debug, Clone, Copy)]
pub struct ContextOverflowInfo {
/// 当前输入 token 数
pub input_length: usize,
/// 请求的 max_tokens 数
pub requested_max_tokens: usize,
/// 模型上下文窗口总容量
pub context_limit: usize,
}
/// 上下文溢出检测的最小可接受 token 数。
/// 低于此值说明即使调整 max_tokens 也没有意义，应触发压缩。
const MIN_ACCEPTABLE_TOKENS: usize = 500;
/// 上下文溢出调整的安全边际（token 数）。
/// 保留少量余量以防止重试时再次溢出。
const OVERFLOW_SAFETY_MARGIN: usize = 1000;
/// 从错误消息中解析 Anthropic-style context overflow 信息。
///
/// 匹配模式:
/// "input length and max_tokens exceed context limit: 180000 + 32000 > 200000"
/// "prompt too long: X tokens + Y max_tokens > Z context_window"
/// "context length exceeded: X + Y > Z"
/// "maximum context length is Z tokens, but X + Y = ..."
pub fn parse_context_overflow(error_str: &str) -> Option<ContextOverflowInfo> {
let lower = error_str.to_lowercase();
// Pattern 1: Anthropic standard format
// "input length and max_tokens exceed context limit: A + B > C"
if let Some(info) = parse_anthropic_overflow(&lower) {
return Some(info);
}
// Pattern 2: OpenAI format
// "maximum context length is C tokens. however X tokens ... and Y max_tokens"
if let Some(info) = parse_openai_overflow(&lower) {
return Some(info);
}
// Pattern 3: Generic "X + Y > Z" pattern
if let Some(info) = parse_generic_overflow(&lower) {
return Some(info);
}
None
}
/// 解析 Anthropic API 格式: "input length and max_tokens exceed context limit: A + B > C"
fn parse_anthropic_overflow(error_str: &str) -> Option<ContextOverflowInfo> {
// Find the "A + B > C" pattern after "context limit"
let marker = "input length and max_tokens exceed context limit";
let pos = error_str.find(marker)?;
let after_marker = &error_str[pos + marker.len()..];
// Extract numbers: look for pattern like " 180000 + 32000 > 200000"
let numbers = extract_three_numbers(after_marker)?;
Some(ContextOverflowInfo {
input_length: numbers.0,
requested_max_tokens: numbers.1,
context_limit: numbers.2,
})
}
/// 解析 OpenAI API 格式: "maximum context length is C tokens..."
fn parse_openai_overflow(error_str: &str) -> Option<ContextOverflowInfo> {
// "maximum context length is C tokens"
let context_re = regex::Regex::new(r"maximum context length is (\d+)").ok()?;
let caps = context_re.captures(error_str)?;
let context_limit: usize = caps.get(1)?.as_str().parse().ok()?;
// "you requested X tokens" or "messages resulted in X tokens"
let input_re = regex::Regex::new(r"(?:requested|resulted in|totals? to) (\d+) tokens").ok()?;
let input_length: usize = if let Some(caps) = input_re.captures(error_str) {
caps.get(1)?.as_str().parse().ok()?
} else {
// Fallback: find any substantial number and use it as input_length
return None;
};
// max_tokens might not be explicitly stated; use the context_limit - input_length - margin
let requested_max_tokens = context_limit.saturating_sub(input_length);
Some(ContextOverflowInfo {
input_length,
requested_max_tokens,
context_limit,
})
}
/// 解析通用格式: "X + Y > Z"
fn parse_generic_overflow(error_str: &str) -> Option<ContextOverflowInfo> {
// Look for patterns like "130000 + 32000 > 200000"
let (a, b, c) = extract_three_numbers(error_str)?;
Some(ContextOverflowInfo {
input_length: a,
requested_max_tokens: b,
context_limit: c,
})
}
/// 从字符串中提取三个连续的数字，匹配 "A + B > C" 模式。
fn extract_three_numbers(s: &str) -> Option<(usize, usize, usize)> {
let re = regex::Regex::new(r"(\d{4,})\s*\+\s*(\d+)\s*>\s*(\d{4,})").ok()?;
let caps = re.captures(s)?;
let a: usize = caps.get(1)?.as_str().parse().ok()?;
let b: usize = caps.get(2)?.as_str().parse().ok()?;
let c: usize = caps.get(3)?.as_str().parse().ok()?;
Some((a, b, c))
}
/// 根据 ContextOverflow 信息计算安全的 max_tokens 值。
///
/// 公式: `new_max_tokens = context_limit - input_length - SAFETY_MARGIN`
///
/// 返回 `None` 如果计算出的值低于最小可接受阈值（说明必须压缩上下文）。
pub fn calculate_safe_max_tokens(info: &ContextOverflowInfo) -> Option<usize> {
// 如果输入本身已经超过上下文窗口，max_tokens 调整无意义
if info.input_length >= info.context_limit {
return None;
}
let available = info.context_limit.saturating_sub(info.input_length);
let adjusted = available.saturating_sub(OVERFLOW_SAFETY_MARGIN);
if adjusted >= MIN_ACCEPTABLE_TOKENS {
Some(adjusted)
} else {
// 可用空间太小，调整 max_tokens 无意义 → 应触发压缩
None
}
}
// ── 向后兼容 ──
/// 解析 Retry-After header 值
pub fn parse_retry_after(error_str: &str) -> Option<u64> {
if let Some(pos) = error_str.find("retry_after=Some(") {
let prefix_len = "retry_after=Some(".len();
let rest = &error_str[pos + prefix_len..];
if let Some(end) = rest.find(')') {
return rest[..end].parse().ok();
}
}
if let Some(pos) = error_str.find("Retry-After:") {
let rest = &error_str[pos + "Retry-After:".len()..];
if let Some(end) = rest.find('\r').or_else(|| rest.find('\n')) {
return rest[..end].trim().parse().ok();
}
return rest.trim().parse().ok();
}
None
}
src/agent/runtime/events.rs
+83
View File
@@ -0,0 +1,83 @@
// src/agent/runtime/events.rs
//
// SSE 流式事件定义与运行指标。
use serde::Serialize;
use std::collections::HashMap;
/// SSE 流式事件（发送给前端）
#[derive(Debug, Clone, Serialize)]
#[serde(tag = "type")]
pub enum AgentStreamEvent {
/// 会话创建/恢复
#[serde(rename = "session")]
Session { session_id: String, title: String },
/// 智能体思考过程
#[serde(rename = "thought")]
Thought { content: String, step: usize },
/// 工具调用开始
#[serde(rename = "tool_call")]
ToolCall {
/// LLM 生成的工具调用 ID，用于全链路关联（前端/审计/持久化）
id: String,
name: String,
arguments: serde_json::Value,
step: usize,
},
/// 工具执行结果（Observation）
#[serde(rename = "tool_result")]
ToolResult {
/// 对应的工具调用 ID，前端凭此精确匹配 tool_call 条目
tool_call_id: String,
name: String,
output: String,
is_error: bool,
metadata: serde_json::Value,
step: usize,
},
/// 文本增量流式输出（最终回答或工具流式输出）
#[serde(rename = "text_delta")]
TextDelta {
content: String,
/// 可选：工具调用 ID。当 set 时，此增量属于对应工具的流式输出，
/// 前端应将其渲染到工具结果区域而非主文本区。
#[serde(skip_serializing_if = "Option::is_none")]
tool_call_id: Option<String>,
},
/// Token 使用统计
#[serde(rename = "usage")]
Usage {
prompt_tokens: u32,
completion_tokens: u32,
total_tokens: u32,
},
/// 错误通知
#[serde(rename = "error")]
Error { message: String },
/// 权限请求（需要用户确认工具执行）
#[serde(rename = "permission_request")]
PermissionRequest {
tool_call_id: String,
tool_name: String,
message: String,
arguments: serde_json::Value,
/// 可选的权限风险解释（参考 Claude Code permissionExplainer）
explanation: Option<serde_json::Value>,
},
/// 权限响应已处理
#[serde(rename = "permission_response")]
PermissionResponse { tool_call_id: String, allowed: bool },
/// 完成标记
#[serde(rename = "done")]
Done,
}
/// Agent 运行指标
#[derive(Debug, Default, Serialize)]
pub struct AgentMetrics {
pub total_steps: usize,
pub compression_count: usize,
pub duplicate_detections: usize,
/// 各工具调用次数统计
pub tool_calls: HashMap<String, usize>,
}
src/agent/runtime/executor/helpers.rs
+257
View File
@@ -0,0 +1,257 @@
// src/agent/runtime/executor/helpers.rs
//
// 执行器辅助类型与函数：PreparedCall, ToolResultMessage, ToolExecutionResult,
// execute_single_tool, process_single_result, save_tool_message_sync。
use sqlx::SqlitePool;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use tokio::sync::mpsc;
use tracing::warn;
use crate::clients::llm::ChatMessage;
use super::AgentStreamEvent;
use crate::agent::hooks::{
event_label, HookRegistry, PostToolUseContext, PostToolUseFailureContext,
};
#[derive(Debug, Clone)]
pub struct PreparedCall {
pub tool_call_id: String,
pub tool_name: String,
pub args: serde_json::Value,
}
/// 单次工具执行后的消息 + 元数据
pub struct ToolResultMessage {
pub chat_message: ChatMessage,
pub was_error: bool,
}
/// 工具执行结果摘要
pub struct ToolExecutionResult {
/// 每条工具调用对应的 tool_result 消息（供调用方 push 到 messages）
pub tool_messages: Vec<ToolResultMessage>,
pub was_cancelled: bool,
pub had_duplicate: bool,
/// Hook 注入的附加上下文（PreToolUse + PostToolUse），需注入 LLM 消息列表
pub hook_contexts: Vec<String>,
/// Hook 的阻塞错误详情（用于日志和诊断）
pub blocking_errors: Vec<String>,
}
/// 执行单个工具调用（含超时和取消检测）。
///
/// 从原 `execute_parallel` 的闭包提取，供分区后的批次执行复用。
pub(super) async fn execute_single_tool(
tool_opt: Option<&dyn crate::agent::tools::AgentTool>,
args: serde_json::Value,
tool_ctx: &crate::agent::tools::ToolContext,
cancelled: &Arc<AtomicBool>,
timeout_dur: std::time::Duration,
tool_name: &str,
) -> crate::agent::tools::ToolOutput {
let tool = match tool_opt {
Some(t) => t,
None => return crate::agent::tools::ToolOutput::error(format!("未知工具: {}", tool_name)),
};
let interrupt_behavior = tool.interrupt_behavior();
let is_blocking = interrupt_behavior == crate::agent::tools::InterruptBehavior::Block;
let tool_fut = tool.execute(args, tool_ctx);
let cancelled = cancelled.clone();
let cancel_fut = async {
loop {
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
if !is_blocking && cancelled.load(Ordering::SeqCst) {
return;
}
}
};
tokio::select! {
res = tokio::time::timeout(timeout_dur, tool_fut) => {
match res {
Ok(output) => output,
Err(_) => crate::agent::tools::ToolOutput::error(format!(
"工具 {} 执行超时（{}秒）",
tool_name,
timeout_dur.as_secs()
)),
}
}
_ = cancel_fut => {
crate::agent::tools::ToolOutput::error("执行已被用户取消")
}
}
}
/// 处理单个工具执行结果（SSE 事件、PostToolUse hooks、持久化）。
///
/// 从原 `execute_parallel` 的结果处理循环提取。
#[allow(clippy::too_many_arguments)]
pub(super) async fn process_single_result(
tool_call_id: &str,
tool_name: &str,
tool_args: &serde_json::Value,
output: &crate::agent::tools::ToolOutput,
cancelled_flag: bool,
exec_start: std::time::Instant,
tx: &mpsc::UnboundedSender<AgentStreamEvent>,
hook_registry: &HookRegistry,
library_dir: &std::path::Path,
sid: &str,
agent_name: &str,
step: usize,
max_output_chars: usize,
tool_messages: &mut Vec<ToolResultMessage>,
additional_contexts: &mut Vec<String>,
db: &SqlitePool,
turn_index: i32,
) {
use crate::agent::tools::persist::maybe_persist_tool_result;
let elapsed_ms = exec_start.elapsed().as_millis() as u64;
// SSE 事件 — 立即推送到前端
let _ = tx.send(AgentStreamEvent::ToolResult {
tool_call_id: tool_call_id.to_string(),
name: tool_name.to_string(),
output: output.content.clone(),
is_error: output.is_error,
metadata: output.metadata.clone(),
step,
});
// 输出处理：小结果直接传递，大结果持久化到磁盘并返回 stub
// 但对于已从磁盘读取内容的工具（如 read_file），跳过持久化以防止级联
let tool_results_dir = library_dir.join("tool-results");
let (processed_content, _persisted_path) = if output.skip_persist {
(output.content.clone(), None)
} else {
maybe_persist_tool_result(
&output.content,
tool_call_id,
max_output_chars,
&tool_results_dir,
)
};
// PostToolUse hook
let post_ctx = PostToolUseContext {
session_id: sid.to_string(),
agent_name: agent_name.to_string(),
tool_name: tool_name.to_string(),
tool_args: tool_args.clone(),
output_content: processed_content.clone(),
is_error: output.is_error,
step,
elapsed_ms,
};
let post_result = hook_registry.run_post_tool_use(&post_ctx).await;
let final_content = post_result.final_content;
// 非可信内容包裹（间接 prompt 注入防御）
let llm_content =
crate::agent::runtime::untrusted::wrap_untrusted_content(tool_name, &final_content);
// 收集 PostToolUse hook 注入的上下文
if !post_result.tagged_contexts.is_empty() {
for tc in &post_result.tagged_contexts {
additional_contexts.push(format!(
"[Hook: {} | {}] {}",
tc.hook_name,
event_label(tc.source_event),
tc.content,
));
}
} else {
for ctx in &post_result.additional_contexts {
additional_contexts.push(ctx.clone());
}
}
// 收集 PostToolUse 的警告
for warning in &post_result.warnings {
additional_contexts.push(format!("[Hook Warning] {}", warning));
}
// 事后权限请求（audit trail）
for (perm_tool, perm) in &post_result.post_permission_requests {
warn!(
"[Executor] Hook 事后请求权限: tool={} permission={}",
perm_tool, perm
);
}
// PostToolUseFailure hook
if output.is_error {
let failure_ctx = PostToolUseFailureContext {
session_id: sid.to_string(),
agent_name: agent_name.to_string(),
tool_name: tool_name.to_string(),
tool_args: tool_args.clone(),
error_message: output.content.clone(),
is_interrupt: cancelled_flag,
step,
elapsed_ms,
};
hook_registry
.run_on_post_tool_use_failure(&failure_ctx)
.await;
}
// 发送给 LLM 使用包裹后的内容（安全防御）
let chat_message = ChatMessage::tool_result(tool_call_id, &llm_content);
// 持久化到数据库（fire-and-forget）
save_tool_message_sync(db, sid, turn_index, step, &chat_message);
tool_messages.push(ToolResultMessage {
chat_message,
was_error: output.is_error,
});
}
/// 同步保存 tool 角色消息到数据库。
pub(super) fn save_tool_message_sync(
db: &SqlitePool,
session_id: &str,
turn_index: i32,
step_index: usize,
msg: &ChatMessage,
) {
let db_clone = db.clone();
let session_id = session_id.to_string();
let content = msg.text().unwrap_or("").to_string();
let tool_call_id = msg.tool_call_id.clone();
// 提前序列化，避免闭包内的生命周期问题
let metadata_str =
serde_json::to_string(&serde_json::json!({ "role": "tool" })).unwrap_or_default();
let raw_json = serde_json::to_string(&msg).unwrap_or_default();
// fire-and-forget: tool 消息保存失败不影响主流程
tokio::spawn(async move {
let token_count = content.len() as i32 / 4;
if let Err(e) = sqlx::query(
"INSERT INTO agent_messages (session_id, turn_index, step_index, role, content, tool_call_id, token_count, metadata, raw_json, agent_name) \
VALUES (?, ?, ?, 'tool', ?, ?, ?, ?, ?, ?)",
)
.bind(&session_id)
.bind(turn_index)
.bind(step_index as i32)
.bind(&content)
.bind(&tool_call_id)
.bind(token_count)
.bind(&metadata_str)
.bind(&raw_json)
.bind("lead")
.execute(&db_clone)
.await
{
warn!("[Executor] 保存 tool 消息失败（非致命）: {}", e);
}
});
}
src/agent/runtime/executor.rs → src/agent/runtime/executor/mod.rs
+9 -244
View File
@@ -1,6 +1,11 @@
// src/agent/runtime/executor.rs
// src/agent/runtime/executor/mod.rs
//
// 工具调用执行器：验证 → PreToolUse hooks → 并行执行 → 结果收集 → PostToolUse hooks。
// 工具调用验证与并行执行器。
mod helpers;
// Re-export 公共类型
pub use helpers::{PreparedCall, ToolExecutionResult, ToolResultMessage};
use futures_util::stream::FuturesUnordered;
use futures_util::StreamExt;
@@ -21,36 +26,10 @@ use super::partitioner::ToolPartitioner;
use super::permission::{PermissionChecker, PermissionResult};
use super::permission_explainer::explain_permission;
use super::{AgentStreamEvent, DuplicateDetector};
use crate::agent::hooks::{
event_label, HookRegistry, PostToolUseContext, PostToolUseFailureContext, PreToolUseContext,
};
use crate::agent::hooks::{event_label, HookRegistry, PreToolUseContext};
use crate::agent::tools::{ToolContext, ToolRegistry};
/// 准备好的工具调用
#[derive(Debug, Clone)]
pub struct PreparedCall {
pub tool_call_id: String,
pub tool_name: String,
pub args: serde_json::Value,
}
/// 单次工具执行后的消息 + 元数据
pub struct ToolResultMessage {
pub chat_message: ChatMessage,
pub was_error: bool,
}
/// 工具执行结果摘要
pub struct ToolExecutionResult {
/// 每条工具调用对应的 tool_result 消息（供调用方 push 到 messages）
pub tool_messages: Vec<ToolResultMessage>,
pub was_cancelled: bool,
pub had_duplicate: bool,
/// Hook 注入的附加上下文（PreToolUse + PostToolUse），需注入 LLM 消息列表
pub hook_contexts: Vec<String>,
/// Hook 的阻塞错误详情（用于日志和诊断）
pub blocking_errors: Vec<String>,
}
use helpers::{execute_single_tool, process_single_result, save_tool_message_sync};
/// 验证工具调用：死循环检测 + 参数解析。
///
@@ -799,217 +778,3 @@ pub async fn execute_parallel(
blocking_errors: hook_blocking_errors,
}
}
/// 执行单个工具调用（含超时和取消检测）。
///
/// 从原 `execute_parallel` 的闭包提取，供分区后的批次执行复用。
async fn execute_single_tool(
tool_opt: Option<&dyn crate::agent::tools::AgentTool>,
args: serde_json::Value,
tool_ctx: &crate::agent::tools::ToolContext,
cancelled: &Arc<AtomicBool>,
timeout_dur: std::time::Duration,
tool_name: &str,
) -> crate::agent::tools::ToolOutput {
let tool = match tool_opt {
Some(t) => t,
None => return crate::agent::tools::ToolOutput::error(format!("未知工具: {}", tool_name)),
};
let interrupt_behavior = tool.interrupt_behavior();
let is_blocking = interrupt_behavior == crate::agent::tools::InterruptBehavior::Block;
let tool_fut = tool.execute(args, tool_ctx);
let cancelled = cancelled.clone();
let cancel_fut = async {
loop {
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
if !is_blocking && cancelled.load(Ordering::SeqCst) {
return;
}
}
};
tokio::select! {
res = tokio::time::timeout(timeout_dur, tool_fut) => {
match res {
Ok(output) => output,
Err(_) => crate::agent::tools::ToolOutput::error(format!(
"工具 {} 执行超时（{}秒）",
tool_name,
timeout_dur.as_secs()
)),
}
}
_ = cancel_fut => {
crate::agent::tools::ToolOutput::error("执行已被用户取消")
}
}
}
/// 处理单个工具执行结果（SSE 事件、PostToolUse hooks、持久化）。
///
/// 从原 `execute_parallel` 的结果处理循环提取。
#[allow(clippy::too_many_arguments)]
async fn process_single_result(
tool_call_id: &str,
tool_name: &str,
tool_args: &serde_json::Value,
output: &crate::agent::tools::ToolOutput,
cancelled_flag: bool,
exec_start: std::time::Instant,
tx: &mpsc::UnboundedSender<AgentStreamEvent>,
hook_registry: &HookRegistry,
library_dir: &std::path::Path,
sid: &str,
agent_name: &str,
step: usize,
max_output_chars: usize,
tool_messages: &mut Vec<ToolResultMessage>,
additional_contexts: &mut Vec<String>,
db: &SqlitePool,
turn_index: i32,
) {
use crate::agent::tools::persist::maybe_persist_tool_result;
let elapsed_ms = exec_start.elapsed().as_millis() as u64;
// SSE 事件 — 立即推送到前端
let _ = tx.send(AgentStreamEvent::ToolResult {
tool_call_id: tool_call_id.to_string(),
name: tool_name.to_string(),
output: output.content.clone(),
is_error: output.is_error,
metadata: output.metadata.clone(),
step,
});
// 输出处理：小结果直接传递，大结果持久化到磁盘并返回 stub
// 但对于已从磁盘读取内容的工具（如 read_file），跳过持久化以防止级联
let tool_results_dir = library_dir.join("tool-results");
let (processed_content, _persisted_path) = if output.skip_persist {
(output.content.clone(), None)
} else {
maybe_persist_tool_result(
&output.content,
tool_call_id,
max_output_chars,
&tool_results_dir,
)
};
// PostToolUse hook
let post_ctx = PostToolUseContext {
session_id: sid.to_string(),
agent_name: agent_name.to_string(),
tool_name: tool_name.to_string(),
tool_args: tool_args.clone(),
output_content: processed_content.clone(),
is_error: output.is_error,
step,
elapsed_ms,
};
let post_result = hook_registry.run_post_tool_use(&post_ctx).await;
let final_content = post_result.final_content;
// 非可信内容包裹（间接 prompt 注入防御）
let llm_content = super::untrusted::wrap_untrusted_content(tool_name, &final_content);
// 收集 PostToolUse hook 注入的上下文
if !post_result.tagged_contexts.is_empty() {
for tc in &post_result.tagged_contexts {
additional_contexts.push(format!(
"[Hook: {} | {}] {}",
tc.hook_name,
event_label(tc.source_event),
tc.content,
));
}
} else {
for ctx in &post_result.additional_contexts {
additional_contexts.push(ctx.clone());
}
}
// 收集 PostToolUse 的警告
for warning in &post_result.warnings {
additional_contexts.push(format!("[Hook Warning] {}", warning));
}
// 事后权限请求（audit trail）
for (perm_tool, perm) in &post_result.post_permission_requests {
warn!(
"[Executor] Hook 事后请求权限: tool={} permission={}",
perm_tool, perm
);
}
// PostToolUseFailure hook
if output.is_error {
let failure_ctx = PostToolUseFailureContext {
session_id: sid.to_string(),
agent_name: agent_name.to_string(),
tool_name: tool_name.to_string(),
tool_args: tool_args.clone(),
error_message: output.content.clone(),
is_interrupt: cancelled_flag,
step,
elapsed_ms,
};
hook_registry
.run_on_post_tool_use_failure(&failure_ctx)
.await;
}
// 发送给 LLM 使用包裹后的内容（安全防御）
let chat_message = ChatMessage::tool_result(tool_call_id, &llm_content);
// 持久化到数据库（fire-and-forget）
save_tool_message_sync(db, sid, turn_index, step, &chat_message);
tool_messages.push(ToolResultMessage {
chat_message,
was_error: output.is_error,
});
}
/// 同步保存 tool 角色消息到数据库。
fn save_tool_message_sync(
db: &SqlitePool,
session_id: &str,
turn_index: i32,
step_index: usize,
msg: &ChatMessage,
) {
let db_clone = db.clone();
let session_id = session_id.to_string();
let content = msg.text().unwrap_or("").to_string();
let tool_call_id = msg.tool_call_id.clone();
// 提前序列化，避免闭包内的生命周期问题
let metadata_str =
serde_json::to_string(&serde_json::json!({ "role": "tool" })).unwrap_or_default();
let raw_json = serde_json::to_string(&msg).unwrap_or_default();
// fire-and-forget: tool 消息保存失败不影响主流程
tokio::spawn(async move {
let token_count = content.len() as i32 / 4;
if let Err(e) = sqlx::query(
"INSERT INTO agent_messages (session_id, turn_index, step_index, role, content, tool_call_id, token_count, metadata, raw_json, agent_name) \
VALUES (?, ?, ?, 'tool', ?, ?, ?, ?, ?, ?)",
)
.bind(&session_id)
.bind(turn_index)
.bind(step_index as i32)
.bind(&content)
.bind(&tool_call_id)
.bind(token_count)
.bind(&metadata_str)
.bind(&raw_json)
.bind("lead")
.execute(&db_clone)
.await
{
warn!("[Executor] 保存 tool 消息失败（非致命）: {}", e);
}
});
}
src/agent/runtime/mod.rs
+10 -196
View File
@@ -13,9 +13,12 @@
pub mod checkpoint;
pub mod circuit_breaker;
pub mod config;
pub mod context;
pub mod denial_tracker;
pub mod duplicate_detector;
pub mod error_recovery;
pub mod events;
pub mod executor;
pub mod file_cache;
pub mod finalize;
@@ -31,9 +34,7 @@ pub mod system_prompt;
pub mod token_budget;
pub mod untrusted;
use serde::Serialize;
use sqlx::SqlitePool;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::mpsc;
use tracing::{error, info, warn};
@@ -55,203 +56,16 @@ use self::streaming::{StreamOutput, StreamStatus};
use self::system_prompt::SystemPromptCache;
use self::token_budget::TokenBudget;
/// Agent 配置参数
#[derive(Debug, Clone)]
pub struct AgentConfig {
/// 最大 ReAct 迭代次数
pub max_steps: usize,
/// 同质调用检测阈值（连续相同调用次数）
pub duplicate_call_threshold: usize,
/// 工具执行超时时间（秒）
pub tool_timeout_secs: u64,
/// 工具输出最大字符数
pub max_tool_output_chars: usize,
/// Token 预算软限制 — 各压缩层统一触发阈值
pub token_soft_limit: usize,
/// Token 预算硬限制（触发强制动作）
pub token_hard_limit: usize,
/// 最大消息数（超过此阈值触发 snip_compact 层压缩）
pub max_messages: usize,
/// 是否启用 LLM 思考模式（前端可控，默认关闭）
pub enable_thinking: bool,
/// 权限拒绝规则（逗号分隔，格式: ToolName 或 ToolName(content_pattern)）
pub permission_deny_rules: Vec<String>,
/// 权限允许规则（逗号分隔）
pub permission_allow_rules: Vec<String>,
/// 权限询问规则（逗号分隔）
pub permission_ask_rules: Vec<String>,
/// 权限模式: "default" | "accept_edits" | "bypass" | "dont_ask"
pub permission_mode: String,
/// 拒绝追踪：连续拒绝上限（默认 3）
pub denial_max_consecutive: usize,
/// 拒绝追踪：总拒绝上限（默认 20）
pub denial_max_total: usize,
/// 附加允许目录（逗号分隔，扩展文件沙箱范围）
pub additional_allowed_dirs: Vec<String>,
/// 子代理工具白名单（逗号分隔，空=全部工具可用）
pub subagent_allowed_tools: Vec<String>,
/// Agent 运行模式 ID（"default" / "deep-research" / "literature-reader"）
pub mode: String,
}
impl AgentConfig {
/// 从环境变量加载配置，缺失时使用默认值。
pub fn from_env_optional() -> Self {
AgentConfig {
max_steps: 8,
duplicate_call_threshold: 3,
tool_timeout_secs: 120,
max_tool_output_chars: 4000,
token_soft_limit: std::env::var("AGENT_TOKEN_SOFT_LIMIT")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(80000),
token_hard_limit: std::env::var("AGENT_TOKEN_HARD_LIMIT")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(100000),
max_messages: 50,
enable_thinking: false,
permission_deny_rules: parse_comma_list("AGENT_PERMISSIONS_DENY"),
permission_allow_rules: parse_comma_list("AGENT_PERMISSIONS_ALLOW"),
permission_ask_rules: parse_comma_list("AGENT_PERMISSIONS_ASK"),
permission_mode: std::env::var("AGENT_PERMISSION_MODE")
.unwrap_or_else(|_| "default".to_string()),
denial_max_consecutive: 3,
denial_max_total: 20,
additional_allowed_dirs: parse_comma_list("AGENT_ADDITIONAL_DIRS"),
subagent_allowed_tools: parse_comma_list("AGENT_SUBAGENT_ALLOWED_TOOLS"),
mode: "default".to_string(),
}
}
}
/// 解析逗号分隔的环境变量为字符串列表
fn parse_comma_list(env_key: &str) -> Vec<String> {
std::env::var(env_key)
.ok()
.map(|v| {
v.split(',')
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect()
})
.unwrap_or_default()
}
impl Default for AgentConfig {
fn default() -> Self {
Self::from_env_optional()
}
}
// ── SSE Stream Events ──
/// SSE 流式事件（发送给前端）
#[derive(Debug, Clone, Serialize)]
#[serde(tag = "type")]
pub enum AgentStreamEvent {
/// 会话创建/恢复
#[serde(rename = "session")]
Session { session_id: String, title: String },
/// 智能体思考过程
#[serde(rename = "thought")]
Thought { content: String, step: usize },
/// 工具调用开始
#[serde(rename = "tool_call")]
ToolCall {
/// LLM 生成的工具调用 ID，用于全链路关联（前端/审计/持久化）
id: String,
name: String,
arguments: serde_json::Value,
step: usize,
},
/// 工具执行结果（Observation）
#[serde(rename = "tool_result")]
ToolResult {
/// 对应的工具调用 ID，前端凭此精确匹配 tool_call 条目
tool_call_id: String,
name: String,
output: String,
is_error: bool,
metadata: serde_json::Value,
step: usize,
},
/// 文本增量流式输出（最终回答或工具流式输出）
#[serde(rename = "text_delta")]
TextDelta {
content: String,
/// 可选：工具调用 ID。当 set 时，此增量属于对应工具的流式输出，
/// 前端应将其渲染到工具结果区域而非主文本区。
#[serde(skip_serializing_if = "Option::is_none")]
tool_call_id: Option<String>,
},
/// Token 使用统计
#[serde(rename = "usage")]
Usage {
prompt_tokens: u32,
completion_tokens: u32,
total_tokens: u32,
},
/// 错误通知
#[serde(rename = "error")]
Error { message: String },
/// 权限请求（需要用户确认工具执行）
#[serde(rename = "permission_request")]
PermissionRequest {
tool_call_id: String,
tool_name: String,
message: String,
arguments: serde_json::Value,
/// 可选的权限风险解释（参考 Claude Code permissionExplainer）
explanation: Option<serde_json::Value>,
},
/// 权限响应已处理
#[serde(rename = "permission_response")]
PermissionResponse { tool_call_id: String, allowed: bool },
/// 完成标记
#[serde(rename = "done")]
Done,
}
// ── Metrics & Detection ──
/// Agent 运行指标
#[derive(Debug, Default, Serialize)]
pub struct AgentMetrics {
pub total_steps: usize,
pub compression_count: usize,
pub duplicate_detections: usize,
/// 各工具调用次数统计
pub tool_calls: HashMap<String, usize>,
}
/// 同质调用检测器
#[derive(Debug, Default)]
pub struct DuplicateDetector {
last_call: Option<(String, String)>, // (tool_name, arguments)
consecutive_count: usize,
}
impl DuplicateDetector {
/// 记录一次调用，返回是否检测到死循环
pub fn record(&mut self, tool_name: &str, arguments: &str, threshold: usize) -> bool {
let key = (tool_name.to_string(), arguments.to_string());
if self.last_call.as_ref() == Some(&key) {
self.consecutive_count += 1;
if self.consecutive_count >= threshold {
return true;
}
} else {
self.last_call = Some(key);
self.consecutive_count = 1;
}
false
}
}
// Re-export 公共类型，保持外部引用路径不变
// （use crate::agent::runtime::{AgentConfig, AgentStreamEvent, ...}）
// pub use 同时将这些名称引入当前作用域，供本文件内使用
pub use self::config::AgentConfig;
pub use self::duplicate_detector::DuplicateDetector;
pub use self::events::{AgentMetrics, AgentStreamEvent};
// ── Agent Runtime ──
/// Agent 运行时核心，协调 ReAct 循环的执行。
/// 智能体运行时
pub struct AgentRuntime {
app_state: Arc<AppState>,