diff --git a/cmake_ext/IndividualTests.cmake b/cmake_ext/IndividualTests.cmake index daa453d9..b72bffb8 100644 --- a/cmake_ext/IndividualTests.cmake +++ b/cmake_ext/IndividualTests.cmake @@ -237,7 +237,7 @@ target_link_libraries(test_text_rerank_node PRIVATE llm_edgeflow::internal_runti add_test(NAME TextRerankNodeTest COMMAND test_text_rerank_node) add_executable(test_text_template_node ${EDGEFLOW_SOURCE_test_text_template_node}) -target_link_libraries(test_text_template_node PRIVATE llm_edgeflow::internal_runtime GTest::gtest GTest::gtest_main) +target_link_libraries(test_text_template_node PRIVATE llm_edgeflow::internal_runtime GTest::gtest GTest::gtest_main edgeflow_test_allocation_failure) add_test(NAME TextTemplateNodeTest COMMAND test_text_template_node) add_executable(test_llm_generate_node ${EDGEFLOW_SOURCE_test_llm_generate_node}) @@ -253,7 +253,7 @@ target_link_libraries(test_ocr_detect_node PRIVATE llm_edgeflow::internal_runtim add_test(NAME OcrDetectNodeTest COMMAND test_ocr_detect_node) add_executable(test_text_rule_match_node ${EDGEFLOW_SOURCE_test_text_rule_match_node}) -target_link_libraries(test_text_rule_match_node PRIVATE llm_edgeflow::internal_runtime GTest::gtest GTest::gtest_main) +target_link_libraries(test_text_rule_match_node PRIVATE llm_edgeflow::internal_runtime GTest::gtest GTest::gtest_main edgeflow_test_allocation_failure) add_test(NAME TextRuleMatchNodeTest COMMAND test_text_rule_match_node) add_executable(test_structured_json_parse_node ${EDGEFLOW_SOURCE_test_structured_json_parse_node}) @@ -271,7 +271,7 @@ target_link_libraries(test_common_nodes PRIVATE llm_edgeflow::internal_runtime G add_test(NAME CommonNodesTest COMMAND test_common_nodes) add_executable(test_function_node ${EDGEFLOW_SOURCE_test_function_node}) -target_link_libraries(test_function_node PRIVATE llm_edgeflow::internal_runtime GTest::gtest GTest::gtest_main) +target_link_libraries(test_function_node PRIVATE llm_edgeflow::internal_runtime GTest::gtest GTest::gtest_main edgeflow_test_allocation_failure) add_test(NAME FunctionNodeTest COMMAND test_function_node) add_executable(test_parameter_binding ${EDGEFLOW_SOURCE_test_parameter_binding}) diff --git a/cmake_ext/Tests.cmake b/cmake_ext/Tests.cmake index 551a6ed6..645d6e6f 100644 --- a/cmake_ext/Tests.cmake +++ b/cmake_ext/Tests.cmake @@ -142,7 +142,8 @@ add_executable(edgeflow_test_nodes_runner $ $) target_link_libraries(edgeflow_test_nodes_runner PRIVATE - llm_edgeflow::internal_runtime GTest::gtest GTest::gtest_main) + llm_edgeflow::internal_runtime GTest::gtest GTest::gtest_main + edgeflow_test_allocation_failure) edgeflow_enable_test_pch(edgeflow_test_nodes_runner) set(EDGEFLOW_TEST_ADAPTER_SRCS @@ -284,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.*" "${_edgeflow_tier1}") + "FunctionNodeTest.*:ConfigurationSnapshotTest.*" "${_edgeflow_tier1}") edgeflow_add_runner_test(ParameterBindingTest edgeflow_test_nodes_runner "ParameterBindingTest.*" "${_edgeflow_tier1}") diff --git a/dev_support/benchmarks/control_snapshots.cpp b/dev_support/benchmarks/control_snapshots.cpp new file mode 100644 index 00000000..97d568af --- /dev/null +++ b/dev_support/benchmarks/control_snapshots.cpp @@ -0,0 +1,124 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "core/alg_context.h" +#include "core/common_contracts.h" +#include "core/node_registry.h" +#include "core/session_context.h" +#include "tests/support/node_test_utils.h" +// Counts ordinary C++ new/new[] on the calling thread only. Aligned +// allocations, direct malloc calls, and allocations on other threads are not +// included. +thread_local bool count_on = false; +thread_local size_t alloc_count = 0, alloc_bytes = 0; +void* operator new(size_t n) { + if (count_on) { + ++alloc_count; + alloc_bytes += n; + } + if (auto p = std::malloc(n ? n : 1)) return p; + throw std::bad_alloc(); +} +void* operator new[](size_t n) { return ::operator new(n); } +void operator delete(void* p) noexcept { std::free(p); } +void operator delete[](void* p) noexcept { std::free(p); } +void operator delete(void* p, size_t) noexcept { std::free(p); } +void operator delete[](void* p, size_t) noexcept { std::free(p); } +using namespace llm_edgeflow; +int main(int argc, char** argv) { + if (argc != 3 || + (std::string(argv[1]) != "template" && std::string(argv[1]) != "rules") || + (std::string(argv[2]) != "0" && std::string(argv[2]) != "1")) { + std::cerr << "Usage: " << argv[0] << " template|rules 0|1\n"; + return 1; + } + bool tpl = std::string(argv[1]) == "template", + concurrent = std::atoi(argv[2]); + const char* type = tpl ? "TextTemplateNode" : "TextRuleMatchNode"; + auto node = NodeRegistry::Instance().Create(type); + SessionContext session; + nlohmann::json cfg = + tpl ? nlohmann::json{{"template", "V0: {{primary}} / {{role}}"}, + {"values", {{"role", "assistant"}}}} + : nlohmann::json{{"categories", {{"GREETING", {"hello", "hi"}}}}, + {"rules", nlohmann::json::array( + {{{"id", "world"}, + {"strategy", "regex"}, + {"pattern", "hello (?world)"}, + {"category", "WORLD"}}})}}; + if (!node || !InitNodeForTest(*node, cfg, &session)) return 2; + int cmd = tpl ? kControlCmdUpdatePrompt : kControlCmdUpdateRules; + std::string update = + tpl ? R"({"template":"V1: {{primary}} / {{role}}","prompt_id":"pid_1"})" + : R"({"categories":{"GREETING":["hello","hi"],"EXTRA":["absent"]}})"; + node->Control(cmd, update); // warm schema statics + for (int a = 0; a < 5; ++a) { + alloc_count = alloc_bytes = 0; + count_on = true; + auto r = node->Control(cmd, update); + count_on = false; + if (r.status != NodeControlStatus::kHandled) return 3; + std::cout << "ALLOC " << alloc_count << " " << alloc_bytes << "\n"; + } + TextBatch input; + for (int i = 0; i < 50; ++i) + input.emplace_back(100, i, "hello world sample " + std::to_string(i)); + for (int w = 0; w < 100; ++w) { + AlgContext ctx; + ctx.Publish(tpl ? "primary" : "text", input); + if (node->Process(&ctx)) return 4; + } + constexpr int n = 2000; + std::vector> contexts; + for (int i = 0; i < n; ++i) { + auto ctx = std::make_unique(); + ctx->Publish(tpl ? "primary" : "text", input); + contexts.push_back(std::move(ctx)); + } + std::atomic stop{false}, ready{false}; + std::atomic updates{0}; + std::thread writer; + if (concurrent) { + writer = std::thread([&] { + ready = true; + while (!stop) { + if (node->Control(cmd, update).status != NodeControlStatus::kHandled) + std::abort(); + ++updates; + std::this_thread::sleep_for(std::chrono::microseconds(100)); + } + }); + while (!ready) std::this_thread::yield(); + } + auto start = std::chrono::steady_clock::now(); + for (int i = 0; i < n; ++i) + if (node->Process(contexts[i].get())) std::abort(); + auto end = std::chrono::steady_clock::now(); + stop = true; + if (concurrent) writer.join(); + for (auto& context : contexts) { + auto& ctx = *context; + if (tpl) { + auto* out = ctx.Read("text"); + if (!out || out->size() != 50 || + (*out)[0].data != "V1: hello world sample 0 / assistant") + return 5; + } else { + auto* out = ctx.Read("matches"); + if (!out || out->size() != 50 || (*out)[0].data.category != "GREETING" || + (*out)[0].data.captures.at("tail") != "world") + return 6; + } + } + std::cout << "RESULT " << argv[1] << " " << concurrent << " " + << std::chrono::duration(end - start).count() / + n + << " " << updates << "\n"; +} diff --git a/dev_support/benchmarks/control_snapshots.py b/dev_support/benchmarks/control_snapshots.py new file mode 100644 index 00000000..22cfef38 --- /dev/null +++ b/dev_support/benchmarks/control_snapshots.py @@ -0,0 +1,165 @@ +#!/usr/bin/env python3 +"""Compare RFC-0054 node implementations on an otherwise idle machine. + +Run the canonical gate/build first. This script reuses its Ninja node-runner +runtime objects and libraries without building the repository. Only the two node +translation units are replaced for the baseline; this is not a full historical +checkout benchmark. Both versions use C++17, -O3 and -DNDEBUG. Each invocation +processes 2,000 requests of 50 samples; writer updates are spaced by 100 us. +Ordinary C++ allocation counts are measured separately from request timing. +""" + +import argparse +import json +from pathlib import Path +import shlex +import statistics +import subprocess +import sys + + +NODES = ("text_template_node", "text_rule_match_node") +ROOT = Path(__file__).resolve().parents[2] + + +def positive_integer(value): + number = int(value) + if number < 1: + raise argparse.ArgumentTypeError("must be at least 1") + return number + + +def runner_link_command(commands): + """Keep compiler/launcher arguments while stripping Ninja shell wrappers.""" + for line in reversed(commands.splitlines()): + tokens = shlex.split(line) + if "-o" not in tokens or not all( + any(token.endswith(f"/{node}.cpp.o") for token in tokens) + for node in NODES + ): + continue + if tokens[:2] == [":", "&&"]: + tokens = tokens[2:] + if tokens[-2:] == ["&&", ":"]: + tokens = tokens[:-2] + if any(token in ("&&", ";", "|") for token in tokens): + raise RuntimeError("Unsupported shell wrapper in the node-runner link command") + return [ + token for token in tokens + if not token.startswith("tests/CMakeFiles/") + and not ( + token.endswith(".o") + and "edgeflow_test_allocation_failure" in token + ) + ] + raise RuntimeError("Cannot find the Ninja node-runner link command; run the gate/build first") + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--build-dir", type=Path, default=ROOT / "build") + parser.add_argument("--output-dir", type=Path, required=True, + help="empty temporary directory for binaries and all evidence") + parser.add_argument("--baseline", default="7a6ca02") + parser.add_argument("--rounds", type=positive_integer, default=7) + args = parser.parse_args() + build = args.build_dir.resolve() + output = args.output_dir.resolve() + if not (build / "build.ninja").is_file(): + parser.error("Ninja build directory is missing; run ./scripts/run_all_tests.sh first") + if output.exists() and any(output.iterdir()): + parser.error("--output-dir must be empty to preserve previous evidence") + output.mkdir(parents=True, exist_ok=True) + print("Run only with an idle machine; compiling isolated benchmark objects.", flush=True) + + with (output / "commands.log").open("w") as log: + def run(command, cwd=ROOT): + command = [str(token) for token in command] + log.write(f"cwd={cwd}\n{shlex.join(command)}\n") + log.flush() + result = subprocess.run(command, cwd=cwd, text=True, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + log.write(result.stdout) + log.flush() + if result.returncode: + raise RuntimeError( + f"Command failed ({result.returncode}): {shlex.join(command)}; " + f"see {output / 'commands.log'}. Ensure the gate/build completed first." + ) + return result.stdout + + environment = "".join(run(command) for command in ( + ["uname", "-a"], ["c++", "--version"], ["lscpu"], + ["git", "rev-parse", "HEAD"], ["git", "status", "--short"], + ["git", "rev-parse", args.baseline], + )) + environment += f"\narguments: {vars(args)}\n" + (output / "environment.txt").write_text(environment) + link = runner_link_command(run([ + "ninja", "-C", build, "-t", "commands", "edgeflow_test_nodes_runner" + ])) + flags = ["c++", "-O3", "-DNDEBUG", "-std=c++17", "-fPIC", "-fopenmp"] + flags.extend(f"-I{path}" for path in ( + build / "layer_includes/capability_nodes", ROOT / "include", ROOT, + build / "generated/include", ROOT / "3rdparty/nlohmann_json/include", + )) + benchmark_object = output / "bench.o" + run(flags + ["-c", Path(__file__).with_suffix(".cpp"), "-o", benchmark_object]) + for version in ("baseline", "current"): + objects = [] + for node in NODES: + relative_source = f"src/common_nodes/{node}.cpp" + source = output / f"{version}_{node}.cpp" + source.write_text( + run(["git", "show", f"{args.baseline}:{relative_source}"]) + if version == "baseline" else (ROOT / relative_source).read_text() + ) + obj = output / f"{version}_{node}.o" + run(flags + ["-c", source, "-o", obj]) + objects.append(str(obj)) + command = [token for token in link if not any( + token.endswith(f"/{node}.cpp.o") for node in NODES + )] + command[command.index("-o") + 1] = str(output / f"bench_{version}") + # Put replacement objects ahead of static libraries for normal linkers. + command[command.index("-o"):command.index("-o")] = objects + [str(benchmark_object)] + run(command, cwd=build) + + records = [] + for round_index in range(args.rounds): + for node in ("template", "rules"): + for concurrent in (0, 1): + versions = ("baseline", "current") if round_index % 2 == 0 else ("current", "baseline") + for version in versions: + stdout = run([output / f"bench_{version}", node, concurrent]) + (output / f"{round_index}_{node}_{concurrent}_{version}.log").write_text(stdout) + result = next(line.split() for line in stdout.splitlines() if line.startswith("RESULT ")) + allocation = next(line.split() for line in stdout.splitlines() if line.startswith("ALLOC ")) + records.append(dict( + round=round_index, node=node, concurrent=concurrent, version=version, + us=float(result[3]), updates=int(result[4]), + allocations=int(allocation[1]), bytes=int(allocation[2]), + )) + (output / "results.json").write_text(json.dumps(records, indent=2) + "\n") + print(f"Completed round {round_index + 1}/{args.rounds}", flush=True) + + summary = [] + for node in ("template", "rules"): + for concurrent in (0, 1): + medians = {version: statistics.median( + record["us"] for record in records + if record["node"] == node and record["concurrent"] == concurrent + and record["version"] == version + ) for version in ("baseline", "current")} + change = 100 * (medians["current"] / medians["baseline"] - 1) + summary.append(f"{node} concurrent={concurrent}: {medians}, change_pct={change}\n") + (output / "summary.txt").write_text("".join(summary)) + print("".join(summary), end="") + + +if __name__ == "__main__": + try: + main() + except (OSError, RuntimeError, StopIteration, ValueError) as error: + print(f"Benchmark failed: {error}", file=sys.stderr) + sys.exit(1) diff --git a/dev_support/node_authoring/starter_control_node.cpp b/dev_support/node_authoring/starter_control_node.cpp index 5f79233c..1d231697 100644 --- a/dev_support/node_authoring/starter_control_node.cpp +++ b/dev_support/node_authoring/starter_control_node.cpp @@ -1,155 +1,55 @@ -#include -#include -#include #include -#include -#include -#include "contracts/config_schema_validation.h" -#include "contracts/control_payload.h" -#include "core/node_registry.h" -#include "nodes/node_base.h" +#include "nodes/authoring.h" -namespace llm_edgeflow::custom_nodes { -namespace { +namespace llm_edgeflow { +namespace custom_nodes { +namespace StarterControlNode_impl { -// Generated into the author's Node; this template is not a production Node. -class StarterControlNode final : public NodeBase { - public: - inline static constexpr char kNodeType[] = "StarterControlNode"; - // Choose a stable, unused custom ID using the current Catalog. - inline static constexpr int kUpdatePrefix = 1001; - inline static constexpr BlackboardKey kInput{"input", "TextBatch"}; - inline static constexpr BlackboardKey kOutput{"output", - "TextBatch"}; - - static const std::vector& PrefixConfigFields() { - static const std::vector fields = { - {"prefix", - ConfigValueKind::kString, - false, - "", - std::nullopt, - std::nullopt, - {}, - "Text prepended to each input; at most 64 UTF-8 bytes. " - "Control replaces this initial value."}}; - return fields; - } - - // Share parsing and business checks across initial configuration and Control. - static bool ReadPrefix(const nlohmann::json& config, std::string* prefix, - std::string* diagnostic) { - nlohmann::json normalized; - std::vector errors; - if (!ValidateAndNormalizeFields(PrefixConfigFields(), config, &normalized, - &errors)) { - if (diagnostic && !errors.empty()) *diagnostic = errors.front().message; - return false; - } - std::string next = normalized.at("prefix").get(); - if (next.size() > 64) { - if (diagnostic) *diagnostic = "prefix exceeds 64 UTF-8 bytes"; - return false; - } - *prefix = std::move(next); - return true; - } - - static const ControlCommandDefinition& PrefixCommand() { - static const ControlCommandDefinition command( - kUpdatePrefix, "set_prefix", - "Replace the text prefix (at most 64 bytes)", - {{"type", "object"}, - {"required", {"prefix"}}, - {"additionalProperties", false}, - {"properties", {{"prefix", {{"type", "string"}}}}}}, - true); - return command; - } - - StarterControlNode() - : NodeBase(kNodeType), input_(kInput.name), output_(kOutput.name) {} - - protected: - bool InitNode(const NodeInitContext& ctx, const nlohmann::json& config, - SessionContext&) override { - std::string next; - std::string error; - if (!ReadPrefix(config, &next, &error)) return ctx.Fail(error); - BindPort(ctx, input_); - BindPort(ctx, output_); - std::unique_lock lock(config_mutex_); - prefix_.swap(next); - return true; - } - - NodeControlResult ControlNode(int cmd, const std::string& text) override { - if (cmd != kUpdatePrefix) return NodeControlResult::Unsupported(); - nlohmann::json payload; - std::string error; - if (!ParseControlPayload(text, PrefixCommand().payload_schema, &payload, - &error)) { - return NodeControlResult::Failed(-1, std::move(error)); - } - std::string next; - if (!ReadPrefix(payload, &next, &error)) { - return NodeControlResult::Failed(-1, std::move(error)); - } - std::unique_lock lock(config_mutex_); - prefix_.swap(next); - return NodeControlResult::Handled(); - } +struct StarterControlNodeParams { + std::string prefix; +}; - int ProcessNode(AlgContext& ctx) override { - const auto* inputs = input_.Require(ctx, -8101); - if (!inputs) return -8101; - std::string prefix; - { - std::shared_lock lock(config_mutex_); - prefix = prefix_; // One consistent value for the whole request batch. - } - TextBatch outputs; - outputs.reserve(inputs->size()); - for (const auto& item : *inputs) { - outputs.emplace_back(item.req_id, item.sub_id, prefix + item.data); - } - output_.Set(ctx, std::move(outputs)); - return 0; - } +// Choose a stable, unused custom ID using the current Catalog. +inline constexpr int kUpdatePrefix = 1001; - private: - std::shared_mutex config_mutex_; - std::string prefix_; - BoundInput input_; - BoundOutput output_; -}; +// Business logic works on ordinary data, not platform structures. +static std::string ApplyPrefix(const std::string& input, + const StarterControlNodeParams& params) { + return params.prefix + input; +} -NodeDefinition MakeStarterControlNodeDefinition() { - NodeDefinition def; - def.node_type = StarterControlNode::kNodeType; - def.category = "custom"; - def.description = "Control authoring starter"; - def.config_fields = StarterControlNode::PrefixConfigFields(); - def.validate_config = [](const nlohmann::json& config, const auto&, - std::string* diagnostic) { - std::string prefix; - return StarterControlNode::ReadPrefix(config, &prefix, diagnostic); - }; - def.inputs = {RequiredInputPort(StarterControlNode::kInput.name, - StarterControlNode::kInput, "1:1", "preserve", - "request")}; - def.outputs = {OutputPort(StarterControlNode::kOutput.name, - StarterControlNode::kOutput, "1:1", "preserve", - "request")}; - def.control_commands = {StarterControlNode::PrefixCommand()}; - // Review any changes to shared state before enabling parallel scheduling. - def.parallel_safe = false; - return def; +auto StarterControlNodeSpec() { + return MakeMapSpec( + Input("input"), Output("output"), + Parameters( + { + Field("prefix", &StarterControlNodeParams::prefix) + .Default("") + .Description( + "Text prepended to each input; at most 64 UTF-8 " + "bytes. Control replaces this initial value."), + }) + .Validate([](const StarterControlNodeParams& params, + std::string* diagnostic) { + if (params.prefix.size() > 64) { + if (diagnostic) { + *diagnostic = "prefix exceeds 64 UTF-8 bytes"; + } + return false; + } + return true; + }), + &ApplyPrefix) + .Description("Control authoring starter") + .WithControls({ + ReplaceFields(kUpdatePrefix, "set_prefix", {"prefix"}, + "Replace the text prefix (at most 64 bytes)"), + }); } -REGISTER_NODE_WITH_DEFINITION(StarterControlNode, - MakeStarterControlNodeDefinition()); +REGISTER_FUNCTION_NODE(StarterControlNode, StarterControlNodeSpec()); -} // namespace -} // namespace llm_edgeflow::custom_nodes +} // namespace StarterControlNode_impl +} // namespace custom_nodes +} // namespace llm_edgeflow diff --git a/doc/CHANGELOG.md b/doc/CHANGELOG.md index f70cea8a..6952f119 100644 --- a/doc/CHANGELOG.md +++ b/doc/CHANGELOG.md @@ -1,5 +1,15 @@ # Changelog +## 2026-09-14 Control 作者接口与不可变配置快照收敛(RFC-0054) + +- **不可变配置快照组件**:引入 `ConfigurationSnapshot` 模板,提供原子读与写事务互斥保护,实现单请求整批不可变快照隔离与更新失败零状态泄露;支持 Move-Only 状态类型与原子快照发布;C++17 shared_ptr 原子操作不承诺无锁或无等待。 +- **声明式 Control 命令与参数复用**:提供 `ReplaceFields` 与 `PatchFields` 声明函数,自动投影 Control Payload Schema 并复用 `ParameterBinding` 的字段类型、默认值与业务语义校验规则;通过 `WithControls` 为函数式 Node(`MapSpec` / `BatchSpec`)提供开箱即用的运行时受控参数热更新能力。 +- **高级复杂节点试点迁移**: + - `TextTemplateNode` 迁移至不可变快照,统一 writer 串行更新事务并规范化三参数 `BuildNextTemplate` typed 更新签名。 + - `TextRuleMatchNode` 迁移至快照管理,保持现有类别规则增量补丁语义并统一错误码与异常安全屏障。 + - `StarterControlNode` 模板全面精简为声明式字段更新,消除裸手写互斥锁与轮询样板代码。 +- **契约测试与 Harness 增强**:`NodeHarness` 扩展支持 `Control` 接口与状态化执行;新增无丢更新、失败原子回滚、旧快照生命周期、并发 Control 与 Process 交织压测用例;更新 `doc/dev_guide/first_control.md` 实践教程与 `scaffold_custom_node.py` 生成脚本。 + ## 2026-09-14 业务 Adapter 函数式作者接口与载体机制收敛(RFC-0053) - **函数式 Adapter 作者接口**:引入 `AdapterResult` 显式失败传递协议与 `OneToOneTextAdapterSpec` / `OneToOneTextAdapter` 框架模板。单输入单输出业务开发者仅需编写纯函数(`DecodeRequest` 与 `EncodeResponse`),无需处理黑板读写、批内编号分配、局部批次发布或生命周期管理;框架保证输入不可变性、请求原始 ID 对齐与整批校验前置,在全部样本转换成功前不向黑板发布中间数据。 diff --git a/doc/dev_guide/first_control.md b/doc/dev_guide/first_control.md index 7e1a25cd..c50cfa58 100644 --- a/doc/dev_guide/first_control.md +++ b/doc/dev_guide/first_control.md @@ -12,9 +12,7 @@ | 控制命令的行为断言 | 已有节点测试套件 | | 新平台专有结构的转换和拷贝 | Integration;普通 JSON Control 使用已有通用入口 | -已有节点增加命令时,只需移植下面模板中的声明和 `ControlNode` 片段。继续使用原有 -`NodeBase` / `ModelBoundNode` 等基类,保持端口、模型绑定和请求逻辑。一个节点支持多个 -命令时,可在 `ControlNode` 中分支调用普通成员函数,每个函数只处理一种更新。 +函数式节点直接通过 Spec 的 `WithControls` 声明可受控字段,框架自动管理不可变快照与并发更新。已有基于 `NodeBase` 的高级节点增加命令时,可组合使用 `ConfigurationSnapshot`,在 `ControlNode` 中构建候选、验证后发布,并在 `ProcessNode` 单次读取快照处理整批请求。 ## 2. 生成能直接编译的例子 @@ -37,33 +35,20 @@ hot-swap 声明一致;通常直接复用同一份命令声明。重复使用 1:1 保留来源的纯计算例子,不会改造任意已有 C++ 类。已有文件默认拒绝覆盖。 `--generate-test` 打印注册及业务测试代码,请将它加入现有套件;不会自动修改测试文件。 -## 3. 阅读四个编辑点 +## 3. 阅读受控参数声明 -| 编辑点 | 作用 | +| 声明点 | 作用 | | --- | --- | -| `kUpdatePrefix` / `PrefixCommand()` | 同文件的具名 ID、命令说明和参数 schema | -| `PrefixConfigFields()` / `ReadPrefix()` | 声明初值默认空字符串和字段说明,共享类型、默认值与 64 字节业务校验 | -| `ControlNode()` | 解析拥有数据的 JSON 值,调用同一校验函数,成功后替换前缀 | -| `ProcessNode()` | 为整批请求读取一次前缀,在保留 `(req_id, sub_id)` 的输出中使用它 | - -`def.control_commands` 引用 `PrefixCommand()`;`ParseControlPayload` 也引用其中的 -schema,避免重复维护字段检查。`ReadPrefix` 同时用于 Definition 的 `validate_config`、 -`InitNode` 与 Control;初始配置写在节点的 `config`,例如 `{"prefix":"BASE:"}`。 -未设置时使用空字符串,Control 成功后替换该值;非法初始配置会在预检拒绝,直接 Init -也通过 `ctx.Fail` 返回具体原因。初值和在线更新共享业务规则,不需要再写一套解析器。 - -现有校验子集包括 `type`、`enum`、`required`、 -`properties`、`minProperties`、`additionalProperties`、同类型 `items`、数值 -`minimum` / `maximum`。注册时拒绝无效或未支持的 schema 关键字;只使用这些 -关键字;字符串长度、字段关系、规则编译等约束使用普通 C++ 语义校验。它不是完整的 -JSON Schema 实现。 - -参数校验完成前不修改在线配置。模板先构造拥有数据的 `std::string next`,再在写锁中 -`swap`;失败保留旧配置。输入平台字符串仅借用到同步调用返回,Integration 和节点 -负责形成各自拥有的内部值。不要把平台指针、请求 Context 或本次输入存进节点成员。 - -模板只展示值类型配置。含模型句柄、外部资源的更新需要单独设计所有权;不能假设复制 -结构体就能深拷贝资源或撤销外部副作用。 +| `PrefixControlNodeParams` / `Field("prefix", ...)` | 声明业务参数结构体,绑定初值默认空字符串、字段说明与 64 字节业务校验 | +| `kUpdatePrefix` / `ReplaceFields(...)` | 声明具名命令 ID 与受控字段集合,自动投影 Control payload schema | +| `ApplyPrefix(...)` | 纯业务转换函数,接收普通数据与参数,无需接触锁或平台结构 | +| `WithControls(...)` | 将受控命令挂载到 Spec,框架自动管理不可变快照与并发更新事务 | + +`WithControls` 引用 `ReplaceFields(kUpdatePrefix, "set_prefix", {"prefix"})`,框架复用 `Parameters` 已绑定的字段类型、默认值和业务校验规则自动生成 Control payload schema。初始配置写在节点的 `config`(例如 `{"prefix":"BASE:"}`),未设置时使用默认空字符串;Control 下发新值时通过相同校验规则验证,并通过不可变快照原子发布。非法初始配置会在预检拒绝,直接 Init 也返回具体原因。 + +框架采用 `ConfigurationSnapshot` 管理节点状态:更新在独立的 writer 锁内构建候选、校验成功后原子发布;正在执行的 Process 读取单次快照处理整批请求,互不干扰;更新失败保留旧配置。开发者只需关注普通参数绑定与业务逻辑,不需要手写互斥锁、JSON 解析或快照轮询。 + +模板展示值类型参数的控制。含模型句柄、外部资源的更新需要单独设计所有权;不能假设复制结构体就能深拷贝资源或撤销外部副作用。 ## 4. 编译并检查实际注册 diff --git a/doc/rfcs/0054-controlled-configuration-snapshots.md b/doc/rfcs/0054-controlled-configuration-snapshots.md index 7cfbbf3b..595c3f34 100644 --- a/doc/rfcs/0054-controlled-configuration-snapshots.md +++ b/doc/rfcs/0054-controlled-configuration-snapshots.md @@ -2,7 +2,7 @@ - **RFC 编号**:0054-controlled-configuration-snapshots - **创建日期**:2026-09-14 -- **文档状态**:Proposed +- **文档状态**:In Implementation - **关联分支**:`docs/framework-authoring-rfcs`;建议实施分支 `refactor/control-snapshots` - **目标版本**:下一次投产前开发接口版本 - **负责人 / 作者**:LLM-EdgeFlow contributors @@ -278,14 +278,100 @@ Map/Batch 继续使用同一套输入绑定、模型绑定、异常屏障、结 生命周期、并发和 AuthorNode 变更需独立 Reviewer。只有所有必需工程出口满足才能声明工程 交付;真实试用尚未进行时保留状态和待办,不以 Agent 自测关闭体验验收。 -## 9. 实施与最终结果记录 +## 9. 实施与验证记录 -| 项目 | 当前状态 | -| --- | --- | -| 设计文档 | Proposed;未改变运行时并发契约 | -| 生产实现与迁移 | 未开始 | -| 工程、并发与性能验证 | 待实施后填写命令、基线和结果 | -| 开发者试用 | 待记录 | -| 完成条件 | M0–M5 必需交付、验证、现行指南及体验结果记录完成;按 CONTRIBUTING 更新状态 | +生产实现与迁移已完成:`ConfigurationSnapshot`、`PatchFields` / `ReplaceFields`、 +函数式 Node opt-in,以及 starter、TextTemplate、TextRuleMatch 的迁移。未启用 Control 的 +不可复制参数仍可使用;Init 与 Control 使用真实 `BindingFacts` 执行参数语义链。 +真实开发者试用仍为待办,本文暂保留 `In Implementation`,不以工程自测关闭体验验收。 + +### 9.1 并发与生命周期证据 + +Map、Batch、TextTemplate、TextRuleMatch 的直接节点测试使用同一确定性握手: +在 reader 已取得快照后的首次分配处暂停,主线程完成 Control 后才放行 reader; +断言整个在途批次仍使用旧值、后续批次使用新值,并检查 `(req_id, sub_id)`。 +规则节点的分类、规则 ID 和正则捕获均可区分 OLD/NEW,避免所有版本产生相同结果。 + +同步设施仅链接进测试可执行文件,使用线程局部一次性分配回调;生产 Node 不增加测试钩子。 +等待超时会使 Process 失败,另有零等待预算的失败回归,不能把超时当作发布成功。 +组件测试检查旧 reader 在 owner 销毁后仍可读、最后 reader 释放后 `weak_ptr` 过期; +两个 writer 的合并检查不使用 sleep 推测锁状态。压力测试作为补充,不替代确定时序证明。 + +### 9.2 性能与分配实测 + +2026-09-14,在 Linux 6.17.0-1020-oracle **aarch64**、GCC 13.3.0 的同一共享工作机上, +对比 `7a6ca02` 的两个迁移前 Node 翻译单元与当前迁移实现。两版以相同 `-O3` 参数重新编译, +复用当前运行时对象;这是节点迁移对比,不是完整历史 SDK 重建。 + +每次 Process 50 样本,每轮 2000 次,7 轮交替 baseline/current。计时前准备请求, +计时后检查返回值、输出数量、模板文本及实际命中的正则捕获。并发 writer 每次 Control +完成后等待 100 µs,两个版本完成的更新数量可以不同。单位为每批中位耗时 µs: + +| 场景 | 迁移前 | 快照实现 | +| --- | ---: | ---: | +| TextTemplate,无并发 Control | 23.8713 | 23.2394 | +| TextTemplate,并发 Control | 27.3272 | 25.0500 | +| TextRuleMatch,无并发 Control | 400.5160 | 394.1910 | +| TextRuleMatch,并发 Control | 434.1460 | 429.2030 | + +该负载未观察到明显回退;共享机器上的小幅差异不应解释为统计显著的加速。 +原文未记录负载的估计表已撤回。56 次原始记录见 +[benchmark JSON](reviews/0054-control-snapshot-benchmark.json);完整负载定义见 +[基准程序](../../dev_support/benchmarks/control_snapshots.cpp)。在完成默认构建且没有其他 +构建/压力测试负载时,可重新执行: + +```bash +python3 dev_support/benchmarks/control_snapshots.py \ + --build-dir build --output-dir /tmp/edgeflow-control-benchmark \ + --baseline 7a6ca02 +``` + +完整预热 Control 的普通全局 C++ `new/new[]` 分配计数为: + +| 节点 | 迁移前次数 / 累计请求字节 | 快照实现次数 / 累计请求字节 | +| --- | ---: | ---: | +| TextTemplate | 24 / 1015 | 26 / 1443 | +| TextRuleMatch(更新 categories,保留编译 regex) | 39 / 1427 | 45 / 2024 | + +计数不含直接 malloc 或 aligned allocation,因此只是所有堆分配的下界;累计申请字节 +不是峰值或持有内存,也不代表泄漏。`make_shared` 的分配不能代表包含 JSON 解析、状态复制、 +容器和派生值构造的完整 Control 成本。 + +`Read()` 获取 shared_ptr 引用并由 Process 局部 guard 持有。C++17 shared_ptr 原子自由函数 +不保证 lock-free;本机 libstdc++ 使用内部锁。读者不持有组件的 writer mutex, +但不能声称完全无锁、零阻塞或无等待发布。 + +### 9.3 Sanitizer 与门禁 + +修复前复核曾实际执行默认门禁(97 CTest)、ASan/UBSan + 泄漏检测(87 个相关测试), +以及仅对快照组件插桩的 TSAN 检查(9 个测试 × 20 轮,含临时补充用例);均无报告。 +这些是当时的覆盖范围,不代表 TSAN 已检查整个 SDK,也不构成修复后的自动豁免。 +本轮测试修复必须重新运行相关用例及 `./scripts/run_all_tests.sh`。 + +项目 sanitizer 选项是 `LLM_EDGEFLOW_SANITIZERS`,不存在 `ENABLE_TSAN` 开关。 +使用独立构建目录,避免污染默认门禁的构建;以下为完整节点 runner 的 TSAN 复现配置: + +```bash +cmake -S . -B build-control-tsan -DBUILD_TESTING=ON \ + -DCMAKE_BUILD_TYPE=Debug -DENABLE_SANITIZERS=ON \ + -DLLM_EDGEFLOW_SANITIZERS=thread \ + -DENABLE_KITELLM=OFF -DENABLE_WHISPERCPP=OFF \ + -DENABLE_LLAMACPP=OFF -DENABLE_ONNXRUNTIME=OFF +cmake --build build-control-tsan --target edgeflow_test_nodes_runner -j2 +TSAN_OPTIONS=halt_on_error=1 ./build-control-tsan/edgeflow_test_nodes_runner \ + --gtest_filter='ConfigurationSnapshotTest.*:FunctionNodeTest.*:TextTemplateNodeTest.*:TextRuleMatchNodeTest.*' +``` + +本机默认 TSAN 启动曾报 `unexpected memory mapping`;按 `scripts/run_sanitizers.sh` +的 aarch64 分支使用 `setarch aarch64 -R` 后,真实组件 TSAN 检查成功。因此应记录具体 +命令与错误,不能仅以“部分容器可能受限”代替尝试。其他平台按实际环境处理。 +不引用不存在的 suppression 文件。ASan/UBSan 使用相同独立构建方式,将 sanitizer 集合 +改为 `address,undefined`,运行时使用 `ASAN_OPTIONS=detect_leaks=1:halt_on_error=1` 和 +`UBSAN_OPTIONS=halt_on_error=1:print_stacktrace=1`。未安装/执行 Valgrind,不作相关通过声明。 + +### 9.4 真实开发者试用待办 -当前文档通过检查不代表快照组件已经实现。实施差异、验证结果和保留边界直接维护在本文。 +工程材料包括 [Control 指南](../dev_guide/first_control.md)、编译 starter 与 +`scripts/scaffold_custom_node.py --control-id`。邀请业务线开发者在真实 Node 中增加普通 +可更新字段,记录参数声明、Patch/Replace、语义校验和诊断体验。收集真实反馈后再关闭 +体验待办;教程、脚手架与 Agent 自测不替代真实试用。 diff --git a/doc/rfcs/README.md b/doc/rfcs/README.md index dffda672..8051861f 100644 --- a/doc/rfcs/README.md +++ b/doc/rfcs/README.md @@ -23,7 +23,7 @@ | **RFC-0036** | Whisper ASR 与 whisper.cpp Backend 接入设计及实施指南 | `In Implementation` | `v10.x` | 模型执行层 / Config / Demo / Build | [0036-whisper-asr-backend.md](0036-whisper-asr-backend.md) | | **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 作者接口与不可变配置快照 | `Proposed` | 投产前 | 能力节点层 / Tooling / Docs | [0054-controlled-configuration-snapshots.md](0054-controlled-configuration-snapshots.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 的实施规格。两篇没有相互编译依赖, diff --git a/doc/rfcs/reviews/0054-control-snapshot-benchmark.json b/doc/rfcs/reviews/0054-control-snapshot-benchmark.json new file mode 100644 index 00000000..6eb87f12 --- /dev/null +++ b/doc/rfcs/reviews/0054-control-snapshot-benchmark.json @@ -0,0 +1,562 @@ +[ + { + "round": 0, + "node": "template", + "concurrent": 0, + "version": "baseline", + "us": 23.4861, + "updates": 0, + "allocations": 24, + "bytes": 1015 + }, + { + "round": 0, + "node": "template", + "concurrent": 0, + "version": "current", + "us": 23.9669, + "updates": 0, + "allocations": 26, + "bytes": 1443 + }, + { + "round": 0, + "node": "template", + "concurrent": 1, + "version": "baseline", + "us": 27.3272, + "updates": 294, + "allocations": 24, + "bytes": 1015 + }, + { + "round": 0, + "node": "template", + "concurrent": 1, + "version": "current", + "us": 26.1707, + "updates": 309, + "allocations": 26, + "bytes": 1443 + }, + { + "round": 0, + "node": "rules", + "concurrent": 0, + "version": "baseline", + "us": 416.808, + "updates": 0, + "allocations": 39, + "bytes": 1427 + }, + { + "round": 0, + "node": "rules", + "concurrent": 0, + "version": "current", + "us": 396.365, + "updates": 0, + "allocations": 45, + "bytes": 2024 + }, + { + "round": 0, + "node": "rules", + "concurrent": 1, + "version": "baseline", + "us": 428.922, + "updates": 2000, + "allocations": 39, + "bytes": 1427 + }, + { + "round": 0, + "node": "rules", + "concurrent": 1, + "version": "current", + "us": 408.13, + "updates": 4797, + "allocations": 45, + "bytes": 2024 + }, + { + "round": 1, + "node": "template", + "concurrent": 0, + "version": "current", + "us": 22.9291, + "updates": 0, + "allocations": 26, + "bytes": 1443 + }, + { + "round": 1, + "node": "template", + "concurrent": 0, + "version": "baseline", + "us": 22.4435, + "updates": 0, + "allocations": 24, + "bytes": 1015 + }, + { + "round": 1, + "node": "template", + "concurrent": 1, + "version": "current", + "us": 24.9005, + "updates": 296, + "allocations": 26, + "bytes": 1443 + }, + { + "round": 1, + "node": "template", + "concurrent": 1, + "version": "baseline", + "us": 27.9497, + "updates": 298, + "allocations": 24, + "bytes": 1015 + }, + { + "round": 1, + "node": "rules", + "concurrent": 0, + "version": "current", + "us": 393.381, + "updates": 0, + "allocations": 45, + "bytes": 2024 + }, + { + "round": 1, + "node": "rules", + "concurrent": 0, + "version": "baseline", + "us": 411.737, + "updates": 0, + "allocations": 39, + "bytes": 1427 + }, + { + "round": 1, + "node": "rules", + "concurrent": 1, + "version": "current", + "us": 409.647, + "updates": 4814, + "allocations": 45, + "bytes": 2024 + }, + { + "round": 1, + "node": "rules", + "concurrent": 1, + "version": "baseline", + "us": 433.294, + "updates": 1999, + "allocations": 39, + "bytes": 1427 + }, + { + "round": 2, + "node": "template", + "concurrent": 0, + "version": "baseline", + "us": 23.8713, + "updates": 0, + "allocations": 24, + "bytes": 1015 + }, + { + "round": 2, + "node": "template", + "concurrent": 0, + "version": "current", + "us": 23.2394, + "updates": 0, + "allocations": 26, + "bytes": 1443 + }, + { + "round": 2, + "node": "template", + "concurrent": 1, + "version": "baseline", + "us": 26.9093, + "updates": 294, + "allocations": 24, + "bytes": 1015 + }, + { + "round": 2, + "node": "template", + "concurrent": 1, + "version": "current", + "us": 26.5008, + "updates": 318, + "allocations": 26, + "bytes": 1443 + }, + { + "round": 2, + "node": "rules", + "concurrent": 0, + "version": "baseline", + "us": 398.386, + "updates": 0, + "allocations": 39, + "bytes": 1427 + }, + { + "round": 2, + "node": "rules", + "concurrent": 0, + "version": "current", + "us": 398.767, + "updates": 0, + "allocations": 45, + "bytes": 2024 + }, + { + "round": 2, + "node": "rules", + "concurrent": 1, + "version": "baseline", + "us": 434.146, + "updates": 2000, + "allocations": 39, + "bytes": 1427 + }, + { + "round": 2, + "node": "rules", + "concurrent": 1, + "version": "current", + "us": 429.203, + "updates": 4907, + "allocations": 45, + "bytes": 2024 + }, + { + "round": 3, + "node": "template", + "concurrent": 0, + "version": "current", + "us": 23.3498, + "updates": 0, + "allocations": 26, + "bytes": 1443 + }, + { + "round": 3, + "node": "template", + "concurrent": 0, + "version": "baseline", + "us": 23.8728, + "updates": 0, + "allocations": 24, + "bytes": 1015 + }, + { + "round": 3, + "node": "template", + "concurrent": 1, + "version": "current", + "us": 25.322, + "updates": 306, + "allocations": 26, + "bytes": 1443 + }, + { + "round": 3, + "node": "template", + "concurrent": 1, + "version": "baseline", + "us": 26.2741, + "updates": 281, + "allocations": 24, + "bytes": 1015 + }, + { + "round": 3, + "node": "rules", + "concurrent": 0, + "version": "current", + "us": 394.191, + "updates": 0, + "allocations": 45, + "bytes": 2024 + }, + { + "round": 3, + "node": "rules", + "concurrent": 0, + "version": "baseline", + "us": 394.916, + "updates": 0, + "allocations": 39, + "bytes": 1427 + }, + { + "round": 3, + "node": "rules", + "concurrent": 1, + "version": "current", + "us": 424.502, + "updates": 4991, + "allocations": 45, + "bytes": 2024 + }, + { + "round": 3, + "node": "rules", + "concurrent": 1, + "version": "baseline", + "us": 434.459, + "updates": 2000, + "allocations": 39, + "bytes": 1427 + }, + { + "round": 4, + "node": "template", + "concurrent": 0, + "version": "baseline", + "us": 23.8871, + "updates": 0, + "allocations": 24, + "bytes": 1015 + }, + { + "round": 4, + "node": "template", + "concurrent": 0, + "version": "current", + "us": 24.2985, + "updates": 0, + "allocations": 26, + "bytes": 1443 + }, + { + "round": 4, + "node": "template", + "concurrent": 1, + "version": "baseline", + "us": 27.2498, + "updates": 296, + "allocations": 24, + "bytes": 1015 + }, + { + "round": 4, + "node": "template", + "concurrent": 1, + "version": "current", + "us": 24.5608, + "updates": 298, + "allocations": 26, + "bytes": 1443 + }, + { + "round": 4, + "node": "rules", + "concurrent": 0, + "version": "baseline", + "us": 400.516, + "updates": 0, + "allocations": 39, + "bytes": 1427 + }, + { + "round": 4, + "node": "rules", + "concurrent": 0, + "version": "current", + "us": 391.373, + "updates": 0, + "allocations": 45, + "bytes": 2024 + }, + { + "round": 4, + "node": "rules", + "concurrent": 1, + "version": "baseline", + "us": 427.545, + "updates": 2000, + "allocations": 39, + "bytes": 1427 + }, + { + "round": 4, + "node": "rules", + "concurrent": 1, + "version": "current", + "us": 432.237, + "updates": 4933, + "allocations": 45, + "bytes": 2024 + }, + { + "round": 5, + "node": "template", + "concurrent": 0, + "version": "current", + "us": 22.9768, + "updates": 0, + "allocations": 26, + "bytes": 1443 + }, + { + "round": 5, + "node": "template", + "concurrent": 0, + "version": "baseline", + "us": 22.9588, + "updates": 0, + "allocations": 24, + "bytes": 1015 + }, + { + "round": 5, + "node": "template", + "concurrent": 1, + "version": "current", + "us": 25.05, + "updates": 305, + "allocations": 26, + "bytes": 1443 + }, + { + "round": 5, + "node": "template", + "concurrent": 1, + "version": "baseline", + "us": 27.3507, + "updates": 294, + "allocations": 24, + "bytes": 1015 + }, + { + "round": 5, + "node": "rules", + "concurrent": 0, + "version": "current", + "us": 398.096, + "updates": 0, + "allocations": 45, + "bytes": 2024 + }, + { + "round": 5, + "node": "rules", + "concurrent": 0, + "version": "baseline", + "us": 398.42, + "updates": 0, + "allocations": 39, + "bytes": 1427 + }, + { + "round": 5, + "node": "rules", + "concurrent": 1, + "version": "current", + "us": 432.916, + "updates": 5077, + "allocations": 45, + "bytes": 2024 + }, + { + "round": 5, + "node": "rules", + "concurrent": 1, + "version": "baseline", + "us": 439.731, + "updates": 1997, + "allocations": 39, + "bytes": 1427 + }, + { + "round": 6, + "node": "template", + "concurrent": 0, + "version": "baseline", + "us": 24.2007, + "updates": 0, + "allocations": 24, + "bytes": 1015 + }, + { + "round": 6, + "node": "template", + "concurrent": 0, + "version": "current", + "us": 22.9217, + "updates": 0, + "allocations": 26, + "bytes": 1443 + }, + { + "round": 6, + "node": "template", + "concurrent": 1, + "version": "baseline", + "us": 27.4704, + "updates": 295, + "allocations": 24, + "bytes": 1015 + }, + { + "round": 6, + "node": "template", + "concurrent": 1, + "version": "current", + "us": 24.3614, + "updates": 297, + "allocations": 26, + "bytes": 1443 + }, + { + "round": 6, + "node": "rules", + "concurrent": 0, + "version": "baseline", + "us": 414.98, + "updates": 0, + "allocations": 39, + "bytes": 1427 + }, + { + "round": 6, + "node": "rules", + "concurrent": 0, + "version": "current", + "us": 393.141, + "updates": 0, + "allocations": 45, + "bytes": 2024 + }, + { + "round": 6, + "node": "rules", + "concurrent": 1, + "version": "baseline", + "us": 449.59, + "updates": 2000, + "allocations": 39, + "bytes": 1427 + }, + { + "round": 6, + "node": "rules", + "concurrent": 1, + "version": "current", + "us": 431.644, + "updates": 4902, + "allocations": 45, + "bytes": 2024 + } +] \ No newline at end of file diff --git a/include/nodes/authoring.h b/include/nodes/authoring.h index ab7caef9..0f9049f9 100644 --- a/include/nodes/authoring.h +++ b/include/nodes/authoring.h @@ -2,6 +2,8 @@ #include "contracts/inference_payloads.h" #include "contracts/traceable_item.h" +#include "nodes/configuration_snapshot.h" +#include "nodes/control_authoring.h" #include "nodes/function_node.h" #include "nodes/model_calls.h" #include "nodes/node_error_codes.h" diff --git a/include/nodes/configuration_snapshot.h b/include/nodes/configuration_snapshot.h new file mode 100644 index 00000000..e55bc6b2 --- /dev/null +++ b/include/nodes/configuration_snapshot.h @@ -0,0 +1,193 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +#include "core/node_interface.h" +#include "core/validated_node_plan.h" +#include "nodes/node_error_codes.h" +#include "nodes/node_result.h" + +namespace llm_edgeflow { + +/** + * @brief Connection facts of input ports captured defensively during Init. + * + * Immutable after initialization; retains distinction between "no plan" + * and "plan present with explicit bindings". + */ +struct BindingFacts { + bool has_plan = false; + bool has_bindings = false; + std::unordered_set connected_inputs; + + bool IsConnected(const std::string& port_name) const noexcept { + return connected_inputs.count(port_name) > 0; + } +}; + +inline BindingFacts MakeBindingFacts( + const NodeInitContext& ctx, + std::unordered_set connected_inputs) { + BindingFacts facts; + facts.has_plan = (ctx.plan != nullptr); + facts.has_bindings = true; + facts.connected_inputs = std::move(connected_inputs); + return facts; +} + +inline BindingFacts MakeBindingFacts(const NodeInitContext& ctx) { + BindingFacts facts; + facts.has_bindings = true; + if (ctx.plan) { + facts.has_plan = true; + for (const auto& port : ctx.plan->ports) { + if (port.direction == PortDirection::kInput && + !port.blackboard_key.empty()) { + facts.connected_inputs.insert(port.logical_name); + } + } + } + return facts; +} + +inline BindingFacts MakeBindingFacts(const ValidatedNodePlan* plan) { + BindingFacts facts; + facts.has_bindings = true; + if (plan) { + facts.has_plan = true; + for (const auto& port : plan->ports) { + if (port.direction == PortDirection::kInput && + !port.blackboard_key.empty()) { + facts.connected_inputs.insert(port.logical_name); + } + } + } + return facts; +} + +/** + * @brief Thread-safe configuration snapshot manager for node instances + * (RFC-0054). + * + * Enforces: + * - Atomic snapshot acquisition for readers via atomic load acquire (without + * the writer mutex; shared_ptr atomic operations may use internal locks). + * - Serialized, transaction-safe candidate building and atomic publication for + * writers. + * - Readers safely retain old snapshots for arbitrary batch duration. + * - Failed candidate building / validation never overwrites active + * configuration. + */ +template +class ConfigurationSnapshot { + public: + ConfigurationSnapshot() = default; + + explicit ConfigurationSnapshot(State initial_state) { + Initialize(std::move(initial_state)); + } + + explicit ConfigurationSnapshot(std::shared_ptr initial_state) { + Initialize(std::move(initial_state)); + } + + ~ConfigurationSnapshot() = default; + + ConfigurationSnapshot(const ConfigurationSnapshot&) = delete; + ConfigurationSnapshot& operator=(const ConfigurationSnapshot&) = delete; + ConfigurationSnapshot(ConfigurationSnapshot&&) = delete; + ConfigurationSnapshot& operator=(ConfigurationSnapshot&&) = delete; + + bool Initialize(State initial_state) { + std::lock_guard lock(writer_mutex_); + auto ptr = std::make_shared(std::move(initial_state)); + std::atomic_store_explicit(&state_, + std::shared_ptr(std::move(ptr)), + std::memory_order_release); + return true; + } + + bool Initialize(std::shared_ptr initial_state) { + if (!initial_state) return false; + std::lock_guard lock(writer_mutex_); + std::atomic_store_explicit(&state_, std::move(initial_state), + std::memory_order_release); + return true; + } + + bool IsInitialized() const noexcept { + return std::atomic_load_explicit(&state_, std::memory_order_acquire) != + nullptr; + } + + std::shared_ptr Read() const noexcept { + return std::atomic_load_explicit(&state_, std::memory_order_acquire); + } + + template + NodeControlResult Update(BuildNextFn&& build_next) { + std::unique_lock lock(writer_mutex_); + auto current = + std::atomic_load_explicit(&state_, std::memory_order_acquire); + if (!current) { + return NodeControlResult::Failed( + node_error::control::kInvalidRequest, + "ConfigurationSnapshot is not initialized"); + } + try { + auto run_build = [&]() { + if constexpr (std::is_invocable_v) { + return build_next(*current); + } else { + return build_next(current); + } + }; + auto res = run_build(); + if (!res.ok()) { + const auto& failure = res.failure(); + int code = failure.cause_code != 0 + ? failure.cause_code + : node_error::control::kInvalidRequest; + return NodeControlResult::Failed( + code, failure.message.empty() ? "Configuration update failed" + : failure.message); + } + std::shared_ptr next_ptr; + using ResValType = std::decay_t; + if constexpr (std::is_same_v> || + std::is_same_v>) { + next_ptr = std::move(res).value(); + } else { + next_ptr = std::make_shared(std::move(res).value()); + } + if (!next_ptr) { + return NodeControlResult::Failed( + node_error::control::kInvalidRequest, + "Configuration update produced null state"); + } + auto success_res = NodeControlResult::Handled(); + std::atomic_store_explicit(&state_, std::move(next_ptr), + std::memory_order_release); + return success_res; + } catch (const std::exception& e) { + return NodeControlResult::Failed(node_error::control::kInvalidRequest, + e.what()); + } catch (...) { + return NodeControlResult::Failed( + node_error::control::kInvalidRequest, + "Unknown exception during configuration update"); + } + } + + private: + mutable std::mutex writer_mutex_; + std::shared_ptr state_; +}; + +} // namespace llm_edgeflow diff --git a/include/nodes/control_authoring.h b/include/nodes/control_authoring.h new file mode 100644 index 00000000..6562028d --- /dev/null +++ b/include/nodes/control_authoring.h @@ -0,0 +1,283 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "contracts/control_payload.h" +#include "core/common_contracts.h" +#include "core/node_definition.h" +#include "nodes/configuration_snapshot.h" +#include "nodes/node_error_codes.h" +#include "nodes/node_result.h" +#include "nodes/parameter_binding.h" + +namespace llm_edgeflow { + +enum class ControlFieldStrategy { + kPatch, + kReplace, +}; + +class FieldControlCommand { + public: + FieldControlCommand(ControlFieldStrategy strategy, int cmd_id, + std::string name, std::vector field_names, + std::string description = "") + : strategy_(strategy), + cmd_id_(cmd_id), + name_(std::move(name)), + field_names_(std::move(field_names)), + description_(std::move(description)) {} + + ControlFieldStrategy Strategy() const noexcept { return strategy_; } + int Id() const noexcept { return cmd_id_; } + const std::string& Name() const noexcept { return name_; } + const std::vector& FieldNames() const noexcept { + return field_names_; + } + const std::string& Description() const noexcept { return description_; } + + FieldControlCommand& Description(std::string desc) { + description_ = std::move(desc); + return *this; + } + + FieldControlCommand& SharedId(bool shared) { + shared_id_ = shared; + return *this; + } + + bool IsSharedId() const noexcept { return shared_id_; } + + template + nlohmann::json GenerateSchema(const Parameters& params) const { + nlohmann::json schema = { + {"type", "object"}, + {"additionalProperties", false}, + }; + nlohmann::json props = nlohmann::json::object(); + nlohmann::json req = nlohmann::json::array(); + for (const auto& fname : field_names_) { + const auto* binding = params.FindBinding(fname); + if (!binding) continue; + ConfigFieldDefinition def = binding->ToFieldDefinition(); + nlohmann::json prop = nlohmann::json::object(); + switch (def.kind) { + case ConfigValueKind::kString: + prop["type"] = "string"; + if (!def.enum_values.empty()) { + prop["enum"] = def.enum_values; + } + break; + case ConfigValueKind::kBoolean: + prop["type"] = "boolean"; + break; + case ConfigValueKind::kInteger: + prop["type"] = "integer"; + if (def.minimum.has_value()) prop["minimum"] = *def.minimum; + if (def.maximum.has_value()) prop["maximum"] = *def.maximum; + break; + case ConfigValueKind::kNumber: + prop["type"] = "number"; + if (def.minimum.has_value()) prop["minimum"] = *def.minimum; + if (def.maximum.has_value()) prop["maximum"] = *def.maximum; + break; + case ConfigValueKind::kArray: + prop["type"] = "array"; + prop["items"] = {{"type", "string"}}; + break; + default: + break; + } + if (!def.semantic.empty()) { + prop["description"] = def.semantic; + } + if (!def.default_value.is_null()) { + prop["default"] = def.default_value; + } + props[fname] = std::move(prop); + if (strategy_ == ControlFieldStrategy::kReplace) { + req.push_back(fname); + } + } + schema["properties"] = std::move(props); + if (strategy_ == ControlFieldStrategy::kReplace) { + schema["required"] = std::move(req); + } else { + schema["minProperties"] = 1; + } + return schema; + } + + template + ControlCommandDefinition ToCommandDefinition( + const Parameters& params) const { + ControlCommandDefinition def(cmd_id_, name_, + description_.empty() ? name_ : description_, + GenerateSchema(params), /*is_hot_swap=*/true); + def.shared_id = shared_id_; + return def; + } + + template + NodeControlResult Execute(const Parameters& params, + const std::string& json_param, + const BindingFacts& facts, + ConfigurationSnapshot& snapshot) const { + if constexpr (!std::is_copy_constructible_v) { + return NodeControlResult::Unsupported(); + } else { + nlohmann::json payload; + std::string err; + auto cmd_def = ToCommandDefinition(params); + if (!ParseControlPayload(json_param, cmd_def.payload_schema, &payload, + &err)) { + return NodeControlResult::Failed(node_error::control::kInvalidRequest, + err); + } + return snapshot.Update( + [&](const ParamsT& current) -> NodeResult { + ParamsT next = current; + for (const auto& field_name : field_names_) { + if (strategy_ == ControlFieldStrategy::kReplace) { + if (!payload.contains(field_name)) { + return NodeResult::Failure( + NodeErrorKind::kBusinessError, + "Missing required field in control payload: " + + field_name, + node_error::control::kInvalidRequest); + } + std::string assign_err; + if (!params.AssignField(field_name, payload[field_name], &next, + &assign_err)) { + return NodeResult::Failure( + NodeErrorKind::kBusinessError, assign_err, + node_error::control::kInvalidRequest); + } + } else { // kPatch + if (payload.contains(field_name)) { + std::string assign_err; + if (!params.AssignField(field_name, payload[field_name], + &next, &assign_err)) { + return NodeResult::Failure( + NodeErrorKind::kBusinessError, assign_err, + node_error::control::kInvalidRequest); + } + } + } + } + std::string val_err; + if (!params.ValidateState(&next, facts, &val_err)) { + return NodeResult::Failure( + NodeErrorKind::kBusinessError, + val_err.empty() + ? "Parameter validation failed after control update" + : val_err, + node_error::control::kInvalidRequest); + } + return NodeResult::Success(std::move(next)); + }); + } + } + + private: + ControlFieldStrategy strategy_; + int cmd_id_; + std::string name_; + std::vector field_names_; + std::string description_; + bool shared_id_ = false; +}; + +inline FieldControlCommand ReplaceFields(int cmd_id, std::string name, + std::vector field_names, + std::string description = "") { + return FieldControlCommand(ControlFieldStrategy::kReplace, cmd_id, + std::move(name), std::move(field_names), + std::move(description)); +} + +inline FieldControlCommand PatchFields(int cmd_id, std::string name, + std::vector field_names, + std::string description = "") { + return FieldControlCommand(ControlFieldStrategy::kPatch, cmd_id, + std::move(name), std::move(field_names), + std::move(description)); +} + +template +inline void ValidateControlCommands( + const std::vector& commands, + const Parameters& params, const ModelsT* models = nullptr) { + std::unordered_set seen_ids; + std::unordered_set seen_names; + for (const auto& cmd : commands) { + if (cmd.Id() <= 0) { + throw std::invalid_argument( + "Control command ID must be a positive integer"); + } + if (cmd.Name().empty()) { + throw std::invalid_argument("Control command name cannot be empty"); + } + if (!seen_ids.insert(cmd.Id()).second) { + throw std::invalid_argument("Duplicate control command ID: " + + std::to_string(cmd.Id())); + } + if (!seen_names.insert(cmd.Name()).second) { + throw std::invalid_argument("Duplicate control command name: " + + cmd.Name()); + } + if (cmd.FieldNames().empty()) { + throw std::invalid_argument("Control command '" + cmd.Name() + + "' must specify at least one field"); + } + std::unordered_set seen_fields; + for (const auto& f : cmd.FieldNames()) { + if (!seen_fields.insert(f).second) { + throw std::invalid_argument("Duplicate field '" + f + + "' in control command '" + cmd.Name() + + "'"); + } + const auto* binding = params.FindBinding(f); + if (!binding) { + throw std::invalid_argument("Field '" + f + "' in control command '" + + cmd.Name() + + "' is not bound in Parameters"); + } + ConfigFieldDefinition def = binding->ToFieldDefinition(); + if (def.kind != ConfigValueKind::kString && + def.kind != ConfigValueKind::kBoolean && + def.kind != ConfigValueKind::kInteger && + def.kind != ConfigValueKind::kNumber && + def.kind != ConfigValueKind::kArray) { + throw std::invalid_argument( + "Field '" + f + "' has unsupported kind for control command"); + } + if constexpr (!std::is_void_v) { + if (models) { + for (const auto& b : models->Bindings()) { + if (b && b->ConfigField() == f) { + throw std::invalid_argument( + "Field '" + f + + "' is bound to a model and cannot be controlled"); + } + } + } + } + } + } + if (params.HasParser() && !commands.empty() && !params.HasPrepare()) { + throw std::invalid_argument( + "Spec with WithParser and WithControls requires an explicit Prepare " + "function"); + } +} + +} // namespace llm_edgeflow diff --git a/include/nodes/function_node.h b/include/nodes/function_node.h index 756e2f89..11c792e5 100644 --- a/include/nodes/function_node.h +++ b/include/nodes/function_node.h @@ -17,6 +17,8 @@ #include "core/port_definition.h" #include "core/session_context.h" #include "core/validated_node_plan.h" +#include "nodes/configuration_snapshot.h" +#include "nodes/control_authoring.h" #include "nodes/model_calls.h" #include "nodes/node_base.h" #include "nodes/node_error_codes.h" @@ -113,9 +115,19 @@ auto InvokeBatch(const Fn& fn, const InputsT& inputs, const ParamsT& params, if constexpr (std::is_member_function_pointer_v) { using ClassT = typename MemberFunctionTraits::ClassType; ClassT logic{}; - return (logic.*fn)(inputs, params, models); + if constexpr (std::is_invocable_v) { + return (logic.*fn)(inputs, params, models); + } else { + return (logic.*fn)(inputs, params); + } } else { - return fn(inputs, params, models); + if constexpr (std::is_invocable_v) { + return fn(inputs, params, models); + } else { + return fn(inputs, params); + } } } @@ -183,6 +195,25 @@ class MapSpec { return *this; } + MapSpec WithControls(std::vector commands) && { + static_assert(std::is_copy_constructible_v, + "WithControls requires copy-constructible ParametersType"); + ValidateControlCommands(commands, params_); + control_commands_ = std::move(commands); + return std::move(*this); + } + + MapSpec WithControls(std::initializer_list commands) && { + return std::move(*this).WithControls( + std::vector(commands)); + } + + bool HasControls() const noexcept { return !control_commands_.empty(); } + + const std::vector& ControlCommands() const noexcept { + return control_commands_; + } + const std::string& InputName() const noexcept { return in_.name; } const std::string& OutputName() const noexcept { return out_.name; } const Parameters& ParametersSpec() const noexcept { return params_; } @@ -208,6 +239,9 @@ class MapSpec { std::string* err) { return params.ValidateWithBindings(cfg, conn, err); }; + for (const auto& cmd : control_commands_) { + def.control_commands.push_back(cmd.ToCommandDefinition(params_)); + } return def; } @@ -216,6 +250,7 @@ class MapSpec { Output out_; Parameters params_; MapFnT fn_; + std::vector control_commands_; std::string category_ = "custom"; std::string description_; bool parallel_safe_ = false; @@ -873,6 +908,11 @@ class ModelsOf { const nlohmann::json&, NoModels*, std::string*) { return true; } + static const std::vector>>& + Bindings() noexcept { + static const std::vector>> empty; + return empty; + } }; template commands) && { + static_assert(std::is_copy_constructible_v, + "WithControls requires copy-constructible ParametersType"); + ValidateControlCommands(commands, params_, &models_); + control_commands_ = std::move(commands); + return std::move(*this); + } + + BatchSpec WithControls( + std::initializer_list commands) && { + return std::move(*this).WithControls( + std::vector(commands)); + } + + bool HasControls() const noexcept { return !control_commands_.empty(); } + + const std::vector& ControlCommands() const noexcept { + return control_commands_; + } + InputsOf& Inputs() noexcept { return inputs_; } const InputsOf& Inputs() const noexcept { return inputs_; } const PreservedOutput& Output() const noexcept { @@ -982,6 +1042,9 @@ class BatchSpec { std::string* err) { return params.ValidateWithBindings(cfg, conn, err); }; + for (const auto& cmd : control_commands_) { + def.control_commands.push_back(cmd.ToCommandDefinition(params_)); + } return def; } @@ -991,6 +1054,7 @@ class BatchSpec { Parameters params_; ModelsOf models_; RunFnT fn_; + std::vector control_commands_; std::string category_ = "custom"; std::string description_; bool parallel_safe_ = false; @@ -1120,19 +1184,39 @@ class AuthorNode> connected_inputs.insert(spec_.InputName()); } + binding_facts_ = MakeBindingFacts(init_ctx, std::move(connected_inputs)); + std::string err; - if (!spec_.ParametersSpec().ValidateWithBindings(normalized, - connected_inputs, &err)) { - return init_ctx.Fail(err.empty() ? "Invalid node configuration" : err); - } - auto parsed = spec_.ParametersSpec().ParseNormalized(normalized, &err); + auto parsed = spec_.ParametersSpec().ParseNormalized(normalized, + binding_facts_, &err); if (!parsed) { return init_ctx.Fail(err.empty() ? "Invalid node configuration" : err); } - parameters_ = std::move(*parsed); + if (spec_.HasControls()) { + snapshot_.Initialize(std::move(*parsed)); + } else { + parameters_ = std::move(*parsed); + } return true; } + NodeControlResult ControlNode(int cmd, + const std::string& json_param) override { + if (!spec_.HasControls()) { + return NodeControlResult::Unsupported(); + } + if constexpr (std::is_copy_constructible_v< + typename SpecType::ParametersType>) { + for (const auto& command : spec_.ControlCommands()) { + if (command.Id() == cmd) { + return command.Execute(spec_.ParametersSpec(), json_param, + binding_facts_, snapshot_); + } + } + } + return NodeControlResult::Unsupported(); + } + int ProcessNode(AlgContext& req_ctx) override { using OutputBatch = typename SpecType::OutputBatch; @@ -1147,13 +1231,26 @@ class AuthorNode> return 0; } + std::shared_ptr snapshot_guard; + const typename SpecType::ParametersType* params_ptr = nullptr; + if (spec_.HasControls()) { + snapshot_guard = snapshot_.Read(); + if (!snapshot_guard) { + return this->Fail(req_ctx, node_error::author_node::kInternalError, + this->Name() + ": snapshot not initialized"); + } + params_ptr = snapshot_guard.get(); + } else { + params_ptr = ¶meters_; + } + const auto& params = *params_ptr; + OutputBatch outputs; outputs.reserve(inputs->size()); for (const auto& item : *inputs) { if constexpr (SpecType::kReturnsNodeResult) { - auto res = - detail::InvokeMapItem(spec_.Function(), item.data, parameters_); + auto res = detail::InvokeMapItem(spec_.Function(), item.data, params); if (!res.ok()) { auto failure = std::move(res).ExtractFailure(); int code = failure.cause_code != 0 @@ -1168,7 +1265,7 @@ class AuthorNode> } else { outputs.emplace_back( item.req_id, item.sub_id, - detail::InvokeMapItem(spec_.Function(), item.data, parameters_)); + detail::InvokeMapItem(spec_.Function(), item.data, params)); } } @@ -1193,6 +1290,8 @@ class AuthorNode> BoundInput in_port_; BoundOutput out_port_; typename SpecType::ParametersType parameters_{}; + ConfigurationSnapshot snapshot_; + BindingFacts binding_facts_; }; // BatchSpec Specialization @@ -1246,17 +1345,19 @@ class AuthorNode> } auto connected_inputs = spec_.Inputs().ConnectedInputs(init_ctx.plan); - std::string err; - if (!spec_.ParametersSpec().ValidateWithBindings(normalized, - connected_inputs, &err)) { - return init_ctx.Fail(err.empty() ? "Invalid node configuration" : err); - } + binding_facts_ = MakeBindingFacts(init_ctx, std::move(connected_inputs)); - auto parsed = spec_.ParametersSpec().ParseNormalized(normalized, &err); + std::string err; + auto parsed = spec_.ParametersSpec().ParseNormalized(normalized, + binding_facts_, &err); if (!parsed) { return init_ctx.Fail(err.empty() ? "Invalid node configuration" : err); } - parameters_ = std::move(*parsed); + if (spec_.HasControls()) { + snapshot_.Initialize(std::move(*parsed)); + } else { + parameters_ = std::move(*parsed); + } if (!spec_.Models().BindModels(init_ctx, session_ctx, normalized, &models_, &err)) { @@ -1266,6 +1367,23 @@ class AuthorNode> return true; } + NodeControlResult ControlNode(int cmd, + const std::string& json_param) override { + if (!spec_.HasControls()) { + return NodeControlResult::Unsupported(); + } + if constexpr (std::is_copy_constructible_v< + typename SpecType::ParametersType>) { + for (const auto& command : spec_.ControlCommands()) { + if (command.Id() == cmd) { + return command.Execute(spec_.ParametersSpec(), json_param, + binding_facts_, snapshot_); + } + } + } + return NodeControlResult::Unsupported(); + } + int ProcessNode(AlgContext& req_ctx) override { InputsT inputs{}; std::string err; @@ -1274,8 +1392,21 @@ class AuthorNode> err.empty() ? "Failed to populate inputs" : err); } + std::shared_ptr snapshot_guard; + const typename SpecType::ParametersType* params_ptr = nullptr; + if (spec_.HasControls()) { + snapshot_guard = snapshot_.Read(); + if (!snapshot_guard) { + return this->Fail(req_ctx, node_error::author_node::kInternalError, + this->Name() + ": snapshot not initialized"); + } + params_ptr = snapshot_guard.get(); + } else { + params_ptr = ¶meters_; + } + auto res = - detail::InvokeBatch(spec_.Function(), inputs, parameters_, models_); + detail::InvokeBatch(spec_.Function(), inputs, *params_ptr, models_); if (!res.ok()) { auto failure = std::move(res).ExtractFailure(); int code = failure.cause_code != 0 @@ -1328,6 +1459,8 @@ class AuthorNode> BoundOutput out_port_; typename SpecType::ParametersType parameters_{}; typename SpecType::ModelsType models_{}; + ConfigurationSnapshot snapshot_; + BindingFacts binding_facts_; }; // --------------------------------------------------------------------------- diff --git a/include/nodes/parameter_binding.h b/include/nodes/parameter_binding.h index 80addaa2..d7b33791 100644 --- a/include/nodes/parameter_binding.h +++ b/include/nodes/parameter_binding.h @@ -16,6 +16,7 @@ #include "contracts/config_schema.h" #include "contracts/config_schema_validation.h" #include "contracts/diagnostic.h" +#include "nodes/configuration_snapshot.h" #include "nodes/node_config_parser.h" namespace llm_edgeflow { @@ -423,6 +424,8 @@ class Parameters { using SemanticValidator = std::function; using BindingValidator = std::function&, std::string*)>; + using PrepareFunction = + std::function; Parameters() = default; @@ -460,6 +463,7 @@ class Parameters { Parameters(const Parameters& other) : definitions_(other.definitions_), complex_parser_(other.complex_parser_), + prepare_fn_(other.prepare_fn_), semantic_validator_(other.semantic_validator_), binding_validator_(other.binding_validator_) { bindings_.reserve(other.bindings_.size()); @@ -473,6 +477,7 @@ class Parameters { if (this != &other) { definitions_ = other.definitions_; complex_parser_ = other.complex_parser_; + prepare_fn_ = other.prepare_fn_; semantic_validator_ = other.semantic_validator_; binding_validator_ = other.binding_validator_; bindings_.clear(); @@ -485,6 +490,79 @@ class Parameters { } Parameters& operator=(Parameters&&) noexcept = default; + Parameters& Prepare(PrepareFunction prepare) { + prepare_fn_ = std::move(prepare); + return *this; + } + + Parameters& Prepare(std::function prepare) { + prepare_fn_ = [fn = std::move(prepare)](ParamsT* p, const BindingFacts&, + std::string* diag) { + return fn(p, diag); + }; + return *this; + } + + bool HasPrepare() const noexcept { return static_cast(prepare_fn_); } + + bool HasParser() const noexcept { return complex_parser_.has_value(); } + + const std::vector>>& Bindings() + const noexcept { + return bindings_; + } + + const ParameterFieldBinding* FindBinding( + const std::string& name) const noexcept { + for (const auto& b : bindings_) { + if (b && b->Name() == name) return b.get(); + } + return nullptr; + } + + bool AssignField(const std::string& name, const nlohmann::json& val, + ParamsT* out, std::string* err) const { + const auto* binding = FindBinding(name); + if (!binding) { + if (err) *err = "Field '" + name + "' not found in parameter bindings"; + return false; + } + nlohmann::json obj = nlohmann::json::object(); + obj[name] = val; + return binding->Assign(obj, out, err); + } + + bool ValidateState(ParamsT* state, const BindingFacts& facts, + std::string* err) const noexcept { + try { + if (prepare_fn_) { + if (!prepare_fn_(state, facts, err)) { + if (err && err->empty()) *err = "Prepare failed"; + return false; + } + } + if (semantic_validator_) { + if (!semantic_validator_(*state, err)) { + if (err && err->empty()) *err = "Semantic validation failed"; + return false; + } + } + if (binding_validator_ && (facts.has_plan || facts.has_bindings)) { + if (!binding_validator_(*state, facts.connected_inputs, err)) { + if (err && err->empty()) *err = "Binding validation failed"; + return false; + } + } + return true; + } catch (const std::exception& e) { + SetDiagnosticNoexcept(err, e.what()); + return false; + } catch (...) { + SetDiagnosticNoexcept(err, "Unknown exception validating parameters"); + return false; + } + } + Parameters& Validate(SemanticValidator validator) { semantic_validator_ = std::move(validator); return *this; @@ -541,7 +619,7 @@ class Parameters { } std::optional ParseNormalized( - const nlohmann::json& normalized, + const nlohmann::json& normalized, const BindingFacts& facts, std::string* error = nullptr) const noexcept { if (error) error->clear(); try { @@ -556,13 +634,8 @@ class Parameters { return std::nullopt; } } - if (semantic_validator_) { - if (!semantic_validator_(params, error)) { - if (error && error->empty()) { - *error = "Semantic validation failed"; - } - return std::nullopt; - } + if (!ValidateState(¶ms, facts, error)) { + return std::nullopt; } return params; } catch (const std::exception& e) { @@ -574,35 +647,30 @@ class Parameters { } } + std::optional ParseNormalized( + const nlohmann::json& normalized, + std::string* error = nullptr) const noexcept { + BindingFacts facts; + return ParseNormalized(normalized, facts, error); + } + bool ValidateWithBindings( const nlohmann::json& normalized, const std::unordered_set& connected_inputs, std::string* error = nullptr) const noexcept { - try { - auto parsed = ParseNormalized(normalized, error); - if (!parsed) return false; - if (binding_validator_) { - if (!binding_validator_(*parsed, connected_inputs, error)) { - if (error && error->empty()) { - *error = "Binding validation failed"; - } - return false; - } - } - return true; - } catch (const std::exception& e) { - SetDiagnosticNoexcept(error, e.what()); - return false; - } catch (...) { - SetDiagnosticNoexcept(error, "Unknown exception in binding validator"); - return false; - } + BindingFacts facts; + facts.has_plan = true; + facts.has_bindings = true; + facts.connected_inputs = connected_inputs; + auto parsed = ParseNormalized(normalized, facts, error); + return parsed.has_value(); } private: std::vector>> bindings_; std::vector definitions_; std::optional> complex_parser_; + PrepareFunction prepare_fn_; SemanticValidator semantic_validator_; BindingValidator binding_validator_; }; @@ -638,11 +706,35 @@ class Parameters { return NoParameters{}; } + std::optional ParseNormalized( + const nlohmann::json&, const BindingFacts&, + std::string* = nullptr) const noexcept { + return NoParameters{}; + } + bool ValidateWithBindings(const nlohmann::json&, const std::unordered_set&, std::string* = nullptr) const noexcept { return true; } + + bool HasPrepare() const noexcept { return false; } + bool HasParser() const noexcept { return false; } + + const ParameterFieldBinding* FindBinding( + const std::string&) const noexcept { + return nullptr; + } + + bool AssignField(const std::string&, const nlohmann::json&, NoParameters*, + std::string*) const { + return false; + } + + bool ValidateState(NoParameters*, const BindingFacts&, + std::string*) const noexcept { + return true; + } }; } // namespace llm_edgeflow diff --git a/src/common_nodes/text_rule_match_node.cpp b/src/common_nodes/text_rule_match_node.cpp index ea1a0e24..b9763738 100644 --- a/src/common_nodes/text_rule_match_node.cpp +++ b/src/common_nodes/text_rule_match_node.cpp @@ -15,6 +15,7 @@ #include "core/common_contracts.h" #include "core/node_registry.h" #include "edgeflow/log.h" +#include "nodes/configuration_snapshot.h" #include "nodes/node_base.h" #include "nodes/node_error_codes.h" @@ -105,6 +106,9 @@ class TextRuleMatchNode final : public NodeBase { public: inline static constexpr char kNodeType[] = "TextRuleMatchNode"; + using CategoryList = + std::vector>>; + struct RuleSpec { std::string id; std::string strategy; // "contains", "exact", "regex" @@ -113,7 +117,14 @@ class TextRuleMatchNode final : public NodeBase { float score = kDefaultScore; std::unordered_map constants; std::unordered_map constants_json; - CompiledTextRegex compiled_regex; + std::shared_ptr compiled_regex; + }; + + struct RuleMatchState { + CategoryList category_keywords_list; + std::vector rules_list; + std::string default_category; + float default_score = kDefaultScore; }; TextRuleMatchNode() @@ -141,10 +152,14 @@ class TextRuleMatchNode final : public NodeBase { return NodeControlResult::Failed(node_error::control::kInvalidRequest, error); } - std::unique_lock lock(rw_mutex_); - if (has_categories) category_keywords_list_ = std::move(new_categories); - if (has_rules) rules_list_ = std::move(new_rules); - return NodeControlResult::Handled(); + return snapshot_.Update( + [&](const RuleMatchState& current) -> NodeResult { + RuleMatchState next = current; + if (has_categories) + next.category_keywords_list = std::move(new_categories); + if (has_rules) next.rules_list = std::move(new_rules); + return NodeResult::Success(std::move(next)); + }); } static bool ValidateConfig(const nlohmann::json& config, @@ -168,10 +183,13 @@ class TextRuleMatchNode final : public NodeBase { BindPort(init_ctx, in_text_); BindPort(init_ctx, out_matches_); - default_category_ = normalized["default_category"].get(); - default_score_ = normalized["default_score"].get(); - category_keywords_list_ = std::move(categories); - rules_list_ = std::move(rules); + RuleMatchState initial_state; + initial_state.default_category = + normalized["default_category"].get(); + initial_state.default_score = normalized["default_score"].get(); + initial_state.category_keywords_list = std::move(categories); + initial_state.rules_list = std::move(rules); + snapshot_.Initialize(std::move(initial_state)); return true; } @@ -183,7 +201,12 @@ class TextRuleMatchNode final : public NodeBase { return node_error::text_rule_match::kMissingInput; } - std::shared_lock lock(rw_mutex_); + auto state_guard = snapshot_.Read(); + if (!state_guard) { + return Fail(req_ctx, node_error::author_node::kInternalError, + "Snapshot uninitialized"); + } + const auto& state = *state_guard; RuleMatchBatch output_matches; output_matches.reserve(text_items->size()); @@ -204,7 +227,7 @@ class TextRuleMatchNode final : public NodeBase { nlohmann::json slots_obj = nlohmann::json::object(); // 1. 匹配 categories (词表模式) - for (const auto& [category, words] : category_keywords_list_) { + for (const auto& [category, words] : state.category_keywords_list) { for (const auto& w : words) { if (w.empty()) continue; if (sentence.find(w) != std::string::npos) { @@ -223,14 +246,16 @@ class TextRuleMatchNode final : public NodeBase { } // 2. 匹配 rules (结构化规则模式,支持 regex, exact, contains) - for (const auto& rule : rules_list_) { + for (const auto& rule : state.rules_list) { bool rule_matched = false; std::unordered_map rule_captures; if (rule.strategy == "regex") { std::string diagnostic; const TextRegexSearchStatus status = - rule.compiled_regex.Search(sentence, &rule_captures, &diagnostic); + rule.compiled_regex ? rule.compiled_regex->Search( + sentence, &rule_captures, &diagnostic) + : TextRegexSearchStatus::kNotMatched; if (status == TextRegexSearchStatus::kError) { ALG_LOG_ERROR( "[TextRuleMatchNode] Regex execution failed for rule '%s': " @@ -280,10 +305,10 @@ class TextRuleMatchNode final : public NodeBase { } } - if (!is_hit && !default_category_.empty()) { + if (!is_hit && !state.default_category.empty()) { is_hit = 1; - first_hit_category = default_category_; - first_hit_score = default_score_; + first_hit_category = state.default_category; + first_hit_score = state.default_score; first_hit_word = ""; slots_obj["raw_query"] = sentence; } @@ -312,9 +337,6 @@ class TextRuleMatchNode final : public NodeBase { } private: - using CategoryList = - std::vector>>; - static bool BuildCategories(const nlohmann::json& categories_json, CategoryList* out_categories, std::string* diagnostic) { @@ -372,7 +394,8 @@ class TextRuleMatchNode final : public NodeBase { } if (spec.strategy == "regex" && !spec.pattern.empty()) { - if (!spec.compiled_regex.Compile(spec.pattern, &detail)) { + auto compiled = std::make_shared(); + if (!compiled->Compile(spec.pattern, &detail)) { if (diagnostic) { *diagnostic = "rules[" + std::to_string(index) + "].pattern" + (spec.id.empty() ? "" : " (id='" + spec.id + "')") + @@ -380,6 +403,7 @@ class TextRuleMatchNode final : public NodeBase { } return false; } + spec.compiled_regex = std::move(compiled); } temp_rules.push_back(std::move(spec)); } @@ -410,13 +434,7 @@ class TextRuleMatchNode final : public NodeBase { BuildRules((*normalized)["rules"], rules, diagnostic)); } - mutable std::shared_mutex rw_mutex_; - std::vector>> - category_keywords_list_; - std::vector rules_list_; - std::string default_category_; - float default_score_ = kDefaultScore; - + ConfigurationSnapshot snapshot_; BoundInput in_text_; BoundOutput out_matches_; }; diff --git a/src/common_nodes/text_template_node.cpp b/src/common_nodes/text_template_node.cpp index 93ff6141..45a47e61 100644 --- a/src/common_nodes/text_template_node.cpp +++ b/src/common_nodes/text_template_node.cpp @@ -15,8 +15,8 @@ #include "contracts/control_payload.h" #include "core/common_contracts.h" #include "core/node_registry.h" -#include "edgeflow/log.h" #include "engine/text/utf8.h" +#include "nodes/configuration_snapshot.h" #include "nodes/node_base.h" #include "nodes/node_error_codes.h" #include "nodes/text_template.h" @@ -147,6 +147,95 @@ const nlohmann::json& TemplateControlSchema() { {{"type", "string"}, {"enum", {"fail", "empty", "preserve"}}}}}}}; return schema; } +struct TemplateState { + std::string template_str = kDefaultTemplate; + std::string separator = kDefaultSeparator; + size_t max_length = kDefaultMaxLength; + std::string overflow_policy = "fail"; + std::string missing_variable_policy = kDefaultMissingPolicy; + std::string prompt_id; + bool allow_dynamic_attrs = false; + std::unordered_map static_values; + std::vector compiled_tokens; +}; + +struct TemplateUpdate { + std::optional template_str; + std::optional prompt_id; + std::optional> values; + std::optional allow_dynamic_attributes; + std::optional missing_variable_policy; +}; + +inline bool CompileTemplate( + const std::string& tmpl, + const std::unordered_map& static_vals, + bool allow_dynamic_attrs, std::vector* out_tokens, + std::string* diagnostic = nullptr) { + std::string error; + bool ok = ParseTextTemplate(tmpl, out_tokens, &error); + if (ok) { + for (const auto& token : *out_tokens) { + if (token.type == TextTemplateTokenType::kVariable && + !allow_dynamic_attrs && !BuiltinInputs().count(token.value) && + !static_vals.count(token.value)) { + error = "Unknown template placeholder: " + token.value + + "; connect attributes or declare values"; + ok = false; + break; + } + } + } + if (!ok) { + out_tokens->clear(); + ALG_LOG_ERROR("[TextTemplateNode] %s\n", error.c_str()); + if (diagnostic) *diagnostic = std::move(error); + } + return ok; +} + +inline NodeResult BuildNextTemplate( + const TemplateState& current, const TemplateUpdate& update, + const BindingFacts& bindings) { + TemplateState next = current; + if (update.template_str) next.template_str = *update.template_str; + if (update.prompt_id) next.prompt_id = *update.prompt_id; + if (update.missing_variable_policy) { + next.missing_variable_policy = *update.missing_variable_policy; + } + if (update.allow_dynamic_attributes) { + next.allow_dynamic_attrs = + bindings.IsConnected("attributes") || *update.allow_dynamic_attributes; + } + if (update.values) { + for (const auto& [k, v] : *update.values) { + next.static_values[k] = v; + } + } + std::vector new_tokens; + std::string diagnostic; + if (!CompileTemplate(next.template_str, next.static_values, + next.allow_dynamic_attrs, &new_tokens, &diagnostic)) { + return NodeResult::Failure( + NodeErrorKind::kBusinessError, + diagnostic.empty() + ? "Invalid template placeholders or syntax in Control" + : diagnostic, + node_error::control::kInvalidRequest); + } + if (bindings.has_plan && + !ValidateTemplateInputs(new_tokens, bindings.connected_inputs, + next.missing_variable_policy, &diagnostic)) { + return NodeResult::Failure( + NodeErrorKind::kBusinessError, + diagnostic.empty() + ? "Invalid template placeholders or syntax in Control" + : diagnostic, + node_error::control::kInvalidRequest); + } + next.compiled_tokens = std::move(new_tokens); + return NodeResult::Success(std::move(next)); +} } // namespace /** @@ -205,58 +294,61 @@ class TextTemplateNode final : public NodeBase { return false; } - std::unique_lock lock(rw_mutex_); - template_str_ = normalized_config.value("template", kDefaultTemplate); - separator_ = normalized_config.value("separator", kDefaultSeparator); + TemplateState initial_state; + initial_state.template_str = + normalized_config.value("template", kDefaultTemplate); + initial_state.separator = + normalized_config.value("separator", kDefaultSeparator); const int64_t configured_max_length = normalized_config.value("max_length", kDefaultMaxLength); if (configured_max_length < 1 || configured_max_length > 1048576) { return false; } - max_length_ = static_cast(configured_max_length); - overflow_policy_ = normalized_config.value("overflow_policy", "fail"); - if (overflow_policy_ != "fail" && overflow_policy_ != "truncate") { + initial_state.max_length = static_cast(configured_max_length); + initial_state.overflow_policy = + normalized_config.value("overflow_policy", "fail"); + if (initial_state.overflow_policy != "fail" && + initial_state.overflow_policy != "truncate") { return false; } - static_values_.clear(); if (normalized_config.contains("values")) { if (!normalized_config["values"].is_object()) return false; for (auto it = normalized_config["values"].begin(); it != normalized_config["values"].end(); ++it) { if (!it.value().is_string()) return false; - static_values_[it.key()] = it.value().get(); + initial_state.static_values[it.key()] = it.value().get(); } } - missing_variable_policy_ = normalized_config.value( + initial_state.missing_variable_policy = normalized_config.value( "missing_variable_policy", kDefaultMissingPolicy); - if (missing_variable_policy_ != "fail" && - missing_variable_policy_ != "empty" && - missing_variable_policy_ != "preserve") { + if (initial_state.missing_variable_policy != "fail" && + initial_state.missing_variable_policy != "empty" && + initial_state.missing_variable_policy != "preserve") { return false; } - allow_dynamic_attrs_ = - in_attributes_.IsBound() || + binding_facts_ = MakeBindingFacts(init_ctx); + if (in_attributes_.IsBound()) { + binding_facts_.connected_inputs.insert("attributes"); + } + initial_state.allow_dynamic_attrs = + binding_facts_.IsConnected("attributes") || normalized_config.value("allow_dynamic_attributes", false); std::vector compiled; - if (!CompileTemplate(template_str_, static_values_, allow_dynamic_attrs_, - &compiled)) { + if (!CompileTemplate(initial_state.template_str, + initial_state.static_values, + initial_state.allow_dynamic_attrs, &compiled)) { return false; } - connected_inputs_.reset(); - if (init_ctx.plan) { - connected_inputs_.emplace(); - for (const auto& port : init_ctx.plan->ports) { - if (port.direction == PortDirection::kInput) - connected_inputs_->insert(port.logical_name); - } - if (!ValidateTemplateInputs(compiled, *connected_inputs_, - missing_variable_policy_)) + if (binding_facts_.has_plan) { + if (!ValidateTemplateInputs(compiled, binding_facts_.connected_inputs, + initial_state.missing_variable_policy)) return false; } - compiled_tokens_ = std::move(compiled); + initial_state.compiled_tokens = std::move(compiled); + snapshot_.Initialize(std::move(initial_state)); return true; } @@ -270,57 +362,38 @@ class TextTemplateNode final : public NodeBase { return NodeControlResult::Failed(node_error::control::kInvalidRequest, error); } - std::string new_tmpl; - std::unordered_map new_values; - bool new_allow_dynamic; - std::string new_missing_policy; - std::string new_prompt_id; - { - std::shared_lock lock(rw_mutex_); - new_tmpl = template_str_; - new_values = static_values_; - new_allow_dynamic = allow_dynamic_attrs_; - new_missing_policy = missing_variable_policy_; - new_prompt_id = prompt_id_; - } + TemplateUpdate update; if (root.contains("template")) - new_tmpl = root["template"].get(); + update.template_str = root["template"].get(); if (root.contains("allow_dynamic_attributes")) { - new_allow_dynamic = in_attributes_.IsBound() || - root["allow_dynamic_attributes"].get(); + update.allow_dynamic_attributes = + root["allow_dynamic_attributes"].get(); } if (root.contains("missing_variable_policy")) { - new_missing_policy = root["missing_variable_policy"].get(); + update.missing_variable_policy = + root["missing_variable_policy"].get(); } if (root.contains("prompt_id")) - new_prompt_id = root["prompt_id"].get(); + update.prompt_id = root["prompt_id"].get(); if (root.contains("values")) { + std::unordered_map vals; for (auto it = root["values"].begin(); it != root["values"].end(); ++it) { - new_values[it.key()] = it.value().get(); + vals[it.key()] = it.value().get(); } + update.values = std::move(vals); } - std::vector new_tokens; - if (!CompileTemplate(new_tmpl, new_values, new_allow_dynamic, - &new_tokens) || - (connected_inputs_ && - !ValidateTemplateInputs(new_tokens, *connected_inputs_, - new_missing_policy))) { - return NodeControlResult::Failed( - node_error::control::kInvalidRequest, - "Invalid template placeholders or syntax in Control"); - } - std::unique_lock lock(rw_mutex_); - template_str_ = std::move(new_tmpl); - static_values_ = std::move(new_values); - allow_dynamic_attrs_ = new_allow_dynamic; - missing_variable_policy_ = std::move(new_missing_policy); - prompt_id_ = std::move(new_prompt_id); - compiled_tokens_ = std::move(new_tokens); - return NodeControlResult::Handled(); + return snapshot_.Update([&](const TemplateState& current) { + return BuildNextTemplate(current, update, binding_facts_); + }); } int ProcessNode(AlgContext& req_ctx) override { - std::shared_lock lock(rw_mutex_); + auto state_guard = snapshot_.Read(); + if (!state_guard) { + return Fail(req_ctx, node_error::author_node::kInternalError, + "Snapshot uninitialized"); + } + const auto& state = *state_guard; const auto* primary_items = in_primary_.Get(req_ctx); const auto* context_items = in_context_.Get(req_ctx); @@ -438,7 +511,7 @@ class TextTemplateNode final : public NodeBase { auto c_it = context_by_req.find(req_id); if (c_it != context_by_req.end()) { for (size_t i = 0; i < c_it->second.size(); ++i) { - if (i > 0) context_str += separator_; + if (i > 0) context_str += state.separator; context_str += c_it->second[i]; } } @@ -456,7 +529,7 @@ class TextTemplateNode final : public NodeBase { auto d_it = document_by_req.find(req_id); if (d_it != document_by_req.end()) { for (size_t i = 0; i < d_it->second.size(); ++i) { - if (i > 0) doc_str += separator_; + if (i > 0) doc_str += state.separator; doc_str += d_it->second[i]; } } @@ -470,7 +543,7 @@ class TextTemplateNode final : public NodeBase { std::string rendered; rendered.reserve(256); - for (const auto& token : compiled_tokens_) { + for (const auto& token : state.compiled_tokens) { if (token.type == TokenType::kLiteral) { rendered += token.value; } else { @@ -488,28 +561,29 @@ class TextTemplateNode final : public NodeBase { if (document_items || document_text_items) value = &doc_str; } else if (attrs_ptr && attrs_ptr->find(var) != attrs_ptr->end()) { value = &attrs_ptr->at(var); - } else if (static_values_.find(var) != static_values_.end()) { - value = &static_values_.at(var); + } else if (state.static_values.find(var) != + state.static_values.end()) { + value = &state.static_values.at(var); } if (value) { rendered += *value; } else { - if (missing_variable_policy_ == "fail") { + if (state.missing_variable_policy == "fail") { return Fail(req_ctx, node_error::text_template::kMissingVariable, "Missing required template variable: " + var); - } else if (missing_variable_policy_ == "preserve") { + } else if (state.missing_variable_policy == "preserve") { rendered += "{" + var + "}"; } } } } - if (rendered.size() > max_length_) { - if (overflow_policy_ == "fail") { + if (rendered.size() > state.max_length) { + if (state.overflow_policy == "fail") { return Fail(req_ctx, node_error::text_template::kRenderedOutputTooLong, "Rendered prompt exceeds max_length of " + - std::to_string(max_length_)); + std::to_string(state.max_length)); } std::vector boundaries; size_t invalid_offset = 0; @@ -519,8 +593,8 @@ class TextTemplateNode final : public NodeBase { "Rendered prompt contains invalid UTF-8 at byte offset " + std::to_string(invalid_offset)); } - const auto boundary = - std::upper_bound(boundaries.begin(), boundaries.end(), max_length_); + const auto boundary = std::upper_bound( + boundaries.begin(), boundaries.end(), state.max_length); rendered.resize(*(boundary - 1)); } @@ -532,44 +606,8 @@ class TextTemplateNode final : public NodeBase { } private: - static bool CompileTemplate( - const std::string& tmpl, - const std::unordered_map& static_vals, - bool allow_dynamic_attrs, std::vector* out_tokens, - std::string* diagnostic = nullptr) { - std::string error; - bool ok = ParseTextTemplate(tmpl, out_tokens, &error); - if (ok) { - for (const auto& token : *out_tokens) { - if (token.type == TokenType::kVariable && !allow_dynamic_attrs && - !BuiltinInputs().count(token.value) && - !static_vals.count(token.value)) { - error = "Unknown template placeholder: " + token.value + - "; connect attributes or declare values"; - ok = false; - break; - } - } - } - if (!ok) { - out_tokens->clear(); - ALG_LOG_ERROR("[TextTemplateNode] %s\n", error.c_str()); - if (diagnostic) *diagnostic = std::move(error); - } - return ok; - } - - mutable std::shared_mutex rw_mutex_; - std::string template_str_ = kDefaultTemplate; - std::string separator_ = kDefaultSeparator; - size_t max_length_ = kDefaultMaxLength; - std::string overflow_policy_ = "fail"; - std::string missing_variable_policy_ = kDefaultMissingPolicy; - std::optional> connected_inputs_; - std::string prompt_id_; - bool allow_dynamic_attrs_ = false; - std::unordered_map static_values_; - std::vector compiled_tokens_; + ConfigurationSnapshot snapshot_; + BindingFacts binding_facts_; BoundInput in_primary_; BoundInput in_context_; diff --git a/tests/support/node_harness.h b/tests/support/node_harness.h index cf127858..36b9e160 100644 --- a/tests/support/node_harness.h +++ b/tests/support/node_harness.h @@ -110,9 +110,19 @@ class NodeHarness { NodeHarness& Config(nlohmann::json config) { config_ = std::move(config); + Reset(); return *this; } + void Reset() { + node_.reset(); + session_ctx_.reset(); + input_keys_.clear(); + output_keys_.clear(); + init_diagnostic_.clear(); + initialized_ = false; + } + NodeHarness& TextInput(const std::string& logical_port_name, const std::vector& payloads, uint64_t start_req_id = 101) { @@ -151,49 +161,53 @@ class NodeHarness { NodeHarness& BindModel(std::string model_id, std::shared_ptr model) { models_[std::move(model_id)] = std::move(model); + Reset(); return *this; } NodeHarness& DisablePlan() { use_plan_ = false; + Reset(); return *this; } NodeHarness& OmitPortFromPlan(std::string logical_port_name) { omitted_ports_.insert(std::move(logical_port_name)); + Reset(); return *this; } - NodeHarnessResult Run() { + bool EnsureInitialized() { + if (initialized_) return true; + if (!NodeRegistry::Instance().Has(node_type_)) { - return NodeHarnessResult::InitFailed("Node type '" + node_type_ + - "' is not registered"); + init_diagnostic_ = "Node type '" + node_type_ + "' is not registered"; + return false; } - auto node = NodeRegistry::Instance().Create(node_type_); - if (!node) { - return NodeHarnessResult::InitFailed("Failed to create node '" + - node_type_ + "'"); + node_ = NodeRegistry::Instance().Create(node_type_); + if (!node_) { + init_diagnostic_ = "Failed to create node '" + node_type_ + "'"; + return false; } - SessionContext session_ctx; + session_ctx_ = std::make_unique(); for (const auto& [mid, model] : models_) { - session_ctx.GetModelManager().RegisterModel( + session_ctx_->GetModelManager().RegisterModel( mid, model, "harness_rev", model ? model->ModelType() : "mock", model ? model->Capability() : "llm", "mock"); } const auto definition = PipelineCatalog::FindNode(node_type_); - std::unordered_map input_keys; - std::unordered_map output_keys; + input_keys_.clear(); + output_keys_.clear(); NodeInitContext init_ctx; init_ctx.config = &config_; - init_ctx.session_ctx = &session_ctx; - std::string diagnostic; - init_ctx.diagnostic = &diagnostic; + init_ctx.session_ctx = session_ctx_.get(); + init_diagnostic_.clear(); + init_ctx.diagnostic = &init_diagnostic_; - ValidatedNodePlan plan; if (use_plan_) { if (definition) { nlohmann::json doc = nlohmann::json::object(); @@ -201,8 +215,6 @@ class NodeHarness { doc["biz_name"] = biz; doc["models"] = nlohmann::json::array(); - // Resolve defaults before synthesizing model declarations. Keep the - // original config in the document so Validator owns full diagnostics. nlohmann::json normalized_config; if (!ValidateAndNormalizeFields(definition->config_fields, config_, &normalized_config, nullptr)) { @@ -215,8 +227,6 @@ class NodeHarness { std::string mid = normalized_config[dep.config_field].get(); if (!declared_models.insert(mid).second) continue; - // Mock plans use a stable test-only schema. Selecting an arbitrary - // production model can introduce unrelated required model fields. const std::string mtype = "harness_dummy_m_" + dep.capability; const std::string mbackend = "harness_dummy_b_" + dep.capability; if (!ModelRegistry::Instance().Has(mtype)) { @@ -280,58 +290,87 @@ class NodeHarness { for (const auto& d : plan_res.report.diagnostics) { err_msg += " [" + d.path + "] " + d.message; } - return NodeHarnessResult::InitFailed(err_msg); + init_diagnostic_ = err_msg; + return false; } auto it = plan_res.node_plans.find("harness_node"); if (it != plan_res.node_plans.end()) { - plan = std::move(it->second); - for (const auto& p : plan.ports) { + plan_ = std::move(it->second); + for (const auto& p : plan_.ports) { if (p.direction == PortDirection::kInput) { - input_keys[p.logical_name] = p.blackboard_key; + input_keys_[p.logical_name] = p.blackboard_key; } else if (p.direction == PortDirection::kOutput) { - output_keys[p.logical_name] = p.blackboard_key; + output_keys_[p.logical_name] = p.blackboard_key; } } } } - init_ctx.plan = &plan; + init_ctx.plan = &plan_; } else { if (definition) { for (const auto& in_def : definition->inputs) { - input_keys[in_def.logical_name] = in_def.logical_name; + input_keys_[in_def.logical_name] = in_def.logical_name; } for (const auto& out_def : definition->outputs) { - output_keys[out_def.logical_name] = out_def.logical_name; + output_keys_[out_def.logical_name] = out_def.logical_name; } } } - if (!node->Init(init_ctx)) { + if (!node_->Init(init_ctx)) { + if (init_diagnostic_.empty()) init_diagnostic_ = "Node Init failed"; + return false; + } + + initialized_ = true; + return true; + } + + NodeControlResult Control(int cmd, const std::string& json_param) { + if (!EnsureInitialized()) { + return NodeControlResult::Failed( + node_error::control::kInvalidRequest, + init_diagnostic_.empty() ? "Node Init failed" : init_diagnostic_); + } + return node_->Control(cmd, json_param); + } + + NodeControlResult Control(int cmd, const char* json_param) { + return Control(cmd, std::string(json_param ? json_param : "")); + } + + NodeControlResult Control(int cmd, const nlohmann::json& payload) { + return Control(cmd, payload.dump()); + } + + INode* GetNode() const noexcept { return node_.get(); } + + NodeHarnessResult Run() { + if (!EnsureInitialized()) { return NodeHarnessResult::InitFailed( - diagnostic.empty() ? "Node Init failed" : diagnostic); + init_diagnostic_.empty() ? "Node Init failed" : init_diagnostic_); } auto req_ctx = std::make_unique(); for (auto& [logical_name, publisher] : custom_inputs_) { - std::string key = input_keys.count(logical_name) - ? input_keys[logical_name] + std::string key = input_keys_.count(logical_name) + ? input_keys_[logical_name] : logical_name; publisher(*req_ctx, key); } - int ret = node->Process(req_ctx.get()); + int ret = node_->Process(req_ctx.get()); if (ret != 0) { std::string msg = "Process returned " + std::to_string(ret); if (!req_ctx->IsOk()) { msg += ": " + req_ctx->GetErrorMessage(); } return NodeHarnessResult::ProcessFailed(ret, msg, std::move(req_ctx), - std::move(output_keys)); + output_keys_); } - return NodeHarnessResult::Success(std::move(req_ctx), - std::move(output_keys)); + return NodeHarnessResult::Success(std::move(req_ctx), output_keys_); } private: @@ -343,6 +382,14 @@ class NodeHarness { custom_inputs_; std::unordered_map> models_; std::unordered_set omitted_ports_; + + std::unique_ptr node_; + std::unique_ptr session_ctx_; + ValidatedNodePlan plan_; + std::unordered_map input_keys_; + std::unordered_map output_keys_; + std::string init_diagnostic_; + bool initialized_ = false; }; } // namespace llm_edgeflow diff --git a/tests/support/node_process_pause.h b/tests/support/node_process_pause.h new file mode 100644 index 00000000..bbee7962 --- /dev/null +++ b/tests/support/node_process_pause.h @@ -0,0 +1,51 @@ +#pragma once + +#include +#include +#include +#include + +namespace llm_edgeflow::test_support { + +// A bounded handshake: timeout must fail the caller, never silently allow a +// Process to finish before the publication under test. Always resume and join +// the reader before fatal GoogleTest assertions. +class NodeProcessPause { + public: + explicit NodeProcessPause( + std::chrono::milliseconds timeout = std::chrono::seconds(5)) + : timeout_(timeout) {} + + static void OnAllocation(void* user_data) { + static_cast(user_data)->Pause(); + } + + void Pause() { + std::unique_lock lock(mutex_); + paused_ = true; + condition_.notify_all(); + if (!condition_.wait_for(lock, timeout_, [this] { return resumed_; })) { + throw std::runtime_error("Process publication handshake timed out"); + } + } + + bool WaitUntilPaused() { + std::unique_lock lock(mutex_); + return condition_.wait_for(lock, timeout_, [this] { return paused_; }); + } + + void Resume() { + std::lock_guard lock(mutex_); + resumed_ = true; + condition_.notify_all(); + } + + private: + const std::chrono::milliseconds timeout_; + std::mutex mutex_; + std::condition_variable condition_; + bool paused_ = false; + bool resumed_ = false; +}; + +} // namespace llm_edgeflow::test_support diff --git a/tests/support/scoped_allocation_failure.cpp b/tests/support/scoped_allocation_failure.cpp index e6fae359..77c42add 100644 --- a/tests/support/scoped_allocation_failure.cpp +++ b/tests/support/scoped_allocation_failure.cpp @@ -6,6 +6,26 @@ namespace llm_edgeflow::test_support { +thread_local ScopedNextAllocationCallback* + ScopedNextAllocationCallback::current_ = nullptr; + +ScopedNextAllocationCallback::ScopedNextAllocationCallback( + Callback callback, void* user_data) noexcept + : previous_(current_), callback_(callback), user_data_(user_data) { + current_ = this; +} + +ScopedNextAllocationCallback::~ScopedNextAllocationCallback() { + current_ = previous_; +} + +void ScopedNextAllocationCallback::BeforeAllocation() { + if (!current_ || !current_->callback_) return; + auto callback = current_->callback_; + current_->callback_ = nullptr; + callback(current_->user_data_); +} + thread_local ScopedAllocationFailure* ScopedAllocationFailure::current_ = nullptr; @@ -53,6 +73,7 @@ namespace { using llm_edgeflow::test_support::ScopedAllocationFailure; void* Allocate(size_t size, size_t alignment = 0) { + llm_edgeflow::test_support::ScopedNextAllocationCallback::BeforeAllocation(); ScopedAllocationFailure::BeforeAllocation(); if (size == 0) size = 1; if (alignment) { diff --git a/tests/support/scoped_allocation_failure.h b/tests/support/scoped_allocation_failure.h index 33b968b9..7d20c4c9 100644 --- a/tests/support/scoped_allocation_failure.h +++ b/tests/support/scoped_allocation_failure.h @@ -5,6 +5,25 @@ namespace llm_edgeflow::test_support { +// One-shot callback on this thread's next replacement-new allocation. Clear +// before invoking so the callback may itself allocate without recursion. +class ScopedNextAllocationCallback { + public: + using Callback = void (*)(void*); + ScopedNextAllocationCallback(Callback callback, void* user_data) noexcept; + ~ScopedNextAllocationCallback(); + ScopedNextAllocationCallback(const ScopedNextAllocationCallback&) = delete; + ScopedNextAllocationCallback& operator=(const ScopedNextAllocationCallback&) = + delete; + static void BeforeAllocation(); + + private: + static thread_local ScopedNextAllocationCallback* current_; + ScopedNextAllocationCallback* previous_; + Callback callback_; + void* user_data_; +}; + // Test-executable-only replacement new/delete support. Arm only around a // synchronous operation, outside GoogleTest assertions. For leak assertions, // destroy tracked allocations on this thread before checking Outstanding(). diff --git a/tests/unit/nodes/test_function_node.cpp b/tests/unit/nodes/test_function_node.cpp index 2316e7f3..474866c0 100644 --- a/tests/unit/nodes/test_function_node.cpp +++ b/tests/unit/nodes/test_function_node.cpp @@ -1,8 +1,14 @@ #include +#include #include +#include +#include +#include #include +#include #include +#include #include #include "core/alg_context.h" @@ -13,7 +19,9 @@ #include "nodes/node_config_parser.h" #include "nodes/node_error_codes.h" #include "tests/support/node_harness.h" +#include "tests/support/node_process_pause.h" #include "tests/support/node_test_utils.h" +#include "tests/support/scoped_allocation_failure.h" namespace llm_edgeflow { namespace { @@ -547,6 +555,245 @@ inline auto BindingMapSpec() { } REGISTER_FUNCTION_NODE(BindingMapNode, BindingMapSpec()); +struct ControlledMapParams { + std::string prefix; + std::string suffix; + int multiplier = 1; +}; + +inline constexpr int kCmdReplaceMap = 3001; +inline constexpr int kCmdPatchMap = 3002; + +inline std::string ControlledMapFn(const std::string& in, + const ControlledMapParams& p) { + std::string res = p.prefix; + for (int i = 0; i < p.multiplier; ++i) { + res += in; + } + return res + p.suffix; +} + +inline auto ControlledMapSpec() { + return MakeMapSpec( + Input("input"), Output("output"), + Parameters( + { + Field("prefix", &ControlledMapParams::prefix).Default(""), + Field("suffix", &ControlledMapParams::suffix).Default(""), + Field("multiplier", &ControlledMapParams::multiplier) + .Default(1) + .Minimum(1) + .Maximum(10), + }) + .Validate([](const ControlledMapParams& p, std::string* err) { + if (p.prefix == "INVALID") { + if (err) *err = "Invalid prefix disallowed"; + return false; + } + return true; + }), + &ControlledMapFn) + .WithControls({ + ReplaceFields(kCmdReplaceMap, "replace_map", + {"prefix", "multiplier"}), + PatchFields(kCmdPatchMap, "patch_map", + {"prefix", "suffix", "multiplier"}), + }); +} +REGISTER_FUNCTION_NODE(ControlledMapNode, ControlledMapSpec()); + +// --------------------------------------------------------------------------- +// Non-Copyable Params (holding unique_ptr) without Controls (Problem 1) +// --------------------------------------------------------------------------- +struct NonCopyableMapParams { + std::string prefix; + std::unique_ptr extra_counter; +}; + +inline auto NonCopyableMapSpec() { + return MakeMapSpec( + Input("input"), Output("output"), + Parameters( + { + Field("prefix", &NonCopyableMapParams::prefix).Default("nc:"), + }) + .Prepare( + [](NonCopyableMapParams* p, const BindingFacts&, std::string*) { + p->extra_counter = std::make_unique(100); + return true; + }), + [](const std::string& in, const NonCopyableMapParams& p) { + return p.prefix + in + "_" + + (p.extra_counter ? std::to_string(*p.extra_counter) : "null"); + }); +} +REGISTER_FUNCTION_NODE(NonCopyableMapNode, NonCopyableMapSpec()); + +struct NonCopyableBatchInputs { + const TextBatch* texts = nullptr; +}; + +struct NonCopyableBatchParams { + std::string tag; + std::unique_ptr extra_val; +}; + +inline auto NonCopyableBatchSpec() { + return MakeBatchSpec( + InputsOf({ + Required("texts", &NonCopyableBatchInputs::texts), + }), + PreservedOutput("output", "texts"), + Parameters( + { + Field("tag", &NonCopyableBatchParams::tag).Default("batch_nc:"), + }) + .Prepare( + [](NonCopyableBatchParams* p, const BindingFacts&, std::string*) { + p->extra_val = std::make_unique(200); + return true; + }), + [](const NonCopyableBatchInputs& in, + const NonCopyableBatchParams& p) -> NodeResult { + TextBatch out; + if (!in.texts) return out; + for (const auto& item : *in.texts) { + out.emplace_back( + item.req_id, item.sub_id, + p.tag + item.data + "_" + + (p.extra_val ? std::to_string(*p.extra_val) : "null")); + } + return out; + }); +} +REGISTER_FUNCTION_NODE(NonCopyableBatchNode, NonCopyableBatchSpec()); + +// --------------------------------------------------------------------------- +// Strict Unplanned Fact Checking Node (Problem 2) +// --------------------------------------------------------------------------- +struct StrictUnplannedMapParams { + std::string name; +}; + +inline auto StrictUnplannedMapSpec() { + return MakeMapSpec( + Input("input"), Output("output"), + Parameters( + { + Field("name", &StrictUnplannedMapParams::name) + .Default("unplanned"), + }) + .Prepare([](StrictUnplannedMapParams*, const BindingFacts& facts, + std::string* err) { + if (facts.has_plan) { + if (err) + *err = "StrictUnplannedMapNode expects has_plan == false"; + return false; + } + return true; + }), + [](const std::string& in, const StrictUnplannedMapParams& p) { + return p.name + ":" + in; + }); +} +REGISTER_FUNCTION_NODE(StrictUnplannedMapNode, StrictUnplannedMapSpec()); + +struct StrictUnplannedBatchInputs { + const TextBatch* texts = nullptr; +}; + +struct StrictUnplannedBatchParams { + std::string name; +}; + +inline auto StrictUnplannedBatchSpec() { + return MakeBatchSpec( + InputsOf({ + Required("texts", &StrictUnplannedBatchInputs::texts), + }), + PreservedOutput("output", "texts"), + Parameters( + { + Field("name", &StrictUnplannedBatchParams::name) + .Default("unplanned_batch"), + }) + .Prepare([](StrictUnplannedBatchParams*, const BindingFacts& facts, + std::string* err) { + if (facts.has_plan) { + if (err) + *err = "StrictUnplannedBatchNode expects has_plan == false"; + return false; + } + return true; + }), + [](const StrictUnplannedBatchInputs& in, + const StrictUnplannedBatchParams& p) -> NodeResult { + TextBatch out; + if (!in.texts) return out; + for (const auto& item : *in.texts) { + out.emplace_back(item.req_id, item.sub_id, p.name + ":" + item.data); + } + return out; + }); +} +REGISTER_FUNCTION_NODE(StrictUnplannedBatchNode, StrictUnplannedBatchSpec()); + +struct ControlledBatchInputs { + const TextBatch* texts = nullptr; +}; + +struct ControlledBatchParams { + std::string header; + bool uppercase = false; +}; + +inline constexpr int kCmdReplaceBatch = 3003; + +inline NodeResult ControlledBatchFn( + const ControlledBatchInputs& inputs, const ControlledBatchParams& params) { + TextBatch out; + if (!inputs.texts) return out; + out.reserve(inputs.texts->size()); + for (const auto& item : *inputs.texts) { + std::string text = params.header + item.data; + if (params.uppercase) { + for (char& c : text) + c = static_cast(std::toupper(static_cast(c))); + } + out.emplace_back(item.req_id, item.sub_id, std::move(text)); + } + return out; +} + +inline auto ControlledBatchSpec() { + return MakeBatchSpec( + InputsOf({ + Required("texts", &ControlledBatchInputs::texts), + }), + PreservedOutput("output", "texts"), + Parameters( + { + Field("header", &ControlledBatchParams::header) + .Default(""), + Field("uppercase", &ControlledBatchParams::uppercase) + .Default(false), + }) + .Validate( + [](const ControlledBatchParams& p, std::string* err) { + if (p.header == "REJECT") { + if (err) *err = "Rejected header"; + return false; + } + return true; + }), + &ControlledBatchFn) + .WithControls({ + ReplaceFields(kCmdReplaceBatch, "set_batch_params", + {"header", "uppercase"}), + }); +} +REGISTER_FUNCTION_NODE(ControlledBatchNode, ControlledBatchSpec()); + } // namespace // =========================================================================== @@ -1127,4 +1374,811 @@ TEST(FunctionNodeTest, LogicObjectIsRecreatedForEachProcessOnSameNode) { EXPECT_EQ(local_logic_constructions, before + 3); } +// --------------------------------------------------------------------------- +// RFC-0054 Tests: ConfigurationSnapshot & Direct Concurrency (Section 7.1) +// --------------------------------------------------------------------------- + +TEST(ConfigurationSnapshotTest, UninitializedAndNullStateHandled) { + ConfigurationSnapshot snapshot; + EXPECT_FALSE(snapshot.IsInitialized()); + EXPECT_EQ(snapshot.Read(), nullptr); + + auto res = snapshot.Update([](const std::string& s) { + return NodeResult::Success(s + "_next"); + }); + EXPECT_EQ(res.status, NodeControlStatus::kFailed); + EXPECT_EQ(res.code, node_error::control::kInvalidRequest); + + EXPECT_FALSE( + snapshot.Initialize(std::shared_ptr(nullptr))); + EXPECT_FALSE(snapshot.IsInitialized()); +} + +TEST(ConfigurationSnapshotTest, InitializeAndRead) { + ConfigurationSnapshot snapshot; + EXPECT_TRUE(snapshot.Initialize("initial_val")); + EXPECT_TRUE(snapshot.IsInitialized()); + auto read_val = snapshot.Read(); + ASSERT_NE(read_val, nullptr); + EXPECT_EQ(*read_val, "initial_val"); +} + +TEST(ConfigurationSnapshotTest, WriterSerializationAndIndependentPatchMerging) { + // Verifies RFC-0054 Section 7.1: + // writer A updates prefix, writer B updates suffix. + // Both successful updates are preserved; B cannot submit based on stale pre-A + // state. + struct TwoFields { + std::string prefix; + std::string suffix; + }; + ConfigurationSnapshot snapshot( + TwoFields{"init_pre:", ":init_suf"}); + + std::promise a_entered_lock; + std::promise release_a; + std::promise b_called; + + std::atomic b_saw_a_prefix{false}; + + // Writer A holds the writer lock while B attempts to run + std::thread thread_a([&]() { + snapshot.Update([&](const TwoFields& cur) { + a_entered_lock.set_value(); + release_a.get_future().wait(); + TwoFields next = cur; + next.prefix = "A_pre:"; + return NodeResult::Success(next); + }); + }); + + a_entered_lock.get_future().wait(); + + // Writer B attempts to update suffix while A is in Update callback + std::thread thread_b([&]() { + b_called.set_value(); + snapshot.Update([&](const TwoFields& cur) { + if (cur.prefix == "A_pre:") { + b_saw_a_prefix = true; + } + TwoFields next = cur; + next.suffix = ":B_suf"; + return NodeResult::Success(next); + }); + }); + + b_called.get_future().wait(); + // A owns the transaction before B is launched. No timing assumption is + // needed: B must observe A's published value whenever it acquires the lock. + + // Release A so it publishes its update + release_a.set_value(); + + thread_a.join(); + thread_b.join(); + + EXPECT_TRUE(b_saw_a_prefix); + auto final_state = snapshot.Read(); + ASSERT_NE(final_state, nullptr); + EXPECT_EQ(final_state->prefix, "A_pre:"); + EXPECT_EQ(final_state->suffix, ":B_suf"); +} + +TEST(ConfigurationSnapshotTest, FailedWriterRollbackPreservesActiveState) { + // Verifies RFC-0054 Section 7.1: + // Writer fails during validation / candidate generation -> no new snapshot + // published; old state remains active and intact. + struct State { + std::string val; + int rev; + }; + ConfigurationSnapshot snapshot(State{"original", 1}); + + auto res = snapshot.Update([](const State&) -> NodeResult { + return NodeResult::Failure(NodeErrorKind::kBusinessError, + "validation failed", + node_error::control::kInvalidRequest); + }); + EXPECT_EQ(res.status, NodeControlStatus::kFailed); + EXPECT_EQ(res.code, node_error::control::kInvalidRequest); + + auto cur = snapshot.Read(); + ASSERT_NE(cur, nullptr); + EXPECT_EQ(cur->val, "original"); + EXPECT_EQ(cur->rev, 1); + + // Also test exception in candidate building + auto res_ex = snapshot.Update([](const State&) -> NodeResult { + throw std::runtime_error("candidate throw"); + }); + EXPECT_EQ(res_ex.status, NodeControlStatus::kFailed); + EXPECT_EQ(snapshot.Read()->val, "original"); +} + +TEST(ConfigurationSnapshotTest, ReaderHoldsOldSnapshotWhileWriterPublishes) { + // Verifies RFC-0054 Section 7.1: + // Reader holding old snapshot is isolated from concurrent writer publication; + // old reader completes safely with old version; subsequent reader sees new + // version. + struct State { + std::string val; + int rev; + }; + ConfigurationSnapshot snapshot(State{"v1", 1}); + + std::promise reader_acquired; + std::promise writer_published; + std::promise reader_done; + + std::shared_ptr reader_held_state; + + std::thread reader_thread([&]() { + reader_held_state = snapshot.Read(); + reader_acquired.set_value(); + writer_published.get_future().wait(); + EXPECT_EQ(reader_held_state->val, "v1"); + EXPECT_EQ(reader_held_state->rev, 1); + reader_done.set_value(); + }); + + reader_acquired.get_future().wait(); + + auto res = snapshot.Update( + [](const State&) { return NodeResult::Success(State{"v2", 2}); }); + EXPECT_EQ(res.status, NodeControlStatus::kHandled); + + writer_published.set_value(); + reader_done.get_future().wait(); + reader_thread.join(); + + auto fresh = snapshot.Read(); + ASSERT_NE(fresh, nullptr); + EXPECT_EQ(fresh->val, "v2"); + EXPECT_EQ(fresh->rev, 2); +} + +TEST(ConfigurationSnapshotTest, TwoWritersSameFieldOrdered) { + struct State { + std::string tag; + }; + ConfigurationSnapshot snapshot(State{"init"}); + + auto res1 = snapshot.Update( + [](const State&) { return NodeResult::Success(State{"first"}); }); + EXPECT_EQ(res1.status, NodeControlStatus::kHandled); + + auto res2 = snapshot.Update( + [](const State&) { return NodeResult::Success(State{"second"}); }); + EXPECT_EQ(res2.status, NodeControlStatus::kHandled); + + EXPECT_EQ(snapshot.Read()->tag, "second"); +} + +TEST(ConfigurationSnapshotTest, OldReaderOutlivesOwnerAndReleasesState) { + auto owner = std::make_unique>("old"); + auto reader = owner->Read(); + std::weak_ptr old_state = reader; + ASSERT_EQ(owner + ->Update([](const std::string&) { + return NodeResult::Success("new"); + }) + .status, + NodeControlStatus::kHandled); + EXPECT_FALSE(old_state.expired()); + owner.reset(); + ASSERT_FALSE(old_state.expired()); + EXPECT_EQ(*reader, "old"); + reader.reset(); + EXPECT_TRUE(old_state.expired()); +} + +TEST(ConfigurationSnapshotTest, MoveOnlyStateHandled) { + struct MoveOnlyState { + std::unique_ptr val; + explicit MoveOnlyState(std::string s) + : val(std::make_unique(std::move(s))) {} + MoveOnlyState(MoveOnlyState&&) noexcept = default; + MoveOnlyState& operator=(MoveOnlyState&&) noexcept = default; + MoveOnlyState(const MoveOnlyState&) = delete; + MoveOnlyState& operator=(const MoveOnlyState&) = delete; + }; + + ConfigurationSnapshot snapshot(MoveOnlyState("init")); + EXPECT_TRUE(snapshot.IsInitialized()); + auto cur = snapshot.Read(); + ASSERT_NE(cur, nullptr); + EXPECT_EQ(*cur->val, "init"); + + auto res = snapshot.Update([](const MoveOnlyState& current) { + return NodeResult::Success( + MoveOnlyState(*current.val + "_updated")); + }); + EXPECT_EQ(res.status, NodeControlStatus::kHandled); + auto next = snapshot.Read(); + ASSERT_NE(next, nullptr); + EXPECT_EQ(*next->val, "init_updated"); +} + +// --------------------------------------------------------------------------- +// Functional Spec WithControls & NodeHarness Tests +// --------------------------------------------------------------------------- + +TEST(FunctionNodeTest, FunctionalMapSpecWithControlsReplaceAndPatch) { + NodeHarness harness("ControlledMapNode"); + harness.Config( + {{"prefix", "init_p:"}, {"suffix", ":init_s"}, {"multiplier", 1}}); + harness.TextInput("input", {"payload"}); + + auto res1 = harness.Run(); + ASSERT_TRUE(res1.ok()) << res1.diagnostic(); + EXPECT_EQ(res1.TextValues("output"), + (std::vector{"init_p:payload:init_s"})); + + // 1. ReplaceFields missing multiplier -> rejected + auto bad_replace = harness.Control(kCmdReplaceMap, R"({"prefix":"new_p:"})"); + EXPECT_EQ(bad_replace.status, NodeControlStatus::kFailed); + EXPECT_NE(bad_replace.message.find("multiplier"), std::string::npos); + + // State preserved + auto res2 = harness.Run(); + ASSERT_TRUE(res2.ok()); + EXPECT_EQ(res2.TextValues("output"), + (std::vector{"init_p:payload:init_s"})); + + // 2. ReplaceFields with all declared fields -> handled + auto good_replace = + harness.Control(kCmdReplaceMap, R"({"prefix":"rep_p:","multiplier":2})"); + EXPECT_EQ(good_replace.status, NodeControlStatus::kHandled); + + // Undeclared suffix remains ":init_s", prefix and multiplier updated + auto res3 = harness.Run(); + ASSERT_TRUE(res3.ok()); + EXPECT_EQ(res3.TextValues("output"), + (std::vector{"rep_p:payloadpayload:init_s"})); + + // 3. PatchFields with subset of fields -> handled + auto good_patch = harness.Control(kCmdPatchMap, R"({"suffix":":patch_s"})"); + EXPECT_EQ(good_patch.status, NodeControlStatus::kHandled); + + auto res4 = harness.Run(); + ASSERT_TRUE(res4.ok()); + EXPECT_EQ(res4.TextValues("output"), + (std::vector{"rep_p:payloadpayload:patch_s"})); + + // 4. PatchFields with empty object -> rejected + auto empty_patch = harness.Control(kCmdPatchMap, R"({})"); + EXPECT_EQ(empty_patch.status, NodeControlStatus::kFailed); + + // 5. Semantic validator failure -> rejected and state rolled back + auto invalid_prefix = + harness.Control(kCmdPatchMap, R"({"prefix":"INVALID"})"); + EXPECT_EQ(invalid_prefix.status, NodeControlStatus::kFailed); + EXPECT_NE(invalid_prefix.message.find("Invalid prefix disallowed"), + std::string::npos); + + auto res5 = harness.Run(); + ASSERT_TRUE(res5.ok()); + EXPECT_EQ(res5.TextValues("output"), + (std::vector{"rep_p:payloadpayload:patch_s"})); + + // 6. Unknown command -> unsupported + auto unk = harness.Control(9999, R"({})"); + EXPECT_EQ(unk.status, NodeControlStatus::kUnsupported); +} + +TEST(FunctionNodeTest, FunctionalBatchSpecWithControlsAndValidation) { + NodeHarness harness("ControlledBatchNode"); + harness.Config({{"header", "H:"}, {"uppercase", false}}); + harness.TextInput("texts", {"abc", "def"}); + + auto res1 = harness.Run(); + ASSERT_TRUE(res1.ok()) << res1.diagnostic(); + EXPECT_EQ(res1.TextValues("output"), + (std::vector{"H:abc", "H:def"})); + + // Replace update with uppercase = true + auto ctrl1 = + harness.Control(kCmdReplaceBatch, R"({"header":"G:","uppercase":true})"); + EXPECT_EQ(ctrl1.status, NodeControlStatus::kHandled); + + auto res2 = harness.Run(); + ASSERT_TRUE(res2.ok()); + EXPECT_EQ(res2.TextValues("output"), + (std::vector{"G:ABC", "G:DEF"})); + + // Semantic rejection + auto ctrl2 = harness.Control(kCmdReplaceBatch, + R"({"header":"REJECT","uppercase":true})"); + EXPECT_EQ(ctrl2.status, NodeControlStatus::kFailed); + EXPECT_NE(ctrl2.message.find("Rejected header"), std::string::npos); + + // Preserved on failure + auto res3 = harness.Run(); + ASSERT_TRUE(res3.ok()); + EXPECT_EQ(res3.TextValues("output"), + (std::vector{"G:ABC", "G:DEF"})); +} + +TEST(FunctionNodeTest, SnapshotPauseTimeoutFailsProcess) { + NodeHarness harness("ControlledMapNode"); + harness.DisablePlan(); + ASSERT_TRUE(harness.EnsureInitialized()); + AlgContext ctx; + ctx.Publish("input", TextBatch{{101, 3, "sample"}}); + test_support::NodeProcessPause pause(std::chrono::milliseconds(0)); + int result = 0; + { + test_support::ScopedNextAllocationCallback callback( + &test_support::NodeProcessPause::OnAllocation, &pause); + result = harness.GetNode()->Process(&ctx); + } + EXPECT_NE(result, 0); + EXPECT_NE(ctx.GetErrorMessage().find("handshake timed out"), + std::string::npos); + EXPECT_EQ(ctx.Read("output"), nullptr); +} + +TEST(FunctionNodeTest, WholeBatchProcessConsistencyDuringControl) { + NodeHarness harness("ControlledMapNode"); + harness.DisablePlan(); + harness.Config({{"prefix", "v1:"}, {"suffix", ":s1"}, {"multiplier", 1}}); + ASSERT_TRUE(harness.EnsureInitialized()); + auto* node = harness.GetNode(); + ASSERT_NE(node, nullptr); + + TextBatch batch; + for (uint64_t i = 0; i < 50; ++i) { + batch.emplace_back(100 + i, i, "sample_" + std::to_string(i)); + } + AlgContext old_ctx; + old_ctx.Publish("input", batch); + test_support::NodeProcessPause pause; + auto reader = std::async(std::launch::async, [&] { + // The first allocation is outputs.reserve, after AuthorNode has acquired + // its parameter snapshot. The callback is confined to this reader thread. + test_support::ScopedNextAllocationCallback callback( + &test_support::NodeProcessPause::OnAllocation, &pause); + return node->Process(&old_ctx); + }); + const bool paused = pause.WaitUntilPaused(); + NodeControlResult control = NodeControlResult::Unsupported(); + if (paused) { + control = + node->Control(kCmdReplaceMap, R"({"prefix":"v2:","multiplier":2})"); + } + pause.Resume(); + const int process_result = reader.get(); + ASSERT_TRUE(paused) << "Reader did not reach its snapshot pause"; + ASSERT_EQ(control.status, NodeControlStatus::kHandled); + ASSERT_EQ(process_result, 0) << old_ctx.GetErrorMessage(); + + const auto* old_output = old_ctx.Read("output"); + ASSERT_NE(old_output, nullptr); + ASSERT_EQ(old_output->size(), batch.size()); + for (size_t i = 0; i < batch.size(); ++i) { + EXPECT_EQ((*old_output)[i].data, "v1:" + batch[i].data + ":s1"); + EXPECT_EQ((*old_output)[i].req_id, batch[i].req_id); + EXPECT_EQ((*old_output)[i].sub_id, batch[i].sub_id); + } + AlgContext new_ctx; + new_ctx.Publish("input", batch); + ASSERT_EQ(node->Process(&new_ctx), 0); + const auto* new_output = new_ctx.Read("output"); + ASSERT_NE(new_output, nullptr); + ASSERT_EQ(new_output->size(), batch.size()); + for (size_t i = 0; i < batch.size(); ++i) { + EXPECT_EQ((*new_output)[i].data, + "v2:" + batch[i].data + batch[i].data + ":s1"); + EXPECT_EQ((*new_output)[i].req_id, batch[i].req_id); + EXPECT_EQ((*new_output)[i].sub_id, batch[i].sub_id); + } +} + +TEST(FunctionNodeTest, WholeBatchProcessConsistencyDuringControlForBatchSpec) { + NodeHarness harness("ControlledBatchNode"); + harness.DisablePlan(); + harness.Config({{"header", "old:"}, {"uppercase", false}}); + ASSERT_TRUE(harness.EnsureInitialized()); + auto* node = harness.GetNode(); + ASSERT_NE(node, nullptr); + + TextBatch batch; + for (uint64_t i = 0; i < 50; ++i) { + batch.emplace_back(300 + i, i, "sample"); + } + AlgContext old_ctx; + old_ctx.Publish("texts", batch); + test_support::NodeProcessPause pause; + auto reader = std::async(std::launch::async, [&] { + // ControlledBatchFn reserves output after AuthorNode acquires its snapshot. + test_support::ScopedNextAllocationCallback callback( + &test_support::NodeProcessPause::OnAllocation, &pause); + return node->Process(&old_ctx); + }); + const bool paused = pause.WaitUntilPaused(); + NodeControlResult control = NodeControlResult::Unsupported(); + if (paused) { + control = node->Control(kCmdReplaceBatch, + R"({"header":"new:","uppercase":true})"); + } + pause.Resume(); + const int process_result = reader.get(); + ASSERT_TRUE(paused) << "Reader did not reach its snapshot pause"; + ASSERT_EQ(control.status, NodeControlStatus::kHandled); + ASSERT_EQ(process_result, 0) << old_ctx.GetErrorMessage(); + + const auto* old_output = old_ctx.Read("output"); + ASSERT_NE(old_output, nullptr); + ASSERT_EQ(old_output->size(), batch.size()); + for (size_t i = 0; i < batch.size(); ++i) { + EXPECT_EQ((*old_output)[i].data, "old:sample"); + EXPECT_EQ((*old_output)[i].req_id, batch[i].req_id); + EXPECT_EQ((*old_output)[i].sub_id, batch[i].sub_id); + } + AlgContext new_ctx; + new_ctx.Publish("texts", batch); + ASSERT_EQ(node->Process(&new_ctx), 0); + const auto* new_output = new_ctx.Read("output"); + ASSERT_NE(new_output, nullptr); + ASSERT_EQ(new_output->size(), batch.size()); + for (size_t i = 0; i < batch.size(); ++i) { + EXPECT_EQ((*new_output)[i].data, "NEW:SAMPLE"); + EXPECT_EQ((*new_output)[i].req_id, batch[i].req_id); + EXPECT_EQ((*new_output)[i].sub_id, batch[i].sub_id); + } +} + +TEST(FunctionNodeTest, + SpecWithNonCopyableParamsCompilesAndExecutesWithoutControls) { + // Verifies MapSpec with non-copyable ParamsT (containing unique_ptr) + NodeHarness map_harness("NonCopyableMapNode"); + map_harness.DisablePlan(); + map_harness.Config({{"prefix", "map_nc:"}}); + map_harness.TextInput("input", {"hello", "world"}); + auto map_res = map_harness.Run(); + ASSERT_TRUE(map_res.ok()) << map_res.diagnostic(); + EXPECT_EQ(map_res.TextValues("output"), + (std::vector{"map_nc:hello_100", "map_nc:world_100"})); + + auto ctrl_map = map_harness.Control(1001, R"({})"); + EXPECT_EQ(ctrl_map.status, NodeControlStatus::kUnsupported); + + // Verifies BatchSpec with non-copyable ParamsT (containing unique_ptr) + NodeHarness batch_harness("NonCopyableBatchNode"); + batch_harness.DisablePlan(); + batch_harness.Config({{"tag", "batch_nc:"}}); + batch_harness.TextInput("texts", {"foo", "bar"}); + auto batch_res = batch_harness.Run(); + ASSERT_TRUE(batch_res.ok()) << batch_res.diagnostic(); + EXPECT_EQ(batch_res.TextValues("output"), + (std::vector{"batch_nc:foo_200", "batch_nc:bar_200"})); + + auto ctrl_batch = batch_harness.Control(1001, R"({})"); + EXPECT_EQ(ctrl_batch.status, NodeControlStatus::kUnsupported); + + // Also verify planned execution works with non-copyable ParamsT + NodeHarness map_planned("NonCopyableMapNode"); + map_planned.Config({{"prefix", "map_nc_p:"}}); + map_planned.TextInput("input", {"hello"}); + auto map_res_p = map_planned.Run(); + ASSERT_TRUE(map_res_p.ok()) << map_res_p.diagnostic(); + EXPECT_EQ(map_res_p.TextValues("output"), + (std::vector{"map_nc_p:hello_100"})); + + NodeHarness batch_planned("NonCopyableBatchNode"); + batch_planned.Config({{"tag", "batch_nc_p:"}}); + batch_planned.TextInput("texts", {"foo"}); + auto batch_res_p = batch_planned.Run(); + ASSERT_TRUE(batch_res_p.ok()) << batch_res_p.diagnostic(); + EXPECT_EQ(batch_res_p.TextValues("output"), + (std::vector{"batch_nc_p:foo_200"})); +} + +TEST(FunctionNodeTest, + UnplannedInitPassesCorrectBindingFactsToPrepareWithoutPlan) { + // Map node verifying facts.has_plan is false during unplanned init + NodeHarness map_harness("StrictUnplannedMapNode"); + map_harness.DisablePlan(); + map_harness.TextInput("input", {"item1"}); + auto map_res = map_harness.Run(); + ASSERT_TRUE(map_res.ok()) << map_res.diagnostic(); + EXPECT_EQ(map_res.TextValues("output"), + (std::vector{"unplanned:item1"})); + + // Batch node verifying facts.has_plan is false during unplanned init + NodeHarness batch_harness("StrictUnplannedBatchNode"); + batch_harness.DisablePlan(); + batch_harness.TextInput("texts", {"item2"}); + auto batch_res = batch_harness.Run(); + ASSERT_TRUE(batch_res.ok()) << batch_res.diagnostic(); + EXPECT_EQ(batch_res.TextValues("output"), + (std::vector{"unplanned_batch:item2"})); + + // Verify that both fail if run WITH a plan (facts.has_plan == true) + NodeHarness map_planned("StrictUnplannedMapNode"); + map_planned.TextInput("input", {"item_p"}); + auto map_res_p = map_planned.Run(); + EXPECT_FALSE(map_res_p.ok()); + EXPECT_TRUE(map_res_p.init_failed()); + + NodeHarness batch_planned("StrictUnplannedBatchNode"); + batch_planned.TextInput("texts", {"item_p"}); + auto batch_res_p = batch_planned.Run(); + EXPECT_FALSE(batch_res_p.ok()); + EXPECT_TRUE(batch_res_p.init_failed()); +} + +TEST(FunctionNodeTest, SpecWithoutWithControlsReturnsUnsupported) { + NodeHarness harness("UpperMapNode"); + harness.TextInput("input", {"hello"}); + ASSERT_TRUE(harness.EnsureInitialized()); + auto ctrl = harness.Control(1001, R"({})"); + EXPECT_EQ(ctrl.status, NodeControlStatus::kUnsupported); +} + +TEST(FunctionNodeTest, DeclarationValidationRejectsInvalidControlCommands) { + struct DummyParams { + std::string text; + int count = 0; + }; + auto make_params = []() { + return Parameters({ + Field("text", &DummyParams::text).Default(""), + Field("count", &DummyParams::count).Default(0), + }); + }; + + // 1. Invalid command ID (<= 0) + EXPECT_THROW(ValidateControlCommands({ReplaceFields(0, "set_text", {"text"})}, + make_params()), + std::invalid_argument); + + // 2. Empty command name + EXPECT_THROW(ValidateControlCommands({ReplaceFields(1001, "", {"text"})}, + make_params()), + std::invalid_argument); + + // 3. Duplicate command ID + EXPECT_THROW( + ValidateControlCommands({ReplaceFields(1001, "cmd_a", {"text"}), + ReplaceFields(1001, "cmd_b", {"count"})}, + make_params()), + std::invalid_argument); + + // 4. Duplicate command name + EXPECT_THROW( + ValidateControlCommands({ReplaceFields(1001, "same_name", {"text"}), + ReplaceFields(1002, "same_name", {"count"})}, + make_params()), + std::invalid_argument); + + // 5. Empty field names + EXPECT_THROW( + ValidateControlCommands({ReplaceFields(1001, "cmd", {})}, make_params()), + std::invalid_argument); + + // 6. Duplicate field name in same command + EXPECT_THROW( + ValidateControlCommands({ReplaceFields(1001, "cmd", {"text", "text"})}, + make_params()), + std::invalid_argument); + + // 7. Unbound field name + EXPECT_THROW( + ValidateControlCommands({ReplaceFields(1001, "cmd", {"non_existent"})}, + make_params()), + std::invalid_argument); +} + +TEST(FunctionNodeTest, WithParserWithControlsRequiresExplicitPrepare) { + struct DummyParams { + std::string text; + }; + NodeConfigParser parser( + {ConfigFieldDefinition{"nested", ConfigValueKind::kObject, true}}, + [](const nlohmann::json& c, DummyParams* p, std::string*) { + if (c.contains("nested") && c["nested"].contains("text")) { + p->text = c["nested"]["text"].get(); + } + return true; + }); + auto params = + Parameters({ + Field("text", &DummyParams::text).Default(""), + }) + .WithParser(std::move(parser)); + + // HasParser() == true, commands not empty, HasPrepare() == false -> throws + EXPECT_THROW(ValidateControlCommands( + {ReplaceFields(1001, "set_text", {"text"})}, params), + std::invalid_argument); + + // Adding Prepare allows validation to pass + params.Prepare( + [](DummyParams*, const BindingFacts&, std::string*) { return true; }); + EXPECT_NO_THROW(ValidateControlCommands( + {ReplaceFields(1001, "set_text", {"text"})}, params)); +} + +TEST(FunctionNodeTest, TypedPrepareHookExecutesAndCanReject) { + struct PreparedParams { + std::string raw; + std::string derived; + }; + + auto params = Parameters( + { + Field("raw", &PreparedParams::raw).Default(""), + }) + .Prepare([](PreparedParams* p, const BindingFacts&, + std::string* err) { + if (p->raw == "FAIL_PREPARE") { + if (err) *err = "Prepare rejected"; + return false; + } + p->derived = "prepared:" + p->raw; + return true; + }); + + BindingFacts facts; + std::string err; + auto ok_res = params.ParseNormalized({{"raw", "hello"}}, facts, &err); + ASSERT_TRUE(ok_res.has_value()); + EXPECT_EQ(ok_res->derived, "prepared:hello"); + + auto fail_res = + params.ParseNormalized({{"raw", "FAIL_PREPARE"}}, facts, &err); + EXPECT_FALSE(fail_res.has_value()); + EXPECT_NE(err.find("Prepare rejected"), std::string::npos); +} + +struct BindingFactsProbeParams { + std::string mode; + bool plan_seen = false; + bool input_connected = false; +}; + +inline auto BindingFactsProbeSpec() { + return MakeMapSpec( + Input("input"), Output("output"), + Parameters( + { + Field("mode", &BindingFactsProbeParams::mode).Default("base"), + }) + .Prepare([](BindingFactsProbeParams* p, const BindingFacts& facts, + std::string*) { + p->plan_seen = facts.has_plan; + p->input_connected = facts.IsConnected("input"); + return true; + }), + [](const std::string& in, const BindingFactsProbeParams& p) { + return std::string(p.plan_seen ? "PLAN:" : "NO_PLAN:") + + (p.input_connected ? "CONN:" : "DISCONN:") + in; + }); +} +REGISTER_FUNCTION_NODE(BindingFactsProbeNode, BindingFactsProbeSpec()); + +TEST(FunctionNodeTest, AuthorNodeInitPassesRealBindingFactsToPrepare) { + // Test planned execution: plan_seen must be true, input must be connected + NodeHarness harness_planned("BindingFactsProbeNode"); + harness_planned.TextInput("input", {"hello"}); + auto res_planned = harness_planned.Run(); + ASSERT_TRUE(res_planned.ok()) << res_planned.diagnostic(); + EXPECT_EQ(res_planned.TextValues("output"), + (std::vector{"PLAN:CONN:hello"})); + + // Test unplanned execution: plan_seen must be false, input is still connected + // logically + NodeHarness harness_unplanned("BindingFactsProbeNode"); + harness_unplanned.DisablePlan(); + harness_unplanned.TextInput("input", {"world"}); + auto res_unplanned = harness_unplanned.Run(); + ASSERT_TRUE(res_unplanned.ok()) << res_unplanned.diagnostic(); + EXPECT_EQ(res_unplanned.TextValues("output"), + (std::vector{"NO_PLAN:CONN:world"})); +} + +TEST(FunctionNodeTest, RapidInterleavedControlsAndConcurrentProcesses) { + NodeHarness harness("ControlledMapNode"); + harness.DisablePlan(); + harness.Config({{"prefix", "p0:"}, {"suffix", ":s0"}, {"multiplier", 1}}); + ASSERT_TRUE(harness.EnsureInitialized()); + auto* node = harness.GetNode(); + ASSERT_NE(node, nullptr); + + std::atomic stop{false}; + std::atomic successful_processes{0}; + + // Writer thread 1: rapid ReplaceFields + std::thread writer1([&]() { + for (int i = 1; i <= 30; ++i) { + std::string payload = + "{\"prefix\":\"p" + std::to_string(i) + + ":\",\"multiplier\":" + std::to_string((i % 3) + 1) + "}"; + auto res = node->Control(kCmdReplaceMap, payload); + EXPECT_EQ(res.status, NodeControlStatus::kHandled); + std::this_thread::yield(); + } + }); + + // Writer thread 2: rapid PatchFields + std::thread writer2([&]() { + for (int i = 1; i <= 30; ++i) { + std::string payload = "{\"suffix\":\":s" + std::to_string(i) + "\"}"; + auto res = node->Control(kCmdPatchMap, payload); + EXPECT_EQ(res.status, NodeControlStatus::kHandled); + std::this_thread::yield(); + } + }); + + // Multiple reader threads: concurrent Process calls with multi-item batches + std::vector readers; + for (int r = 0; r < 4; ++r) { + readers.emplace_back([&, r]() { + uint64_t req_id = 1000 + r * 10000; + while (!stop.load(std::memory_order_relaxed)) { + AlgContext ctx; + TextBatch input; + for (int i = 0; i < 8; ++i) { + input.emplace_back(req_id, i, "payload_" + std::to_string(i)); + } + req_id++; + ctx.Publish("input", std::move(input)); + int rc = node->Process(&ctx); + EXPECT_EQ(rc, 0); + const auto* out = ctx.Read("output"); + ASSERT_NE(out, nullptr); + ASSERT_EQ(out->size(), 8u); + + // Verify intra-batch snapshot consistency: + // Item format: prefix + (multiplier * "payload_i") + suffix + // Extract prefix, suffix, multiplier from item 0: + const std::string& item0 = (*out)[0].data; + auto pos0 = item0.find("payload_0"); + ASSERT_NE(pos0, std::string::npos); + std::string prefix = item0.substr(0, pos0); + auto last_pos0 = item0.rfind("payload_0"); + std::string suffix = + item0.substr(last_pos0 + std::string("payload_0").size()); + int multiplier = 0; + size_t sp = 0; + while ((sp = item0.find("payload_0", sp)) != std::string::npos) { + multiplier++; + sp += std::string("payload_0").size(); + } + // Every other item in this batch must match the exact same snapshot + // parameters: + for (int i = 1; i < 8; ++i) { + std::string expected = prefix; + for (int m = 0; m < multiplier; ++m) { + expected += "payload_" + std::to_string(i); + } + expected += suffix; + EXPECT_EQ((*out)[i].data, expected) + << "Batch mixed configuration snapshots between items!"; + } + successful_processes.fetch_add(1, std::memory_order_relaxed); + } + }); + } + + writer1.join(); + writer2.join(); + stop.store(true, std::memory_order_relaxed); + for (auto& t : readers) { + t.join(); + } + + EXPECT_GT(successful_processes.load(), 0); + + // Verify node remains in a coherent final state + AlgContext final_ctx; + final_ctx.Publish("input", TextBatch{{999, 0, "final"}}); + ASSERT_EQ(node->Process(&final_ctx), 0); + const auto* final_out = final_ctx.Read("output"); + ASSERT_NE(final_out, nullptr); + ASSERT_EQ(final_out->size(), 1u); + EXPECT_EQ(final_out->at(0).data, "p30:final:s30"); +} + } // namespace llm_edgeflow diff --git a/tests/unit/nodes/test_parameter_binding.cpp b/tests/unit/nodes/test_parameter_binding.cpp index d9dff0fb..c93a5f4d 100644 --- a/tests/unit/nodes/test_parameter_binding.cpp +++ b/tests/unit/nodes/test_parameter_binding.cpp @@ -274,5 +274,29 @@ TEST(ParameterBindingTest, BindingValidatorExceptionDoesNotCrash) { std::string::npos); } +TEST(ParameterBindingTest, + ValidateWithBindingsEnforcesConstraintsEvenWhenConnectedInputsIsEmpty) { + auto schema = Parameters({ + Field("mode", &SampleParams::mode).Default("custom"), + }); + + schema.ValidateBindings([](const SampleParams& p, + const std::unordered_set& conn, + std::string* err) -> bool { + if (p.mode == "custom" && conn.count("context") == 0) { + if (err) *err = "custom mode requires context port"; + return false; + } + return true; + }); + + std::string err; + nlohmann::json norm = {{"mode", "custom"}}; + // Connected inputs is empty set {} - must still enforce binding validation! + bool bind_ok = schema.ValidateWithBindings(norm, {}, &err); + EXPECT_FALSE(bind_ok); + EXPECT_NE(err.find("custom mode requires context port"), std::string::npos); +} + } // namespace } // namespace llm_edgeflow diff --git a/tests/unit/nodes/test_text_rule_match_node.cpp b/tests/unit/nodes/test_text_rule_match_node.cpp index bc622d28..74e7401e 100644 --- a/tests/unit/nodes/test_text_rule_match_node.cpp +++ b/tests/unit/nodes/test_text_rule_match_node.cpp @@ -2,10 +2,12 @@ #include #include +#include #include #include #include #include +#include #include #include "adapter/shared_algorithm_runtime.h" @@ -14,7 +16,9 @@ #include "core/node_registry.h" #include "core/pipeline_validator.h" #include "core/session_context.h" +#include "tests/support/node_process_pause.h" #include "tests/support/node_test_utils.h" +#include "tests/support/scoped_allocation_failure.h" namespace llm_edgeflow { @@ -431,4 +435,87 @@ TEST_F(TextRuleMatchNodeTest, MissingInputFailsClosed) { EXPECT_EQ(node->Process(&empty_ctx), -5001); } +TEST_F(TextRuleMatchNodeTest, DirectConcurrentProcessAndControl) { + auto node = NodeRegistry::Instance().Create("TextRuleMatchNode"); + ASSERT_NE(node, nullptr); + auto configuration = [](const std::string& version) { + return nlohmann::json{{"categories", {{version, {"hello"}}}}, + {"rules", + {{{"id", version + "_rule"}, + {"strategy", "regex"}, + {"pattern", "(?<" + version + ">world)"}, + {"category", version}}}}}; + }; + ASSERT_TRUE(InitNodeForTest(*node, configuration("OLD"), session_ctx_.get())); + const auto update = configuration("NEW"); + // Both category keywords and compiled regex captures distinguish versions. + const TextBatch inputs{{101, 2, "hello"}, + {101, 7, "world"}, + {202, 3, "hello"}, + {202, 8, "world"}}; + AlgContext in_flight; + in_flight.Publish("text", inputs); + test_support::NodeProcessPause pause; + int process_result = -1; + std::exception_ptr reader_error; + // On this valid Process path, the first heap allocation follows + // snapshot.Read: the output batch reserve. Pause with that old snapshot + // retained. + std::thread reader([&] { + try { + test_support::ScopedNextAllocationCallback callback( + &test_support::NodeProcessPause::OnAllocation, &pause); + process_result = node->Process(&in_flight); + } catch (...) { + reader_error = std::current_exception(); + } + }); + const bool paused = pause.WaitUntilPaused(); + NodeControlStatus control_status = NodeControlStatus::kFailed; + std::exception_ptr control_error; + if (paused) { + try { + control_status = + node->Control(kControlCmdUpdateRules, update.dump()).status; + } catch (...) { + control_error = std::current_exception(); + } + } + pause.Resume(); + reader.join(); + ASSERT_TRUE(paused) << "Process never reached the snapshot pause"; + ASSERT_EQ(reader_error, nullptr); + ASSERT_EQ(control_error, nullptr); + ASSERT_EQ(control_status, NodeControlStatus::kHandled); + ASSERT_EQ(process_result, 0); + + auto expect_batch = [&](const AlgContext& ctx, const std::string& version) { + const auto* output = ctx.Read("matches"); + ASSERT_NE(output, nullptr); + ASSERT_EQ(output->size(), inputs.size()); + for (size_t i = 0; i < inputs.size(); ++i) { + EXPECT_EQ((*output)[i].req_id, inputs[i].req_id); + EXPECT_EQ((*output)[i].sub_id, inputs[i].sub_id); + EXPECT_EQ((*output)[i].data.is_hit, 1); + EXPECT_EQ((*output)[i].data.category, version); + if (inputs[i].data == "world") { + EXPECT_EQ((*output)[i].data.rule_id, version + "_rule"); + EXPECT_EQ((*output)[i].data.captures.count(version), 1u); + const auto capture = (*output)[i].data.captures.find(version); + if (capture != (*output)[i].data.captures.end()) { + EXPECT_EQ(capture->second, "world"); + } + EXPECT_EQ( + (*output)[i].data.captures.count(version == "OLD" ? "NEW" : "OLD"), + 0u); + } + } + }; + expect_batch(in_flight, "OLD"); + AlgContext subsequent; + subsequent.Publish("text", inputs); + ASSERT_EQ(node->Process(&subsequent), 0); + expect_batch(subsequent, "NEW"); +} + } // namespace llm_edgeflow diff --git a/tests/unit/nodes/test_text_template_node.cpp b/tests/unit/nodes/test_text_template_node.cpp index e3337fb1..8784fe59 100644 --- a/tests/unit/nodes/test_text_template_node.cpp +++ b/tests/unit/nodes/test_text_template_node.cpp @@ -2,9 +2,11 @@ #include #include +#include #include #include #include +#include #include #include "adapter/shared_algorithm_runtime.h" @@ -13,7 +15,9 @@ #include "core/node_registry.h" #include "core/pipeline.h" #include "core/session_context.h" +#include "tests/support/node_process_pause.h" #include "tests/support/node_test_utils.h" +#include "tests/support/scoped_allocation_failure.h" namespace llm_edgeflow { @@ -356,4 +360,65 @@ TEST_F(TextTemplateNodeTest, ConnectedAttributesRemainAvailableAcrossControl) { EXPECT_EQ(ctx.Read("text")->front().data, "Alice"); } +TEST_F(TextTemplateNodeTest, DirectConcurrentProcessAndControl) { + auto node = NodeRegistry::Instance().Create("TextTemplateNode"); + ASSERT_NE(node, nullptr); + ASSERT_TRUE(InitNodeForTest(*node, {{"template", "OLD: {{primary}}"}}, + session_ctx_.get())); + const nlohmann::json update = {{"template", "NEW: {{primary}}"}}; + const TextBatch inputs{ + {101, 2, "first"}, {101, 7, "second"}, {202, 3, "third"}}; + AlgContext in_flight; + in_flight.Publish("primary", inputs); + test_support::NodeProcessPause pause; + int process_result = -1; + std::exception_ptr reader_error; + // On this valid Process path, the first heap allocation follows + // snapshot.Read: the template grouping container. Pause with that old + // snapshot retained. + std::thread reader([&] { + try { + test_support::ScopedNextAllocationCallback callback( + &test_support::NodeProcessPause::OnAllocation, &pause); + process_result = node->Process(&in_flight); + } catch (...) { + reader_error = std::current_exception(); + } + }); + const bool paused = pause.WaitUntilPaused(); + NodeControlStatus control_status = NodeControlStatus::kFailed; + std::exception_ptr control_error; + if (paused) { + try { + control_status = + node->Control(kControlCmdUpdatePrompt, update.dump()).status; + } catch (...) { + control_error = std::current_exception(); + } + } + pause.Resume(); + reader.join(); + ASSERT_TRUE(paused) << "Process never reached the snapshot pause"; + ASSERT_EQ(reader_error, nullptr); + ASSERT_EQ(control_error, nullptr); + ASSERT_EQ(control_status, NodeControlStatus::kHandled); + ASSERT_EQ(process_result, 0); + + auto expect_batch = [&](const AlgContext& ctx, const std::string& version) { + const auto* output = ctx.Read("text"); + ASSERT_NE(output, nullptr); + ASSERT_EQ(output->size(), inputs.size()); + for (size_t i = 0; i < inputs.size(); ++i) { + EXPECT_EQ((*output)[i].req_id, inputs[i].req_id); + EXPECT_EQ((*output)[i].sub_id, inputs[i].sub_id); + EXPECT_EQ((*output)[i].data, version + ": " + inputs[i].data); + } + }; + expect_batch(in_flight, "OLD"); + AlgContext subsequent; + subsequent.Publish("primary", inputs); + ASSERT_EQ(node->Process(&subsequent), 0); + expect_batch(subsequent, "NEW"); +} + } // namespace llm_edgeflow