diff --git a/cmake_ext/ScaffoldFixtures.cmake b/cmake_ext/ScaffoldFixtures.cmake index be631085..471491f3 100644 --- a/cmake_ext/ScaffoldFixtures.cmake +++ b/cmake_ext/ScaffoldFixtures.cmake @@ -34,4 +34,7 @@ add_custom_command( list(APPEND EDGEFLOW_SCAFFOLD_FIXTURE_SOURCE "${PROJECT_SOURCE_DIR}/dev_support/node_authoring/starter_llm_node_advanced.cpp" "${PROJECT_SOURCE_DIR}/dev_support/node_authoring/starter_batch_node.cpp" - "${PROJECT_SOURCE_DIR}/dev_support/node_authoring/starter_multi_model_node.cpp") + "${PROJECT_SOURCE_DIR}/dev_support/node_authoring/starter_multi_model_node.cpp" + "${PROJECT_SOURCE_DIR}/dev_support/node_authoring/starter_batch_join_node.cpp" + "${PROJECT_SOURCE_DIR}/dev_support/node_authoring/starter_batch_group_node.cpp" + "${PROJECT_SOURCE_DIR}/dev_support/node_authoring/starter_batch_select_scatter_node.cpp") diff --git a/cmake_ext/TestInventory.cmake b/cmake_ext/TestInventory.cmake index 1e536069..37bbc60f 100644 --- a/cmake_ext/TestInventory.cmake +++ b/cmake_ext/TestInventory.cmake @@ -105,7 +105,9 @@ set(EDGEFLOW_SOURCE_test_log "${PROJECT_SOURCE_DIR}/tests/unit/logging/test_log. set(EDGEFLOW_SOURCE_test_log_name_override "${PROJECT_SOURCE_DIR}/tests/unit/logging/test_log_name_override.cpp") set(EDGEFLOW_SOURCE_test_asr_transcribe_node "${PROJECT_SOURCE_DIR}/tests/unit/nodes/test_asr_transcribe_node.cpp") set(EDGEFLOW_SOURCE_test_common_nodes "${PROJECT_SOURCE_DIR}/tests/unit/nodes/test_common_nodes.cpp") -set(EDGEFLOW_SOURCE_test_function_node "${PROJECT_SOURCE_DIR}/tests/unit/nodes/test_function_node.cpp") +set(EDGEFLOW_SOURCE_test_function_node + "${PROJECT_SOURCE_DIR}/tests/unit/nodes/test_function_node.cpp" + "${PROJECT_SOURCE_DIR}/tests/unit/nodes/test_traceable_batch_operations.cpp") set(EDGEFLOW_SOURCE_test_parameter_binding "${PROJECT_SOURCE_DIR}/tests/unit/nodes/test_parameter_binding.cpp") set(EDGEFLOW_SOURCE_test_llm_generate_node "${PROJECT_SOURCE_DIR}/tests/unit/nodes/test_llm_generate_node.cpp") set(EDGEFLOW_SOURCE_test_ocr_detect_node "${PROJECT_SOURCE_DIR}/tests/unit/nodes/test_ocr_detect_node.cpp") diff --git a/cmake_ext/Tests.cmake b/cmake_ext/Tests.cmake index 645d6e6f..fc8e4251 100644 --- a/cmake_ext/Tests.cmake +++ b/cmake_ext/Tests.cmake @@ -285,7 +285,7 @@ edgeflow_add_runner_test(TextCorpusSourceNodeTest edgeflow_test_nodes_runner edgeflow_add_runner_test(CommonNodesTest edgeflow_test_nodes_runner "CommonNodesTest.*:CustomNodeCatalogTest.*" "${_edgeflow_tier1}") edgeflow_add_runner_test(FunctionNodeTest edgeflow_test_nodes_runner - "FunctionNodeTest.*:ConfigurationSnapshotTest.*" "${_edgeflow_tier1}") + "FunctionNodeTest.*:ConfigurationSnapshotTest.*:TraceableBatchOperationsTest.*" "${_edgeflow_tier1}") edgeflow_add_runner_test(ParameterBindingTest edgeflow_test_nodes_runner "ParameterBindingTest.*" "${_edgeflow_tier1}") diff --git a/dev_support/benchmarks/traceable_batch_operations.cpp b/dev_support/benchmarks/traceable_batch_operations.cpp new file mode 100644 index 00000000..ccce7ae2 --- /dev/null +++ b/dev_support/benchmarks/traceable_batch_operations.cpp @@ -0,0 +1,97 @@ +// Standalone Linux measurement; build/run commands and limits are recorded in +// doc/rfcs/reviews/0055-traceable-batch-verification.md. +#include "nodes/traceable_batch_operations.h" + +#include + +#include +#include +#include +#include +#include +#include +#include + +using namespace llm_edgeflow; + +template +double Measure(Fn fn, size_t expected_size) { + std::vector times; + for (int round = 0; round < 8; ++round) { + const auto start = std::chrono::steady_clock::now(); + auto result = fn(); + const auto end = std::chrono::steady_clock::now(); + if (!result.ok() || result.value().size() != expected_size) { + throw std::runtime_error("Unexpected benchmark output"); + } + if (round > 0) { // One warmup, then seven measured constructions. + times.push_back( + std::chrono::duration(end - start).count()); + } + } + std::sort(times.begin(), times.end()); + return times[times.size() / 2]; +} + +int main(int argc, char** argv) { + if (argc != 3) { + std::cerr << "Usage: " << argv[0] + << " join|group|select|scatter|split item_count\n"; + return 1; + } + const std::string operation = argv[1]; + const size_t count = std::stoull(argv[2]); + if (count == 0 || count > 1000000 || count % 4 != 0) return 1; + TextBatch anchor; + anchor.reserve(count); + for (size_t i = 0; i < count; ++i) { + anchor.emplace_back(static_cast(i / 4), + static_cast(i % 4), + std::string(128, i % 2 == 0 ? 'a' : 'b')); + } + auto other = anchor; + std::mt19937 rng(55); + std::shuffle(other.begin(), other.end(), rng); + auto predicate = [](const std::string& s) { return s.front() == 'a'; }; + double median_ms = 0; + if (operation == "join") { + median_ms = Measure([&] { return JoinByItem(anchor, other); }, count); + } else if (operation == "group") { + median_ms = + Measure([&] { return GroupByRequest(anchor, other); }, count / 4); + } else if (operation == "select") { + median_ms = + Measure([&] { return SelectBatch(anchor, predicate); }, count / 2); + } else if (operation == "scatter") { + auto selection = SelectBatch(anchor, predicate); + if (!selection.ok()) return 2; + auto replacements = selection.value().Materialize(); + std::shuffle(replacements.begin(), replacements.end(), rng); + median_ms = Measure( + [&] { return ScatterReplace(selection.value(), replacements); }, count); + } else if (operation == "split") { + median_ms = Measure( + [&]() -> NodeResult { + auto result = SplitPayloads(anchor, [](const std::string& s) { + return std::vector{s.substr(0, 64), s.substr(64)}; + }); + if (!result.ok()) { + return NodeResult::Failure( + std::move(result).ExtractFailure()); + } + return NodeResult::Success( + std::move(result.value().children)); + }, + count * 2); + } else { + return 1; + } + rusage usage{}; + if (getrusage(RUSAGE_SELF, &usage) != 0) return 2; + std::cout << nlohmann::json({{"operation", operation}, + {"items", count}, + {"median_ms", median_ms}, + {"process_peak_rss_kib", usage.ru_maxrss}}) + .dump() + << '\n'; +} diff --git a/dev_support/node_authoring/starter_batch_group_node.cpp b/dev_support/node_authoring/starter_batch_group_node.cpp new file mode 100644 index 00000000..0281e54d --- /dev/null +++ b/dev_support/node_authoring/starter_batch_group_node.cpp @@ -0,0 +1,76 @@ +#include + +#include "nodes/authoring.h" + +namespace llm_edgeflow { +namespace custom_nodes { +namespace starter_batch_group { + +struct Inputs { + const TextBatch* queries = nullptr; + const TextBatch* references = nullptr; +}; + +struct Options {}; + +struct Models { + LlmCall generator; +}; + +NodeResult Run(const Inputs& inputs, const Options& /*options*/, + const Models& models) { + if (!inputs.queries || inputs.queries->empty()) { + return NodeResult::Success(TextBatch{}); + } + + TextBatch empty_refs; + const auto& refs = inputs.references ? *inputs.references : empty_refs; + auto group_res = GroupByRequest(*inputs.queries, refs); + if (!group_res.ok()) { + return NodeResult::Failure( + std::move(group_res).ExtractFailure()); + } + + const auto& view = group_res.value(); + TextBatch prompts; + prompts.reserve(inputs.queries->size()); + + // Preserve anchor order for 1:1 PreservedOutput alignment + for (size_t i = 0; i < inputs.queries->size(); ++i) { + const auto& query_item = (*inputs.queries)[i]; + const auto& group = view.GroupByAnchorIndex(i); + + std::string context_text; + for (const auto& ref_item : group.members()) { + context_text += ref_item.get().data + "\n"; + } + + std::string full_prompt = context_text + query_item.data; + prompts.emplace_back(query_item.req_id, query_item.sub_id, + std::move(full_prompt)); + } + + return models.generator.Generate(prompts); +} + +auto Spec() { + return MakeBatchSpec(InputsOf({ + Required("queries", &Inputs::queries), + Optional("references", &Inputs::references, + InputFlow::AggregateByRequest), + }), + PreservedOutput("output", "queries"), + Parameters({}), + ModelsOf({ + Llm("generator", "bind_model", &Models::generator), + }), + &Run) + .Description( + "Batch starter with traceable GroupByRequest reference aggregation"); +} + +REGISTER_FUNCTION_NODE(StarterBatchGroupNode, Spec()); + +} // namespace starter_batch_group +} // namespace custom_nodes +} // namespace llm_edgeflow diff --git a/dev_support/node_authoring/starter_batch_join_node.cpp b/dev_support/node_authoring/starter_batch_join_node.cpp new file mode 100644 index 00000000..2340c010 --- /dev/null +++ b/dev_support/node_authoring/starter_batch_join_node.cpp @@ -0,0 +1,64 @@ +#include + +#include "nodes/authoring.h" + +namespace llm_edgeflow { +namespace custom_nodes { +namespace starter_batch_join { + +struct Inputs { + const TextBatch* questions = nullptr; + const TextBatch* attributes = nullptr; +}; + +struct Options {}; + +struct Models { + LlmCall generator; +}; + +NodeResult Run(const Inputs& inputs, const Options& /*options*/, + const Models& models) { + if (!inputs.questions || inputs.questions->empty()) { + return NodeResult::Success(TextBatch{}); + } + + TextBatch empty_attrs; + const auto& attrs = inputs.attributes ? *inputs.attributes : empty_attrs; + auto joined = JoinByItem(*inputs.questions, attrs, JoinMode::kLeft); + if (!joined.ok()) { + return NodeResult::Failure(std::move(joined).ExtractFailure()); + } + + TextBatch prompts; + prompts.reserve(joined.value().size()); + for (const auto& row : joined.value()) { + std::string prompt = row.left_payload(); + if (row.has_right()) { + prompt += " [attr: " + *row.right_payload() + "]"; + } + prompts.emplace_back(row.req_id(), row.sub_id(), std::move(prompt)); + } + + return models.generator.Generate(prompts); +} + +auto Spec() { + return MakeBatchSpec(InputsOf({ + Required("questions", &Inputs::questions), + Optional("attributes", &Inputs::attributes), + }), + PreservedOutput("output", "questions"), + Parameters({}), + ModelsOf({ + Llm("generator", "bind_model", &Models::generator), + }), + &Run) + .Description("Batch starter with traceable Left Join across inputs"); +} + +REGISTER_FUNCTION_NODE(StarterBatchJoinNode, Spec()); + +} // namespace starter_batch_join +} // namespace custom_nodes +} // namespace llm_edgeflow diff --git a/dev_support/node_authoring/starter_batch_select_scatter_node.cpp b/dev_support/node_authoring/starter_batch_select_scatter_node.cpp new file mode 100644 index 00000000..975dfa4f --- /dev/null +++ b/dev_support/node_authoring/starter_batch_select_scatter_node.cpp @@ -0,0 +1,87 @@ +#include + +#include "nodes/authoring.h" + +namespace llm_edgeflow { +namespace custom_nodes { +namespace starter_batch_select_scatter { + +struct Inputs { + const TextBatch* input = nullptr; +}; + +struct Options { + std::string polish_tag = "[POLISH]"; +}; + +struct Models { + LlmCall generator; + LlmCall polisher; +}; + +NodeResult Run(const Inputs& inputs, const Options& options, + const Models& models) { + if (!inputs.input || inputs.input->empty()) { + return NodeResult::Success(TextBatch{}); + } + + // Round 1: Generate initial drafts for all inputs + auto first_res = models.generator.Generate(*inputs.input); + if (!first_res.ok()) { + return first_res; + } + + const auto& drafts = first_res.value(); + + // Step 2: Select items that require polishing (contain polish_tag) + auto selection_res = SelectBatch(drafts, [&](const std::string& text) { + return text.find(options.polish_tag) != std::string::npos; + }); + if (!selection_res.ok()) { + return NodeResult::Failure( + std::move(selection_res).ExtractFailure()); + } + + const auto& selection = selection_res.value(); + if (selection.empty()) { + // Skip second model call when no items need polishing + return first_res; + } + + // Materialize owned sub-batch for the second model call + TextBatch sub_batch = selection.Materialize(); + + // Round 2: Call polisher model only on selected sub-batch + auto second_res = models.polisher.Generate(sub_batch); + if (!second_res.ok()) { + return second_res; + } + + // Step 3: Scatter-replace polished items back into full draft batch + return ScatterReplace(selection, second_res.value()); +} + +auto Spec() { + return MakeBatchSpec( + InputsOf({ + Required("input", &Inputs::input), + }), + PreservedOutput("output", "input"), + Parameters({ + Field("polish_tag", &Options::polish_tag).Default("[POLISH]"), + }), + ModelsOf({ + Llm("generator", "bind_model", &Models::generator), + Llm("polisher", "polish_model", &Models::polisher), + }), + &Run) + .Description( + "Batch starter with conditional sub-batch LLM refinement and " + "scatter replacement"); +} + +REGISTER_FUNCTION_NODE(StarterBatchSelectScatterNode, Spec()); + +} // namespace starter_batch_select_scatter +} // namespace custom_nodes +} // namespace llm_edgeflow diff --git a/doc/CHANGELOG.md b/doc/CHANGELOG.md index 6952f119..879e768b 100644 --- a/doc/CHANGELOG.md +++ b/doc/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## 2026-09-14 批次关联、分组、选择回填与拆分公共工具（RFC-0055） + +- **可追踪批次公共操作**：在能力节点层引入 `include/nodes/traceable_batch_operations.h`（通过 `include/nodes/authoring.h` 导出），提供普通函数与轻量借用视图： + - `JoinByItem`：按 `(req_id, sub_id)` 完整来源键实现精确匹配（`exact`）与左外关联（`left`）视图，左侧批次原始顺序严格稳定，右侧缺失或冗余 Fail-Closed。 + - `GroupByRequest`：以显式 `anchor` 批次首次出现顺序作为分组顺序，保留子批次内部元素相对顺序，保留零子项空组。 + - `SelectBatch` 与 `ScatterReplace`：支持基于谓词筛选子批次，通过 `Selection::Materialize()` 安全物化；`ScatterReplace` 严格校验元素数量与完整来源键一致性，将二次推理结果按原序恢复回填到原始批次，保持未选中项不可变。 + - `SplitPayloads`：多项拆分并按请求分配单调递增连续 `sub_id`，生成 `SplitResult`（含拆分后批次与按父项来源键索引的子项计数），内置 `uint32` 与 `Int32` 边界容量防溢出校验。 + - 编译期临时对象保护：对全部借用视图构造器及公共操作禁用右值重载（`= delete`），防止悬垂引用。 +- **结构化批次失败详情**：在 `NodeResult` 中扩展可选的 `BatchFailureDetail` 与 `BatchFailureReason`（`duplicate`, `missing`, `unknown`, `count_mismatch`, `sub_id_overflow`, `count_overflow`, `callback_failed`），精准记录失败原因、请求项键及诊断信息，消除脆弱的字符串解析。 +- **TextChunkNode 试点迁移**：迁移至纯拆分算法 `SplitText` 与 `SplitPayloads`，将结构化批次失败原因确定性映射为历史业务错误码（`-4003` 重复输入、`-4004` 子编号溢出、`-4005` 计数溢出、`-4002` UTF-8 校验失败），保持黑板发布与端口契约完全不变。 +- **作者示例与自动化验证**：在 `dev_support/node_authoring/` 提供三个典型作者范例（Join, Group, Select/Scatter），新增 45 项独立单元测试覆盖乱序对齐、Int32/uint32 边界、const 临时对象拒绝、完整来源诊断与桩模型多轮推理；补充 ASan/UBSan、代表性性能测量及其环境限制记录。 + ## 2026-09-14 Control 作者接口与不可变配置快照收敛（RFC-0054） - **不可变配置快照组件**：引入 `ConfigurationSnapshot` 模板，提供原子读与写事务互斥保护，实现单请求整批不可变快照隔离与更新失败零状态泄露；支持 Move-Only 状态类型与原子快照发布；C++17 shared_ptr 原子操作不承诺无锁或无等待。 diff --git a/doc/dev_guide/custom_node_concepts.md b/doc/dev_guide/custom_node_concepts.md index 07a2127e..e17f045a 100644 --- a/doc/dev_guide/custom_node_concepts.md +++ b/doc/dev_guide/custom_node_concepts.md @@ -180,7 +180,9 @@ Definition 会帮助原生校验发现类型、字段和连线错误，但不会 ## 复杂算法仍按普通 C++ 函数组织 -当算法包含多个输入、循环推理或复杂后处理时，继续使用 `NodeBase`；需要模型句柄时用 +多个输入、条件二次推理或复杂后处理可组织在普通自由 Batch 函数中：显式声明 anchor， +最终返回与它等长、同序、同来源的 `PreservedOutput`。局部选择后须回填完整批次。 +需要发布拆分后的子项和父项计数等不同来源/数量的端口时，继续使用 `NodeBase`；需要模型句柄时用 `ModelBoundNode`。`ProcessNode` 负责读端口、取得本次配置快照、调用你的算法函数、检查 结果并发布。算法函数使用普通值、容器和局部变量；函数较多时拆成操作相关的 `.h/.cpp`， 再登记同目录 CMake，保持 `custom_nodes` 按操作组织。复杂程度本身不要求修改 Core。 @@ -190,6 +192,10 @@ Definition 会帮助原生校验发现类型、字段和连线错误，但不会 | 具体问题 | 可复用代码与边界 | | --- | --- | | 输入与模型输出一一对应 | [ValidatePreservedTraceableAlignment](../../include/nodes/traceable_batch_validation.h) 检查数量、顺序和两个来源编号；真实过滤/聚合不能套用 1:1 校验 | +| 两批数据按完整来源关联 | [Join 示例](../../dev_support/node_authoring/starter_batch_join_node.cpp) 使用 `JoinByItem`；显式选择 exact/left，右侧未知 key 均失败 | +| 按请求收集参考内容并保留空组 | [Group 示例](../../dev_support/node_authoring/starter_batch_group_node.cpp) 使用 `GroupByRequest`；按原 anchor 位置查询组，保持 A0/B0/A1 原序 | +| 只对部分结果再次推理 | [Select/Scatter 示例](../../dev_support/node_authoring/starter_batch_select_scatter_node.cpp) 先 `SelectBatch`、显式 `Materialize()`、调用模型，再 `ScatterReplace`；无选中项时跳过第二次调用 | +| 拆分载荷并分配子编号 | [TextChunkNode](../../src/common_nodes/text_chunk_node.cpp) 使用 `SplitPayloads`；每个请求连续分配子编号，counts 保留父 key，载荷回调只负责切分 | | 多个问题各自配多段材料 | [PromptGuidedLlmNode::ProcessNode](../../src/custom_nodes/prompt_guided_llm_node.cpp) 按 `req_id` 收集 context，主输出沿用 input 的 `(req_id, sub_id)` | | 候选打分、按请求分组、保留原候选来源 | [TextRerankNode::ProcessNode](../../src/common_nodes/text_rerank_node.cpp) 展示来源检查后再排序；新 rank 与原候选编号分别保存 | | 字段、默认值与范围 | [ValidateAndNormalizeFields](../../include/contracts/config_schema_validation.h)，Definition 与 Init 共用一份字段列表 | @@ -197,6 +203,12 @@ Definition 会帮助原生校验发现类型、字段和连线错误，但不会 | 初值与运行时更新使用同一业务校验 | [Control 模板](../../dev_support/node_authoring/starter_control_node.cpp) 的局部解析函数，失败不替换旧配置 | | 提示词变量替换 | [现有模板工具](../../include/nodes/text_template.h)，只在实际需要模板语义时使用 | +批次工具由 `nodes/authoring.h` 提供，返回 `NodeResult`。Join/Group/Selection 借用输入， +拒绝临时批次；使用期间输入必须存活且不修改、不移动。视图仅用于本次请求内的同步算法， +不能保存到 Node/Session 或异步任务。`Materialize`、Scatter 和 Split 的输出拥有数据。 +错误在 AuthorNode 边界统一写入诊断，保留回调错误码、内容及完整来源 key；普通算法不提前 +写 Context。工具要求完整 key 唯一，不改变未使用这些工具的 Map/模型重复 key 行为。 + 先声明结果数量和来源，再编码。例如“两条输入各输出一条”必须保留两组编号；“每个问题 取前三个候选”要按请求分组并声明排名来源，不能用整个 batch 的前三项代替。 多输入不能仅凭数组下标配对：一对一数据用两个编号关联，片段聚合按声明的请求关系处理。 diff --git a/doc/rfcs/0055-traceable-batch-operations.md b/doc/rfcs/0055-traceable-batch-operations.md index a012195b..43e54338 100644 --- a/doc/rfcs/0055-traceable-batch-operations.md +++ b/doc/rfcs/0055-traceable-batch-operations.md @@ -2,14 +2,14 @@ - **RFC 编号**：0055-traceable-batch-operations - **创建日期**：2026-09-14 -- **文档状态**：Proposed -- **关联分支**：`docs/framework-authoring-rfcs`；建议实施分支 `refactor/traceable-batch-operations` +- **文档状态**：Completed +- **关联分支**：`refactor/traceable-batch-operations` - **目标版本**：下一次投产前开发接口版本；保持 Catalog v3 - **负责人 / 作者**：LLM-EdgeFlow contributors - **设计基线**：`3fb4ba5be18f00fd8855b7d2de900e80ad203335` - **关联决策**：补充 RFC-0012、0022、0039、0052；实现 RFC-0052 第 8.3/10 节的一部分局部来源工具，保留独立 Filter/Group Node 作者策略的延期边界。 -本文是待实施规格，新增 API 均为拟议名称。 +本文规格已完成实施，新增公共工具位于 `include/nodes/traceable_batch_operations.h`。 与 [RFC-0053](0053-function-oriented-adapter-authoring.md)、 [RFC-0054](0054-controlled-configuration-snapshots.md) 独立。 开发与交付流程遵循 [CONTRIBUTING](../../CONTRIBUTING.md)。 @@ -287,10 +287,23 @@ policy 插槽；本轮不为局部 helper 重构它。这样可以让普通批 | 项目 | 当前状态 | | --- | --- | -| 设计文档 | Proposed；范围限定为局部工具及 TextChunk 试点 | -| 工具与试点实现 | 未开始 | -| 工程、所有权与性能验证 | 待实施后填写实际命令、基线与结果 | -| 开发者试用 | 待记录 | -| 完成条件 | M0–M5 必需交付、验证、现行指南及体验结果记录完成；按 CONTRIBUTING 更新状态 | - -当前文档门禁不构成工具已实现的证据。实施结果直接更新本文，不另建重复接续计划。 +| 设计与工具实现 | Completed；范围限定为局部工具及 TextChunk 试点 | +| 工程、所有权与性能验证 | 45 项批次工具测试、TextChunk/函数式 Node 验证、ASan/UBSan、代表性成本测量与 canonical gate；具体命令、结果与限制见下方记录 | +| 作者指南 | 已补普通 Batch 工具入口、借用生命周期、失败诊断及三个编译示例导航 | +| 真实开发者试用 | 待办；三个 starter 的 Agent 工程检查不替代组合任务的真实试用 | +| 工程完成条件 | M0–M5 工程实现和验证完成；真实试用按第 10 节保留待办，不宣称已通过 | + +[修复、验证与测量记录](reviews/0055-traceable-batch-verification.md) 包含运行环境、 +const 右值及非法谓词的编译拒绝、诊断完整 key 回归、Int32/uint32 边界、独立 Reviewer +复核、sanitizer、Pipeline/Demo、哈希索引次数、耗时/内存与未覆盖范围。 + +- **M0–M1**：Catalog 确认 TextChunk 既有端口；Join/Group 覆盖重复、缺失、空组、乱序及 + const/non-const 临时对象拒绝，保持 anchor 顺序与借用生命周期规则。 +- **M2**：Selection 只能由工厂产生，Materialize 拥有数据；Scatter 按完整 key 恢复全量。 + 谓词错误返回类型编译失败，回调失败在 AuthorNode 边界补完整来源且保留原错误码/内容。 +- **M3**：TextChunk 复用 SplitPayloads，保持 UTF-8、空字符串、overlap、来源及历史错误码。 + Int32 受检转换与 UINT32_MAX 接缝分别验证容量边界，不分配数十亿子项。 +- **M4**：Join、Group、Select/Scatter 编译示例由 NodeHarness 与 Mock 模型测试，断言输入、 + 实际调用次数及最终文本；缺失真实开发者组合任务体验明确列为待办。 +- **M5**：补齐作者概念/示例导航、方案执行与成本记录。Debug ASan/UBSan 聚焦检查和 + Release 默认后端 canonical gate 分别运行，不将单次门禁写成 Debug/Release 双配置验收。 diff --git a/doc/rfcs/README.md b/doc/rfcs/README.md index 8051861f..14368f1f 100644 --- a/doc/rfcs/README.md +++ b/doc/rfcs/README.md @@ -24,11 +24,8 @@ | **RFC-0051** | 开发者任务路径、测试生成与修复诊断 | `In Implementation` | `v10.x` | 流程编排层、能力节点层、模型执行层 / Tooling / Docs | [0051-developer-task-experience.md](0051-developer-task-experience.md) | | **RFC-0052** | 面向基础 C++ 开发者的 Node 作者接口重构 | `In Implementation` | 投产前 / Catalog v3 | 流程编排层、能力节点层 / Tooling / Docs | [0052-function-oriented-node-authoring.md](0052-function-oriented-node-authoring.md) | | **RFC-0054** | Control 作者接口与不可变配置快照 | `In Implementation` | 投产前 | 能力节点层 / Tooling / Docs | [0054-controlled-configuration-snapshots.md](0054-controlled-configuration-snapshots.md) | -| **RFC-0055** | 批次关联、分组、选择回填与拆分公共工具 | `Proposed` | 投产前 / Catalog v3 | 能力节点层 / Tooling / Docs | [0055-traceable-batch-operations.md](0055-traceable-batch-operations.md) | -RFC-0054–0055 是接续 RFC-0052 与已交付 RFC-0053 的实施规格。两篇没有相互编译依赖， -可以独立交付。`Proposed` 表示设计方案已形成，其中拟议 API、迁移与验证尚未实施， -不作为当前 SDK 功能说明。 +RFC-0054 是接续 RFC-0052 与已交付 RFC-0053、RFC-0055 的实施规格。`Proposed` 表示设计方案已形成，其中拟议 API、迁移与验证尚未实施，不作为当前 SDK 功能说明。 ## 已完成的 RFC @@ -83,6 +80,7 @@ RFC-0054–0055 是接续 RFC-0052 与已交付 RFC-0053 的实施规格。两 | **RFC-0049** | Operator 多输出与可注册的单份分配方案 | `Completed` | `v10.x` | 接入适配层 / Tooling | [0049-operator-output-allocation-strategies.md](0049-operator-output-allocation-strategies.md) | | **RFC-0050** | Operator 配置读取边界与普通参数结构 | `Completed` | `v10.x` | 接入适配层 / Tooling | [0050-operator-configuration-text-boundary.md](0050-operator-configuration-text-boundary.md) | | **RFC-0053** | 业务 Adapter 函数式作者接口与载体机制收敛 | `Completed` | 投产前 | 接入适配层 / Tooling / Docs | [0053-function-oriented-adapter-authoring.md](0053-function-oriented-adapter-authoring.md) | +| **RFC-0055** | 批次关联、分组、选择回填与拆分公共工具 | `Completed` | 投产前 / Catalog v3 | 能力节点层 / Tooling / Docs | [0055-traceable-batch-operations.md](0055-traceable-batch-operations.md) | ## 专项验收与评审归档 diff --git a/doc/rfcs/reviews/0055-traceable-batch-benchmark.json b/doc/rfcs/reviews/0055-traceable-batch-benchmark.json new file mode 100644 index 00000000..3440f06a --- /dev/null +++ b/doc/rfcs/reviews/0055-traceable-batch-benchmark.json @@ -0,0 +1,96 @@ +{ + "environment": "Ubuntu 24.04.4 LTS; Linux 6.17.0-1020-oracle aarch64; GCC 13.3.0; C++17 -O3 -DNDEBUG", + "method": "Independent process per workload; no concurrent project build; 1 warmup + 7 timed constructions, median; 128-byte payload, 4 items per request; shuffle seed 55; half selected; split into two 64-byte payloads. Process peak RSS includes inputs, preparation and allocator retention.", + "measurements": [ + { + "items": 1000, + "median_ms": 0.193442, + "operation": "join", + "process_peak_rss_kib": 11592 + }, + { + "items": 1000, + "median_ms": 0.276962, + "operation": "group", + "process_peak_rss_kib": 11592 + }, + { + "items": 1000, + "median_ms": 0.081521, + "operation": "select", + "process_peak_rss_kib": 11592 + }, + { + "items": 1000, + "median_ms": 0.144762, + "operation": "scatter", + "process_peak_rss_kib": 11592 + }, + { + "items": 1000, + "median_ms": 0.240122, + "operation": "split", + "process_peak_rss_kib": 11592 + }, + { + "items": 10000, + "median_ms": 2.251021, + "operation": "join", + "process_peak_rss_kib": 11592 + }, + { + "items": 10000, + "median_ms": 3.319312, + "operation": "group", + "process_peak_rss_kib": 11592 + }, + { + "items": 10000, + "median_ms": 1.08413, + "operation": "select", + "process_peak_rss_kib": 11592 + }, + { + "items": 10000, + "median_ms": 2.12002, + "operation": "scatter", + "process_peak_rss_kib": 11592 + }, + { + "items": 10000, + "median_ms": 2.885067, + "operation": "split", + "process_peak_rss_kib": 11592 + }, + { + "items": 100000, + "median_ms": 42.177126, + "operation": "join", + "process_peak_rss_kib": 48236 + }, + { + "items": 100000, + "median_ms": 100.475607, + "operation": "group", + "process_peak_rss_kib": 53912 + }, + { + "items": 100000, + "median_ms": 22.90506, + "operation": "select", + "process_peak_rss_kib": 44212 + }, + { + "items": 100000, + "median_ms": 39.727142, + "operation": "scatter", + "process_peak_rss_kib": 70344 + }, + { + "items": 100000, + "median_ms": 45.235635, + "operation": "split", + "process_peak_rss_kib": 76704 + } + ] +} diff --git a/doc/rfcs/reviews/0055-traceable-batch-verification.md b/doc/rfcs/reviews/0055-traceable-batch-verification.md new file mode 100644 index 00000000..8c6f9652 --- /dev/null +++ b/doc/rfcs/reviews/0055-traceable-batch-verification.md @@ -0,0 +1,122 @@ +# RFC-0055 修复与验证记录（2026-09-14） + +本记录区分工程验证、Mock 方案执行与真实开发者试用。测量环境为 Ubuntu 24.04.4 LTS、 +Linux 6.17.0-1020-oracle aarch64、GCC 13.3.0、CMake 3.28.3；不含公司内网 SDK 或目标硬件。 + +## 修复与覆盖 + +- Join/Group 的工厂和直接视图构造器，以及 Selection 工厂，拒绝 const/non-const 临时批次。 + `CompileTimeRejectionOfRvalues` 覆盖左右输入的组合；原 const 值返回工厂复现现在编译失败。 +- SelectBatch 的载荷谓词与 TraceableItem 谓词分支都只接受 bool/NodeResult。 + 原返回 int 的谓词复现现在由 static_assert 拒绝，不能静默跳过回调。 +- AuthorNode 统一格式化 BatchFailureDetail，保留回调数值码和原消息。缺少 sub_id、数字前缀、 + `xreq_id` 等非完整 token 不会阻止补充真正的父 key。聚焦测试包括 Select、Split、Map 的实际 + AuthorNode 诊断，以及没有输出发布的保序失败路径。 +- `detail::CheckedSplitCount` 被生产 SplitPayloads 复用。测试直接传入 0、INT32_MAX-1、 + INT32_MAX、INT32_MAX+1 和 SIZE_MAX，验证受检转换及失败详情，不分配巨量子批次。 + 子编号测试覆盖 UINT32_MAX 处一个子项成功、随后零子项成功、额外子项失败。 +- TextChunk 的 children 仍为 `1:N/generate_sub_id`，counts 为 `1:1/preserve`；历史错误码、 + UTF-8、空字符串、overlap 及交错父项来源由现有 TextChunk 套件验证。 +- 独立 Reviewer 只读复核上述修复和新增测试，未发现剩余确定缺陷。 + +测试来源：[批次工具](../../../tests/unit/nodes/test_traceable_batch_operations.cpp)、 +[TextChunk](../../../tests/unit/nodes/test_text_chunk_node.cpp)。 + +## ASan/UBSan 与门禁 + +复用隔离的 sanitizer 构建目录，只构建节点 runner；关闭 vendor backend，覆盖本次公共工具、 +函数式 Node 与 TextChunk。完整默认后端配置由最后一次 canonical gate 单独验证。 + +```bash +cmake -S . -B build-sanitizers-address-undefined-fast \ + -DBUILD_TESTING=ON -DENABLE_SANITIZERS=ON \ + -DLLM_EDGEFLOW_SANITIZERS=address,undefined -DCMAKE_BUILD_TYPE=Debug \ + -DENABLE_LLAMACPP=OFF -DENABLE_ONNXRUNTIME=OFF \ + -DENABLE_KITELLM=OFF -DENABLE_WHISPERCPP=OFF \ + -DLLM_EDGEFLOW_SHARDED_TEST_RUNNERS=ON -DLLM_EDGEFLOW_TEST_PCH=OFF +cmake --build build-sanitizers-address-undefined-fast \ + --target edgeflow_test_nodes_runner -j2 +ASAN_OPTIONS=detect_leaks=1:halt_on_error=1 UBSAN_OPTIONS=halt_on_error=1:print_stacktrace=1 \ + ./build-sanitizers-address-undefined-fast/edgeflow_test_nodes_runner \ + --gtest_filter='TraceableBatchOperationsTest.*:TextChunkNodeTest.*:FunctionNodeTest.*' +./scripts/run_all_tests.sh +``` + +执行结果：Debug ASan/UBSan/LSan 聚焦检查 **95/95** 通过（45 项批次工具、11 项 TextChunk、 +39 项函数式 Node），无 sanitizer 报告；Release canonical gate **97/97** CTest 项通过。 +canonical gate 的 sanitizer 关闭，不将其描述为同时完成 Debug/Release 或 sanitizer 验证。 + +## Pipeline 与 Demo + +```bash +./build/alg_pipeline_tool describe-node TextChunkNode +./build/alg_pipeline_tool validate configs/pipeline_doc_qa_default.json +./build/alg_pipeline_tool plan configs/pipeline_doc_qa_default.json +./build/alg_pipeline_tool_test validate demo/fixtures/mock/pipeline_doc_qa.json +./build/alg_pipeline_tool_test plan demo/fixtures/mock/pipeline_doc_qa.json +./build/alg_demo --profile doc_qa_mock --output-dir /tmp/rfc0055-demo +``` + +静态 validate/plan 成功。Mock Profile 使用自己的注册模型和配置，通过 Demo 的 Operator/C ABI +路径执行；不是实际模型效果验收。两条输出均 status=0，request_id 分别为 10001、10002， +chunk_count 均为 2。答案逐项检查为： + +1. `【LLM总结】文档核心为现代软件工程化设计，包含松耦合、状态隔离与跨平台编译。` +2. `【LLM意图分析】检测到售后退款诉求。建议操作：7天无理由退货审核流程。` + +counts 在 Node 内保存父来源计数，经 DocQA Adapter 组装为外部 `chunk_count`，两项均与已知 +语料的拆分期望一致。默认真实模型 Pipeline 仅完成静态验证，没有声明真实推理效果。 + +## 批次耗时、内存与索引 + +使用[独立测量程序](../../../dev_support/benchmarks/traceable_batch_operations.cpp)，不增加测试 +runner 或生产 Catalog 注册。构造 1,000/10,000/100,000 项批次，每项 128 字节、每请求四项； +右侧用种子 55 打乱，选择一半项，Split 每父项生成两段 64 字节文本。 + +```bash +c++ -std=c++17 -O3 -DNDEBUG -Iinclude -I3rdparty/nlohmann_json/include \ + dev_support/benchmarks/traceable_batch_operations.cpp -o /tmp/rfc0055-benchmark +for count in 1000 10000 100000; do + for operation in join group select scatter split; do + /tmp/rfc0055-benchmark "$operation" "$count" + done +done +``` + +每个操作独立进程，一轮预热、七轮测量取中位数。输入构造不计时，Scatter 的 Selection 与 +Materialize 不计时；输出构造及复制计时，析构在计时外（Split 的 counts 析构包含在包装中）。 +RSS 为 Linux 进程峰值，包含输入、预备数据及分配器保留内存，不等于 helper 净分配量。 +这是一台外部开发机的代表性成本记录，不是跨版本加速比或生产性能承诺。 + +原始测量见 [JSON 记录](0055-traceable-batch-benchmark.json)。耗时单位 ms，RSS 单位 KiB： + +| 操作 | 1,000 项 | 10,000 项 | 100,000 项 | 100,000 项进程峰值 RSS | +| --- | ---: | ---: | ---: | ---: | +| join | 0.193 | 2.251 | 42.177 | 48236 | +| group | 0.277 | 3.319 | 100.476 | 53912 | +| select | 0.082 | 1.084 | 22.905 | 44212 | +| scatter | 0.145 | 2.120 | 39.727 | 70344 | +| split | 0.240 | 2.885 | 45.236 | 76704 | + +测量期间无并发项目构建。小规模 RSS 含进程启动基线；更大批次的缓存/分配器成本也体现在 +耗时中，不用三个测量点声称严格线性性能或绝对内存开销。 + +| 操作 | 每次调用的哈希索引构建次数（源码检查） | 用途 | +| --- | --- | --- | +| Join | 2 | 左 key 位置、右 key 引用 | +| Group | 3 | anchor 去重、member 去重、req 到组位置；返回视图持有后者 | +| Select | 1 | anchor 去重 | +| Scatter | 2 | replacement key 索引、selected key 集合 | +| Split | 2 | 输入去重、每请求下一个 sub_id | + +这些索引均在单次调用内建立，循环内只增量查询/插入，不为每个请求重新扫描或建立全批次 +索引。预期时间、额外空间均随项数线性增长；哈希容器不承诺最坏情况常数查询。 + +## 作者体验与剩余边界 + +工程检查使用三个编译 starter：Join、参考 Group、Select/Scatter。分组示例无需作者手写 +key 索引；条件生成示例正常路径调用 generator 一次、polisher 一次，第二轮输入恰为选中项； +全不选时 polisher 零次，失败直接传播。测试同时断言 prompt、来源和最终文本。 + +真实开发者将参考分组和条件二次推理组合为一个 Node 的试用尚未进行，求助次数、编辑位置、 +耗时等真实体验数据保留待办。上述 Agent 工程检查不能代替该记录，不将其写成“真实试用通过”。 diff --git a/include/nodes/authoring.h b/include/nodes/authoring.h index 0f9049f9..019b3cb1 100644 --- a/include/nodes/authoring.h +++ b/include/nodes/authoring.h @@ -10,3 +10,4 @@ #include "nodes/node_result.h" #include "nodes/parameter_binding.h" #include "nodes/traceable_algorithms.h" +#include "nodes/traceable_batch_operations.h" diff --git a/include/nodes/function_node.h b/include/nodes/function_node.h index 11c792e5..10756003 100644 --- a/include/nodes/function_node.h +++ b/include/nodes/function_node.h @@ -1253,13 +1253,17 @@ class AuthorNode> auto res = detail::InvokeMapItem(spec_.Function(), item.data, params); if (!res.ok()) { auto failure = std::move(res).ExtractFailure(); + if (!failure.batch_detail.has_value()) { + failure.batch_detail = BatchFailureDetail{ + this->Name(), BatchFailureReason::kCallbackFailed, + TraceableItemKey{item.req_id, item.sub_id}}; + } int code = failure.cause_code != 0 ? failure.cause_code : node_error::author_node::kBusinessError; - return this->Fail(req_ctx, code, - failure.message.empty() - ? (this->Name() + " map function failed") - : failure.message); + return this->Fail( + req_ctx, code, + failure.FormatDiagnostic(this->Name() + " map function failed")); } outputs.emplace_back(item.req_id, item.sub_id, std::move(res).value()); } else { @@ -1412,10 +1416,9 @@ class AuthorNode> int code = failure.cause_code != 0 ? failure.cause_code : node_error::author_node::kBusinessError; - return this->Fail(req_ctx, code, - failure.message.empty() - ? (this->Name() + " process failed") - : failure.message); + return this->Fail( + req_ctx, code, + failure.FormatDiagnostic(this->Name() + " process failed")); } OutputBatchT output = std::move(res).value(); diff --git a/include/nodes/node_result.h b/include/nodes/node_result.h index ed19bf68..fab5d812 100644 --- a/include/nodes/node_result.h +++ b/include/nodes/node_result.h @@ -1,7 +1,12 @@ #pragma once +#include +#include +#include +#include #include #include +#include #include #include @@ -34,6 +39,87 @@ inline const char* NodeErrorKindName(NodeErrorKind kind) noexcept { return "UnknownError"; } +enum class BatchFailureReason { + kDuplicate, + kMissing, + kUnknown, + kCountMismatch, + kSubIdOverflow, + kCountOverflow, + kCallbackFailed, +}; + +inline const char* BatchFailureReasonName(BatchFailureReason reason) noexcept { + switch (reason) { + case BatchFailureReason::kDuplicate: + return "duplicate"; + case BatchFailureReason::kMissing: + return "missing"; + case BatchFailureReason::kUnknown: + return "unknown"; + case BatchFailureReason::kCountMismatch: + return "count_mismatch"; + case BatchFailureReason::kSubIdOverflow: + return "sub_id_overflow"; + case BatchFailureReason::kCountOverflow: + return "count_overflow"; + case BatchFailureReason::kCallbackFailed: + return "callback_failed"; + } + return "unknown"; +} + +struct TraceableItemKey { + uint32_t req_id = 0; + uint32_t sub_id = 0; + + constexpr bool operator==(const TraceableItemKey& other) const noexcept { + return req_id == other.req_id && sub_id == other.sub_id; + } + constexpr bool operator!=(const TraceableItemKey& other) const noexcept { + return !(*this == other); + } + constexpr bool operator<(const TraceableItemKey& other) const noexcept { + if (req_id != other.req_id) return req_id < other.req_id; + return sub_id < other.sub_id; + } +}; + +struct TraceableItemKeyHash { + std::size_t operator()(const TraceableItemKey& k) const noexcept { + uint64_t x = (static_cast(k.req_id) << 32) | k.sub_id; + x ^= x >> 30; + x *= 0xbf58476d1ce4e5b9ULL; + x ^= x >> 27; + x *= 0x94d049bb133111ebULL; + x ^= x >> 31; + return static_cast(x); + } +}; + +struct BatchFailureDetail { + std::string operation; + BatchFailureReason reason = BatchFailureReason::kUnknown; + std::optional key; +}; + +inline std::string FormatBatchFailureDetail(const BatchFailureDetail& detail) { + std::string out; + if (!detail.operation.empty()) { + out += detail.operation; + } + if (detail.reason != BatchFailureReason::kUnknown) { + if (!out.empty()) out += " "; + out += BatchFailureReasonName(detail.reason); + } + if (detail.key.has_value()) { + if (!out.empty()) out += " for "; + out += "req_id=" + std::to_string(detail.key->req_id) + + ", sub_id=" + std::to_string(detail.key->sub_id); + } + return out; +} + struct NodeFailure { NodeErrorKind kind = NodeErrorKind::kBusinessError; std::string message; @@ -41,6 +127,48 @@ struct NodeFailure { std::string stage; std::string model_slot; std::string source_location; + std::optional batch_detail; + + std::string FormatDiagnostic(std::string_view fallback_message = {}) const { + std::string base = + message.empty() ? std::string(fallback_message) : message; + if (!batch_detail.has_value()) { + return base; + } + const auto& detail = *batch_detail; + std::string structured = FormatBatchFailureDetail(detail); + if (structured.empty()) { + return base; + } + if (base.empty()) { + return structured; + } + const auto contains_token = [&base](const std::string& token) { + const auto is_identifier = [](unsigned char ch) { + return std::isalnum(ch) != 0 || ch == '_'; + }; + size_t pos = base.find(token); + while (pos != std::string::npos) { + const size_t end = pos + token.size(); + if ((pos == 0 || !is_identifier(base[pos - 1])) && + (end == base.size() || !is_identifier(base[end]))) { + return true; + } + pos = base.find(token, pos + 1); + } + return false; + }; + // Only omit the prefix when the complete key is already present. A request + // alone, or a numeric prefix of another item's key, does not identify it. + if (detail.operation.empty() || contains_token(detail.operation)) { + if (!detail.key.has_value() || + contains_token("req_id=" + std::to_string(detail.key->req_id) + + ", sub_id=" + std::to_string(detail.key->sub_id))) { + return base; + } + } + return structured + ": " + base; + } NodeFailure() = default; NodeFailure(NodeErrorKind k, std::string msg, int cause = 0, @@ -51,6 +179,17 @@ struct NodeFailure { stage(std::move(stg)), model_slot(std::move(slot)), source_location(std::move(loc)) {} + + NodeFailure(NodeErrorKind k, std::string msg, BatchFailureDetail detail, + int cause = 0, std::string stg = {}, std::string slot = {}, + std::string loc = {}) + : kind(k), + message(std::move(msg)), + cause_code(cause), + stage(std::move(stg)), + model_slot(std::move(slot)), + source_location(std::move(loc)), + batch_detail(std::move(detail)) {} }; template @@ -80,6 +219,18 @@ class [[nodiscard]] NodeResult { std::move(model_slot), std::move(source_location))); } + static NodeResult Failure(NodeErrorKind kind, std::string message, + BatchFailureDetail detail, int cause_code = 0, + std::string stage = {}, + std::string model_slot = {}, + std::string source_location = {}) { + return NodeResult( + FailureTag{}, + NodeFailure(kind, std::move(message), std::move(detail), cause_code, + std::move(stage), std::move(model_slot), + std::move(source_location))); + } + bool ok() const noexcept { return std::holds_alternative(storage_); } explicit operator bool() const noexcept { return ok(); } @@ -130,3 +281,13 @@ class [[nodiscard]] NodeResult { }; } // namespace llm_edgeflow + +namespace std { +template <> +struct hash { + std::size_t operator()( + const llm_edgeflow::TraceableItemKey& k) const noexcept { + return llm_edgeflow::TraceableItemKeyHash{}(k); + } +}; +} // namespace std diff --git a/include/nodes/traceable_batch_operations.h b/include/nodes/traceable_batch_operations.h new file mode 100644 index 00000000..9703306b --- /dev/null +++ b/include/nodes/traceable_batch_operations.h @@ -0,0 +1,878 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "contracts/traceable_item.h" +#include "core/common_contracts.h" +#include "nodes/node_result.h" +#include "nodes/traceable_algorithms.h" + +namespace llm_edgeflow { + +// TraceableItemKeyHash and std::hash are defined in +// node_result.h for repository-wide availability. + +// ============================================================================ +// 1. JoinByItem +// ============================================================================ + +enum class JoinMode { + kExact, + kLeft, +}; + +template +struct JoinedRow { + const TraceableItem& left; + const TraceableItem* right = nullptr; + + JoinedRow(const TraceableItem& l, + const TraceableItem* r = nullptr) + : left(l), right(r) {} + JoinedRow(TraceableItem&&, + const TraceableItem* = nullptr) = delete; + JoinedRow(const TraceableItem&&, + const TraceableItem* = nullptr) = delete; + + JoinedRow(const JoinedRow&) = default; + JoinedRow(JoinedRow&&) = default; + + uint32_t req_id() const noexcept { return left.req_id; } + uint32_t sub_id() const noexcept { return left.sub_id; } + const LeftPayload& left_payload() const noexcept { return left.data; } + const RightPayload* right_payload() const noexcept { + return right ? &right->data : nullptr; + } + bool has_right() const noexcept { return right != nullptr; } +}; + +template +class ItemJoinView { + public: + using RowType = JoinedRow; + + ItemJoinView(const std::vector>& left, + const std::vector>& right, + std::vector rows) + : left_(&left), right_(&right), rows_(std::move(rows)) {} + + // Disallow construction from temporary / rvalue batches (both non-const and + // const) + ItemJoinView(std::vector>&&, + const std::vector>&, + std::vector) = delete; + ItemJoinView(const std::vector>&&, + const std::vector>&, + std::vector) = delete; + ItemJoinView(const std::vector>&, + std::vector>&&, + std::vector) = delete; + ItemJoinView(const std::vector>&, + const std::vector>&&, + std::vector) = delete; + ItemJoinView(std::vector>&&, + std::vector>&&, + std::vector) = delete; + ItemJoinView(std::vector>&&, + const std::vector>&&, + std::vector) = delete; + ItemJoinView(const std::vector>&&, + std::vector>&&, + std::vector) = delete; + ItemJoinView(const std::vector>&&, + const std::vector>&&, + std::vector) = delete; + + size_t size() const noexcept { return rows_.size(); } + bool empty() const noexcept { return rows_.empty(); } + + const RowType& operator[](size_t index) const { return rows_[index]; } + const RowType& at(size_t index) const { return rows_.at(index); } + + auto begin() const noexcept { return rows_.begin(); } + auto end() const noexcept { return rows_.end(); } + + const std::vector& rows() const noexcept { return rows_; } + const std::vector>& left_batch() const noexcept { + return *left_; + } + const std::vector>& right_batch() const noexcept { + return *right_; + } + + private: + const std::vector>* left_ = nullptr; + const std::vector>* right_ = nullptr; + std::vector rows_; +}; + +template +NodeResult> JoinByItem( + const std::vector>& left, + const std::vector>& right, + JoinMode mode = JoinMode::kExact) { + std::unordered_map left_key_to_idx; + left_key_to_idx.reserve(left.size()); + for (size_t i = 0; i < left.size(); ++i) { + TraceableItemKey key{left[i].req_id, left[i].sub_id}; + if (!left_key_to_idx.emplace(key, i).second) { + return NodeResult>::Failure( + NodeErrorKind::kInputError, + "JoinByItem left batch contains duplicate key: req_id=" + + std::to_string(key.req_id) + + ", sub_id=" + std::to_string(key.sub_id), + BatchFailureDetail{"JoinByItem", BatchFailureReason::kDuplicate, + key}); + } + } + + std::unordered_map*> + right_key_to_ptr; + right_key_to_ptr.reserve(right.size()); + for (size_t j = 0; j < right.size(); ++j) { + TraceableItemKey key{right[j].req_id, right[j].sub_id}; + if (!right_key_to_ptr.emplace(key, &right[j]).second) { + return NodeResult>::Failure( + NodeErrorKind::kInputError, + "JoinByItem right batch contains duplicate key: req_id=" + + std::to_string(key.req_id) + + ", sub_id=" + std::to_string(key.sub_id), + BatchFailureDetail{"JoinByItem", BatchFailureReason::kDuplicate, + key}); + } + } + + // Check for unknown keys in right (in both kExact and kLeft) + for (const auto& r_item : right) { + TraceableItemKey r_key{r_item.req_id, r_item.sub_id}; + if (left_key_to_idx.find(r_key) == left_key_to_idx.end()) { + return NodeResult>::Failure( + NodeErrorKind::kInputError, + "JoinByItem right batch contains unknown key not in left: req_id=" + + std::to_string(r_key.req_id) + + ", sub_id=" + std::to_string(r_key.sub_id), + BatchFailureDetail{"JoinByItem", BatchFailureReason::kUnknown, + r_key}); + } + } + + std::vector> rows; + rows.reserve(left.size()); + for (const auto& l_item : left) { + TraceableItemKey key{l_item.req_id, l_item.sub_id}; + auto it = right_key_to_ptr.find(key); + if (it == right_key_to_ptr.end()) { + if (mode == JoinMode::kExact) { + return NodeResult>::Failure( + NodeErrorKind::kInputError, + "JoinByItem right batch missing key present in left: req_id=" + + std::to_string(key.req_id) + + ", sub_id=" + std::to_string(key.sub_id), + BatchFailureDetail{"JoinByItem", BatchFailureReason::kMissing, + key}); + } + rows.push_back(JoinedRow{l_item, nullptr}); + } else { + rows.push_back(JoinedRow{l_item, it->second}); + } + } + + return NodeResult>::Success( + ItemJoinView(left, right, std::move(rows))); +} + +template +void JoinByItem(std::vector>&&, + const std::vector>&, + JoinMode = JoinMode::kExact) = delete; + +template +void JoinByItem(const std::vector>&&, + const std::vector>&, + JoinMode = JoinMode::kExact) = delete; + +template +void JoinByItem(const std::vector>&, + std::vector>&&, + JoinMode = JoinMode::kExact) = delete; + +template +void JoinByItem(const std::vector>&, + const std::vector>&&, + JoinMode = JoinMode::kExact) = delete; + +template +void JoinByItem(std::vector>&&, + std::vector>&&, + JoinMode = JoinMode::kExact) = delete; + +template +void JoinByItem(std::vector>&&, + const std::vector>&&, + JoinMode = JoinMode::kExact) = delete; + +template +void JoinByItem(const std::vector>&&, + std::vector>&&, + JoinMode = JoinMode::kExact) = delete; + +template +void JoinByItem(const std::vector>&&, + const std::vector>&&, + JoinMode = JoinMode::kExact) = delete; + +// ============================================================================ +// 2. GroupByRequest +// ============================================================================ + +template +class RequestGroup { + public: + using AnchorItem = TraceableItem; + using MemberItem = TraceableItem; + + explicit RequestGroup(uint32_t req_id) : req_id_(req_id) {} + + uint32_t req_id() const noexcept { return req_id_; } + + const std::vector>& anchors() + const noexcept { + return anchors_; + } + const std::vector>& members() + const noexcept { + return members_; + } + + size_t anchor_count() const noexcept { return anchors_.size(); } + size_t member_count() const noexcept { return members_.size(); } + bool has_members() const noexcept { return !members_.empty(); } + + void AddAnchor(const AnchorItem& item) { + anchors_.push_back(std::cref(item)); + } + void AddAnchor(AnchorItem&&) = delete; + void AddAnchor(const AnchorItem&&) = delete; + + void AddMember(const MemberItem& item) { + members_.push_back(std::cref(item)); + } + void AddMember(MemberItem&&) = delete; + void AddMember(const MemberItem&&) = delete; + + private: + uint32_t req_id_ = 0; + std::vector> anchors_; + std::vector> members_; +}; + +template +class RequestGroupView { + public: + using GroupType = RequestGroup; + + RequestGroupView(const std::vector>& anchor, + const std::vector>& members, + std::vector groups, + std::vector anchor_to_group_index, + std::unordered_map req_id_to_group_index) + : anchor_(&anchor), + members_(&members), + groups_(std::move(groups)), + anchor_to_group_index_(std::move(anchor_to_group_index)), + req_id_to_group_index_(std::move(req_id_to_group_index)) {} + + RequestGroupView(std::vector>&&, + const std::vector>&, + std::vector, std::vector, + std::unordered_map) = delete; + RequestGroupView(const std::vector>&&, + const std::vector>&, + std::vector, std::vector, + std::unordered_map) = delete; + RequestGroupView(const std::vector>&, + std::vector>&&, + std::vector, std::vector, + std::unordered_map) = delete; + RequestGroupView(const std::vector>&, + const std::vector>&&, + std::vector, std::vector, + std::unordered_map) = delete; + RequestGroupView(std::vector>&&, + std::vector>&&, + std::vector, std::vector, + std::unordered_map) = delete; + RequestGroupView(std::vector>&&, + const std::vector>&&, + std::vector, std::vector, + std::unordered_map) = delete; + RequestGroupView(const std::vector>&&, + std::vector>&&, + std::vector, std::vector, + std::unordered_map) = delete; + RequestGroupView(const std::vector>&&, + const std::vector>&&, + std::vector, std::vector, + std::unordered_map) = delete; + + size_t size() const noexcept { return groups_.size(); } + bool empty() const noexcept { return groups_.empty(); } + + const GroupType& operator[](size_t index) const { return groups_[index]; } + const GroupType& at(size_t index) const { return groups_.at(index); } + + auto begin() const noexcept { return groups_.begin(); } + auto end() const noexcept { return groups_.end(); } + + const std::vector& groups() const noexcept { return groups_; } + + const GroupType* FindByReqId(uint32_t req_id) const noexcept { + auto it = req_id_to_group_index_.find(req_id); + if (it == req_id_to_group_index_.end()) return nullptr; + return &groups_[it->second]; + } + + const GroupType* Find(uint32_t req_id) const noexcept { + return FindByReqId(req_id); + } + + bool HasReqId(uint32_t req_id) const noexcept { + return req_id_to_group_index_.find(req_id) != req_id_to_group_index_.end(); + } + + bool Contains(uint32_t req_id) const noexcept { return HasReqId(req_id); } + + const GroupType& GroupByAnchorIndex(size_t anchor_index) const { + return groups_.at(anchor_to_group_index_.at(anchor_index)); + } + + size_t GroupIndexForAnchor(size_t anchor_index) const { + return anchor_to_group_index_.at(anchor_index); + } + + const std::vector>& anchor_batch() + const noexcept { + return *anchor_; + } + const std::vector>& members_batch() + const noexcept { + return *members_; + } + + private: + const std::vector>* anchor_ = nullptr; + const std::vector>* members_ = nullptr; + std::vector groups_; + std::vector anchor_to_group_index_; + std::unordered_map req_id_to_group_index_; +}; + +template +NodeResult> GroupByRequest( + const std::vector>& anchor, + const std::vector>& members) { + std::unordered_set anchor_keys; + std::vector> groups; + std::unordered_map req_id_to_group_index; + std::vector anchor_to_group_index; + anchor_to_group_index.reserve(anchor.size()); + + for (size_t i = 0; i < anchor.size(); ++i) { + const auto& item = anchor[i]; + TraceableItemKey key{item.req_id, item.sub_id}; + if (!anchor_keys.insert(key).second) { + return NodeResult>:: + Failure(NodeErrorKind::kInputError, + "GroupByRequest anchor contains duplicate key: req_id=" + + std::to_string(key.req_id) + + ", sub_id=" + std::to_string(key.sub_id), + BatchFailureDetail{"GroupByRequest", + BatchFailureReason::kDuplicate, key}); + } + + auto it = req_id_to_group_index.find(item.req_id); + size_t group_idx = 0; + if (it == req_id_to_group_index.end()) { + group_idx = groups.size(); + groups.emplace_back(item.req_id); + req_id_to_group_index.emplace(item.req_id, group_idx); + } else { + group_idx = it->second; + } + groups[group_idx].AddAnchor(item); + anchor_to_group_index.push_back(group_idx); + } + + std::unordered_set member_keys; + for (const auto& item : members) { + TraceableItemKey key{item.req_id, item.sub_id}; + if (!member_keys.insert(key).second) { + return NodeResult>:: + Failure(NodeErrorKind::kInputError, + "GroupByRequest members contains duplicate key: req_id=" + + std::to_string(key.req_id) + + ", sub_id=" + std::to_string(key.sub_id), + BatchFailureDetail{"GroupByRequest", + BatchFailureReason::kDuplicate, key}); + } + + auto it = req_id_to_group_index.find(item.req_id); + if (it == req_id_to_group_index.end()) { + return NodeResult>:: + Failure(NodeErrorKind::kInputError, + "GroupByRequest member contains unknown req_id=" + + std::to_string(item.req_id) + " not present in anchor", + BatchFailureDetail{"GroupByRequest", + BatchFailureReason::kUnknown, key}); + } + groups[it->second].AddMember(item); + } + + return NodeResult>::Success( + RequestGroupView( + anchor, members, std::move(groups), std::move(anchor_to_group_index), + std::move(req_id_to_group_index))); +} + +template +void GroupByRequest(std::vector>&&, + const std::vector>&) = delete; + +template +void GroupByRequest(const std::vector>&&, + const std::vector>&) = delete; + +template +void GroupByRequest(const std::vector>&, + std::vector>&&) = delete; + +template +void GroupByRequest(const std::vector>&, + const std::vector>&&) = delete; + +template +void GroupByRequest(std::vector>&&, + std::vector>&&) = delete; + +template +void GroupByRequest(std::vector>&&, + const std::vector>&&) = delete; + +template +void GroupByRequest(const std::vector>&&, + std::vector>&&) = delete; + +template +void GroupByRequest(const std::vector>&&, + const std::vector>&&) = delete; + +// ============================================================================ +// 3. SelectBatch and ScatterReplace +// ============================================================================ + +template +class Selection { + public: + using ItemType = TraceableItem; + using BatchType = std::vector; + + Selection(std::vector&&, std::vector) = delete; + Selection(const std::vector&&, std::vector) = delete; + + size_t size() const noexcept { return selected_indices_.size(); } + bool empty() const noexcept { return selected_indices_.empty(); } + size_t total_anchor_size() const noexcept { + return anchor_ ? anchor_->size() : 0; + } + + const ItemType& operator[](size_t index) const { + return (*anchor_)[selected_indices_.at(index)]; + } + const ItemType& at(size_t index) const { + return (*anchor_)[selected_indices_.at(index)]; + } + + class const_iterator { + public: + using iterator_category = std::forward_iterator_tag; + using value_type = ItemType; + using difference_type = std::ptrdiff_t; + using pointer = const ItemType*; + using reference = const ItemType&; + + const_iterator(const BatchType* anchor, + std::vector::const_iterator it) + : anchor_(anchor), it_(it) {} + + reference operator*() const { return (*anchor_)[*it_]; } + pointer operator->() const { return &((*anchor_)[*it_]); } + + const_iterator& operator++() { + ++it_; + return *this; + } + const_iterator operator++(int) { + const_iterator tmp = *this; + ++it_; + return tmp; + } + + bool operator==(const const_iterator& other) const noexcept { + return it_ == other.it_; + } + bool operator!=(const const_iterator& other) const noexcept { + return !(*this == other); + } + + private: + const BatchType* anchor_ = nullptr; + std::vector::const_iterator it_; + }; + + const_iterator begin() const noexcept { + return const_iterator(anchor_, selected_indices_.cbegin()); + } + const_iterator end() const noexcept { + return const_iterator(anchor_, selected_indices_.cend()); + } + + BatchType Materialize() const { + BatchType result; + result.reserve(selected_indices_.size()); + for (size_t idx : selected_indices_) { + result.push_back((*anchor_)[idx]); + } + return result; + } + + const BatchType& original_batch() const noexcept { return *anchor_; } + const std::vector& selected_indices() const noexcept { + return selected_indices_; + } + + private: + template + friend NodeResult> SelectBatch( + const std::vector>& anchor, Pred&& predicate); + + Selection(const BatchType& anchor, std::vector indices) + : anchor_(&anchor), selected_indices_(std::move(indices)) {} + + const BatchType* anchor_ = nullptr; + std::vector selected_indices_; +}; + +template +NodeResult> SelectBatch( + const std::vector>& anchor, Predicate&& predicate) { + std::unordered_set seen_keys; + for (const auto& item : anchor) { + TraceableItemKey key{item.req_id, item.sub_id}; + if (!seen_keys.insert(key).second) { + return NodeResult>::Failure( + NodeErrorKind::kInputError, + "SelectBatch anchor contains duplicate key: req_id=" + + std::to_string(key.req_id) + + ", sub_id=" + std::to_string(key.sub_id), + BatchFailureDetail{"SelectBatch", BatchFailureReason::kDuplicate, + key}); + } + } + + std::vector selected_indices; + for (size_t i = 0; i < anchor.size(); ++i) { + const auto& item = anchor[i]; + try { + if constexpr (std::is_invocable_v) { + using Ret = std::invoke_result_t; + if constexpr (std::is_same_v) { + if (predicate(item.data)) { + selected_indices.push_back(i); + } + } else if constexpr (std::is_same_v>) { + auto res = predicate(item.data); + if (!res.ok()) { + auto failure = std::move(res).ExtractFailure(); + failure.batch_detail = BatchFailureDetail{ + "SelectBatch", BatchFailureReason::kCallbackFailed, + TraceableItemKey{item.req_id, item.sub_id}}; + return NodeResult>::Failure(std::move(failure)); + } + if (res.value()) { + selected_indices.push_back(i); + } + } else { + static_assert(std::is_same_v || + std::is_same_v>, + "SelectBatch predicate must return bool or " + "NodeResult"); + } + } else if constexpr (std::is_invocable_v&>) { + using Ret = + std::invoke_result_t&>; + if constexpr (std::is_same_v) { + if (predicate(item)) { + selected_indices.push_back(i); + } + } else if constexpr (std::is_same_v>) { + auto res = predicate(item); + if (!res.ok()) { + auto failure = std::move(res).ExtractFailure(); + failure.batch_detail = BatchFailureDetail{ + "SelectBatch", BatchFailureReason::kCallbackFailed, + TraceableItemKey{item.req_id, item.sub_id}}; + return NodeResult>::Failure(std::move(failure)); + } + if (res.value()) { + selected_indices.push_back(i); + } + } else { + static_assert(std::is_same_v || + std::is_same_v>, + "SelectBatch predicate must return bool or " + "NodeResult"); + } + } else { + static_assert( + std::is_invocable_v || + std::is_invocable_v&>, + "SelectBatch predicate must accept (const Payload&) or (const " + "TraceableItem&)"); + } + } catch (const std::exception& e) { + return NodeResult>::Failure( + NodeErrorKind::kBusinessError, + std::string("SelectBatch predicate threw exception: ") + e.what(), + BatchFailureDetail{"SelectBatch", BatchFailureReason::kCallbackFailed, + TraceableItemKey{item.req_id, item.sub_id}}); + } catch (...) { + return NodeResult>::Failure( + NodeErrorKind::kBusinessError, + "SelectBatch predicate threw unknown exception", + BatchFailureDetail{"SelectBatch", BatchFailureReason::kCallbackFailed, + TraceableItemKey{item.req_id, item.sub_id}}); + } + } + + return NodeResult>::Success( + Selection(anchor, std::move(selected_indices))); +} + +template +void SelectBatch(std::vector>&&, Predicate&&) = delete; + +template +void SelectBatch(const std::vector>&&, + Predicate&&) = delete; + +template +NodeResult>> ScatterReplace( + const Selection& selection, + const std::vector>& replacements) { + std::unordered_map*> repl_map; + repl_map.reserve(replacements.size()); + for (const auto& item : replacements) { + TraceableItemKey key{item.req_id, item.sub_id}; + if (!repl_map.emplace(key, &item).second) { + return NodeResult>>::Failure( + NodeErrorKind::kInputError, + "ScatterReplace replacements contains duplicate key: req_id=" + + std::to_string(key.req_id) + + ", sub_id=" + std::to_string(key.sub_id), + BatchFailureDetail{"ScatterReplace", BatchFailureReason::kDuplicate, + key}); + } + } + + std::unordered_set selected_keys; + const auto& anchor = selection.original_batch(); + for (size_t idx : selection.selected_indices()) { + selected_keys.insert( + TraceableItemKey{anchor[idx].req_id, anchor[idx].sub_id}); + } + + for (const auto& item : replacements) { + TraceableItemKey key{item.req_id, item.sub_id}; + if (selected_keys.find(key) == selected_keys.end()) { + return NodeResult>>::Failure( + NodeErrorKind::kInputError, + "ScatterReplace replacement contains unselected or unknown key: " + "req_id=" + + std::to_string(key.req_id) + + ", sub_id=" + std::to_string(key.sub_id), + BatchFailureDetail{"ScatterReplace", BatchFailureReason::kUnknown, + key}); + } + } + + for (size_t idx : selection.selected_indices()) { + TraceableItemKey sel_key{anchor[idx].req_id, anchor[idx].sub_id}; + if (repl_map.find(sel_key) == repl_map.end()) { + return NodeResult>>::Failure( + NodeErrorKind::kInputError, + "ScatterReplace replacement missing selected key: req_id=" + + std::to_string(sel_key.req_id) + + ", sub_id=" + std::to_string(sel_key.sub_id), + BatchFailureDetail{"ScatterReplace", BatchFailureReason::kMissing, + sel_key}); + } + } + + std::vector> full_batch = anchor; + for (size_t idx : selection.selected_indices()) { + TraceableItemKey key{anchor[idx].req_id, anchor[idx].sub_id}; + auto it = repl_map.find(key); + full_batch[idx].data = it->second->data; + } + + return NodeResult>>::Success( + std::move(full_batch)); +} + +// ============================================================================ +// 4. SplitPayloads +// ============================================================================ + +template +struct SplitResult { + std::vector> children; + Int32Batch counts; +}; + +namespace detail { + +inline NodeResult CheckedSplitCount(size_t count, + TraceableItemKey parent) { + if (count > static_cast(std::numeric_limits::max())) { + return NodeResult::Failure( + NodeErrorKind::kBusinessError, + "SplitPayloads chunk count exceeds Int32 capacity for req_id=" + + std::to_string(parent.req_id) + + ", sub_id=" + std::to_string(parent.sub_id), + BatchFailureDetail{"SplitPayloads", BatchFailureReason::kCountOverflow, + parent}); + } + return NodeResult::Success(static_cast(count)); +} + +template +struct SplitResultTraits { + using RawResult = std::invoke_result_t; + static constexpr bool kReturnsNodeResult = IsNodeResultType::value; + using ValueType = typename std::conditional_t< + kReturnsNodeResult, typename IsNodeResultType::ValueType, + RawResult>; + using OutputPayload = typename ValueType::value_type; +}; + +template +auto SplitPayloadsInternal( + const std::vector>& input, SplitFn&& split_one, + const std::unordered_map& initial_sub_id_by_req) { + using Traits = SplitResultTraits, InputPayload>; + using OutPayload = typename Traits::OutputPayload; + using ResultType = SplitResult; + + std::unordered_set seen_keys; + for (const auto& item : input) { + TraceableItemKey key{item.req_id, item.sub_id}; + if (!seen_keys.insert(key).second) { + return NodeResult::Failure( + NodeErrorKind::kInputError, + "SplitPayloads duplicate input item for req_id=" + + std::to_string(key.req_id) + + ", sub_id=" + std::to_string(key.sub_id), + BatchFailureDetail{"SplitPayloads", BatchFailureReason::kDuplicate, + key}); + } + } + + std::unordered_map next_sub_id_by_req = + initial_sub_id_by_req; + std::vector> children; + Int32Batch counts; + counts.reserve(input.size()); + + for (const auto& parent : input) { + std::vector child_items; + try { + if constexpr (Traits::kReturnsNodeResult) { + auto res = split_one(parent.data); + if (!res.ok()) { + auto failure = std::move(res).ExtractFailure(); + failure.batch_detail = BatchFailureDetail{ + "SplitPayloads", BatchFailureReason::kCallbackFailed, + TraceableItemKey{parent.req_id, parent.sub_id}}; + return NodeResult::Failure(std::move(failure)); + } + child_items = std::move(res).value(); + } else { + child_items = split_one(parent.data); + } + } catch (const std::exception& e) { + return NodeResult::Failure( + NodeErrorKind::kBusinessError, + std::string("SplitPayloads split_one threw exception: ") + e.what(), + BatchFailureDetail{"SplitPayloads", + BatchFailureReason::kCallbackFailed, + TraceableItemKey{parent.req_id, parent.sub_id}}); + } catch (...) { + return NodeResult::Failure( + NodeErrorKind::kBusinessError, + "SplitPayloads split_one threw unknown exception", + BatchFailureDetail{"SplitPayloads", + BatchFailureReason::kCallbackFailed, + TraceableItemKey{parent.req_id, parent.sub_id}}); + } + + auto count = + CheckedSplitCount(child_items.size(), {parent.req_id, parent.sub_id}); + if (!count.ok()) { + return NodeResult::Failure(std::move(count).ExtractFailure()); + } + + uint64_t& next_sub_id = next_sub_id_by_req[parent.req_id]; + for (auto& payload : child_items) { + if (next_sub_id > + static_cast(std::numeric_limits::max())) { + return NodeResult::Failure( + NodeErrorKind::kBusinessError, + "SplitPayloads sub_id overflow for req_id=" + + std::to_string(parent.req_id), + BatchFailureDetail{"SplitPayloads", + BatchFailureReason::kSubIdOverflow, + TraceableItemKey{parent.req_id, parent.sub_id}}); + } + children.emplace_back(parent.req_id, static_cast(next_sub_id++), + std::move(payload)); + } + + counts.emplace_back(parent.req_id, parent.sub_id, count.value()); + } + + return NodeResult::Success( + ResultType{std::move(children), std::move(counts)}); +} + +} // namespace detail + +template +auto SplitPayloads(const std::vector>& input, + SplitFn&& split_one) { + return detail::SplitPayloadsInternal(input, std::forward(split_one), + {}); +} + +} // namespace llm_edgeflow diff --git a/src/common_nodes/text_chunk_node.cpp b/src/common_nodes/text_chunk_node.cpp index 6aeec8cd..95a733c9 100644 --- a/src/common_nodes/text_chunk_node.cpp +++ b/src/common_nodes/text_chunk_node.cpp @@ -12,6 +12,7 @@ #include "engine/text/utf8.h" #include "nodes/node_base.h" #include "nodes/node_error_codes.h" +#include "nodes/traceable_batch_operations.h" namespace llm_edgeflow { @@ -48,6 +49,35 @@ bool ValidChunkConfig(const nlohmann::json& config) { return size > 0 && size <= 1000000 && overlap >= 0 && overlap <= 100000 && overlap < size; } + +NodeResult> SplitText(const std::string& str, + size_t chunk_size, + size_t overlap) { + if (str.empty()) { + return NodeResult>::Success({""}); + } + std::vector boundaries; + size_t invalid_offset = 0; + if (!utf8::BuildCodePointBoundaries(str, &boundaries, &invalid_offset)) { + return NodeResult>::Failure( + NodeErrorKind::kBusinessError, + "TextChunkNode invalid UTF-8 input at byte offset " + + std::to_string(invalid_offset), + node_error::text_chunk::kInvalidUtf8); + } + + const size_t code_point_count = boundaries.size() - 1; + const size_t step = + (chunk_size > overlap) ? (chunk_size - overlap) : chunk_size; + std::vector chunks; + for (size_t pos = 0; pos < code_point_count; pos += step) { + const size_t end = std::min(pos + chunk_size, code_point_count); + chunks.push_back( + str.substr(boundaries[pos], boundaries[end] - boundaries[pos])); + if (end == code_point_count) break; + } + return NodeResult>::Success(std::move(chunks)); +} } // namespace /** @@ -91,78 +121,70 @@ class TextChunkNode final : public NodeBase { return node_error::text_chunk::kMissingInput; } - std::set> seen_inputs; - for (const auto& item : *text_items) { - if (!seen_inputs.insert({item.req_id, item.sub_id}).second) { - return Fail(req_ctx, node_error::text_chunk::kDuplicateInput, - "TextChunkNode duplicate input item for req_id=" + - std::to_string(item.req_id) + - ", sub_id=" + std::to_string(item.sub_id)); - } - } - - TextBatch chunked_items; - Int32Batch chunk_counts; - chunk_counts.reserve(text_items->size()); - size_t step = - (chunk_size_ > overlap_) ? (chunk_size_ - overlap_) : chunk_size_; - - std::unordered_map next_sub_id_by_req; - - for (const auto& item : *text_items) { - const std::string& str = item.data; - uint32_t req_id = item.req_id; - uint64_t& next_sub_id = next_sub_id_by_req[req_id]; - int32_t count_for_req = 0; - - if (str.empty()) { - if (next_sub_id > std::numeric_limits::max()) { - return Fail(req_ctx, node_error::text_chunk::kSubIdOverflow, - "TextChunkNode sub_id overflow for req_id=" + - std::to_string(req_id)); - } - chunked_items.emplace_back(req_id, static_cast(next_sub_id++), - ""); - count_for_req = 1; - } else { - std::vector boundaries; - size_t invalid_offset = 0; - if (!utf8::BuildCodePointBoundaries(str, &boundaries, - &invalid_offset)) { - return Fail(req_ctx, node_error::text_chunk::kInvalidUtf8, - "TextChunkNode invalid UTF-8 input for req_id=" + - std::to_string(req_id) + " at byte offset " + - std::to_string(invalid_offset)); - } - - const size_t code_point_count = boundaries.size() - 1; - for (size_t pos = 0; pos < code_point_count; pos += step) { - const size_t end = std::min(pos + chunk_size_, code_point_count); - std::string slice = - str.substr(boundaries[pos], boundaries[end] - boundaries[pos]); - if (next_sub_id > std::numeric_limits::max()) { + auto split_res = SplitPayloads(*text_items, [this](const std::string& str) { + return SplitText(str, chunk_size_, overlap_); + }); + + if (!split_res.ok()) { + const auto& failure = split_res.failure(); + if (failure.batch_detail.has_value()) { + const auto& detail = *failure.batch_detail; + switch (detail.reason) { + case BatchFailureReason::kDuplicate: { + std::string key_str; + if (detail.key.has_value()) { + key_str = " for req_id=" + std::to_string(detail.key->req_id) + + ", sub_id=" + std::to_string(detail.key->sub_id); + } + return Fail(req_ctx, node_error::text_chunk::kDuplicateInput, + "TextChunkNode duplicate input item" + key_str); + } + case BatchFailureReason::kSubIdOverflow: { + std::string req_str; + if (detail.key.has_value()) { + req_str = " for req_id=" + std::to_string(detail.key->req_id); + } return Fail(req_ctx, node_error::text_chunk::kSubIdOverflow, - "TextChunkNode sub_id overflow for req_id=" + - std::to_string(req_id)); + "TextChunkNode sub_id overflow" + req_str); } - if (count_for_req == std::numeric_limits::max()) { + case BatchFailureReason::kCountOverflow: return Fail(req_ctx, node_error::text_chunk::kCountOverflow, "TextChunkNode chunk count exceeds Int32 capacity"); + case BatchFailureReason::kCallbackFailed: { + if (failure.cause_code == node_error::text_chunk::kInvalidUtf8) { + std::string req_str; + if (detail.key.has_value()) { + req_str = " for req_id=" + std::to_string(detail.key->req_id); + } + size_t off_pos = failure.message.find(" at byte offset "); + std::string off_str = (off_pos != std::string::npos) + ? failure.message.substr(off_pos) + : ""; + return Fail( + req_ctx, node_error::text_chunk::kInvalidUtf8, + "TextChunkNode invalid UTF-8 input" + req_str + off_str); + } + int code = failure.cause_code != 0 + ? failure.cause_code + : node_error::author_node::kBusinessError; + return Fail(req_ctx, code, failure.message); } - chunked_items.emplace_back( - req_id, static_cast(next_sub_id++), std::move(slice)); - count_for_req++; - if (end == code_point_count) break; + default: + break; } } - chunk_counts.emplace_back(req_id, item.sub_id, count_for_req); + int code = failure.cause_code != 0 + ? failure.cause_code + : node_error::author_node::kInternalError; + return Fail(req_ctx, code, failure.message); } + auto result = std::move(split_res).value(); ALG_LOG_DEBUG("[TextChunkNode] Split %zu input texts into %zu chunks.\n", - text_items->size(), chunked_items.size()); + text_items->size(), result.children.size()); - out_chunks_.Set(req_ctx, std::move(chunked_items)); - out_chunk_counts_.Set(req_ctx, std::move(chunk_counts)); + out_chunks_.Set(req_ctx, std::move(result.children)); + out_chunk_counts_.Set(req_ctx, std::move(result.counts)); return 0; } diff --git a/tests/unit/nodes/test_text_chunk_node.cpp b/tests/unit/nodes/test_text_chunk_node.cpp index 4ae421bd..0b53f29c 100644 --- a/tests/unit/nodes/test_text_chunk_node.cpp +++ b/tests/unit/nodes/test_text_chunk_node.cpp @@ -290,4 +290,61 @@ TEST_F(TextChunkNodeTest, EXPECT_EQ((*counts)[2].data, 1); } +TEST_F(TextChunkNodeTest, InterleavedRequestsContinuousSubIdAcrossParents) { + auto node = NodeRegistry::Instance().Create("TextChunkNode"); + ASSERT_NE(node, nullptr); + ASSERT_TRUE(InitNodeForTest(*node, {{"chunk_size", 10}, {"overlap", 0}}, + session_ctx_.get())); + + AlgContext ctx; + TextBatch input_batch; + // Req 10, sub 0: 20 chars -> 2 chunks + input_batch.emplace_back(10, 0, "12345678901234567890"); + // Interleaved Req 20, sub 0: 10 chars -> 1 chunk + input_batch.emplace_back(20, 0, "abcdefghij"); + // Resumed Req 10, sub 1: 10 chars -> 1 chunk + input_batch.emplace_back(10, 1, "klmnopqrst"); + ctx.Publish("text", input_batch); + + ASSERT_EQ(node->Process(&ctx), 0); + + const auto* chunks = ctx.Read("chunks"); + ASSERT_NE(chunks, nullptr); + ASSERT_EQ(chunks->size(), 4u); + + // Req 10 first batch + EXPECT_EQ((*chunks)[0].req_id, 10u); + EXPECT_EQ((*chunks)[0].sub_id, 0u); + EXPECT_EQ((*chunks)[0].data, "1234567890"); + + EXPECT_EQ((*chunks)[1].req_id, 10u); + EXPECT_EQ((*chunks)[1].sub_id, 1u); + EXPECT_EQ((*chunks)[1].data, "1234567890"); + + // Interleaved Req 20 starts at 0 + EXPECT_EQ((*chunks)[2].req_id, 20u); + EXPECT_EQ((*chunks)[2].sub_id, 0u); + EXPECT_EQ((*chunks)[2].data, "abcdefghij"); + + // Resumed Req 10 must continue at sub_id 2 (not reset!) + EXPECT_EQ((*chunks)[3].req_id, 10u); + EXPECT_EQ((*chunks)[3].sub_id, 2u); + EXPECT_EQ((*chunks)[3].data, "klmnopqrst"); + + const auto* counts = ctx.Read("chunk_counts"); + ASSERT_NE(counts, nullptr); + ASSERT_EQ(counts->size(), 3u); + EXPECT_EQ((*counts)[0].req_id, 10u); + EXPECT_EQ((*counts)[0].sub_id, 0u); + EXPECT_EQ((*counts)[0].data, 2); + + EXPECT_EQ((*counts)[1].req_id, 20u); + EXPECT_EQ((*counts)[1].sub_id, 0u); + EXPECT_EQ((*counts)[1].data, 1); + + EXPECT_EQ((*counts)[2].req_id, 10u); + EXPECT_EQ((*counts)[2].sub_id, 1u); + EXPECT_EQ((*counts)[2].data, 1); +} + } // namespace llm_edgeflow diff --git a/tests/unit/nodes/test_traceable_batch_operations.cpp b/tests/unit/nodes/test_traceable_batch_operations.cpp new file mode 100644 index 00000000..bfafc97e --- /dev/null +++ b/tests/unit/nodes/test_traceable_batch_operations.cpp @@ -0,0 +1,1659 @@ +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "adapter/shared_algorithm_runtime.h" +#include "contracts/traceable_item.h" +#include "core/common_contracts.h" +#include "core/node_registry.h" +#include "engine/model_interface.h" +#include "nodes/authoring.h" +#include "nodes/traceable_batch_operations.h" +#include "tests/support/node_harness.h" +#include "tests/support/node_test_utils.h" + +namespace llm_edgeflow { +namespace { + +class CountingMockLlmModel final : public ILlmModel { + public: + const std::string& ModelType() const noexcept override { + static const std::string t = "counting_mock_llm"; + return t; + } + const std::string& Capability() const noexcept override { + static const std::string cap = "llm"; + return cap; + } + InferenceConcurrency Concurrency() const noexcept override { + return InferenceConcurrency::kConcurrent; + } + size_t GetMaxBatchSize() const noexcept override { return 8; } + + int Generate(const TextBatch& prompts, const GenerateOptions& options, + TextBatch* outputs) noexcept override { + ++call_count; + last_options = options; + last_prompts = prompts; + if (fail_first_n > 0 && call_count <= fail_first_n) { + if (outputs) outputs->clear(); + return -8901; + } + if (always_fail) { + if (outputs) outputs->clear(); + return -8902; + } + if (outputs) { + outputs->clear(); + for (const auto& item : prompts) { + outputs->emplace_back(item.req_id, item.sub_id, "ans:" + item.data); + } + if (return_wrong_count && !outputs->empty()) outputs->pop_back(); + if (corrupt_provenance && !outputs->empty()) ++(*outputs)[0].sub_id; + } + return 0; + } + + mutable int call_count = 0; + mutable GenerateOptions last_options; + mutable TextBatch last_prompts; + int fail_first_n = 0; + bool always_fail = false; + bool return_wrong_count = false; + bool corrupt_provenance = false; +}; + +class TraceableBatchOperationsTest : public ::testing::Test { + protected: + void SetUp() override { ASSERT_EQ(SharedAlgorithmRuntime::GlobalInit(), 0); } +}; + +// ============================================================================ +// 1. JoinByItem Tests +// ============================================================================ + +TEST_F(TraceableBatchOperationsTest, JoinByItemExactMatchingOrderPreserved) { + std::vector> left = { + {1, 0, "q1"}, {2, 0, "q2"}, {3, 0, "q3"}}; + std::vector> right = { + {3, 0, "a3"}, {1, 0, "a1"}, {2, 0, "a2"}}; + + auto res = JoinByItem(left, right, JoinMode::kExact); + ASSERT_TRUE(res.ok()) << res.failure().message; + + const auto& view = res.value(); + ASSERT_EQ(view.size(), 3u); + EXPECT_FALSE(view.empty()); + + // Strict left order + EXPECT_EQ(view[0].req_id(), 1u); + EXPECT_EQ(view[0].sub_id(), 0u); + EXPECT_EQ(view[0].left_payload(), "q1"); + ASSERT_TRUE(view[0].has_right()); + EXPECT_EQ(*view[0].right_payload(), "a1"); + + EXPECT_EQ(view[1].req_id(), 2u); + EXPECT_EQ(view[1].sub_id(), 0u); + EXPECT_EQ(view[1].left_payload(), "q2"); + ASSERT_TRUE(view[1].has_right()); + EXPECT_EQ(*view[1].right_payload(), "a2"); + + EXPECT_EQ(view[2].req_id(), 3u); + EXPECT_EQ(view[2].sub_id(), 0u); + EXPECT_EQ(view[2].left_payload(), "q3"); + ASSERT_TRUE(view[2].has_right()); + EXPECT_EQ(*view[2].right_payload(), "a3"); +} + +TEST_F(TraceableBatchOperationsTest, + JoinByItemLeftModeMissingRightNullPointer) { + std::vector> left = { + {1, 0, "q1"}, {2, 0, "q2"}, {3, 0, "q3"}}; + std::vector> right = {{1, 0, "a1"}, {3, 0, "a3"}}; + + auto res = JoinByItem(left, right, JoinMode::kLeft); + ASSERT_TRUE(res.ok()) << res.failure().message; + + const auto& view = res.value(); + ASSERT_EQ(view.size(), 3u); + + EXPECT_TRUE(view[0].has_right()); + EXPECT_EQ(*view[0].right_payload(), "a1"); + + EXPECT_FALSE(view[1].has_right()); + EXPECT_EQ(view[1].right, nullptr); + EXPECT_EQ(view[1].right_payload(), nullptr); + EXPECT_EQ(view[1].left_payload(), "q2"); + + EXPECT_TRUE(view[2].has_right()); + EXPECT_EQ(*view[2].right_payload(), "a3"); +} + +TEST_F(TraceableBatchOperationsTest, JoinByItemExactModeMissingRightFails) { + std::vector> left = {{1, 0, "q1"}, {2, 0, "q2"}}; + std::vector> right = {{1, 0, "a1"}}; + + auto res = JoinByItem(left, right, JoinMode::kExact); + ASSERT_FALSE(res.ok()); + const auto& failure = res.failure(); + ASSERT_TRUE(failure.batch_detail.has_value()); + EXPECT_EQ(failure.batch_detail->operation, "JoinByItem"); + EXPECT_EQ(failure.batch_detail->reason, BatchFailureReason::kMissing); + ASSERT_TRUE(failure.batch_detail->key.has_value()); + EXPECT_EQ(failure.batch_detail->key->req_id, 2u); + EXPECT_EQ(failure.batch_detail->key->sub_id, 0u); +} + +TEST_F(TraceableBatchOperationsTest, JoinByItemRightExtraKeyFailsBothModes) { + std::vector> left = {{1, 0, "q1"}}; + std::vector> right = {{1, 0, "a1"}, + {99, 0, "extra"}}; + + // Exact mode fails + auto exact_res = JoinByItem(left, right, JoinMode::kExact); + ASSERT_FALSE(exact_res.ok()); + ASSERT_TRUE(exact_res.failure().batch_detail.has_value()); + EXPECT_EQ(exact_res.failure().batch_detail->reason, + BatchFailureReason::kUnknown); + ASSERT_TRUE(exact_res.failure().batch_detail->key.has_value()); + EXPECT_EQ(exact_res.failure().batch_detail->key->req_id, 99u); + + // Left mode also fails on unknown right key + auto left_res = JoinByItem(left, right, JoinMode::kLeft); + ASSERT_FALSE(left_res.ok()); + ASSERT_TRUE(left_res.failure().batch_detail.has_value()); + EXPECT_EQ(left_res.failure().batch_detail->reason, + BatchFailureReason::kUnknown); + ASSERT_TRUE(left_res.failure().batch_detail->key.has_value()); + EXPECT_EQ(left_res.failure().batch_detail->key->req_id, 99u); +} + +TEST_F(TraceableBatchOperationsTest, JoinByItemDuplicateKeysFail) { + // Duplicate in left + std::vector> left_dup = {{1, 0, "q1"}, + {1, 0, "q1_dup"}}; + std::vector> right = {{1, 0, "a1"}}; + + auto res1 = JoinByItem(left_dup, right, JoinMode::kExact); + ASSERT_FALSE(res1.ok()); + ASSERT_TRUE(res1.failure().batch_detail.has_value()); + EXPECT_EQ(res1.failure().batch_detail->reason, + BatchFailureReason::kDuplicate); + EXPECT_EQ(res1.failure().batch_detail->key->req_id, 1u); + + // Duplicate in right + std::vector> left = {{1, 0, "q1"}}; + std::vector> right_dup = {{1, 0, "a1"}, + {1, 0, "a1_dup"}}; + auto res2 = JoinByItem(left, right_dup, JoinMode::kExact); + ASSERT_FALSE(res2.ok()); + ASSERT_TRUE(res2.failure().batch_detail.has_value()); + EXPECT_EQ(res2.failure().batch_detail->reason, + BatchFailureReason::kDuplicate); + EXPECT_EQ(res2.failure().batch_detail->key->req_id, 1u); +} + +TEST_F(TraceableBatchOperationsTest, JoinByItemEmptyCombinations) { + std::vector> empty; + std::vector> non_empty = {{1, 0, "q1"}}; + + // Both empty: succeeds in both modes + auto res_both_empty_exact = JoinByItem(empty, empty, JoinMode::kExact); + EXPECT_TRUE(res_both_empty_exact.ok()); + EXPECT_TRUE(res_both_empty_exact.value().empty()); + + auto res_both_empty_left = JoinByItem(empty, empty, JoinMode::kLeft); + EXPECT_TRUE(res_both_empty_left.ok()); + EXPECT_TRUE(res_both_empty_left.value().empty()); + + // Left empty, right non-empty: fails in both modes + auto res_left_empty_exact = JoinByItem(empty, non_empty, JoinMode::kExact); + EXPECT_FALSE(res_left_empty_exact.ok()); + EXPECT_EQ(res_left_empty_exact.failure().batch_detail->reason, + BatchFailureReason::kUnknown); + + auto res_left_empty_left = JoinByItem(empty, non_empty, JoinMode::kLeft); + EXPECT_FALSE(res_left_empty_left.ok()); + EXPECT_EQ(res_left_empty_left.failure().batch_detail->reason, + BatchFailureReason::kUnknown); + + // Left non-empty, right empty: exact fails, left succeeds + auto res_right_empty_exact = JoinByItem(non_empty, empty, JoinMode::kExact); + EXPECT_FALSE(res_right_empty_exact.ok()); + EXPECT_EQ(res_right_empty_exact.failure().batch_detail->reason, + BatchFailureReason::kMissing); + + auto res_right_empty_left = JoinByItem(non_empty, empty, JoinMode::kLeft); + ASSERT_TRUE(res_right_empty_left.ok()); + ASSERT_EQ(res_right_empty_left.value().size(), 1u); + EXPECT_FALSE(res_right_empty_left.value()[0].has_right()); +} + +// ============================================================================ +// 2. GroupByRequest Tests +// ============================================================================ + +TEST_F(TraceableBatchOperationsTest, GroupByRequestAnchorFirstAppearanceOrder) { + // A0 (req 10), B0 (req 20), A1 (req 10) + std::vector> anchor = { + {10, 0, "A0"}, {20, 0, "B0"}, {10, 1, "A1"}}; + std::vector> members = { + {20, 0, "mB0"}, {10, 0, "mA0"}, {10, 1, "mA1"}}; + + auto res = GroupByRequest(anchor, members); + ASSERT_TRUE(res.ok()) << res.failure().message; + + const auto& view = res.value(); + ASSERT_EQ(view.size(), 2u); + + // First appearance in anchor: req 10, then req 20 + EXPECT_EQ(view[0].req_id(), 10u); + EXPECT_EQ(view[1].req_id(), 20u); + + // Group 10 anchors + ASSERT_EQ(view[0].anchor_count(), 2u); + EXPECT_EQ(view[0].anchors()[0].get().data, "A0"); + EXPECT_EQ(view[0].anchors()[1].get().data, "A1"); + + // Group 20 anchors + ASSERT_EQ(view[1].anchor_count(), 1u); + EXPECT_EQ(view[1].anchors()[0].get().data, "B0"); +} + +TEST_F(TraceableBatchOperationsTest, + GroupByRequestMembersRelativeOrderPreserved) { + std::vector> anchor = {{10, 0, "A0"}}; + // Members have sub_id 5 then 2 (unsorted!) + std::vector> members = {{10, 5, "m5"}, + {10, 2, "m2"}}; + + auto res = GroupByRequest(anchor, members); + ASSERT_TRUE(res.ok()) << res.failure().message; + + const auto& view = res.value(); + ASSERT_EQ(view.size(), 1u); + ASSERT_EQ(view[0].member_count(), 2u); + EXPECT_EQ(view[0].members()[0].get().data, "m5"); + EXPECT_EQ(view[0].members()[1].get().data, "m2"); +} + +TEST_F(TraceableBatchOperationsTest, + GroupByRequestAnchorWithNoMembersKeepsEmptyGroup) { + std::vector> anchor = {{10, 0, "A0"}, + {20, 0, "B0"}}; + std::vector> members = {{10, 0, "mA0"}}; + + auto res = GroupByRequest(anchor, members); + ASSERT_TRUE(res.ok()) << res.failure().message; + + const auto& view = res.value(); + ASSERT_EQ(view.size(), 2u); + EXPECT_EQ(view[0].req_id(), 10u); + EXPECT_TRUE(view[0].has_members()); + + EXPECT_EQ(view[1].req_id(), 20u); + EXPECT_FALSE(view[1].has_members()); + EXPECT_EQ(view[1].member_count(), 0u); +} + +TEST_F(TraceableBatchOperationsTest, GroupByRequestUnknownMemberReqIdFails) { + std::vector> anchor = {{10, 0, "A0"}}; + std::vector> members = {{999, 0, "m999"}}; + + auto res = GroupByRequest(anchor, members); + ASSERT_FALSE(res.ok()); + ASSERT_TRUE(res.failure().batch_detail.has_value()); + EXPECT_EQ(res.failure().batch_detail->operation, "GroupByRequest"); + EXPECT_EQ(res.failure().batch_detail->reason, BatchFailureReason::kUnknown); + ASSERT_TRUE(res.failure().batch_detail->key.has_value()); + EXPECT_EQ(res.failure().batch_detail->key->req_id, 999u); +} + +TEST_F(TraceableBatchOperationsTest, GroupByRequestDuplicatesFail) { + // Duplicate in anchor + std::vector> anchor_dup = {{10, 0, "A0"}, + {10, 0, "A0_dup"}}; + std::vector> members = {{10, 0, "m0"}}; + auto res1 = GroupByRequest(anchor_dup, members); + ASSERT_FALSE(res1.ok()); + EXPECT_EQ(res1.failure().batch_detail->reason, + BatchFailureReason::kDuplicate); + + // Duplicate in members + std::vector> anchor = {{10, 0, "A0"}}; + std::vector> members_dup = {{10, 0, "m0"}, + {10, 0, "m0_dup"}}; + auto res2 = GroupByRequest(anchor, members_dup); + ASSERT_FALSE(res2.ok()); + EXPECT_EQ(res2.failure().batch_detail->reason, + BatchFailureReason::kDuplicate); +} + +TEST_F(TraceableBatchOperationsTest, GroupByRequestEmptyCombinations) { + std::vector> empty; + std::vector> members = {{10, 0, "m"}}; + + // Empty anchor + empty members: success + auto res_empty = GroupByRequest(empty, empty); + EXPECT_TRUE(res_empty.ok()); + EXPECT_TRUE(res_empty.value().empty()); + + // Empty anchor + non-empty members: failure + auto res_fail = GroupByRequest(empty, members); + EXPECT_FALSE(res_fail.ok()); + EXPECT_EQ(res_fail.failure().batch_detail->reason, + BatchFailureReason::kUnknown); +} + +TEST_F(TraceableBatchOperationsTest, + GroupByRequestPreservesOriginalAnchorOrderFor1to1) { + // Anchor is A0, B0, A1 + std::vector> anchor = { + {10, 0, "A0"}, {20, 0, "B0"}, {10, 1, "A1"}}; + std::vector> members = { + {10, 0, "ctx10_1"}, {10, 1, "ctx10_2"}, {20, 0, "ctx20_1"}}; + + auto res = GroupByRequest(anchor, members); + ASSERT_TRUE(res.ok()); + + const auto& view = res.value(); + + // Generate 1:1 output using GroupByAnchorIndex + std::vector> output; + output.reserve(anchor.size()); + for (size_t i = 0; i < anchor.size(); ++i) { + const auto& item = anchor[i]; + const auto& group = view.GroupByAnchorIndex(i); + std::string combined = item.data + ":("; + for (const auto& m : group.members()) { + combined += m.get().data + ","; + } + combined += ")"; + output.emplace_back(item.req_id, item.sub_id, std::move(combined)); + } + + // Verify output strictly matches anchor order: A0, B0, A1 + ASSERT_EQ(output.size(), 3u); + EXPECT_EQ(output[0].req_id, 10u); + EXPECT_EQ(output[0].sub_id, 0u); + EXPECT_EQ(output[0].data, "A0:(ctx10_1,ctx10_2,)"); + + EXPECT_EQ(output[1].req_id, 20u); + EXPECT_EQ(output[1].sub_id, 0u); + EXPECT_EQ(output[1].data, "B0:(ctx20_1,)"); + + EXPECT_EQ(output[2].req_id, 10u); + EXPECT_EQ(output[2].sub_id, 1u); + EXPECT_EQ(output[2].data, "A1:(ctx10_1,ctx10_2,)"); +} + +// ============================================================================ +// 3. SelectBatch and ScatterReplace Tests +// ============================================================================ + +TEST_F(TraceableBatchOperationsTest, SelectAndScatterAllSelected) { + std::vector> anchor = { + {1, 0, "a"}, {2, 0, "b"}, {3, 0, "c"}}; + + auto sel_res = SelectBatch(anchor, [](const std::string&) { return true; }); + ASSERT_TRUE(sel_res.ok()); + const auto& selection = sel_res.value(); + EXPECT_EQ(selection.size(), 3u); + + auto sub_batch = selection.Materialize(); + ASSERT_EQ(sub_batch.size(), 3u); + for (auto& item : sub_batch) { + item.data += "_mod"; + } + + auto scatter_res = ScatterReplace(selection, sub_batch); + ASSERT_TRUE(scatter_res.ok()); + const auto& full = scatter_res.value(); + ASSERT_EQ(full.size(), 3u); + EXPECT_EQ(full[0].data, "a_mod"); + EXPECT_EQ(full[1].data, "b_mod"); + EXPECT_EQ(full[2].data, "c_mod"); +} + +TEST_F(TraceableBatchOperationsTest, SelectAndScatterNoneSelected) { + std::vector> anchor = { + {1, 0, "a"}, {2, 0, "b"}, {3, 0, "c"}}; + + auto sel_res = SelectBatch(anchor, [](const std::string&) { return false; }); + ASSERT_TRUE(sel_res.ok()); + const auto& selection = sel_res.value(); + EXPECT_EQ(selection.size(), 0u); + EXPECT_TRUE(selection.empty()); + + auto sub_batch = selection.Materialize(); + EXPECT_TRUE(sub_batch.empty()); + + auto scatter_res = ScatterReplace(selection, sub_batch); + ASSERT_TRUE(scatter_res.ok()); + const auto& full = scatter_res.value(); + ASSERT_EQ(full.size(), 3u); + EXPECT_EQ(full[0].data, "a"); + EXPECT_EQ(full[1].data, "b"); + EXPECT_EQ(full[2].data, "c"); +} + +TEST_F(TraceableBatchOperationsTest, + SelectAndScatterPartialOutOfOrderReplacements) { + std::vector> anchor = { + {1, 0, "apple"}, {2, 0, "banana"}, {3, 0, "cherry"}}; + + // Select items with length > 5: "banana" (index 1) and "cherry" (index 2) + auto sel_res = SelectBatch( + anchor, [](const std::string& text) { return text.size() > 5; }); + ASSERT_TRUE(sel_res.ok()); + const auto& selection = sel_res.value(); + ASSERT_EQ(selection.size(), 2u); + + // Provide replacements in REVERSE order + std::vector> replacements = {{3, 0, "CHERRY"}, + {2, 0, "BANANA"}}; + + auto scatter_res = ScatterReplace(selection, replacements); + ASSERT_TRUE(scatter_res.ok()) << scatter_res.failure().message; + const auto& full = scatter_res.value(); + ASSERT_EQ(full.size(), 3u); + + // Unselected item 0 remains "apple" + EXPECT_EQ(full[0].req_id, 1u); + EXPECT_EQ(full[0].data, "apple"); + + // Selected item 1 updated to "BANANA" + EXPECT_EQ(full[1].req_id, 2u); + EXPECT_EQ(full[1].data, "BANANA"); + + // Selected item 2 updated to "CHERRY" + EXPECT_EQ(full[2].req_id, 3u); + EXPECT_EQ(full[2].data, "CHERRY"); +} + +TEST_F(TraceableBatchOperationsTest, SelectBatchPredicateFailures) { + std::vector> anchor = { + {1, 0, "ok"}, {2, 0, "fail"}, {3, 0, "ok"}}; + + // Predicate returning NodeResult failure + auto res = + SelectBatch(anchor, [](const std::string& text) -> NodeResult { + if (text == "fail") { + return NodeResult::Failure(NodeErrorKind::kBusinessError, + "Predicate check failed", -5555); + } + return NodeResult::Success(true); + }); + ASSERT_FALSE(res.ok()); + EXPECT_EQ(res.failure().cause_code, -5555); + ASSERT_TRUE(res.failure().batch_detail.has_value()); + EXPECT_EQ(res.failure().batch_detail->reason, + BatchFailureReason::kCallbackFailed); + ASSERT_TRUE(res.failure().batch_detail->key.has_value()); + EXPECT_EQ(res.failure().batch_detail->key->req_id, 2u); + + // Predicate throwing exception + auto throw_res = SelectBatch(anchor, [](const std::string& text) -> bool { + if (text == "fail") throw std::runtime_error("Unexpected error"); + return true; + }); + ASSERT_FALSE(throw_res.ok()); + ASSERT_TRUE(throw_res.failure().batch_detail.has_value()); + EXPECT_EQ(throw_res.failure().batch_detail->reason, + BatchFailureReason::kCallbackFailed); + EXPECT_EQ(throw_res.failure().batch_detail->key->req_id, 2u); +} + +TEST_F(TraceableBatchOperationsTest, ScatterReplaceErrorValidations) { + std::vector> anchor = { + {1, 0, "apple"}, {2, 0, "banana"}, {3, 0, "cherry"}}; + auto sel_res = + SelectBatch(anchor, [](const std::string& s) { return s == "banana"; }); + ASSERT_TRUE(sel_res.ok()); + const auto& selection = sel_res.value(); + + // 1. Duplicate in replacements + std::vector> dup_repl = {{2, 0, "b1"}, + {2, 0, "b2"}}; + auto res_dup = ScatterReplace(selection, dup_repl); + ASSERT_FALSE(res_dup.ok()); + EXPECT_EQ(res_dup.failure().batch_detail->reason, + BatchFailureReason::kDuplicate); + + // 2. Replacement has unselected key (e.g. 1:0 which is in anchor but not + // selected) + std::vector> unselected_repl = { + {1, 0, "new_apple"}}; + auto res_unsel = ScatterReplace(selection, unselected_repl); + ASSERT_FALSE(res_unsel.ok()); + EXPECT_EQ(res_unsel.failure().batch_detail->reason, + BatchFailureReason::kUnknown); + + // 3. Replacement has unknown key (not in anchor) + std::vector> unknown_repl = {{99, 0, "ghost"}}; + auto res_unk = ScatterReplace(selection, unknown_repl); + ASSERT_FALSE(res_unk.ok()); + EXPECT_EQ(res_unk.failure().batch_detail->reason, + BatchFailureReason::kUnknown); + + // 4. Replacement missing selected key + std::vector> empty_repl; + auto res_miss = ScatterReplace(selection, empty_repl); + ASSERT_FALSE(res_miss.ok()); + EXPECT_EQ(res_miss.failure().batch_detail->reason, + BatchFailureReason::kMissing); +} + +// ============================================================================ +// 4. SplitPayloads Tests +// ============================================================================ + +TEST_F(TraceableBatchOperationsTest, + SplitPayloadsMultipleParentsContinuousSubId) { + std::vector> input = { + {1, 5, "hello world"}, + {1, 9, "single"}, + {2, 1, "foo bar baz"}, + }; + + auto res = SplitPayloads(input, [](const std::string& str) { + std::vector words; + size_t start = 0; + while (start < str.size()) { + size_t space = str.find(' ', start); + if (space == std::string::npos) { + words.push_back(str.substr(start)); + break; + } + words.push_back(str.substr(start, space - start)); + start = space + 1; + } + return words; + }); + + ASSERT_TRUE(res.ok()) << res.failure().message; + const auto& result = res.value(); + + // Children check + ASSERT_EQ(result.children.size(), 6u); + // req 1 item 0: 2 words -> sub_id 0, 1 + EXPECT_EQ(result.children[0].req_id, 1u); + EXPECT_EQ(result.children[0].sub_id, 0u); + EXPECT_EQ(result.children[0].data, "hello"); + + EXPECT_EQ(result.children[1].req_id, 1u); + EXPECT_EQ(result.children[1].sub_id, 1u); + EXPECT_EQ(result.children[1].data, "world"); + + // req 1 item 1: 1 word -> sub_id 2 (continuous for req 1!) + EXPECT_EQ(result.children[2].req_id, 1u); + EXPECT_EQ(result.children[2].sub_id, 2u); + EXPECT_EQ(result.children[2].data, "single"); + + // req 2 item 0: 3 words -> sub_id starts at 0 for req 2! + EXPECT_EQ(result.children[3].req_id, 2u); + EXPECT_EQ(result.children[3].sub_id, 0u); + EXPECT_EQ(result.children[3].data, "foo"); + + EXPECT_EQ(result.children[4].req_id, 2u); + EXPECT_EQ(result.children[4].sub_id, 1u); + EXPECT_EQ(result.children[4].data, "bar"); + + EXPECT_EQ(result.children[5].req_id, 2u); + EXPECT_EQ(result.children[5].sub_id, 2u); + EXPECT_EQ(result.children[5].data, "baz"); + + // Counts check + ASSERT_EQ(result.counts.size(), 3u); + EXPECT_EQ(result.counts[0].req_id, 1u); + EXPECT_EQ(result.counts[0].sub_id, 5u); + EXPECT_EQ(result.counts[0].data, 2); + + EXPECT_EQ(result.counts[1].req_id, 1u); + EXPECT_EQ(result.counts[1].sub_id, 9u); + EXPECT_EQ(result.counts[1].data, 1); + + EXPECT_EQ(result.counts[2].req_id, 2u); + EXPECT_EQ(result.counts[2].sub_id, 1u); + EXPECT_EQ(result.counts[2].data, 3); +} + +TEST_F(TraceableBatchOperationsTest, + SplitPayloadsInterleavedRequestsCounterNotReset) { + std::vector> input = { + {1, 0, "a b"}, + {2, 0, "x"}, + {1, 1, "c d"}, + }; + + auto res = SplitPayloads(input, [](const std::string& str) { + if (str == "a b") return std::vector{"a", "b"}; + if (str == "x") return std::vector{"x"}; + if (str == "c d") return std::vector{"c", "d"}; + return std::vector{}; + }); + + ASSERT_TRUE(res.ok()); + const auto& children = res.value().children; + ASSERT_EQ(children.size(), 5u); + + EXPECT_EQ(children[0].req_id, 1u); + EXPECT_EQ(children[0].sub_id, 0u); + + EXPECT_EQ(children[1].req_id, 1u); + EXPECT_EQ(children[1].sub_id, 1u); + + EXPECT_EQ(children[2].req_id, 2u); + EXPECT_EQ(children[2].sub_id, 0u); + + // req 1 resumed: sub_id must be 2, 3! + EXPECT_EQ(children[3].req_id, 1u); + EXPECT_EQ(children[3].sub_id, 2u); + + EXPECT_EQ(children[4].req_id, 1u); + EXPECT_EQ(children[4].sub_id, 3u); +} + +TEST_F(TraceableBatchOperationsTest, SplitPayloadsZeroChildrenAndEmptyInput) { + // Zero children is legal + std::vector> input = {{1, 0, "empty"}}; + auto res = SplitPayloads( + input, [](const std::string&) { return std::vector{}; }); + ASSERT_TRUE(res.ok()); + EXPECT_TRUE(res.value().children.empty()); + ASSERT_EQ(res.value().counts.size(), 1u); + EXPECT_EQ(res.value().counts[0].data, 0); + + // Empty input + std::vector> empty_input; + auto empty_res = SplitPayloads(empty_input, [](const std::string&) { + return std::vector{"never"}; + }); + ASSERT_TRUE(empty_res.ok()); + EXPECT_TRUE(empty_res.value().children.empty()); + EXPECT_TRUE(empty_res.value().counts.empty()); +} + +TEST_F(TraceableBatchOperationsTest, + SplitPayloadsCallbackFailurePreservesCause) { + std::vector> input = { + {1, 0, "ok"}, {2, 3, "fail"}, {3, 0, "ok"}}; + + auto res = SplitPayloads( + input, [](const std::string& s) -> NodeResult> { + if (s == "fail") { + return NodeResult>::Failure( + NodeErrorKind::kBusinessError, "custom splitter error", -7788); + } + return NodeResult>::Success( + std::vector{s}); + }); + + ASSERT_FALSE(res.ok()); + EXPECT_EQ(res.failure().cause_code, -7788); + ASSERT_TRUE(res.failure().batch_detail.has_value()); + EXPECT_EQ(res.failure().batch_detail->operation, "SplitPayloads"); + EXPECT_EQ(res.failure().batch_detail->reason, + BatchFailureReason::kCallbackFailed); + ASSERT_TRUE(res.failure().batch_detail->key.has_value()); + EXPECT_EQ(res.failure().batch_detail->key->req_id, 2u); + EXPECT_EQ(res.failure().batch_detail->key->sub_id, 3u); +} + +TEST_F(TraceableBatchOperationsTest, SplitPayloadsSubIdOverflowSeam) { + std::vector> input = {{1, 0, "split"}}; + + // Set near-boundary sub_id using internal test seam + std::unordered_map initial = { + {1, static_cast(std::numeric_limits::max())}}; + + auto res = detail::SplitPayloadsInternal( + input, + [](const std::string&) { + // Generates 2 children: first fits at UINT32_MAX, second overflows! + return std::vector{"chunk1", "chunk2"}; + }, + initial); + + ASSERT_FALSE(res.ok()); + ASSERT_TRUE(res.failure().batch_detail.has_value()); + EXPECT_EQ(res.failure().batch_detail->operation, "SplitPayloads"); + EXPECT_EQ(res.failure().batch_detail->reason, + BatchFailureReason::kSubIdOverflow); + ASSERT_TRUE(res.failure().batch_detail->key.has_value()); + EXPECT_EQ(res.failure().batch_detail->key->req_id, 1u); +} + +TEST_F(TraceableBatchOperationsTest, CheckedSplitCountInt32Boundaries) { + const size_t max_count = + static_cast(std::numeric_limits::max()); + for (size_t count : {size_t{0}, max_count - 1, max_count}) { + SCOPED_TRACE(count); + auto result = detail::CheckedSplitCount(count, TraceableItemKey{42, 9}); + ASSERT_TRUE(result.ok()); + EXPECT_EQ(result.value(), static_cast(count)); + } + for (size_t count : {max_count + 1, std::numeric_limits::max()}) { + SCOPED_TRACE(count); + auto result = detail::CheckedSplitCount(count, TraceableItemKey{42, 9}); + ASSERT_FALSE(result.ok()); + EXPECT_EQ(result.failure().kind, NodeErrorKind::kBusinessError); + EXPECT_EQ(result.failure().cause_code, 0); + EXPECT_NE(result.failure().message.find("Int32 capacity"), + std::string::npos); + ASSERT_TRUE(result.failure().batch_detail.has_value()); + const auto& detail = *result.failure().batch_detail; + EXPECT_EQ(detail.operation, "SplitPayloads"); + EXPECT_EQ(detail.reason, BatchFailureReason::kCountOverflow); + ASSERT_TRUE(detail.key.has_value()); + EXPECT_EQ(detail.key->req_id, 42u); + EXPECT_EQ(detail.key->sub_id, 9u); + EXPECT_NE(result.failure().FormatDiagnostic("").find("req_id=42, sub_id=9"), + std::string::npos); + } +} + +TEST_F(TraceableBatchOperationsTest, + SplitPayloadsMaxSubIdThenZeroChildrenSucceeds) { + const TextBatch input = {{42, 3, "last child"}, {42, 9, "empty"}}; + const std::unordered_map initial = { + {42, static_cast(std::numeric_limits::max())}}; + auto result = detail::SplitPayloadsInternal( + input, + [](const std::string& payload) { + return payload == "empty" ? std::vector{} + : std::vector{payload}; + }, + initial); + ASSERT_TRUE(result.ok()); + ASSERT_EQ(result.value().children.size(), 1u); + EXPECT_EQ(result.value().children[0].req_id, 42u); + EXPECT_EQ(result.value().children[0].sub_id, + std::numeric_limits::max()); + EXPECT_EQ(result.value().children[0].data, "last child"); + ASSERT_EQ(result.value().counts.size(), 2u); + EXPECT_EQ(result.value().counts[0].req_id, 42u); + EXPECT_EQ(result.value().counts[0].sub_id, 3u); + EXPECT_EQ(result.value().counts[0].data, 1); + EXPECT_EQ(result.value().counts[1].req_id, 42u); + EXPECT_EQ(result.value().counts[1].sub_id, 9u); + EXPECT_EQ(result.value().counts[1].data, 0); +} + +// ============================================================================ +// 5. View and Ownership Tests +// ============================================================================ + +// SFINAE probes to detect deleted overloads for factory functions +template +struct CanJoinByItem : std::false_type {}; +template +struct CanJoinByItem< + L, R, + std::void_t(), std::declval()))>> + : std::true_type {}; + +template +struct CanGroupByRequest : std::false_type {}; +template +struct CanGroupByRequest< + A, M, + std::void_t(), std::declval()))>> + : std::true_type {}; + +template +struct CanSelectBatch : std::false_type {}; +template +struct CanSelectBatch< + B, P, + std::void_t(), std::declval

()))>> + : std::true_type {}; + +template +struct CanAddAnchor : std::false_type {}; +template +struct CanAddAnchor< + G, I, std::void_t().AddAnchor(std::declval()))>> + : std::true_type {}; + +template +struct CanAddMember : std::false_type {}; +template +struct CanAddMember< + G, I, std::void_t().AddMember(std::declval()))>> + : std::true_type {}; + +TEST_F(TraceableBatchOperationsTest, CompileTimeRejectionOfRvalues) { + using Batch = std::vector>; + using Pred = bool (*)(const std::string&); + using Item = TraceableItem; + using Group = RequestGroup; + + // ItemJoinView must reject non-const and const rvalues + static_assert( + std::is_constructible_v, + const Batch&, const Batch&, + std::vector>>, + "ItemJoinView must accept const lvalues"); + static_assert( + std::is_constructible_v, Batch&, + Batch&, + std::vector>>, + "ItemJoinView must accept non-const lvalues"); + static_assert( + !std::is_constructible_v< + ItemJoinView, Batch&&, const Batch&, + std::vector>>, + "ItemJoinView must reject rvalue left"); + static_assert( + !std::is_constructible_v< + ItemJoinView, const Batch&&, const Batch&, + std::vector>>, + "ItemJoinView must reject const rvalue left"); + static_assert(!std::is_constructible_v< + ItemJoinView, const Batch&, + Batch&&, std::vector>>, + "ItemJoinView must reject rvalue right"); + static_assert( + !std::is_constructible_v< + ItemJoinView, const Batch&, const Batch&&, + std::vector>>, + "ItemJoinView must reject const rvalue right"); + static_assert(!std::is_constructible_v< + ItemJoinView, Batch&&, Batch&&, + std::vector>>, + "ItemJoinView must reject both non-const rvalues"); + static_assert( + !std::is_constructible_v< + ItemJoinView, Batch&&, const Batch&&, + std::vector>>, + "ItemJoinView must reject rvalue left, const rvalue right"); + static_assert(!std::is_constructible_v< + ItemJoinView, const Batch&&, + Batch&&, std::vector>>, + "ItemJoinView must reject const rvalue left, rvalue right"); + static_assert( + !std::is_constructible_v< + ItemJoinView, const Batch&&, const Batch&&, + std::vector>>, + "ItemJoinView must reject both const rvalues"); + + // JoinByItem factory function must reject non-const and const rvalues + static_assert(CanJoinByItem::value, + "JoinByItem must accept const lvalues"); + static_assert(CanJoinByItem::value, + "JoinByItem must accept non-const lvalues"); + static_assert(CanJoinByItem::value, + "JoinByItem must accept mixed lvalues"); + static_assert(CanJoinByItem::value, + "JoinByItem must accept mixed lvalues"); + static_assert(!CanJoinByItem::value, + "JoinByItem must reject rvalue left"); + static_assert(!CanJoinByItem::value, + "JoinByItem must reject const rvalue left"); + static_assert(!CanJoinByItem::value, + "JoinByItem must reject rvalue right"); + static_assert(!CanJoinByItem::value, + "JoinByItem must reject const rvalue right"); + static_assert(!CanJoinByItem::value, + "JoinByItem must reject both non-const rvalues"); + static_assert(!CanJoinByItem::value, + "JoinByItem must reject rvalue left, const rvalue right"); + static_assert(!CanJoinByItem::value, + "JoinByItem must reject const rvalue left, rvalue right"); + static_assert(!CanJoinByItem::value, + "JoinByItem must reject both const rvalues"); + + // RequestGroupView must reject non-const and const rvalues + static_assert( + std::is_constructible_v< + RequestGroupView, const Batch&, + const Batch&, std::vector>, + std::vector, std::unordered_map>, + "RequestGroupView must accept const lvalues"); + static_assert(std::is_constructible_v< + RequestGroupView, Batch&, Batch&, + std::vector>, + std::vector, std::unordered_map>, + "RequestGroupView must accept non-const lvalues"); + static_assert( + !std::is_constructible_v< + RequestGroupView, Batch&&, const Batch&, + std::vector>, + std::vector, std::unordered_map>, + "RequestGroupView must reject rvalue anchor"); + static_assert( + !std::is_constructible_v< + RequestGroupView, const Batch&&, + const Batch&, std::vector>, + std::vector, std::unordered_map>, + "RequestGroupView must reject const rvalue anchor"); + static_assert( + !std::is_constructible_v< + RequestGroupView, const Batch&, Batch&&, + std::vector>, + std::vector, std::unordered_map>, + "RequestGroupView must reject rvalue members"); + static_assert( + !std::is_constructible_v< + RequestGroupView, const Batch&, + const Batch&&, std::vector>, + std::vector, std::unordered_map>, + "RequestGroupView must reject const rvalue members"); + static_assert( + !std::is_constructible_v< + RequestGroupView, Batch&&, Batch&&, + std::vector>, + std::vector, std::unordered_map>, + "RequestGroupView must reject both non-const rvalues"); + static_assert( + !std::is_constructible_v< + RequestGroupView, Batch&&, const Batch&&, + std::vector>, + std::vector, std::unordered_map>, + "RequestGroupView must reject rvalue anchor, const rvalue members"); + static_assert( + !std::is_constructible_v< + RequestGroupView, const Batch&&, Batch&&, + std::vector>, + std::vector, std::unordered_map>, + "RequestGroupView must reject const rvalue anchor, rvalue members"); + static_assert( + !std::is_constructible_v< + RequestGroupView, const Batch&&, + const Batch&&, std::vector>, + std::vector, std::unordered_map>, + "RequestGroupView must reject both const rvalues"); + + // GroupByRequest factory function must reject non-const and const rvalues + static_assert(CanGroupByRequest::value, + "GroupByRequest must accept const lvalues"); + static_assert(CanGroupByRequest::value, + "GroupByRequest must accept non-const lvalues"); + static_assert(CanGroupByRequest::value, + "GroupByRequest must accept mixed lvalues"); + static_assert(CanGroupByRequest::value, + "GroupByRequest must accept mixed lvalues"); + static_assert(!CanGroupByRequest::value, + "GroupByRequest must reject rvalue anchor"); + static_assert(!CanGroupByRequest::value, + "GroupByRequest must reject const rvalue anchor"); + static_assert(!CanGroupByRequest::value, + "GroupByRequest must reject rvalue members"); + static_assert(!CanGroupByRequest::value, + "GroupByRequest must reject const rvalue members"); + static_assert(!CanGroupByRequest::value, + "GroupByRequest must reject both non-const rvalues"); + static_assert( + !CanGroupByRequest::value, + "GroupByRequest must reject rvalue anchor, const rvalue members"); + static_assert( + !CanGroupByRequest::value, + "GroupByRequest must reject const rvalue anchor, rvalue members"); + static_assert(!CanGroupByRequest::value, + "GroupByRequest must reject both const rvalues"); + + // Selection must reject non-const and const rvalues + static_assert(!std::is_constructible_v, Batch&&, + std::vector>, + "Selection must reject rvalue anchor"); + static_assert(!std::is_constructible_v, const Batch&&, + std::vector>, + "Selection must reject const rvalue anchor"); + + // SelectBatch factory function must reject non-const and const rvalues + static_assert(CanSelectBatch::value, + "SelectBatch must accept const lvalue"); + static_assert(CanSelectBatch::value, + "SelectBatch must accept non-const lvalue"); + static_assert(!CanSelectBatch::value, + "SelectBatch must reject rvalue anchor"); + static_assert(!CanSelectBatch::value, + "SelectBatch must reject const rvalue anchor"); + + // RequestGroup AddAnchor and AddMember must reject rvalue items + static_assert(CanAddAnchor::value, + "AddAnchor must accept const lvalue item"); + static_assert(CanAddAnchor::value, + "AddAnchor must accept non-const lvalue item"); + static_assert(!CanAddAnchor::value, + "AddAnchor must reject rvalue item"); + static_assert(!CanAddAnchor::value, + "AddAnchor must reject const rvalue item"); + + static_assert(CanAddMember::value, + "AddMember must accept const lvalue item"); + static_assert(CanAddMember::value, + "AddMember must accept non-const lvalue item"); + static_assert(!CanAddMember::value, + "AddMember must reject rvalue item"); + static_assert(!CanAddMember::value, + "AddMember must reject const rvalue item"); + + // JoinedRow must reject rvalue left items + static_assert(std::is_constructible_v, + const Item&, const Item*>, + "JoinedRow must accept const lvalue left"); + static_assert(std::is_constructible_v, + Item&, const Item*>, + "JoinedRow must accept non-const lvalue left"); + static_assert(!std::is_constructible_v, + Item&&, const Item*>, + "JoinedRow must reject rvalue left"); + static_assert(!std::is_constructible_v, + const Item&&, const Item*>, + "JoinedRow must reject const rvalue left"); +} + +TEST_F(TraceableBatchOperationsTest, + MaterializeIndependentOfSelectionLifetime) { + std::vector> anchor = {{1, 0, "persist_me"}}; + std::vector> materialized; + { + auto sel_res = SelectBatch(anchor, [](const std::string&) { return true; }); + ASSERT_TRUE(sel_res.ok()); + materialized = sel_res.value().Materialize(); + } + // sel_res and Selection are now out of scope + ASSERT_EQ(materialized.size(), 1u); + EXPECT_EQ(materialized[0].req_id, 1u); + EXPECT_EQ(materialized[0].data, "persist_me"); +} + +// ============================================================================ +// 6. Starter Nodes Verification with Mock LLM +// ============================================================================ + +TEST_F(TraceableBatchOperationsTest, StarterBatchJoinNodeHarness) { + auto mock_llm = std::make_shared(); + NodeHarness harness("StarterBatchJoinNode"); + harness.Config({{"bind_model", "test_llm"}}); + harness.BindModel("test_llm", mock_llm); + + harness.TextInput("questions", {"what is capital", "who wrote hamlet"}); + harness.TextInput("attributes", {"author", "geography"}); + + auto result = harness.Run(); + ASSERT_TRUE(result.ok()) << result.diagnostic(); + + EXPECT_EQ(mock_llm->call_count, 1); + const auto& prompts = mock_llm->last_prompts; + ASSERT_EQ(prompts.size(), 2u); + EXPECT_EQ(prompts[0].data, "what is capital [attr: author]"); + EXPECT_EQ(prompts[1].data, "who wrote hamlet [attr: geography]"); +} + +TEST_F(TraceableBatchOperationsTest, StarterBatchGroupNodeHarness) { + auto mock_llm = std::make_shared(); + NodeHarness harness("StarterBatchGroupNode"); + harness.Config({{"bind_model", "test_llm"}}); + harness.BindModel("test_llm", mock_llm); + + // Interleaved queries + harness.CustomInput("queries", + TextBatch{{10, 0, "q1"}, {20, 0, "q2"}, {10, 1, "q3"}}); + // Aggregated references + harness.CustomInput( + "references", + TextBatch{{10, 0, "ref1"}, {10, 1, "ref2"}, {20, 0, "ref3"}}); + + auto result = harness.Run(); + ASSERT_TRUE(result.ok()) << result.diagnostic(); + + EXPECT_EQ(mock_llm->call_count, 1); + const auto& prompts = mock_llm->last_prompts; + ASSERT_EQ(prompts.size(), 3u); + // req 10 query 0: context ref1 + ref2 + EXPECT_EQ(prompts[0].req_id, 10u); + EXPECT_EQ(prompts[0].sub_id, 0u); + EXPECT_EQ(prompts[0].data, "ref1\nref2\nq1"); + + // req 20 query 0: context ref3 + EXPECT_EQ(prompts[1].req_id, 20u); + EXPECT_EQ(prompts[1].sub_id, 0u); + EXPECT_EQ(prompts[1].data, "ref3\nq2"); + + // req 10 query 1: context ref1 + ref2 + EXPECT_EQ(prompts[2].req_id, 10u); + EXPECT_EQ(prompts[2].sub_id, 1u); + EXPECT_EQ(prompts[2].data, "ref1\nref2\nq3"); +} + +TEST_F(TraceableBatchOperationsTest, + StarterBatchSelectScatterNodeHarnessNoneSelected) { + auto generator = std::make_shared(); + auto polisher = std::make_shared(); + + NodeHarness harness("StarterBatchSelectScatterNode"); + harness.Config( + {{"bind_model", "test_gen_llm"}, {"polish_model", "test_pol_llm"}}); + harness.BindModel("test_gen_llm", generator); + harness.BindModel("test_pol_llm", polisher); + + // Generator produces answers without [POLISH] + harness.TextInput("input", {"hello", "world"}); + + auto result = harness.Run(); + ASSERT_TRUE(result.ok()) << result.diagnostic(); + + EXPECT_EQ(generator->call_count, 1); + EXPECT_EQ(polisher->call_count, 0); // Second call skipped! + EXPECT_EQ(result.TextValues("output"), + (std::vector{"ans:hello", "ans:world"})); +} + +TEST_F(TraceableBatchOperationsTest, + StarterBatchSelectScatterNodeHarnessPartialPolishing) { + class PolishingMockLlm final : public ILlmModel { + public: + const std::string& ModelType() const noexcept override { + static const std::string t = "polishing_mock_llm"; + return t; + } + const std::string& Capability() const noexcept override { + static const std::string cap = "llm"; + return cap; + } + InferenceConcurrency Concurrency() const noexcept override { + return InferenceConcurrency::kConcurrent; + } + size_t GetMaxBatchSize() const noexcept override { return 8; } + + int Generate(const TextBatch& prompts, const GenerateOptions&, + TextBatch* outputs) noexcept override { + ++call_count; + last_prompts = prompts; + if (outputs) { + outputs->clear(); + for (const auto& item : prompts) { + if (is_generator) { + // If item has "bad", output with [POLISH] + std::string text = (item.data.find("bad") != std::string::npos) + ? (item.data + " [POLISH]") + : ("clean:" + item.data); + outputs->emplace_back(item.req_id, item.sub_id, std::move(text)); + } else { + // Polisher: replace [POLISH] with polished version + std::string text = item.data; + size_t tag = text.find(" [POLISH]"); + if (tag != std::string::npos) text.erase(tag); + outputs->emplace_back(item.req_id, item.sub_id, "polished:" + text); + } + } + } + return 0; + } + + bool is_generator = true; + mutable int call_count = 0; + mutable TextBatch last_prompts; + }; + + auto generator = std::make_shared(); + generator->is_generator = true; + auto polisher = std::make_shared(); + polisher->is_generator = false; + + NodeHarness harness("StarterBatchSelectScatterNode"); + harness.Config( + {{"bind_model", "test_gen_llm"}, {"polish_model", "test_pol_llm"}}); + harness.BindModel("test_gen_llm", generator); + harness.BindModel("test_pol_llm", polisher); + + // Input 1 is clean, input 2 is bad, input 3 is clean + harness.TextInput("input", {"good1", "bad2", "good3"}); + + auto result = harness.Run(); + ASSERT_TRUE(result.ok()) << result.diagnostic(); + + // Generator called once with all 3 items + EXPECT_EQ(generator->call_count, 1); + EXPECT_EQ(generator->last_prompts.size(), 3u); + + // Polisher called once ONLY on the 1 selected item ("bad2 [POLISH]") + EXPECT_EQ(polisher->call_count, 1); + ASSERT_EQ(polisher->last_prompts.size(), 1u); + EXPECT_EQ(polisher->last_prompts[0].req_id, 102u); + EXPECT_EQ(polisher->last_prompts[0].sub_id, 0u); + EXPECT_EQ(polisher->last_prompts[0].data, "bad2 [POLISH]"); + + // Full output has clean items unchanged, polished item replaced, in original + // order + auto outputs = result.TextValues("output"); + ASSERT_EQ(outputs.size(), 3u); + EXPECT_EQ(outputs[0], "clean:good1"); + EXPECT_EQ(outputs[1], "polished:bad2"); + EXPECT_EQ(outputs[2], "clean:good3"); +} + +TEST_F(TraceableBatchOperationsTest, + StarterBatchSelectScatterNodePolisherFailure) { + auto generator = std::make_shared(); + auto polisher = std::make_shared(); + polisher->always_fail = true; + + // Custom node with polish_tag = "ans:" so generator outputs get selected + NodeHarness harness("StarterBatchSelectScatterNode"); + harness.Config({{"bind_model", "test_gen_llm"}, + {"polish_model", "test_pol_llm"}, + {"polish_tag", "ans:"}}); + harness.BindModel("test_gen_llm", generator); + harness.BindModel("test_pol_llm", polisher); + + harness.TextInput("input", {"item1"}); + + auto result = harness.Run(); + ASSERT_FALSE(result.ok()); + EXPECT_EQ(generator->call_count, 1); + EXPECT_EQ(polisher->call_count, 1); +} + +// ============================================================================ +// 7. Functional Integration & Additional RFC-0055 Assertions +// ============================================================================ + +struct DirectSubBatchInputs { + const TextBatch* input = nullptr; +}; +struct DirectSubBatchOptions {}; + +NodeResult RunDirectSubBatch(const DirectSubBatchInputs& in, + const DirectSubBatchOptions&) { + if (!in.input || in.input->empty()) { + return NodeResult::Success(TextBatch{}); + } + auto sel = + SelectBatch(*in.input, [](const std::string& s) { return s == "keep"; }); + if (!sel.ok()) { + return NodeResult::Failure(std::move(sel).ExtractFailure()); + } + // Intentionally return sub-batch without ScatterReplace to test fail-closed + // PreservedOutput count validation. + return sel.value().Materialize(); +} + +auto DirectSubBatchSpec() { + return MakeBatchSpec(InputsOf({ + Required("input", &DirectSubBatchInputs::input), + }), + PreservedOutput("output", "input"), + Parameters({}), + &RunDirectSubBatch) + .Description("Test fixture for unscattered sub-batch rejection"); +} + +REGISTER_FUNCTION_NODE(DirectSubBatchTestNode, DirectSubBatchSpec()); + +TEST_F(TraceableBatchOperationsTest, FunctionNodeRejectsDirectSubBatchReturn) { + NodeHarness harness("DirectSubBatchTestNode"); + harness.TextInput("input", {"keep", "drop"}); + + auto result = harness.Run(); + EXPECT_FALSE(result.ok()); + EXPECT_EQ(result.process_code(), + node_error::author_node::kOutputCountMismatch); + EXPECT_EQ(result.Output("output"), nullptr); +} + +TEST_F(TraceableBatchOperationsTest, DeterministicSeedRandomPermutation) { + std::vector> left; + std::vector> right; + for (uint32_t i = 0; i < 20; ++i) { + left.emplace_back(i + 1, 0, "q_" + std::to_string(i)); + right.emplace_back(i + 1, 0, "a_" + std::to_string(i)); + } + + std::mt19937 rng(42); + std::shuffle(right.begin(), right.end(), rng); + + auto join_res = JoinByItem(left, right, JoinMode::kExact); + ASSERT_TRUE(join_res.ok()); + const auto& view = join_res.value(); + ASSERT_EQ(view.size(), 20u); + + for (size_t i = 0; i < 20; ++i) { + EXPECT_EQ(view[i].req_id(), i + 1); + EXPECT_EQ(view[i].sub_id(), 0u); + EXPECT_EQ(view[i].left_payload(), "q_" + std::to_string(i)); + ASSERT_TRUE(view[i].has_right()); + EXPECT_EQ(*view[i].right_payload(), "a_" + std::to_string(i)); + } +} + +TEST_F(TraceableBatchOperationsTest, SelectBatchCatchesNonStdException) { + std::vector> anchor = {{1, 0, "throw_int"}}; + auto res = SelectBatch(anchor, [](const std::string&) -> bool { throw 42; }); + ASSERT_FALSE(res.ok()); + ASSERT_TRUE(res.failure().batch_detail.has_value()); + EXPECT_EQ(res.failure().batch_detail->reason, + BatchFailureReason::kCallbackFailed); + EXPECT_EQ(res.failure().batch_detail->key->req_id, 1u); +} + +TEST_F(TraceableBatchOperationsTest, SplitPayloadsCatchesNonStdException) { + std::vector> input = {{1, 0, "throw_int"}}; + auto res = SplitPayloads( + input, [](const std::string&) -> std::vector { throw 42; }); + ASSERT_FALSE(res.ok()); + ASSERT_TRUE(res.failure().batch_detail.has_value()); + EXPECT_EQ(res.failure().batch_detail->reason, + BatchFailureReason::kCallbackFailed); + EXPECT_EQ(res.failure().batch_detail->key->req_id, 1u); +} + +TEST_F(TraceableBatchOperationsTest, HashDistributionQualityForSubIdZero) { + TraceableItemKeyHash hasher; + std::unordered_set hashes; + for (uint32_t req = 1; req <= 100; ++req) { + hashes.insert(hasher(TraceableItemKey{req, 0})); + } + EXPECT_EQ(hashes.size(), 100u); +} + +TEST_F(TraceableBatchOperationsTest, RequestGroupViewHasReqIdAndContains) { + std::vector> anchor = {{10, 0, "A0"}, + {20, 0, "B0"}}; + std::vector> members = {{10, 0, "m0"}}; + auto res = GroupByRequest(anchor, members); + ASSERT_TRUE(res.ok()); + const auto& view = res.value(); + EXPECT_TRUE(view.HasReqId(10)); + EXPECT_TRUE(view.Contains(20)); + EXPECT_FALSE(view.HasReqId(30)); + EXPECT_FALSE(view.Contains(999)); +} + +TEST_F(TraceableBatchOperationsTest, + SelectBatchTraceableItemPredicateSupported) { + std::vector> anchor = { + {1, 0, "alpha"}, {2, 0, "beta"}, {3, 0, "gamma"}}; + + // 1. Predicate taking const TraceableItem& returning bool + auto res_bool = SelectBatch( + anchor, + [](const TraceableItem& item) { return item.req_id == 2; }); + ASSERT_TRUE(res_bool.ok()); + EXPECT_EQ(res_bool.value().size(), 1u); + EXPECT_EQ(res_bool.value()[0].data, "beta"); + + // 2. Predicate taking const TraceableItem& returning + // NodeResult + auto res_node = SelectBatch( + anchor, [](const TraceableItem& item) -> NodeResult { + if (item.sub_id != 0) { + return NodeResult::Failure(NodeErrorKind::kBusinessError, + "Invalid sub_id"); + } + return NodeResult::Success(item.req_id >= 2); + }); + ASSERT_TRUE(res_node.ok()); + EXPECT_EQ(res_node.value().size(), 2u); + EXPECT_EQ(res_node.value()[0].data, "beta"); + EXPECT_EQ(res_node.value()[1].data, "gamma"); +} + +TEST_F(TraceableBatchOperationsTest, BatchFailureDetailFormatDiagnosticDirect) { + // 1. Failure with batch_detail containing key + NodeFailure f1( + NodeErrorKind::kBusinessError, "predicate failed", + BatchFailureDetail{"SelectBatch", BatchFailureReason::kCallbackFailed, + TraceableItemKey{42, 9}}, + -7788); + EXPECT_EQ( + f1.FormatDiagnostic("fallback"), + "SelectBatch callback_failed for req_id=42, sub_id=9: predicate failed"); + + // 2. Failure where message already contains req_id - avoid redundant + // formatting + NodeFailure f2( + NodeErrorKind::kInputError, + "JoinByItem right batch missing key present in left: req_id=2, sub_id=0", + BatchFailureDetail{"JoinByItem", BatchFailureReason::kMissing, + TraceableItemKey{2, 0}}); + EXPECT_EQ( + f2.FormatDiagnostic("fallback"), + "JoinByItem right batch missing key present in left: req_id=2, sub_id=0"); + + // 3. Failure without batch_detail + NodeFailure f3(NodeErrorKind::kBusinessError, "plain error", -1234); + EXPECT_EQ(f3.FormatDiagnostic("fallback"), "plain error"); + + // 4. Failure with empty message and batch_detail with key + NodeFailure f4( + NodeErrorKind::kBusinessError, "", + BatchFailureDetail{"SplitPayloads", BatchFailureReason::kCallbackFailed, + TraceableItemKey{10, 3}}, + -6677); + EXPECT_EQ(f4.FormatDiagnostic("fallback"), + "SplitPayloads callback_failed for req_id=10, sub_id=3: fallback"); + + // 5. Message with different req_id substring collision (req_id=4 vs + // req_id=400) + NodeFailure f5( + NodeErrorKind::kBusinessError, "failed on req_id=400", + BatchFailureDetail{"SelectBatch", BatchFailureReason::kCallbackFailed, + TraceableItemKey{4, 9}}, + -7788); + EXPECT_EQ(f5.FormatDiagnostic("fallback"), + "SelectBatch callback_failed for req_id=4, sub_id=9: failed on " + "req_id=400"); + + // 6. Message contains req_id but lacks sub_id + NodeFailure f6( + NodeErrorKind::kBusinessError, "failed on req_id=42", + BatchFailureDetail{"SelectBatch", BatchFailureReason::kCallbackFailed, + TraceableItemKey{42, 9}}, + -7788); + EXPECT_EQ(f6.FormatDiagnostic("fallback"), + "SelectBatch callback_failed for req_id=42, sub_id=9: failed on " + "req_id=42"); + + // 7. Message already contains full structured detail - avoids redundant + // double formatting + NodeFailure f7( + NodeErrorKind::kBusinessError, + "SelectBatch callback_failed for req_id=42, sub_id=9: predicate failed", + BatchFailureDetail{"SelectBatch", BatchFailureReason::kCallbackFailed, + TraceableItemKey{42, 9}}, + -7788); + EXPECT_EQ( + f7.FormatDiagnostic("fallback"), + "SelectBatch callback_failed for req_id=42, sub_id=9: predicate failed"); +} + +TEST_F(TraceableBatchOperationsTest, BatchDiagnosticRequiresCompleteKeyTokens) { + for (const std::string message : + {"SelectBatch predicate failed for req_id=42", + "SelectBatch predicate failed for req_id=420, sub_id=9", + "SelectBatch predicate failed for req_id=42, sub_id=90", + "SelectBatch predicate failed for xreq_id=42, sub_id=9", + "SelectBatch predicate failed for req_id=42, xsub_id=9", + "SelectBatch predicate failed for req_id=42x, sub_id=9", + "SelectBatch predicate failed for req_id=42, sub_id=9x"}) { + SCOPED_TRACE(message); + NodeFailure failure( + NodeErrorKind::kBusinessError, message, + BatchFailureDetail{"SelectBatch", BatchFailureReason::kCallbackFailed, + TraceableItemKey{42, 9}}, + -7788); + EXPECT_EQ( + failure.FormatDiagnostic("fallback"), + "SelectBatch callback_failed for req_id=42, sub_id=9: " + message); + EXPECT_EQ(failure.message, message); + EXPECT_EQ(failure.cause_code, -7788); + } +} + +struct BatchSelectFailInputs { + const TextBatch* input = nullptr; +}; +struct BatchSelectFailOptions {}; + +NodeResult RunBatchSelectFail(const BatchSelectFailInputs& in, + const BatchSelectFailOptions&) { + if (!in.input || in.input->empty()) { + return NodeResult::Success(TextBatch{}); + } + // Item (42, 9) triggers failure with cause_code = -7788 and message = + // "predicate failed" + auto sel = + SelectBatch(*in.input, [](const std::string& s) -> NodeResult { + if (s == "trigger_partial_key_failure") { + return NodeResult::Failure( + NodeErrorKind::kBusinessError, + "SelectBatch predicate failed for req_id=42", -7788); + } + if (s == "trigger_failure") { + return NodeResult::Failure(NodeErrorKind::kBusinessError, + "predicate failed", -7788); + } + return NodeResult::Success(true); + }); + if (!sel.ok()) { + return NodeResult::Failure(std::move(sel).ExtractFailure()); + } + return NodeResult::Success(*in.input); +} + +auto BatchSelectFailSpec() { + return MakeBatchSpec(InputsOf({ + Required("input", &BatchSelectFailInputs::input), + }), + PreservedOutput("output", "input"), + Parameters({}), + &RunBatchSelectFail) + .Description( + "Test fixture for SelectBatch failure diagnostic formatting"); +} + +REGISTER_FUNCTION_NODE(BatchSelectFailTestNode, BatchSelectFailSpec()); + +TEST_F(TraceableBatchOperationsTest, + AuthorNodeFormatsSelectBatchFailureDiagnostic) { + NodeHarness harness("BatchSelectFailTestNode"); + TextBatch batch = {{1, 0, "ok"}, {42, 9, "trigger_failure"}}; + harness.TextInputWithBatch("input", std::move(batch)); + + auto result = harness.Run(); + EXPECT_FALSE(result.ok()); + EXPECT_EQ(result.process_code(), -7788); + const auto& diag = result.diagnostic(); + EXPECT_NE(diag.find("Process returned -7788"), std::string::npos); + EXPECT_NE(diag.find("SelectBatch"), std::string::npos); + EXPECT_NE(diag.find("predicate failed"), std::string::npos); + EXPECT_NE(diag.find("req_id=42"), std::string::npos); + EXPECT_NE(diag.find("sub_id=9"), std::string::npos); +} + +TEST_F(TraceableBatchOperationsTest, + AuthorNodeCompletesSelectBatchPartialKeyDiagnostic) { + NodeHarness harness("BatchSelectFailTestNode"); + harness.TextInputWithBatch("input", {{42, 9, "trigger_partial_key_failure"}}); + + auto result = harness.Run(); + EXPECT_FALSE(result.ok()); + EXPECT_EQ(result.process_code(), -7788); + EXPECT_NE(result.diagnostic().find("Process returned -7788"), + std::string::npos); + EXPECT_NE(result.diagnostic().find( + "SelectBatch callback_failed for req_id=42, sub_id=9: " + "SelectBatch predicate failed for req_id=42"), + std::string::npos); +} + +struct BatchSplitFailInputs { + const TextBatch* input = nullptr; +}; +struct BatchSplitFailOptions {}; + +NodeResult RunBatchSplitFail(const BatchSplitFailInputs& in, + const BatchSplitFailOptions&) { + if (!in.input || in.input->empty()) { + return NodeResult::Success(TextBatch{}); + } + auto split = SplitPayloads( + *in.input, + [](const std::string& s) -> NodeResult> { + if (s == "trigger_split_failure") { + return NodeResult>::Failure( + NodeErrorKind::kBusinessError, "split callback failed", -6677); + } + return NodeResult>::Success({s}); + }); + if (!split.ok()) { + return NodeResult::Failure(std::move(split).ExtractFailure()); + } + return NodeResult::Success(*in.input); +} + +auto BatchSplitFailSpec() { + return MakeBatchSpec(InputsOf({ + Required("input", &BatchSplitFailInputs::input), + }), + PreservedOutput("output", "input"), + Parameters({}), + &RunBatchSplitFail) + .Description( + "Test fixture for SplitPayloads failure diagnostic formatting"); +} + +REGISTER_FUNCTION_NODE(BatchSplitFailTestNode, BatchSplitFailSpec()); + +TEST_F(TraceableBatchOperationsTest, + AuthorNodeFormatsSplitPayloadsFailureDiagnostic) { + NodeHarness harness("BatchSplitFailTestNode"); + TextBatch batch = {{10, 3, "trigger_split_failure"}}; + harness.TextInputWithBatch("input", std::move(batch)); + + auto result = harness.Run(); + EXPECT_FALSE(result.ok()); + EXPECT_EQ(result.process_code(), -6677); + const auto& diag = result.diagnostic(); + EXPECT_NE(diag.find("Process returned -6677"), std::string::npos); + EXPECT_NE(diag.find("SplitPayloads"), std::string::npos); + EXPECT_NE(diag.find("split callback failed"), std::string::npos); + EXPECT_NE(diag.find("req_id=10"), std::string::npos); + EXPECT_NE(diag.find("sub_id=3"), std::string::npos); +} + +inline NodeResult RunMapItemFail(const std::string& s) { + if (s == "trigger_map_failure") { + return NodeResult::Failure(NodeErrorKind::kBusinessError, + "map item failed", -5544); + } + return NodeResult::Success(s); +} + +inline auto MapItemFailSpec() { + return MakeMapSpec(Input("input"), Output("output"), + &RunMapItemFail) + .Description("Test fixture for MapSpec failure diagnostic formatting"); +} + +REGISTER_FUNCTION_NODE(MapItemFailTestNode, MapItemFailSpec()); + +TEST_F(TraceableBatchOperationsTest, + AuthorNodeFormatsMapSpecFailureDiagnostic) { + NodeHarness harness("MapItemFailTestNode"); + TextBatch batch = {{1, 0, "ok"}, {7, 3, "trigger_map_failure"}}; + harness.TextInputWithBatch("input", std::move(batch)); + + auto result = harness.Run(); + EXPECT_FALSE(result.ok()); + EXPECT_EQ(result.process_code(), -5544); + const auto& diag = result.diagnostic(); + EXPECT_NE(diag.find("Process returned -5544"), std::string::npos); + EXPECT_NE(diag.find("MapItemFailTestNode"), std::string::npos); + EXPECT_NE(diag.find("map item failed"), std::string::npos); + EXPECT_NE(diag.find("req_id=7"), std::string::npos); + EXPECT_NE(diag.find("sub_id=3"), std::string::npos); +} + +} // namespace +} // namespace llm_edgeflow