chamsechan · chamsechan · Sep 15, 2026 · Sep 15, 2026
diff --git a/cmake_ext/Tests.cmake b/cmake_ext/Tests.cmake
@@ -225,7 +225,7 @@ set(_edgeflow_tier4 "tier4;tooling;dev-fast;sanitizer-compatible")
 edgeflow_add_runner_test(BatchExecutorTest edgeflow_test_core_runner
   "FixedBatchExecutorTest.*" "${_edgeflow_tier1}")
 edgeflow_add_runner_test(FrameworkCoreTest edgeflow_test_core_runner
-  "AlgContextTest.*:TraceableItemTest.*:NodeRegistryTest.*:ModelManagerTest.*:PipelineTest.*"
+  "AlgContextTest.*:TraceableItemTest.*:ModelManagerTest.*:PipelineTest.*:SessionContextTest.*"
   "${_edgeflow_tier1}")
 edgeflow_add_runner_test(CompanyAlgLogTest edgeflow_test_core_runner
   "CompanyAlgLogTest.*:CompanyAlgLogNameOverrideTest.*"
@@ -246,7 +246,7 @@ edgeflow_add_runner_test(TypedBlackboardContractsTest edgeflow_test_core_runner
 edgeflow_add_runner_test(ValidatedPipelinePlanTest edgeflow_test_core_runner
   "ValidatedPipelinePlanTest.*" "${_edgeflow_tier1}")
 edgeflow_add_runner_test(NodeBaseContractsTest edgeflow_test_core_runner
-  "NodeBaseContractsTest.*" "${_edgeflow_tier1}")
+  "NodeBaseContractsTest.*:NodeErrorCodesTest.*" "${_edgeflow_tier1}")
 edgeflow_add_runner_test(NodeOwnershipAndReuseTest edgeflow_test_core_runner
   "NodeOwnershipAndReuseTest.*" "${_edgeflow_tier1}")
 edgeflow_add_runner_test(DefinitionSchemaValidationTest edgeflow_test_core_runner
@@ -316,14 +316,15 @@ edgeflow_add_runner_test(OperatorBizBridgeRegistryTest
 edgeflow_add_runner_test(OperatorGoldenTest edgeflow_test_adapter_runner
   "OperatorGoldenTest.*" "${_edgeflow_tier2}")
 edgeflow_add_runner_test(AdapterPurityTest edgeflow_test_adapter_runner
-  "AdapterPurityTest.*" "${_edgeflow_tier2}")
+  "AdapterPurityTest.*:RequestResultsTest.*:AdapterResultTest.*:ReadMultiWayResultsTest.*:OneToOneTextAdapterTest.*"
+  "${_edgeflow_tier2}")
 
 edgeflow_add_runner_test(DocQaRerankTest edgeflow_test_tooling_runner
   "DocQaRerankPipelineTest.*" "${_edgeflow_tier1}")
 # This suite exercises Validator, typed Blackboard and Pipeline::Execute.
 # Its runner grouping does not make it tooling-only coverage.
 edgeflow_add_runner_test(PipelineStudioTest edgeflow_test_tooling_runner
-  "BlackboardKeyTest.*:PipelineCatalogTest.*:PipelineValidatorTest.*"
+  "PipelineCatalogTest.*:PipelineValidatorTest.*"
   "${_edgeflow_tier3}")
 edgeflow_add_runner_test(DemoRunnerTest edgeflow_test_tooling_runner
   "DemoRunnerTest.*" "${_edgeflow_tier3};kite;kite-real")

diff --git a/demo/biz/ocr_doc_qa_demo.cpp b/demo/biz/ocr_doc_qa_demo.cpp
@@ -42,47 +42,11 @@ int RunOcrDocQaDemo(const DemoOptions& options) {
     }
   }
 
-  if (!ValidateConfigBizMatch(options.config_path, options.biz, &err)) {
-    std::cerr << "[OcrDocQaDemo ERROR] Config validation failed: " << err
-              << std::endl;
-    return 3;
-  }
-
-  llm_edgeflow::operator_api::ComputePlatform chip_type =
-      llm_edgeflow::operator_api::ComputePlatform::kUnknown;
-  if (!ParseComputePlatform(options.chip, &chip_type)) {
-    std::cerr << "[OcrDocQaDemo ERROR] Unsupported chip: " << options.chip
-              << std::endl;
-    return 3;
-  }
-
-  std::string model_root;
-  std::string cfg_rel;
-  ResolveModelRootAndConfig(options.config_path, &model_root, &cfg_rel);
-
-  auto ops = llm_edgeflow::operator_api::Get_LLM_EDGEFLOW_OperatorTable();
-
-  int max_batch_size = options.batch_size > 0 ? options.batch_size : 1;
-  uint32_t requested_depth = options.depth_num > 0 ? options.depth_num : 25;
-  if (requested_depth < static_cast<uint32_t>(max_batch_size)) {
-    requested_depth = static_cast<uint32_t>(max_batch_size);
-  }
-
-  llm_edgeflow::operator_api::CreateParam param{};
-  param.model_path = model_root.c_str();
-  param.cfg_file_name = cfg_rel.c_str();
-  param.device_id = options.device_id;
-  param.compute_platform = chip_type;
-  param.max_frame_depth = requested_depth;
-
+  llm_edgeflow::operator_api::OperatorFunc ops{};
   void* raw_handle = nullptr;
-  int ret = ops.Create(&raw_handle, &param);
-  if (ret != 0 || !raw_handle) {
-    std::cerr << "[OcrDocQaDemo ERROR] Failed ops.Create: "
-              << llm_edgeflow::operator_api::GetOperatorLastError()
-              << std::endl;
-    return 5;
-  }
+  const int init_ret =
+      CreateOperatorInstance(options, "OcrDocQaDemo", &ops, &raw_handle);
+  if (init_ret != 0) return init_ret;
 
   OperatorHandleGuard guard(ops, raw_handle);
   const int control_ret = ApplyOperatorControl(
@@ -106,7 +70,7 @@ int RunOcrDocQaDemo(const DemoOptions& options) {
   out_batch[0]["camera_0.od_out"] = std::shared_ptr<void>();
 
   auto start_time = std::chrono::high_resolution_clock::now();
-  ret = ops.Process(raw_handle, in_batch, out_batch);
+  int ret = ops.Process(raw_handle, in_batch, out_batch);
   auto end_time = std::chrono::high_resolution_clock::now();
 
   double latency_ms =

diff --git a/demo/common/operator_runner.h b/demo/common/operator_runner.h
@@ -203,43 +203,39 @@ inline int ApplyOperatorControl(
 }
 
 /**
- * @brief 通用 Operator 单槽位生命周期与调度执行器
+ * @brief 统一校验配置、解析芯片与模型路径并创建 Operator 实例
+ * @return 0 成功，3 参数/配置错误，5 Operator 创建失败。
  */
-template <typename TInput, typename TOutput, typename TResultExtractor>
-int RunOperatorWithExtractor(
-    const DemoOptions& options, std::string_view input_slot,
-    std::string_view output_slot, const std::vector<TInput>& inputs,
-    TResultExtractor&& extractor,
-    std::vector<double>* out_latencies_ms = nullptr,
-    llm_edgeflow::operator_api::ControlCommand ctrl_cmd =
-        llm_edgeflow::operator_api::ControlCommand::kUpdateRules,
-    const char* default_ctrl_json = nullptr) {
+inline int CreateOperatorInstance(
+    const DemoOptions& options, std::string_view logger_prefix,
+    llm_edgeflow::operator_api::OperatorFunc* out_ops, void** out_handle) {
   using namespace llm_edgeflow::operator_api;
 
-  if (inputs.empty()) {
-    std::cerr << "[OperatorRunner ERROR] Inputs vector is empty." << std::endl;
-    return 4;
+  if (!out_ops || !out_handle) {
+    return 3;
   }
+  *out_handle = nullptr;
 
   std::string err;
   if (!ValidateConfigBizMatch(options.config_path, options.biz, &err)) {
-    std::cerr << "[OperatorRunner ERROR] Config validation failed: " << err
-              << std::endl;
+    std::cerr << "[" << logger_prefix
+              << " ERROR] Config validation failed: " << err << std::endl;
     return 3;
   }
 
   ComputePlatform chip_type = ComputePlatform::kUnknown;
   if (!ParseComputePlatform(options.chip, &chip_type)) {
-    std::cerr << "[OperatorRunner ERROR] Unsupported compute platform / chip: "
-              << options.chip << std::endl;
+    std::cerr << "[" << logger_prefix
+              << " ERROR] Unsupported compute platform / chip: " << options.chip
+              << std::endl;
     return 3;
   }
 
   std::string model_root;
   std::string cfg_rel;
   ResolveModelRootAndConfig(options.config_path, &model_root, &cfg_rel);
 
-  OperatorFunc ops = Get_LLM_EDGEFLOW_OperatorTable();
+  *out_ops = Get_LLM_EDGEFLOW_OperatorTable();
 
   int max_batch_size = options.batch_size > 0 ? options.batch_size : 1;
   uint32_t requested_depth = options.depth_num > 0 ? options.depth_num : 25;
@@ -254,22 +250,50 @@ int RunOperatorWithExtractor(
   param.compute_platform = chip_type;
   param.max_frame_depth = requested_depth;
 
-  void* raw_handle = nullptr;
-  int ret = ops.Create(&raw_handle, &param);
-  if (ret != 0 || !raw_handle) {
+  int ret = out_ops->Create(out_handle, &param);
+  if (ret != 0 || !*out_handle) {
     std::string op_err = GetOperatorLastError();
-    std::cerr << "[OperatorRunner ERROR] Failed ops.Create with conf: "
-              << options.config_path << " (Operator error: " << op_err << ")"
-              << std::endl;
+    std::cerr << "[" << logger_prefix
+              << " ERROR] Failed ops.Create with conf: " << options.config_path
+              << " (Operator error: " << op_err << ")" << std::endl;
     return 5;
   }
 
+  return 0;
+}
+
+/**
+ * @brief 通用 Operator 单槽位生命周期与调度执行器
+ */
+template <typename TInput, typename TOutput, typename TResultExtractor>
+int RunOperatorWithExtractor(
+    const DemoOptions& options, std::string_view input_slot,
+    std::string_view output_slot, const std::vector<TInput>& inputs,
+    TResultExtractor&& extractor,
+    std::vector<double>* out_latencies_ms = nullptr,
+    llm_edgeflow::operator_api::ControlCommand ctrl_cmd =
+        llm_edgeflow::operator_api::ControlCommand::kUpdateRules,
+    const char* default_ctrl_json = nullptr) {
+  using namespace llm_edgeflow::operator_api;
+
+  if (inputs.empty()) {
+    std::cerr << "[OperatorRunner ERROR] Inputs vector is empty." << std::endl;
+    return 4;
+  }
+
+  OperatorFunc ops{};
+  void* raw_handle = nullptr;
+  const int init_ret =
+      CreateOperatorInstance(options, "OperatorRunner", &ops, &raw_handle);
+  if (init_ret != 0) return init_ret;
+
   OperatorHandleGuard guard(ops, raw_handle);
 
   const int control_ret = ApplyOperatorControl(options, ops, raw_handle,
                                                ctrl_cmd, default_ctrl_json);
   if (control_ret != 0) return control_ret;
 
+  const int max_batch_size = options.batch_size > 0 ? options.batch_size : 1;
   size_t total_inputs = inputs.size();
   if (out_latencies_ms) {
     out_latencies_ms->assign(total_inputs, 0.0);
@@ -301,7 +325,7 @@ int RunOperatorWithExtractor(
               << in_key << " -> " << out_key << ")..." << std::endl;
 
     auto start_time = std::chrono::high_resolution_clock::now();
-    ret = ops.Process(raw_handle, in_batch, out_batch);
+    int ret = ops.Process(raw_handle, in_batch, out_batch);
     auto end_time = std::chrono::high_resolution_clock::now();
 
     double chunk_elapsed_ms =

diff --git a/demo/common/result_writer.h b/demo/common/result_writer.h
@@ -20,23 +20,6 @@ struct DemoSampleResult {
   nlohmann::json output;    // 业务自定义结果 JSON
 };
 
-/**
- * @brief 整体运行统计摘要
- */
-struct DemoRunSummary {
-  int schema_version = 1;
-  std::string profile;
-  std::string biz;
-  std::string config_path;
-  std::string dataset_path;
-  int total_samples = 0;
-  int success_count = 0;
-  int failed_count = 0;
-  double total_latency_ms = 0.0;
-  double avg_latency_ms = 0.0;
-  std::string error;
-};
-
 /**
  * @brief 结果落盘写入器 (负责原子落盘 JSONL 记录与 summary.json)
  */

diff --git a/demo/json_prompt_demo.py b/demo/json_prompt_demo.py
@@ -16,13 +16,24 @@
 ROOT = Path(__file__).resolve().parents[1]
 
 
+def dump_compact_json(data):
+    return json.dumps(data, ensure_ascii=False, separators=(",", ":"))
+
+
 def encode_request(payload):
     request = json.loads(payload)
     if not isinstance(request, dict):
         raise ValueError("Input must be a JSON object")
     # Only normalize whitespace for the line-based dataset reader. All fields
     # reach the SDK; query selection and validation belong to its Adapter.
-    return json.dumps(request, ensure_ascii=False, separators=(",", ":"))
+    return dump_compact_json(request)
+
+
+def prepare_requests(payloads):
+    texts = [encode_request(payload) for payload in payloads]
+    if not texts:
+        raise ValueError("Input dataset is empty")
+    return texts
 
 
 def collect_responses(result_file, count):
@@ -40,18 +51,13 @@ def collect_responses(result_file, count):
         if not isinstance(document, dict):
             raise ValueError("Demo result must contain a JSON response object")
         # Forward the complete SDK response. No business field projection here.
-        responses[request_id] = json.dumps(
-            document, ensure_ascii=False, separators=(",", ":")
-        )
+        responses[request_id] = dump_compact_json(document)
     return [responses[30001 + i] for i in range(count)]
 
 
-def run_demo(payloads, config, biz, work_dir, executable):
-    texts = [encode_request(payload) for payload in payloads]
-    if not texts:
-        raise ValueError("Input dataset is empty")
+def _run_demo_impl(requests, config, biz, work_dir, executable):
     dataset = work_dir / "input.txt"
-    dataset.write_text("\n".join(texts) + "\n", encoding="utf-8")
+    dataset.write_text("\n".join(requests) + "\n", encoding="utf-8")
     output_dir = work_dir / "results"
     command = [
         str(executable), "--biz", biz, "--config", str(config),
@@ -63,7 +69,12 @@ def run_demo(payloads, config, biz, work_dir, executable):
         result = subprocess.run(command, cwd=ROOT, stdout=log, stderr=subprocess.STDOUT)
     if result.returncode:
         raise ValueError(f"alg_demo failed (exit {result.returncode}); see {work_dir / 'demo.log'}")
-    return collect_responses(output_dir / biz / "results.jsonl", len(texts))
+    return collect_responses(output_dir / biz / "results.jsonl", len(requests))
+
+
+def run_demo(payloads, config, biz, work_dir, executable):
+    requests = prepare_requests(payloads)
+    return _run_demo_impl(requests, config, biz, work_dir, executable)
 
 
 def main(argv=None):
@@ -83,15 +94,12 @@ def main(argv=None):
         else:
             payloads = [args.input if args.input is not None else sys.stdin.read()]
         # Reject invalid requests before creating run artifacts or starting a model.
-        for payload in payloads:
-            encode_request(payload)
-        if not payloads:
-            raise ValueError("Input dataset is empty")
+        texts = prepare_requests(payloads)
         args.output_dir.mkdir(parents=True, exist_ok=True)
         work_dir = Path(tempfile.mkdtemp(prefix="run-", dir=args.output_dir.resolve()))
         print(f"Demo artifacts: {work_dir}", file=sys.stderr)
-        responses = run_demo(payloads, args.config, args.biz,
-                             work_dir, args.demo_bin.resolve())
+        responses = _run_demo_impl(texts, args.config, args.biz,
+                                   work_dir, args.demo_bin.resolve())
         # Validate the entire run before publishing any response.
         sys.stdout.write("\n".join(responses) + "\n")
         return 0

diff --git a/doc/CHANGELOG.md b/doc/CHANGELOG.md
@@ -1,5 +1,28 @@
 # Changelog
 
+## 2026-09-15 上线前代码与测试精简（批次 0、1、2）
+
+- **测试装配与覆盖收口（批次 0 / V1）**：
+  - 在 `cmake_ext/Tests.cmake` 中为 `FrameworkCoreTest`、`NodeBaseContractsTest` 和 `AdapterPurityTest` 补齐遗漏的 9 个 GoogleTest 套件。
+  - 增强 `test_test_labels_contract.py`，建立 CTest 过滤器与已编译测试二进制 `--gtest_list_tests` 清单集合核对机制，消除漏选测试套件风险。
+- **局部状态与冗余清理（批次 1 / A1–A4）**：
+  - 删除 `Pipeline` 中未读取的 `max_parallel_workers_` 状态与赋值，直接由配置控制线程池。
+  - 删除 `result_writer.h` 中未被消费的结构体 `DemoRunSummary` 及 `scaffold_custom_node.py` 中无调用的辅助包装。
+  - 优化 `PipelineValidator` 计划返回流程，通过 `void finish_plan(plan)` 配合 NRVO 消除全量计划容器深拷贝。
+  - 移除 Validator 中重复的 `if (definition)` 检查与无正常可达路径的 `node.config` 回退，统一以 `normalized_config_by_node.at(id)` 访问。
+  - 清理 7 处 Adapter 在 `IndexResults` 校验成功后回退内部 `req_id` 的冗余分支，简化 ID 恢复。
+- **共享实现与公共组件收口（批次 2 / B1–B3）**：
+  - 提取 `function_node.h` 中的 `ResolveBoundModelId` 与 `model_calls.h` 中的 `ConvertAlignedOutputs`，收口模型槽位绑定与对齐结果错误映射。
+  - 拆分 `adapter_batch.h` 中的前置校验（`ValidatePrimaryAndSpecs`）与多路对齐（`AlignAndIndexResults`），避免带输出参数重载的重复读取与校验，严格保持诊断优先级。
+  - 提取 `operator_runner.h` 的 `CreateOperatorInstance` 公共创建步骤并在 `ocr_doc_qa_demo.cpp` 复用；规范化 `json_prompt_demo.py` 请求预编码流程（`prepare_requests` / `_run_demo_impl`），避免 CLI 重复 parse/dump。
+- **测试重整与归属收口（T1–T6）**：
+  - 删除 4 个确认功能完全重复的旧测试用例并同步清理空 CTest 过滤器。
+  - 将 `NodeRegistryTest` 的 `TextChunkNode` 命名断言与非存在创建断言迁移至 `test_catalog_contract_ssot.cpp`。
+  - 将 `DAG` 缺少 `id`/`depends_on` 负向用例并入 `test_pipeline_config.cpp` 矩阵。
+  - 将 `FixedBatchExecutorStrictOutputsAndRollback` 迁移至专门的 `test_batch_executor.cpp`。
+  - 将 `test_quality_gate_contract.sh` 中的 sanitizer/ccache 规则断言完整并入 Python 契约测试，下线 shell 脚本。
+  - 将 `test_architecture_docs_drift_gate.sh` 的静态版本替换改为从根 `CMakeLists.txt` 动态提取，提升文档治理自测可靠性。
+
 ## 2026-09-14 投产前诊断身份与 Node 注册状态收敛（RFC-0058）
 
 - **统一诊断身份体系（B1）**：

diff --git a/doc/README.md b/doc/README.md
@@ -46,6 +46,7 @@
 | 需要了解什么 | 入口 |
 | --- | --- |
 | 尚待完成的开发者试用与生产验收 | [方案开发者验收计划](plans/solution_developer_acceptance.md) |
+| 上线前精简代码、兼容分支与重复测试 | [代码与测试精简实施计划（2026-09-15）](plans/prelaunch_simplification_2026-09-15.md) |
 | 降低方案编排心智负担的实施顺序与验收 | [RFC-0057：Pipeline 编排体验](rfcs/0057-pipeline-composition-experience.md)（In Implementation） |
 | 架构与接口为何这样设计 | [RFC 索引](rfcs/README.md)，优先列出进行中的 RFC |
 | 用户可感知的版本变化 | [Changelog](CHANGELOG.md) |