test(nodes): validate tutorial code and simplify scaffold tests

.agents/skills/edgeflow-node-developer/SKILL.md

@@ -74,7 +74,7 @@ LLM 生成参数复用 `GenerateParameters()`，默认 `max_tokens = 128`；自
 ## 验证与接入
 
 优先扩展现有测试；`tests/unit/nodes/test_*.cpp` 自动收集到节点 runner，GoogleTest 自动发现
-用例并生成 CTest 入口；脚手架 `--write-test` 沿用 `ScaffoldGeneratedNodeTest`，模型替身来自 `tests/support/recording_models.h`。聚焦测试执行真实 `Run`，
+用例并生成 CTest 入口；脚手架 `--write-test` 沿用 `ScaffoldGeneratedNodeTest`，用 `tests/support/node_probe.h` 的 `NodeUnderTest` 按逻辑端口执行，模型替身来自 `tests/support/recording_models.h`；共享 Session 或串联节点时才用 `NodeProbe`。聚焦测试执行真实 `Run`，
 检查实际业务输出、空/缺失输入、多个请求及非零 `sub_id`、数量与来源、模型失败不发布结果；
 Control/缓存改动补回滚与并发测试。仅创建成功或检查 Catalog 不证明算法正确。
 

doc/dev_guide/custom_node_concepts.md

@@ -381,8 +381,8 @@ TextEmbedding 会话缓存，以及模板和规则的字段 Control。无需按
 `NodeResult::Failure`，原因写明字段或来源。执行包装统一处理诊断和发布，普通算法不保存
 Context 指针。配置更新由包装提供一致快照。含外部资源的更新仍需明确资源生命周期。
 
