修复: 命名空间上限与网络域解析兜底
- 命名空间超容单条拒绝,上限不变式恢复 - usage 部分字段缺失致整帧内容蒸发,补缺省兜底 - tool_call 缺 id 致整批丢弃,对齐流式 fallback 口径 - 视觉模型判定改词素尾,deepseek-v3 不再误判 - 并行分支 panic 任务对账补齐为失败结果 - plan 执行器死代码清理,实况注释校正
This commit is contained in:
@@ -508,16 +508,52 @@ impl Coordinator {
|
||||
});
|
||||
}
|
||||
|
||||
// 记录最近一次 JoinError 详情:JoinError 不携带 subtask 身份,
|
||||
// 缺失任务靠循环结束后按 id 对账补齐,错误文本由此带出
|
||||
let mut join_err_msg: Option<String> = None;
|
||||
|
||||
while let Some(join_result) = join_set.join_next().await {
|
||||
match join_result {
|
||||
Ok(result) => {
|
||||
results_map.insert(result.subtask_id.clone(), result);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!("[COORDINATOR] 子任务 panic: {}", e);
|
||||
// panic 与取消分开留痕:panic 是执行体自身崩溃需排查,取消是外部干预
|
||||
if e.is_panic() {
|
||||
tracing::error!("[COORDINATOR] 子任务 panic: {}", e);
|
||||
} else {
|
||||
tracing::warn!("[COORDINATOR] 子任务被取消: {}", e);
|
||||
}
|
||||
join_err_msg = Some(e.to_string());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// panic/取消的子任务不能凭空消失:返回契约是与 plan.tasks 一一对应,
|
||||
// 缺失会被下游误读为"无此任务";此处按该任务失败语义补失败结果收敛,
|
||||
// merge 按 success 过滤自然跳过,不扩散中断整层
|
||||
for task in layer {
|
||||
if !results_map.contains_key(&task.id) {
|
||||
tracing::error!(
|
||||
subtask = %task.id,
|
||||
"[COORDINATOR] 子任务执行体崩溃,按失败结果收敛(不扩散中断)"
|
||||
);
|
||||
results_map.insert(
|
||||
task.id.clone(),
|
||||
ExecutionResult {
|
||||
subtask_id: task.id.clone(),
|
||||
persona_id: task_persona
|
||||
.get(&task.id)
|
||||
.cloned()
|
||||
.unwrap_or_default(),
|
||||
output: join_err_msg
|
||||
.clone()
|
||||
.unwrap_or_else(|| "子任务 panic/被取消".to_string()),
|
||||
success: false,
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
tracing::info!(
|
||||
layer = layer_idx,
|
||||
@@ -1415,6 +1451,39 @@ mod tests {
|
||||
assert_eq!(results[1].subtask_id, "b");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn par_05_panicked_task_converges_to_failed_result() {
|
||||
// 同层并行中某子任务 panic:JoinSet 返回 JoinError(panic 变体),
|
||||
// 应收敛为该任务的失败结果,而不是让它从结果列表凭空消失
|
||||
let coord = make_coord();
|
||||
let plan = Plan::from_tasks(vec![
|
||||
SubTask::new("t1", "task1"),
|
||||
SubTask::new("t2", "task2"),
|
||||
SubTask::new("t3", "task3"),
|
||||
]);
|
||||
|
||||
let results = coord
|
||||
.dispatch_with_budget(&plan, |task, persona_id: String| async move {
|
||||
if task.id == "t2" {
|
||||
panic!("模拟子任务执行体崩溃");
|
||||
}
|
||||
ExecutionResult {
|
||||
subtask_id: task.id,
|
||||
persona_id,
|
||||
output: format!("done {}", task.intent),
|
||||
success: true,
|
||||
}
|
||||
}, None)
|
||||
.await;
|
||||
|
||||
assert_eq!(results.len(), 3, "panic 子任务也应占一个结果位");
|
||||
let t2 = results.iter().find(|r| r.subtask_id == "t2").expect("t2 应有失败结果");
|
||||
assert!(!t2.success, "panic 子任务应收敛为失败而非成功");
|
||||
assert!(!t2.output.is_empty(), "失败结果应带 JoinError 详情");
|
||||
assert!(results.iter().filter(|r| r.subtask_id != "t2").all(|r| r.success),
|
||||
"其余子任务不受 panic 扩散影响");
|
||||
}
|
||||
|
||||
// -- Plan-driven LLM 规划开关 --
|
||||
|
||||
#[test]
|
||||
|
||||
+21
-6
@@ -134,6 +134,10 @@ pub(crate) fn send_error_class(p: SendErrPreds) -> &'static str {
|
||||
/// 其余（openai / glm / deepseek 等 OpenAI 兼容）走 OpenAICompatProvider。
|
||||
/// 调用方（AI Chat 的 run_agentic_loop、df-nodes 的 AiNode）统一引用此工厂，
|
||||
/// 新增 provider 只改这一处。
|
||||
///
|
||||
/// 未知 provider_type 无法走显式报错（工厂签名不动，调用方横跨 src-tauri/df-nodes），
|
||||
/// 兜底 OpenAI 兼容协议但 warn 留痕，不再静默——协议错配的请求失败可溯源到配置项。
|
||||
/// anthropic_family 口径与 prompt.rs cache_split / chat.rs needs_url_prefetch 对齐。
|
||||
pub fn build_provider(
|
||||
provider_type: &str,
|
||||
base_url: &str,
|
||||
@@ -141,11 +145,22 @@ pub fn build_provider(
|
||||
model: &str,
|
||||
) -> Box<dyn LlmProvider> {
|
||||
match provider_type {
|
||||
"anthropic" => Box::new(anthropic_compat::AnthropicCompatProvider::new(
|
||||
base_url, api_key, model,
|
||||
)),
|
||||
_ => Box::new(openai_compat::OpenAICompatProvider::new(
|
||||
base_url, api_key, model,
|
||||
)),
|
||||
"anthropic" | "anthropic_compat" => Box::new(
|
||||
anthropic_compat::AnthropicCompatProvider::new(base_url, api_key, model),
|
||||
),
|
||||
"openai_compat" | "openai" | "glm" | "deepseek" => {
|
||||
Box::new(openai_compat::OpenAICompatProvider::new(
|
||||
base_url, api_key, model,
|
||||
))
|
||||
}
|
||||
unknown => {
|
||||
tracing::warn!(
|
||||
"未知 provider_type {:?}，按 OpenAI 兼容协议兜底；请检查 ai_providers 配置的 provider_type",
|
||||
unknown
|
||||
);
|
||||
Box::new(openai_compat::OpenAICompatProvider::new(
|
||||
base_url, api_key, model,
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -333,6 +333,36 @@ mod tests {
|
||||
assert_eq!(m.cost_tier, CostTier::Medium);
|
||||
}
|
||||
|
||||
// ── 启发式:版本号后缀不误判 Vision ──
|
||||
|
||||
#[test]
|
||||
fn heuristic_version_suffix_no_vision() {
|
||||
// "-v" 后接数字属版本号(deepseek-v3 为纯文本模型),不得误判 Vision
|
||||
for name in ["deepseek-v3", "acme-chat-v2", "acme-v12"] {
|
||||
let m = probe(name);
|
||||
assert_eq!(m.probe_source, Some(ProbeSource::Heuristic));
|
||||
assert!(
|
||||
!m.modalities.contains(&Modality::Vision),
|
||||
"{name} 版本号后缀不应误判 Vision: {:?}",
|
||||
m.modalities
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn heuristic_v_morpheme_tail_still_vision() {
|
||||
// v 词素尾仍判视觉:独立 v 段(qwen-v-plus 形态)与段尾 v(glm-4v 形态,假名避开预设表)
|
||||
for name in ["acme-chat-v", "acme-4v-plus"] {
|
||||
let m = probe(name);
|
||||
assert_eq!(m.probe_source, Some(ProbeSource::Heuristic));
|
||||
assert!(
|
||||
m.modalities.contains(&Modality::Vision),
|
||||
"{name} 段尾 v 词素应判 Vision: {:?}",
|
||||
m.modalities
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// ── 启发式:默认(Standard/Medium/Text/ToolUse) ──
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -112,8 +112,11 @@ pub(super) fn heuristic_infer(model_id: &str) -> ModelConfig {
|
||||
|
||||
// —— 词素判定助手(命名约定式启发式,集中维护便于增删) ——
|
||||
|
||||
/// Vision 词素:`-v`(尾缀如 glm-4v)、以 `v` 结尾、含 `vl` / `vision`。
|
||||
/// 不直接 `contains('v')`(避免误伤 review/verbal 等)。
|
||||
/// Vision 词素:`vision` / `vl` 子串、`v` 词素尾(任一分隔段以 v 结尾:glm-4v / qwen-v-plus)。
|
||||
/// 裸 "v" 必须是词素尾:`contains("-v")` 会把版本号误判成视觉(deepseek-v3 纯文本,
|
||||
/// "-v" 后接数字属版本号而非视觉词素),按段尾判定后 v3/v2 不命中、4v/v 段仍命中。
|
||||
pub(super) fn has_vision_token(name: &str) -> bool {
|
||||
name.contains("vision") || name.contains("vl") || name.ends_with('v') || name.contains("-v")
|
||||
name.contains("vision")
|
||||
|| name.contains("vl")
|
||||
|| name.split(['-', '.', '_']).any(|seg| seg.ends_with('v'))
|
||||
}
|
||||
|
||||
@@ -81,6 +81,10 @@ impl NamespaceStore {
|
||||
}
|
||||
|
||||
/// 存入 namespace。返回引用路径字符串。
|
||||
///
|
||||
/// 超限行为分两档：常规超限 LRU 淘汰旧条目腾位（既有语义）；单条本身超过
|
||||
/// max_bytes 时淘汰清空也放不下，拒绝入库只返回引用路径——展开点对
|
||||
/// read_only()=None 统一落 EVICTED_PLACEHOLDER，与 LRU 淘汰后的不可解析引用同语义。
|
||||
pub fn store(&mut self, tool_name: &str, content: &str) -> String {
|
||||
let key = simple_hash(content);
|
||||
let path = format!("{}{}/{}", NAMESPACE_REF_PREFIX, tool_name, key);
|
||||
@@ -93,8 +97,15 @@ impl NamespaceStore {
|
||||
return path;
|
||||
}
|
||||
|
||||
// 检查容量，超限淘汰
|
||||
let new_bytes = content.len();
|
||||
|
||||
// 单条超过总容量：硬入库会把 current_bytes 永久顶破上限（evict 清空后仍无条件
|
||||
// insert 的旧路径），拒绝入库保住 current_bytes <= max_bytes 不变式
|
||||
if new_bytes > self.max_bytes {
|
||||
return path;
|
||||
}
|
||||
|
||||
// 检查容量，超限淘汰
|
||||
self.evict_if_needed(new_bytes);
|
||||
|
||||
let entry = NamespaceEntry {
|
||||
@@ -268,6 +279,23 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn store_oversized_single_entry_rejected_cap_intact() {
|
||||
// 回归:单条超过 max_bytes 时旧路径淘汰清空后仍无条件入库,current_bytes
|
||||
// 永久突破上限。现拒绝入库,仅返回引用路径,展开点落 EVICTED_PLACEHOLDER。
|
||||
let mut ns = NamespaceStore::new(100);
|
||||
let path = ns.store("t1", &"x".repeat(200));
|
||||
assert_eq!(ns.len(), 0, "超容量单条应拒绝入库");
|
||||
assert_eq!(ns.current_bytes, 0, "字节账目不得突破上限");
|
||||
assert!(is_namespace_ref(&path), "返回值仍是引用路径");
|
||||
assert_eq!(ns.read_only(&path), None, "不可解析 → 既有淘汰占位语义");
|
||||
|
||||
// 拒绝后不污染状态,常规条目仍可正常入
|
||||
let p2 = ns.store("t2", "ok");
|
||||
assert_eq!(ns.read(&p2).map(str::to_owned), Some("ok".to_string()));
|
||||
assert_eq!(ns.current_bytes, 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn clear_resets_all() {
|
||||
let mut ns = NamespaceStore::new(100_000);
|
||||
|
||||
@@ -310,7 +310,15 @@ impl OpenAICompatProvider {
|
||||
.into_iter()
|
||||
.enumerate()
|
||||
.map(|(i, c)| {
|
||||
let missing_id = c.id.is_empty();
|
||||
let id = tool_call_id_or_fallback(&c.id, i, "gen_tool");
|
||||
if missing_id {
|
||||
tracing::warn!(
|
||||
function = %c.function.name,
|
||||
fallback_id = %id,
|
||||
"[OpenAI] 同步响应 tool_call 缺/空 id,已生成 fallback id(防结果路由串号)"
|
||||
);
|
||||
}
|
||||
ToolCall::new(id, c.function.name, c.function.arguments)
|
||||
})
|
||||
.collect()
|
||||
@@ -1293,6 +1301,26 @@ mod tests {
|
||||
assert_eq!(ids.len(), unique.len(), "id 应全部唯一，实际 {:?}", ids);
|
||||
}
|
||||
|
||||
/// 同步响应 tool_call 缺 id/arguments 字段:serde default 兜底(空串/空串),
|
||||
/// 缺 id 经 parse_tool_calls 生成 fallback。旧行为:id 必填,单条缺字段令整个
|
||||
/// 响应 serde 失败 → complete 报"响应解析失败"整批丢弃。
|
||||
#[test]
|
||||
fn openai_sync_tool_call_missing_fields_default_instead_of_failing() {
|
||||
let calls: Vec<OpenAiToolCallResp> = serde_json::from_str(
|
||||
r#"[{"type":"function","function":{"name":"list_dir"}}]"#,
|
||||
)
|
||||
.expect("缺 id/arguments 的 tool_call 应可反序列化,不再整响应解析失败");
|
||||
let parsed = OpenAICompatProvider::parse_tool_calls(calls);
|
||||
assert_eq!(parsed.len(), 1);
|
||||
assert!(
|
||||
parsed[0].id.starts_with("gen_tool_"),
|
||||
"缺 id → fallback id, got: {}",
|
||||
parsed[0].id
|
||||
);
|
||||
assert_eq!(parsed[0].function.name, "list_dir");
|
||||
assert_eq!(parsed[0].function.arguments, "", "缺 arguments 缺省空串");
|
||||
}
|
||||
|
||||
/// CR-空 id 流式：SSE chunk 携带 `"id":""`（SenseNova 兼容缺陷）→ ToolCallDelta.id
|
||||
/// 转为 `gen_stream_{n}` fallback（n 取自全局计数器，跨轮跨 assistant 递增，非 None），
|
||||
/// 保证下游 accumulate_tool_calls 写入 draft.id 非空。chunk 完全无 id 字段（None）保持
|
||||
|
||||
@@ -102,10 +102,14 @@ pub(crate) struct OpenAiMessageResp {
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(crate) struct OpenAiToolCallResp {
|
||||
/// 缺省容错:个别兼容端点同步响应省略 id → 缺省空串,由 parse_tool_calls 的
|
||||
/// tool_call_id_or_fallback 生成唯一 fallback。此前必填,单条 tool_call 缺 id 令
|
||||
/// 整个响应 serde 失败 → complete 报"响应解析失败"整批丢弃(流式路径早已容错)。
|
||||
#[serde(default)]
|
||||
pub id: String,
|
||||
// 反序列化 #[serde(rename="type")] 字段,保留以对齐 OpenAI 响应结构,标注意图消除 dead_code warning
|
||||
#[allow(dead_code)]
|
||||
#[serde(rename = "type")]
|
||||
#[serde(rename = "type", default)]
|
||||
pub call_type: String,
|
||||
pub function: OpenAiFunctionResp,
|
||||
}
|
||||
@@ -113,6 +117,9 @@ pub(crate) struct OpenAiToolCallResp {
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(crate) struct OpenAiFunctionResp {
|
||||
pub name: String,
|
||||
/// 缺省容错:arguments 缺失 → 空串(对齐 anthropic 同步 tool_use input 缺失 args="" 口径),
|
||||
/// 防单条缺字段令整个响应解析失败;name 保持必填(无名 tool_call 无法派发,无安全缺省)。
|
||||
#[serde(default)]
|
||||
pub arguments: String,
|
||||
}
|
||||
|
||||
@@ -130,8 +137,13 @@ pub(crate) struct PromptTokensDetails {
|
||||
/// 嵌套 cached_tokens 回填,消费方一律读归一后的扁平字段。
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(crate) struct OpenAiUsageRaw {
|
||||
/// 三个计数缺省 0:部分兼容网关 usage 只报部分字段(如仅 total_tokens),必填会让
|
||||
/// 整帧 serde 失败走空 chunk 分支 — 同帧 content 一并蒸发、usage 归零无痕。
|
||||
#[serde(default)]
|
||||
pub prompt_tokens: u32,
|
||||
#[serde(default)]
|
||||
pub completion_tokens: u32,
|
||||
#[serde(default)]
|
||||
pub total_tokens: u32,
|
||||
/// DeepSeek 扩展:缓存命中 token(低价,deepseek-chat/reasoner)。无此字段 → default 0。
|
||||
#[serde(default)]
|
||||
@@ -288,6 +300,14 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option<TokenUsage>)
|
||||
}
|
||||
// 提取 usage（带 include_usage 时末段 chunk 携带，覆盖累积）
|
||||
if let Some(u) = chunk.usage {
|
||||
// 缺省计数的留痕:两个分项均 0 基本可断定 provider 上报不完整(真实完成
|
||||
// 必有 completion 计数),非流路径下游拿到零 usage 无从察觉。
|
||||
if u.prompt_tokens == 0 && u.completion_tokens == 0 {
|
||||
tracing::warn!(
|
||||
total = u.total_tokens,
|
||||
"[OpenAI] usage 缺 prompt/completion 分项(按 0 兜底),provider 上报不完整"
|
||||
);
|
||||
}
|
||||
tracing::info!(
|
||||
prompt = u.prompt_tokens,
|
||||
completion = u.completion_tokens,
|
||||
@@ -316,6 +336,21 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option<TokenUsage>)
|
||||
);
|
||||
|
||||
let tool_calls = choice.delta.tool_calls.map(|tcs| {
|
||||
// 缺 index 的增量缺省 0(单工具调用语义等价);但同帧多 call 且 index
|
||||
// 冲突时缺省会互相踩(下游按 index 累积 → 参数拼错位),warn 留痕定位
|
||||
// provider 兼容缺陷,不做静默改写。
|
||||
if tcs.len() > 1 {
|
||||
let mut idxs: Vec<u32> = tcs.iter().map(|tc| tc.index).collect();
|
||||
idxs.sort_unstable();
|
||||
idxs.dedup();
|
||||
if idxs.len() < tcs.len() {
|
||||
tracing::warn!(
|
||||
calls = tcs.len(),
|
||||
unique = idxs.len(),
|
||||
"[OpenAI] 同帧多 tool_call 增量 index 冲突(缺省 0 互踩),按 index 累积将错位"
|
||||
);
|
||||
}
|
||||
}
|
||||
tcs.into_iter()
|
||||
.map(|tc| {
|
||||
// CR-空 id：流式 chunk 的 id 可能为 Some("")（SenseNova 兼容缺陷）。
|
||||
@@ -423,6 +458,35 @@ mod tests {
|
||||
assert_eq!(usage.prompt_cache_miss_tokens, 36);
|
||||
}
|
||||
|
||||
/// 部分兼容网关 usage 只报部分字段:缺省 0 入累加器,同帧 content 不再因整帧
|
||||
/// serde 失败被吞(旧行为:必填字段缺失 → 空 chunk → 文本蒸发 + usage 归零无痕)
|
||||
#[test]
|
||||
fn sse_partial_usage_frame_parses_with_defaults() {
|
||||
let mut accum = None;
|
||||
let data = r#"{"choices":[{"delta":{"content":"hi"}}],"usage":{"total_tokens":10}}"#;
|
||||
let c = apply_openai_sse(data, &mut accum);
|
||||
assert_eq!(c.delta, "hi", "usage 缺分项不应令整帧解析失败丢 content");
|
||||
let u = accum.expect("部分 usage 应写入累加器");
|
||||
assert_eq!(u.total_tokens, 10);
|
||||
assert_eq!(u.prompt_tokens, 0, "缺省分项按 0 兜底");
|
||||
assert_eq!(u.completion_tokens, 0);
|
||||
}
|
||||
|
||||
/// 同帧多 tool_call 增量缺 index:缺省 0 触发 index 冲突路径(仅 warn 留痕,不改写
|
||||
/// 缺省策略);解析本身不失败,增量仍逐条带出。
|
||||
#[test]
|
||||
fn sse_tool_calls_missing_index_in_multi_call_frame_still_parse() {
|
||||
let mut accum = None;
|
||||
let data = r#"{"choices":[{"delta":{"tool_calls":[
|
||||
{"id":"call_a","function":{"name":"ta","arguments":"{"}},
|
||||
{"id":"call_b","function":{"name":"tb","arguments":"}"}}
|
||||
]}}]}"#;
|
||||
let c = apply_openai_sse(data, &mut accum);
|
||||
let tcs = c.tool_calls.expect("缺 index 的多 call 帧应可解析");
|
||||
assert_eq!(tcs.len(), 2);
|
||||
assert!(tcs.iter().all(|tc| tc.index == 0), "缺 index 缺省 0");
|
||||
}
|
||||
|
||||
/// content_filter 属终态:敏感内容拦截的流部分 provider 不发 [DONE] 帧,
|
||||
/// 必须靠 finish_reason 收敛,否则悬挂;未列入终态的 finish_reason 保持非终态
|
||||
#[test]
|
||||
|
||||
@@ -2,7 +2,8 @@
|
||||
//!
|
||||
//! 按 Plan::to_layers 输出的层级顺序执行子任务：
|
||||
//! - 层间串行：上层全部完成才进下一层（deps 保证）
|
||||
//! - 层内并行：同层 SubTask 用 tokio::JoinSet 并发执行
|
||||
//! - 层内并行：同层 SubTask 用 tokio::JoinSet 并发执行（设计目标；
|
||||
//! 当前为串行循环模拟，见 execute 内 TODO，本执行器未接生产链路）
|
||||
//!
|
||||
//! **当前状态**：骨架实现，PLAN_EXECUTION_ENABLED 门控（默认关）。
|
||||
//! 未接入 agentic loop 主路径，翻 true 启用后仍需与 run_agentic_loop 对接。
|
||||
@@ -16,7 +17,9 @@ use crate::planner::{Plan, SubTask};
|
||||
/// Plan 执行总开关。false = 不启用（走单链 ReAct 旧行为）。
|
||||
///
|
||||
/// 运行时原子门控,支持热切换(通过 IPC 或前端设置开关)。
|
||||
/// 默认关,翻 true 后 process_tool_calls 内以 JoinSet 并行执行同层工具。
|
||||
/// 默认关。现状:PlanExecutor::execute 零生产调用点(仅测试),本执行器未接入主链;
|
||||
/// agentic loop 读本开关走的是 coordinator::dispatch(Phase 1 串行桩),与本文件无关。
|
||||
/// JoinSet 真并行留 Phase 3 对接时接通(见 execute 内 TODO)。
|
||||
/// 与 PLANNING_ENABLED 解耦:plan_hint 独立使能,编排可见性始终可开。
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
pub static PLAN_EXECUTION_ENABLED: AtomicBool = AtomicBool::new(false);
|
||||
@@ -87,16 +90,16 @@ impl PlanExecutor {
|
||||
layer.len()
|
||||
);
|
||||
|
||||
// 层内并行：JoinSet 收集
|
||||
let _join_set: tokio::task::JoinSet<SubTaskResult> = tokio::task::JoinSet::new();
|
||||
let summary_clone = parent_summary.clone();
|
||||
|
||||
for task in layer {
|
||||
let task = Arc::new(task.clone());
|
||||
let summary = summary_clone.clone();
|
||||
// 注意:run_subtask 是 FnMut,不能直接 move 进 JoinSet(多任务并发调用)。
|
||||
// Phase 3 对接时,run_subtask 改为 Arc<Fn> 或 channel 模式。
|
||||
// 当前骨架用串行执行模拟(层内不并行),验证分层逻辑正确性。
|
||||
// TODO(Phase 3 真并行): 当前为串行循环模拟并行(假并行),仅验证分层逻辑。
|
||||
// run_subtask 是 FnMut,不能直接 move 进 JoinSet(多任务并发调用);
|
||||
// 接真并行的改法:run_subtask 换 Arc<Fn> 或 channel 模式,
|
||||
// 本层全部 spawn 进 JoinSet 统一收集后再汇总 parent_summary,替换此串行 await。
|
||||
// 现状无用户可见影响:本执行器未接生产链路(execute 零调用点)。
|
||||
let result = run_subtask(task.clone(), summary).await;
|
||||
let sr = match result {
|
||||
Ok(output) => SubTaskResult {
|
||||
|
||||
Reference in New Issue
Block a user