chamsechan · chamsechan · Sep 14, 2026 · Sep 14, 2026
diff --git 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
@@ -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
@@ -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
@@ -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 <sys/resource.h>
+
+#include <algorithm>
+#include <chrono>
+#include <iostream>
+#include <random>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+using namespace llm_edgeflow;
+
+template <typename Fn>
+double Measure(Fn fn, size_t expected_size) {
+  std::vector<double> 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<double, std::milli>(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<uint32_t>(i / 4),
+                        static_cast<uint32_t>(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<TextBatch> {
+          auto result = SplitPayloads(anchor, [](const std::string& s) {
+            return std::vector<std::string>{s.substr(0, 64), s.substr(64)};
+          });
+          if (!result.ok()) {
+            return NodeResult<TextBatch>::Failure(
+                std::move(result).ExtractFailure());
+          }
+          return NodeResult<TextBatch>::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
@@ -0,0 +1,76 @@
+#include <string>
+
+#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<TextBatch> Run(const Inputs& inputs, const Options& /*options*/,
+                          const Models& models) {
+  if (!inputs.queries || inputs.queries->empty()) {
+    return NodeResult<TextBatch>::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<TextBatch>::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<Inputs>({
+                           Required("queries", &Inputs::queries),
+                           Optional("references", &Inputs::references,
+                                    InputFlow::AggregateByRequest),
+                       }),
+                       PreservedOutput<TextBatch>("output", "queries"),
+                       Parameters<Options>({}),
+                       ModelsOf<Models>({
+                           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
@@ -0,0 +1,64 @@
+#include <string>
+
+#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<TextBatch> Run(const Inputs& inputs, const Options& /*options*/,
+                          const Models& models) {
+  if (!inputs.questions || inputs.questions->empty()) {
+    return NodeResult<TextBatch>::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<TextBatch>::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<Inputs>({
+                           Required("questions", &Inputs::questions),
+                           Optional("attributes", &Inputs::attributes),
+                       }),
+                       PreservedOutput<TextBatch>("output", "questions"),
+                       Parameters<Options>({}),
+                       ModelsOf<Models>({
+                           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
@@ -0,0 +1,87 @@
+#include <string>
+
+#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<TextBatch> Run(const Inputs& inputs, const Options& options,
+                          const Models& models) {
+  if (!inputs.input || inputs.input->empty()) {
+    return NodeResult<TextBatch>::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<TextBatch>::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<Inputs>({
+                 Required("input", &Inputs::input),
+             }),
+             PreservedOutput<TextBatch>("output", "input"),
+             Parameters<Options>({
+                 Field("polish_tag", &Options::polish_tag).Default("[POLISH]"),
+             }),
+             ModelsOf<Models>({
+                 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
@@ -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<T>` 中扩展可选的 `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<State>` 模板，提供原子读与写事务互斥保护，实现单请求整批不可变快照隔离与更新失败零状态泄露；支持 Move-Only 状态类型与原子快照发布；C++17 shared_ptr 原子操作不承诺无锁或无等待。