fix(control): deliver controlled configuration snapshots and review followups (RFC-0054)

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})

cmake_ext/Tests.cmake

@@ -142,7 +142,8 @@ add_executable(edgeflow_test_nodes_runner
   $<TARGET_OBJECTS:edgeflow_test_backend_fixtures>
   $<TARGET_OBJECTS:edgeflow_test_biz_model_fixtures>)
 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}")
 

dev_support/benchmarks/control_snapshots.cpp

@@ -0,0 +1,124 @@
+#include <atomic>
+#include <chrono>
+#include <cstdlib>
+#include <iostream>
+#include <memory>
+#include <new>
+#include <string>
+#include <thread>
+#include <vector>
+
+#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 (?<tail>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<std::unique_ptr<AlgContext>> contexts;
+  for (int i = 0; i < n; ++i) {
+    auto ctx = std::make_unique<AlgContext>();
+    ctx->Publish(tpl ? "primary" : "text", input);
+    contexts.push_back(std::move(ctx));
+  }
+  std::atomic<bool> stop{false}, ready{false};
+  std::atomic<size_t> 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<TextBatch>("text");
+      if (!out || out->size() != 50 ||
+          (*out)[0].data != "V1: hello world sample 0 / assistant")
+        return 5;
+    } else {
+      auto* out = ctx.Read<RuleMatchBatch>("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<double, std::micro>(end - start).count() /
+                   n
+            << " " << updates << "\n";
+}

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)