-第一次测试可复用 [NodeProbe](../../tests/support/node_probe.h)，将断言加入现有
-Node 套件。需要模型时使用 [recording_models.h](../../tests/support/recording_models.h)
+业务测试使用 [NodeUnderTest](../../tests/support/node_probe.h)，只写逻辑端口、配置、输入和断言；
+共享 Session 或串联多个节点的框架测试使用同一文件中的 `NodeProbe`。需要模型时使用 [recording_models.h](../../tests/support/recording_models.h)
 中的固定回答替身，它记录每次调用，也能让指定调用失败。命令见[本地快速验证](../../src/custom_nodes/README.md#本地快速验证)。
 先用两个不同请求、非零 `sub_id` 验证不会串结果，再检查算法自己的正常和失败输出。
 

doc/dev_guide/first_custom_node.md

@@ -29,8 +29,12 @@
 得到 `src/custom_nodes/my_text_normalize_node.cpp` 和
 `tests/unit/nodes/test_my_text_normalize_node.cpp`。已有文件会被保留；如果已做过练习，继续编辑它们。
 
+教程中的每段代码都整体替换源码或测试中的同名定义；新增的结构体放在已有结构体之后。
+仓库测试会把这些代码应用到脚手架输出上，编译并运行三个阶段，教程与实际行为保持一致。
+
 在源码中把 `Transform` 替换为：
 
+<!-- tutorial-code: stage1 node -->
 ```cpp
 std::string Transform(const std::string& text) {
   const std::string from = "开机检查";
@@ -48,8 +52,29 @@ std::string Transform(const std::string& text) {
 `Spec` 声明 `input` 和 `output` 都是文本批次，输入一项就返回一项。
 修改文本时创建新值，不修改共享的只读输入。
 
-在生成的测试中，将输入 `"example"` 改成 `"请做一次开机检查"`，
-将 `item.data` 的期望值改成 `"请做一次自检"`。保留数量与来源编号的断言。
+在生成的测试文件中，保留开头的 `#include` 和命名空间，用下面的测试替换其中的 `TEST`：
+
+<!-- tutorial-code: stage1 test -->
+```cpp
+TEST(ScaffoldGeneratedNodeTest, my_text_normalize_ReplacesFixedPhrase) {
+  test_support::NodeUnderTest node;
+  ASSERT_TRUE(node.Init("my_text_normalize", nlohmann::json::object(),
+                        {"input"}, {"output"}))
+      << node.diagnostic;
+  const auto run =
+      node.Process({{"input", TextBatch{{7, 2, "请做一次开机检查"}}}});
+  ASSERT_EQ(run.status, 0) << run.diagnostic;
+  const auto* result = run.Output<TextBatch>("output");
+  ASSERT_NE(result, nullptr);
+  ASSERT_EQ(result->size(), 1u);
+  EXPECT_EQ(result->front().req_id, 7u);
+  EXPECT_EQ(result->front().sub_id, 2u);
+  EXPECT_EQ(result->front().data, "请做一次自检");
+}
+```
+
+`NodeUnderTest` 按逻辑端口 `input`、`output` 执行真实注册的节点；
+Session、内部数据名和请求上下文由它管理，测试只写配置、输入和期望结果。
 
 每完成一个阶段，都用下面的命令格式化、编译和运行这个测试：
 
@@ -65,21 +90,29 @@ cmake --build build --target edgeflow_test_nodes_runner alg_pipeline_tool -j 4
 
 ## 2. 为同一个节点增加普通配置
 
-现在希望不同方案能选择替换后的词语。在 `Inputs` 后添加：
+现在希望不同方案能选择替换后的词语。添加 `Params`，并让 `Transform` 从参数读取替换文本：
 
+<!-- tutorial-code: stage2 node -->
 ```cpp
 struct Params {
   std::string replacement;
 };
-```
 
-将 `Transform` 的签名改为
-`std::string Transform(const std::string& text, const Params& params)`，
-把函数体中的 `const std::string to = "自检";` 改成
-`const std::string& to = params.replacement;`。其余替换逻辑保持不变。
+std::string Transform(const std::string& text, const Params& params) {
+  const std::string from = "开机检查";
+  const std::string& to = params.replacement;
+  std::string result = text;
+  for (auto pos = result.find(from); pos != std::string::npos;
+       pos = result.find(from, pos + to.size())) {
+    result.replace(pos, from.size(), to);
+  }
+  return result;
+}
+```
 
 用下面的代码替换原来的 `Run` 和 `Spec`，保留原来的注册语句：
 
+<!-- tutorial-code: stage2 node -->
 ```cpp
 NodeResult<TextBatch> Run(const Inputs& inputs, const Params& params) {
   return MapPayloads(*inputs.input, [&params](const std::string& text) {
@@ -102,16 +135,49 @@ auto Spec() {
 框架从 Pipeline 节点的 `params` 读取 `replacement`，未填写时使用 `自检`。
 这些配置在创建节点时生效。只有需要运行时更新，才添加 [Control](first_control.md)。
 
-先运行原来的测试，确认默认值仍然生效。再把测试中 `probe.Init` 的
-`nlohmann::json::object()` 改为 `nlohmann::json{{"replacement", "启动自检"}}`，
-将输出期望值改为 `"请做一次启动自检"`，重新运行测试。
+用下面两个测试替换上一阶段的测试，分别检查默认值和配置值：
+
+<!-- tutorial-code: stage2 test -->
+```cpp
+TEST(ScaffoldGeneratedNodeTest, my_text_normalize_UsesDefaultReplacement) {
+  test_support::NodeUnderTest node;
+  ASSERT_TRUE(node.Init("my_text_normalize", nlohmann::json::object(),
+                        {"input"}, {"output"}))
+      << node.diagnostic;
+  const auto run =
+      node.Process({{"input", TextBatch{{7, 2, "请做一次开机检查"}}}});
+  ASSERT_EQ(run.status, 0) << run.diagnostic;
+  const auto* result = run.Output<TextBatch>("output");
+  ASSERT_NE(result, nullptr);
+  ASSERT_EQ(result->size(), 1u);
+  EXPECT_EQ(result->front().data, "请做一次自检");
+}
+
+TEST(ScaffoldGeneratedNodeTest, my_text_normalize_UsesConfiguredReplacement) {
+  test_support::NodeUnderTest node;
+  ASSERT_TRUE(node.Init("my_text_normalize", {{"replacement", "启动自检"}},
+                        {"input"}, {"output"}))
+      << node.diagnostic;
+  const auto run =
+      node.Process({{"input", TextBatch{{7, 2, "请做一次开机检查"}}}});
+  ASSERT_EQ(run.status, 0) << run.diagnostic;
+  const auto* result = run.Output<TextBatch>("output");
+  ASSERT_NE(result, nullptr);
+  ASSERT_EQ(result->size(), 1u);
+  EXPECT_EQ(result->front().req_id, 7u);
+  EXPECT_EQ(result->front().sub_id, 2u);
+  EXPECT_EQ(result->front().data, "请做一次启动自检");
+}
+```
+
 Catalog 现在应显示节点说明和 `replacement`，仍无模型依赖。
 
 ## 3. 在同一个节点中调用 LLM
 
 继续编辑同一个文件。保留 `Inputs`、`Transform` 和注册语句。
 给现有 `Params` 增加生成参数，并在它后面添加 `Models`：
 
+<!-- tutorial-code: stage3 node -->
 ```cpp
 struct Params {
   std::string replacement;
@@ -125,6 +191,7 @@ struct Models {
 
 用下面的代码替换 `Run` 和 `Spec`。处理顺序是词语替换、构造提示词、调用模型、清理回答：
 
+<!-- tutorial-code: stage3 node -->
 ```cpp
 NodeResult<TextBatch> Run(const Inputs& inputs, const Params& params,
                           const Models& models) {
@@ -161,36 +228,44 @@ auto Spec() {
 
 ### 更新已有测试
 
-继续修改同一个测试文件。加入头文件 `tests/support/recording_models.h`，
-在 `SessionContext session;` 后加入以下模型替身：
-
-```cpp
-auto model = std::make_shared<test_support::RecordingLlm>();
-model->replies = {{"请执行启动自检\n"}};
-ASSERT_TRUE(test_support::RegisterNodeModel(session, "fixture", model));
-```
-
-将 `probe.Init` 的参数对象改为：
-
-```cpp
-nlohmann::json{{"replacement", "启动自检"},
-               {"bind_model", "fixture"},
-               {"max_tokens", 32}}
-```
-
-输入仍为 `"请做一次开机检查"`，将输出期望值改为 `"请执行启动自检"`，
-以确认回答末尾的换行已被清理。在原有结果断言之后补充：
+在测试文件开头加入模型替身的头文件，并用下面的测试替换上一阶段的测试。
+`RecordingLlm` 返回固定回答，并记录每次调用的提示词和生成参数：
 
+<!-- tutorial-code: stage3 test -->
 ```cpp
-ASSERT_EQ(model->calls.size(), 1u);
-ASSERT_EQ(model->calls.front().size(), 1u);
-EXPECT_EQ(model->calls.front().front().data,
-          "将下面的操作说明改写为简洁的规范指令，只输出改写结果：\n请做一次启动自检");
-EXPECT_EQ(model->call_options.front().max_tokens, 32);
+#include "tests/support/recording_models.h"
+
+TEST(ScaffoldGeneratedNodeTest, my_text_normalize_BuildsPromptAndCleansAnswer) {
+  test_support::NodeUnderTest node;
+  auto model = std::make_shared<test_support::RecordingLlm>();
+  model->replies = {{"请执行启动自检\n"}};
+  ASSERT_TRUE(node.AddModel("fixture", model));
+  ASSERT_TRUE(node.Init("my_text_normalize",
+                        {{"replacement", "启动自检"},
+                         {"bind_model", "fixture"},
+                         {"max_tokens", 32}},
+                        {"input"}, {"output"}))
+      << node.diagnostic;
+  const auto run =
+      node.Process({{"input", TextBatch{{7, 2, "请做一次开机检查"}}}});
+  ASSERT_EQ(run.status, 0) << run.diagnostic;
+  const auto* result = run.Output<TextBatch>("output");
+  ASSERT_NE(result, nullptr);
+  ASSERT_EQ(result->size(), 1u);
+  EXPECT_EQ(result->front().req_id, 7u);
+  EXPECT_EQ(result->front().sub_id, 2u);
+  EXPECT_EQ(result->front().data, "请执行启动自检");
+  ASSERT_EQ(model->calls.size(), 1u);
+  ASSERT_EQ(model->calls.front().size(), 1u);
+  EXPECT_EQ(model->calls.front().front().data,
+            "将下面的操作说明改写为简洁的规范指令，只输出改写结果：\n"
+            "请做一次启动自检");
+  EXPECT_EQ(model->call_options.front().max_tokens, 32);
+}
 ```
 
-按第一节的命令重新编译和测试。现在测试验证了替换词、实际提示词、生成参数和回答清理；
-Catalog 应显示更新后的节点说明、`bind_model`、生成参数和一个 LLM 依赖。
+按第一节的命令重新编译和测试。现在测试验证了替换词、实际提示词、生成参数和回答清理，
+其中回答末尾的换行已被清理。Catalog 应显示更新后的节点说明、`bind_model`、生成参数和一个 LLM 依赖。
 模型替身只提供确定的回答，业务函数无需识别测试环境。
 
 ## 4. 把这个节点接入 Pipeline

src/custom_nodes/README.md

@@ -50,9 +50,11 @@ cmake --build build --target edgeflow_test_nodes_runner -j 4
 ./build/edgeflow_test_nodes_runner --gtest_filter='ScaffoldGeneratedNodeTest.*'
 ```
 
-[Node 测试辅助](../../tests/support/node_probe.h)用已校验的 Plan 和 Session 初始化注册的 Node。
-将请求输入放入新的 `AlgContext`。调用 Process 后，断言实际输出和 `(req_id, sub_id)`。
-带模型的节点用 [recording_models.h](../../tests/support/recording_models.h) 中的固定回答替身。
+[Node 测试辅助](../../tests/support/node_probe.h)的 `NodeUnderTest` 用已校验的 Plan 初始化注册的 Node，
+并管理 Session、内部数据名和请求上下文。测试按逻辑端口给出配置和输入，用 `Process` 的结果断言
+返回码、诊断、实际输出和 `(req_id, sub_id)`；失败时输出不发布。
+带模型的节点用 [recording_models.h](../../tests/support/recording_models.h) 中的固定回答替身，经 `AddModel` 注册。
+共享 Session、指定端口生命周期或在同一上下文串联多个节点时，直接使用底层 `NodeProbe`。
 还需覆盖空输入、非法输入和失败行为。脚手架的 `--write-test` 在
 `tests/unit/nodes/` 生成完整测试文件，节点 runner 自动收录，并通过 GoogleTest 原生发现生成 CTest 入口。发现检查核对实际编译用例都有执行归属。
 修改注册或 Definition 后重新构建 `alg_pipeline_tool`，再用同一构建检查组合后的方案。

tests/ScaffoldFixtures.cmake

@@ -20,9 +20,8 @@ add_custom_command(
   DEPENDS
     "${PROJECT_SOURCE_DIR}/tools/scaffold_custom_node.py"
     "${PROJECT_SOURCE_DIR}/doc/dev_guide/first_custom_node.md"
-    "${PROJECT_SOURCE_DIR}/doc/dev_guide/first_control.md"
     "${PROJECT_SOURCE_DIR}/tests/tooling/generate_scaffold_fixtures.py"
-  COMMENT "Generating custom Node snippets and standalone behavioral test fixtures"
+  COMMENT "Generating scaffold and first custom Node tutorial fixtures"
   VERBATIM)
 
 # 在测试 runner 中编译文档里的原样函数示例，绝不进入 SDK Catalog。

tests/support/node_probe.h

@@ -1,12 +1,17 @@
 #pragma once
 
+#include <gtest/gtest.h>
+
 #include <algorithm>
+#include <functional>
+#include <initializer_list>
 #include <memory>
 #include <string>
 #include <utility>
 #include <vector>
 
 #include "contracts/config_schema_validation.h"
+#include "core/alg_context.h"
 #include "core/node_registry.h"
 #include "core/session_context.h"
 #include "core/validated_node_plan.h"
@@ -86,7 +91,13 @@ struct NodeProbe {
       diagnostic = "Node creator returned null: " + type;
       return false;
     }
-    return node->Init({&plan, &session, &diagnostic});
+    // node stays null unless Init succeeds, so callers never reach a node whose
+    // model binding or parameter preparation failed.
+    if (!node->Init({&plan, &session, &diagnostic})) {
+      node.reset();
+      return false;
+    }
+    return true;
   }
 };
 
@@ -100,4 +111,107 @@ inline bool RegisterNodeModel(SessionContext& session, std::string name,
   return session.GetModelManager().RegisterBatch({std::move(registration)});
 }
 
+inline std::string NodeInputKey(const std::string& port) {
+  return "input." + port;
+}
+inline std::string NodeOutputKey(const std::string& port) {
+  return "output." + port;
+}
+
+// One request value for a logical input port.
+struct NodeInput {
+  template <typename Batch>
+  NodeInput(std::string port, Batch batch)
+      : port(std::move(port)),
+        publish([batch = std::move(batch)](AlgContext& context,
+                                           const std::string& key) {
+          return context.Publish(key, batch);
+        }) {}
+
+  std::string port;
+  std::function<bool(AlgContext&, const std::string&)> publish;
+};
+
+// One Process call. Output views belong to the request context owned here and
+// stay valid while this object lives.
+class NodeRun {
+ public:
+  int status = 0;
+  std::string diagnostic;
+
+  template <typename Batch>
+  const Batch* Output(const std::string& port) const {
+    return context_->Read<Batch>(NodeOutputKey(port));
+  }
+  bool Published(const std::string& port) const {
+    return context_->Has(NodeOutputKey(port));
+  }
+
+ private:
+  friend class NodeUnderTest;
+  std::unique_ptr<AlgContext> context_ = std::make_unique<AlgContext>();
+};
+
+// Default entry for business Node tests: the author names logical ports and
+// supplies inputs, params and model doubles; the session, storage keys and
+// request contexts stay here. Tests that share a session, override port
+// lifetimes or chain nodes in one context use NodeProbe directly.
+class NodeUnderTest {
+ public:
+  std::string diagnostic;
+
+  bool AddModel(std::string name, std::shared_ptr<IModel> model) {
+    return RegisterNodeModel(session_, std::move(name), std::move(model));
+  }
+
+  bool Init(const std::string& type, const nlohmann::json& params,
+            const std::vector<std::string>& inputs,
+            const std::vector<std::string>& outputs) {
+    const auto bind = [](const std::vector<std::string>& ports,
+                         std::string (*key)(const std::string&)) {
+      std::vector<NodeBinding> bindings;
+      for (const auto& port : ports) bindings.push_back({port, key(port)});
+      return bindings;
+    };
+    const bool ok =
+        probe_.Init(type, params, session_, bind(inputs, &NodeInputKey),
+                    bind(outputs, &NodeOutputKey));
+    diagnostic = probe_.diagnostic;
+    return ok;
+  }
+
+  NodeRun Process(std::initializer_list<NodeInput> inputs) {
+    NodeRun run;
+    if (!probe_.node) {
+      ADD_FAILURE() << "Process before a successful Init";
+      run.status = -1;
+      return run;
+    }
+    for (const auto& input : inputs) {
+      if (!probe_.plan.FindPort(input.port, PortDirection::kInput) ||
+          !input.publish(*run.context_, NodeInputKey(input.port))) {
+        ADD_FAILURE() << "Input port not connected or given twice: "
+                      << input.port;
+        run.status = -1;
+        return run;
+      }
+    }
+    run.status = probe_.node->Process(run.context_.get());
+    run.diagnostic = run.context_->GetErrorMessage();
+    return run;
+  }
+
+  NodeControlResult Control(int command, const std::string& payload) {
+    if (!probe_.node) {
+      ADD_FAILURE() << "Control before a successful Init";
+      return NodeControlResult::Unsupported();
+    }
+    return probe_.node->Control(command, payload);
+  }
+
+ private:
+  SessionContext session_;
+  NodeProbe probe_;
+};
+
 }  // namespace llm_edgeflow::test_support