refactor(nodes): reuse MapPayloads in common nodes

src/common_nodes/ocr_detect_node.cpp

@@ -18,11 +18,9 @@ NodeResult<Outputs> Run(const Inputs& inputs, const Models& models) {
   if (!result.ok()) return NodeResult<Outputs>::Failure(result.failure());
   Outputs outputs;
   outputs.document = std::move(result).value();
-  outputs.text.reserve(outputs.document.size());
-  for (const auto& document : outputs.document) {
-    outputs.text.emplace_back(document.req_id, document.sub_id,
-                              document.data.combined_text);
-  }
+  outputs.text = MapPayloads(
+      outputs.document,
+      [](const OcrDocumentItem& document) { return document.combined_text; });
   return NodeResult<Outputs>::Success(std::move(outputs));
 }
 

src/common_nodes/structured_json_parse_node.cpp

@@ -133,53 +133,44 @@ bool ParseOrExtractJson(const Params& options, const std::string& input,
   return false;
 }
 
-NodeResult<StructuredDocumentBatch> Run(const Inputs& inputs,
-                                        const Params& options) {
-  const auto* text_items = inputs.text;
-  StructuredDocumentBatch output_docs;
-  output_docs.reserve(text_items->size());
-
-  for (const auto& item : *text_items) {
-    std::string parsed_json_str;
-    nlohmann::json parsed_structured = nlohmann::json::object();
-    JsonParseStatus status = JsonParseStatus::kOk;
-    std::string diag;
+NodeResult<JsonDocumentItem> ParseDocument(const Params& options,
+                                           const std::string& text) {
+  std::string parsed_json_str;
+  nlohmann::json parsed_structured = nlohmann::json::object();
+  JsonParseStatus status = JsonParseStatus::kOk;
+  std::string diag;
 
-    bool ok = ParseOrExtractJson(options, item.data, &parsed_json_str,
-                                 &parsed_structured, &status, &diag);
-    if (ok && !options.ValidateStructuredFields(parsed_structured, &diag)) {
-      ok = false;
-      status = JsonParseStatus::kFailed;
-    }
+  bool ok = ParseOrExtractJson(options, text, &parsed_json_str,
+                               &parsed_structured, &status, &diag);
+  if (ok && !options.ValidateStructuredFields(parsed_structured, &diag)) {
+    ok = false;
+    status = JsonParseStatus::kFailed;
+  }
 
-    if (!ok) {
-      if (options.failure_policy_ == "fail") {
-        return NodeResult<StructuredDocumentBatch>::Failure(
-            "JSON parse failed for sample: " + diag, error::kJsonParse);
-      } else if (options.failure_policy_ == "emit_diagnostic") {
-        output_docs.emplace_back(
-            item.req_id, item.sub_id,
-            JsonDocumentItem(options.fallback_text_, false,
-                             JsonParseStatus::kFailed, diag,
-                             options.fallback_structured_));
-        continue;
-      } else {  // configured_fallback
-        output_docs.emplace_back(
-            item.req_id, item.sub_id,
-            JsonDocumentItem(options.fallback_text_, true,
-                             JsonParseStatus::kFallbackApplied, diag,
-                             options.fallback_structured_));
-        continue;
-      }
+  if (!ok) {
+    if (options.failure_policy_ == "fail") {
+      return NodeResult<JsonDocumentItem>::Failure(
+          "JSON parse failed for sample: " + diag, error::kJsonParse);
+    } else if (options.failure_policy_ == "emit_diagnostic") {
+      return JsonDocumentItem(options.fallback_text_, false,
+                              JsonParseStatus::kFailed, diag,
+                              options.fallback_structured_);
+    } else {  // configured_fallback
+      return JsonDocumentItem(options.fallback_text_, true,
+                              JsonParseStatus::kFallbackApplied, diag,
+                              options.fallback_structured_);
     }
-
-    output_docs.emplace_back(
-        item.req_id, item.sub_id,
-        JsonDocumentItem(std::move(parsed_json_str), true, status, diag,
-                         std::move(parsed_structured)));
   }
 
-  return NodeResult<StructuredDocumentBatch>::Success(std::move(output_docs));
+  return JsonDocumentItem(std::move(parsed_json_str), true, status, diag,
+                          std::move(parsed_structured));
+}
+
+NodeResult<StructuredDocumentBatch> Run(const Inputs& inputs,
+                                        const Params& options) {
+  return MapPayloads(*inputs.text, [&options](const std::string& text) {
+    return ParseDocument(options, text);
+  });
 }
 
 auto Spec() {

src/common_nodes/text_rule_match_node.cpp

@@ -90,83 +90,80 @@ struct Inputs {
   const TextBatch* text = nullptr;
 };
 
-NodeResult<RuleMatchBatch> Run(const Inputs& inputs, const Params& state) {
-  const auto* text_items = inputs.text;
-  RuleMatchBatch output_matches;
-  output_matches.reserve(text_items->size());
-
-  for (const auto& item : *text_items) {
-    const std::string& sentence = item.data;
-    RuleMatchItem result;
-    // 首个带类别的命中决定结果名称。
-    auto record_hit = [&](const std::string& category, const std::string& word,
-                          float score, const std::string& rule_id) {
-      result.is_hit = 1;
-      if (!result.category.empty()) return;
-      result.category = category;
-      result.matched_word = word;
-      result.score = score;
-      result.rule_id = rule_id;
-    };
-
-    // 1. 匹配 categories (词表模式)
-    for (const auto& [category, words] : state.categories) {
-      for (const auto& w : words) {
-        if (w.empty()) continue;
-        if (sentence.find(w) != std::string::npos) {
-          record_hit(category, w, 1.0f, {});
-          result.matches.push_back(
-              {RuleMatchSource::kKeyword, category, {}, w, 1.0f});
-        }
+NodeResult<RuleMatchItem> MatchSentence(const Params& state,
+                                        const std::string& sentence) {
+  RuleMatchItem result;
+  // 首个带类别的命中决定结果名称。
+  auto record_hit = [&](const std::string& category, const std::string& word,
+                        float score, const std::string& rule_id) {
+    result.is_hit = 1;
+    if (!result.category.empty()) return;
+    result.category = category;
+    result.matched_word = word;
+    result.score = score;
+    result.rule_id = rule_id;
+  };
+
+  // 1. 匹配 categories (词表模式)
+  for (const auto& [category, words] : state.categories) {
+    for (const auto& w : words) {
+      if (w.empty()) continue;
+      if (sentence.find(w) != std::string::npos) {
+        record_hit(category, w, 1.0f, {});
+        result.matches.push_back(
+            {RuleMatchSource::kKeyword, category, {}, w, 1.0f});
       }
     }
+  }
 
-    // 2. 匹配 rules (结构化规则模式，支持 regex, exact, contains)
-    for (const auto& rule : state.rules) {
-      bool rule_matched = false;
-      std::unordered_map<std::string, std::string> rule_captures;
-
-      if (rule.strategy == "regex") {
-        std::string diagnostic;
-        const TextRegexSearchStatus status =
-            rule.compiled_regex ? rule.compiled_regex->Search(
-                                      sentence, &rule_captures, &diagnostic)
-                                : TextRegexSearchStatus::kNotMatched;
-        if (status == TextRegexSearchStatus::kError) {
-          return NodeResult<RuleMatchBatch>::Failure(
-              "Regex execution failed for rule '" + rule.id +
-                  "': " + diagnostic,
-              error::kRegexExecutionFailed);
-        }
-        rule_matched = status == TextRegexSearchStatus::kMatched;
-      } else if (rule.strategy == "exact") {
-        if (sentence == rule.pattern) {
-          rule_matched = true;
-        }
-      } else {  // contains
-        if (sentence.find(rule.pattern) != std::string::npos) {
-          rule_matched = true;
-        }
+  // 2. 匹配 rules (结构化规则模式，支持 regex, exact, contains)
+  for (const auto& rule : state.rules) {
+    bool rule_matched = false;
+    std::unordered_map<std::string, std::string> rule_captures;
+
+    if (rule.strategy == "regex") {
+      std::string diagnostic;
+      const TextRegexSearchStatus status =
+          rule.compiled_regex ? rule.compiled_regex->Search(
+                                    sentence, &rule_captures, &diagnostic)
+                              : TextRegexSearchStatus::kNotMatched;
+      if (status == TextRegexSearchStatus::kError) {
+        return NodeResult<RuleMatchItem>::Failure(
+            "Regex execution failed for rule '" + rule.id + "': " + diagnostic,
+            error::kRegexExecutionFailed);
       }
-
-      if (rule_matched) {
-        record_hit(rule.category, rule.pattern, rule.score, rule.id);
-        for (const auto& [k, v] : rule_captures) result.slots[k] = v;
-        for (const auto& [k, v] : rule.constants) result.slots[k] = v;
-        result.matches.push_back({RuleMatchSource::kRule, rule.category,
-                                  rule.id, rule.pattern, rule.score});
+      rule_matched = status == TextRegexSearchStatus::kMatched;
+    } else if (rule.strategy == "exact") {
+      if (sentence == rule.pattern) {
+        rule_matched = true;
+      }
+    } else {  // contains
+      if (sentence.find(rule.pattern) != std::string::npos) {
+        rule_matched = true;
       }
     }
 
-    if (!result.is_hit && !state.default_category.empty()) {
-      record_hit(state.default_category, "", state.default_score, {});
-      result.slots["raw_query"] = sentence;
+    if (rule_matched) {
+      record_hit(rule.category, rule.pattern, rule.score, rule.id);
+      for (const auto& [k, v] : rule_captures) result.slots[k] = v;
+      for (const auto& [k, v] : rule.constants) result.slots[k] = v;
+      result.matches.push_back({RuleMatchSource::kRule, rule.category, rule.id,
+                                rule.pattern, rule.score});
     }
+  }
 
-    output_matches.emplace_back(item.req_id, item.sub_id, std::move(result));
+  if (!result.is_hit && !state.default_category.empty()) {
+    record_hit(state.default_category, "", state.default_score, {});
+    result.slots["raw_query"] = sentence;
   }
 
-  return NodeResult<RuleMatchBatch>::Success(std::move(output_matches));
+  return result;
+}
+
+NodeResult<RuleMatchBatch> Run(const Inputs& inputs, const Params& state) {
+  return MapPayloads(*inputs.text, [&state](const std::string& sentence) {
+    return MatchSentence(state, sentence);
+  });
 }
 
 auto Spec() {

tests/unit/nodes/test_structured_json_parse_node.cpp

@@ -1,5 +1,6 @@
 #include <gtest/gtest.h>
 
+#include "contracts/error_codes.h"
 #include "core/alg_context.h"
 #include "core/common_contracts.h"
 #include "tests/support/node_probe.h"
@@ -109,8 +110,17 @@ TEST(StructuredJsonParseNodeTest,
   ASSERT_TRUE(failed.Publish("input",
                              TextBatch{{1, 0, R"({"name":"ok","score":1})"},
                                        {1, 1, R"({"name":false,"score":1})"}}));
-  EXPECT_NE(fail.node->Process(&failed), 0);
+  EXPECT_EQ(fail.node->Process(&failed), error::kJsonParse);
   EXPECT_FALSE(failed.Has("result"));
+  const std::string message = failed.GetErrorMessage();
+  EXPECT_NE(message.find("JSON parse failed for sample: Field 'name' type "
+                         "mismatch, expected string"),
+            std::string::npos)
+      << message;
+  const std::string source = "req_id=1, sub_id=1";
+  const auto first = message.find(source);
+  ASSERT_NE(first, std::string::npos) << message;
+  EXPECT_EQ(message.find(source, first + 1), std::string::npos) << message;
 }
 
 TEST(StructuredJsonParseNodeTest,

tests/unit/nodes/test_text_rule_match_node.cpp

@@ -1,5 +1,6 @@
 #include <gtest/gtest.h>
 
+#include "contracts/error_codes.h"
 #include "core/alg_context.h"
 #include "core/common_contracts.h"
 #include "tests/support/node_probe.h"
@@ -142,5 +143,31 @@ TEST(TextRuleMatchNodeTest,
   EXPECT_EQ((*output)[0].data.is_hit, 0);
   EXPECT_EQ((*output)[1].data.category, "new");
 }
+
+TEST(TextRuleMatchNodeTest,
+     RegexExecutionFailureAbortsBatchAndNamesFailedItemOnce) {
+  SessionContext session;
+  NodeProbe node;
+  ASSERT_TRUE(node.Init("text_rule_match",
+                        {{"rules",
+                          {{{"id", "limited"},
+                            {"strategy", "regex"},
+                            {"pattern", "(*LIMIT_MATCH=1)(?:a|b)+c"}}}}},
+                        session, {{"text", "input", ""}},
+                        {{"matches", "result", ""}}))
+      << node.diagnostic;
+  AlgContext context;
+  ASSERT_TRUE(
+      context.Publish("input", TextBatch{{4, 1, "plain"}, {4, 6, "ababdc"}}));
+  EXPECT_EQ(node.node->Process(&context), error::kRegexExecutionFailed);
+  EXPECT_FALSE(context.Has("result"));
+  const std::string message = context.GetErrorMessage();
+  EXPECT_NE(message.find("Regex execution failed for rule 'limited'"),
+            std::string::npos);
+  const std::string source = "req_id=4, sub_id=6";
+  const auto first = message.find(source);
+  ASSERT_NE(first, std::string::npos) << message;
+  EXPECT_EQ(message.find(source, first + 1), std::string::npos) << message;
+}
 }  // namespace
 }  // namespace llm_edgeflow