增强配置系统

.gitignore

@@ -94,4 +94,5 @@ run_profiles/*.yaml
 run_profiles/*.yml
 # Sync System Runtime Data
 _runtime/
-test_results/
+test_results/
+*.code-workspace

docs/factory_registry_refactor_plan.md

@@ -0,0 +1,147 @@
+# Factory / Registry / Adapter 初始化流程改造方案
+
+## 目标
+
+这次改造只做三件事：
+
+1. 简化 `factory`，让它只负责编排，不再关心过多介质细节。
+2. 统一 handler 初始化流程，让 `jsonl` / `db` / `api` 三类路径走同一套装配逻辑。
+3. 让适配层尽量极简：适配层只补充少量 handler / datasource 实现，不再复制初始化流程。
+
+## 当前问题
+
+### 1. `factory` 里存在大量按类型分支的重复逻辑
+
+当前 [sync_state_machine/pipeline/factory.py](sync_state_machine/pipeline/factory.py) 同时负责：
+
+- 创建 local / remote datasource
+- 判断 datasource 类型
+- 选择不同 handler getter
+- 注入一部分 handler 配置
+- 注册 handler
+
+这些逻辑对 local / remote 各写了一遍，重复明显。
+
+### 2. `DomainRegistry` 暴露了过多介质细节
+
+当前 registry 暴露：
+
+- `get_jsonl_handler()`
+- `get_db_handler()`
+- `get_api_handler()`
+
+这导致编排层必须知道 handler 是哪一类，从而在 `factory` 里出现大量 `if isinstance(...)`。
+
+### 3. handler 配置注入不统一
+
+目前只有 API handler 在 factory 里显式吃到了 `handler_configs`。
+
+- `api`：支持 per-side 配置
+- `db`：当前没有统一注入入口
+- `jsonl`：当前也没有统一注入入口
+
+这使得“同一类 datasource，两侧配置不同”只能在部分路径下工作。
+
+### 4. 适配层初始化成本偏高
+
+如果适配层要补充一种 handler，目前需要理解：
+
+- factory 的类型分支
+- registry 的不同 getter
+- 某些配置为什么只在 API 生效
+
+这和“适配层极简”的目标不一致。
+
+## 本次改造范围
+
+本次只做最小闭环改造，不碰更大的业务逻辑：
+
+### Step 1. 统一 `DomainRegistry` 的 handler 查询接口
+
+状态：已完成
+
+新增统一接口：
+
+- `get_handler(node_type, datasource_type)`
+
+保留旧接口作为兼容壳，但 factory 新逻辑只使用统一接口。
+
+### Step 2. 给所有 handler 基类增加统一配置注入入口
+
+状态：已完成
+
+新增统一入口：
+
+- `set_handler_config(config)`
+
+这样 `jsonl` / `db` / `api` 三类 handler 都可以在实例化后走同一套配置注入流程。
+
+### Step 3. 重写 factory 的 datasource / handler 初始化流程
+
+状态：已完成
+
+把当前分散的：
+
+- `_create_datasources()`
+- `_register_handlers()`
+- `_api_handler_kwargs()`
+
+收敛成按 endpoint 的统一初始化流程：
+
+- endpoint 创建 datasource
+- endpoint 根据 datasource type 向 registry 查询 handler class
+- endpoint 创建 handler
+- endpoint 统一注入：
+  - `handler_config`
+  - `target_project_ids`
+  - datasource-specific runtime object（如 api client）
+
+最终让 `factory` 只负责：
+
+- persistence 初始化
+- endpoint 初始化
+- strategy 初始化
+- pipeline 组装
+
+## 不在本次范围内
+
+以下内容先不动，只在文档中记为后续方向：
+
+1. 显式 bootstrap 内建 domain 与适配层 domain 的注册时机
+2. 用更强的 `EndpointConfig` / `DataSourceDriver` 抽象进一步收敛 factory
+3. 清理 `NodeHandlerRegistry` 这类历史遗留接口
+
+## 测试策略
+
+每一步都补对应 pytest，并保证通过：
+
+### Step 1 测试
+
+- registry 可以通过统一接口拿到对应 handler class
+- 旧接口仍可兼容使用
+
+### Step 2 测试
+
+- handler 可以收到统一的 `handler_config`
+- 同一 datasource 类型在 local / remote 两侧可以拿到不同配置
+
+### Step 3 测试
+
+- factory 不再依赖 `get_jsonl_handler/get_db_handler/get_api_handler`
+- 统一 endpoint 初始化路径可正常创建并注册 handler
+- 原有 db handler 注册测试仍通过
+
+## 预期结果
+
+改造完成后：
+
+1. `factory` 中关于 datasource / handler 初始化的重复逻辑显著减少。
+2. `DomainRegistry` 对编排层只暴露统一 handler 查询方式。
+3. handler 配置注入完全统一。
+4. 适配层只需：
+   - 提供 datasource（如需要）
+   - 提供 handler
+   - 注册 handler
+   - 调用统一 runner
+
+这就是“适配极简”的基础。

scripts/auto_test_domains/base.py

@@ -296,9 +296,14 @@ def build_temp_run_profile_for_project_auto_bind(
     profile.setdefault("remote_datasource", {})["target_project_ids"] = [remote_project_id]
 
     project_strategy = profile.setdefault("strategies", {}).setdefault("project", {}).setdefault("config", {})
-    project_strategy["auto_bind"] = True
-    project_strategy["auto_bind_fields"] = ["name"]
-    project_strategy["pre_bind_data_id"] = []
+    project_strategy["auto_bind"] = False
+    project_strategy["auto_bind_fields"] = []
+    project_strategy["pre_bind_data_id"] = [
+        {
+            "local_data_id": local_project_id,
+            "remote_data_id": remote_project_id,
+        }
+    ]
     project_strategy["local_orphan_action"] = project_local_orphan_action or "none"
     project_strategy["remote_orphan_action"] = project_remote_orphan_action or "none"
     project_strategy["update_direction"] = project_update_direction or "none"

scripts/auto_test_domains/project_case.py

@@ -141,9 +141,6 @@ def run_case(config: dict[str, Any], report_root) -> dict[str, Any]:
                     "investment_decision_date",
                     "complete_investment",
                     "dynamic_investment",
-                    "implementation_estimate",
-                    "static_total_investment",
-                    "completed_investment",
                 ],
             }
         },
@@ -164,9 +161,6 @@ def run_case(config: dict[str, Any], report_root) -> dict[str, Any]:
                         "investment_decision_date": "2021-08-15",
                         "complete_investment": 125000.0,
                         "dynamic_investment": 1413000.5,
-                        "implementation_estimate": 1225000.0,
-                        "static_total_investment": 1224300.0,
-                        "completed_investment": 194500.0,
                     }
                 }
             }
@@ -180,7 +174,7 @@ def run_case(config: dict[str, Any], report_root) -> dict[str, Any]:
                 "PUT /project/dynamic_investment",
                 "PUT /project/settlement_investment",
             ],
-            "notes": "单轮覆盖 project 剩余 4 条投资类更新接口。",
+            "notes": "单轮覆盖 project 剩余 4 条投资类更新接口。当前回归仅校验后端可稳定读回的投资决策/完成投资/动态投资字段；结算投资接口仍保留覆盖，但扩展字段读回语义以后端修正为准。",
         },
     )
     rounds.append(round_3)

scripts/auto_test_domains/project_detail_data_case.py

@@ -15,12 +15,14 @@ def run_case(config: dict[str, Any], report_root) -> dict[str, Any]:
      - `actual_construction` → `PUT /project/actual_construction`
      - `ensure_production` → `PUT /project/ensure_capacity`
      - `climb_production` → `PUT /project/climb_capacity`
-     - `challenge_production` → `PUT /project/challenge_capacity`
-     - `production` → `PUT /project/production_capacity`
+    - `challenge_production` → `PUT /project/challenge_capacity`
 
-     其中 `production` 在当前水电项目上主要验证“推送后 project 本身字段变更”；
-     `ensure/climb/challenge` 默认策略是 pull，本用例会在临时 profile 中仅对当前 key 切到 push，
+    `ensure/climb/challenge` 默认策略是 pull，本用例会在临时 profile 中仅对当前 key 切到 push，
      以确认对应后端更新接口真实可用，并在报告中展示字段校验结果。
+
+    `production` 当前按文档约束不混入这组统一回归：
+    后端确认该接口会直接更新 `project` 字段，但不保证创建/更新可轮询的
+    `project_detail_data.production` 记录，需单列方案再测。
     """
 
     dataset_root = f"{config['project_root']}/tests/fixtures/auto_sync/datasource/project_detail_data"
@@ -31,14 +33,12 @@ def run_case(config: dict[str, Any], report_root) -> dict[str, Any]:
     ]
     remote_project_id = "f3e02e84-e81c-4aa6-88d4-f5aa108c8227"
     remote_actual_construction_id = "c99ae2e4-809e-47dd-8b86-61674b6a5bea"
-    remote_production_id = "0b3f7e2d-9dc7-4f71-aac9-c9e7093df64f"
     seed_ensure_production_id = "11111111-1111-4111-8111-111111111111"
     seed_climb_production_id = "22222222-2222-4222-8222-222222222222"
     seed_challenge_production_id = "33333333-3333-4333-8333-333333333333"
     seed_plan_construction_id = "4bc5e9db-c9dc-41f1-a77d-574b4fe37b01"
     local_project_id = make_local_data_id("project", remote_project_id)
     local_actual_construction_id = make_local_data_id("project_detail_data", remote_actual_construction_id)
-    local_production_id = make_local_data_id("project_detail_data", remote_production_id)
     local_plan_construction_id = make_local_data_id("project_detail_data", seed_plan_construction_id)
     local_ensure_production_id = make_local_data_id("project_detail_data", seed_ensure_production_id)
     local_climb_production_id = make_local_data_id("project_detail_data", seed_climb_production_id)
@@ -365,63 +365,8 @@ def run_case(config: dict[str, Any], report_root) -> dict[str, Any]:
     )
     rounds.append(round_7)
 
-    round_4 = run_pipeline_round(
-        config=config,
-        dataset_dir=f"{dataset_root}/round_4_production_update",
-        previous_dataset_dir=f"{dataset_root}/round_3_plan_construction_create",
-        base_dataset_dir=filtered_root,
-        base_dataset_files=base_dataset_files,
-        domain_name="project_detail_data",
-        round_name="round_4_production_update",
-        purpose="推送 production，验证 /project/production_capacity 在当前水电项目上会更新 project 投产计划字段",
-        expected_changes={
-            "project": {
-                local_project_id: [
-                    "plan_production_date",
-                    "plan_full_production_date",
-                ],
-            }
-        },
-        required_bindings={
-            "project": [local_project_id],
-            "project_detail_data": [local_production_id],
-        },
-        expected_binding_pairs={
-            "project": {local_project_id: remote_project_id},
-        },
-        require_distinct_binding_ids={
-            "project": [local_project_id],
-            "project_detail_data": [local_production_id],
-        },
-        expected_bound_fields={
-            "remote": {
-                "project": {
-                    local_project_id: {
-                        "plan_production_date": "2026-04-01",
-                        "plan_full_production_date": "2026-06-01",
-                    }
-                },
-                "project_detail_data": {
-                    local_production_id: {
-                        "key": "production",
-                        "value.0.capacity": 51.0,
-                        "value.0.date": "2025-12-27",
-                    }
-                },
-            }
-        },
-        project_id_remap=project_id_remap,
-        report_root=report_root,
-        coverage={
-            "keys": ["production"],
-            "api": ["PUT /project/production_capacity"],
-            "notes": "在可访问项目上复用现有 production 明细记录，验证生产接口会更新 project 投产计划字段并保持 detail 绑定可解析。",
-        },
-    )
-    rounds.append(round_4)
-
     return build_domain_report(
         domain="project_detail_data",
-        description="项目详情 domain 回归。覆盖 plan_construction / actual_construction / ensure_production / climb_production / challenge_production / production 六类 key，并在报告中展示对应接口与 project 字段联动校验。",
+        description="项目详情 domain 回归。覆盖 plan_construction / actual_construction / ensure_production / climb_production / challenge_production 五类 key，并在报告中展示对应接口与 project 字段联动校验。production 按文档约束单列，不混入本组统一回归。",
         rounds=rounds,
     )

sync_state_machine/__init__.py

@@ -14,14 +14,22 @@ from . import domain as _domain
 from .common.registry import DomainRegistry
 from .config import (
 	ApiDataSourceConfig,
+	BaseDataSourceConfig,
+	DbDataSourceConfig,
 	JsonlDataSourceConfig,
 	PipelineRunConfig,
 	build_config,
 	build_default_config,
+	get_datasource_config_model,
+	get_datasource_factory,
+	list_registered_datasource_types,
 	load_named_preset_overrides,
 	load_overrides_from_file,
+	register_datasource_config_model,
+	register_datasource_factory,
+	register_datasource_type,
 )
-from .datasource import ApiClient, ApiDataSource, BaseApiHandler, BaseJsonlHandler, JsonlDataSource
+from .datasource import ApiClient, ApiDataSource, BaseApiHandler, BaseDbHandler, BaseJsonlHandler, DbDataSource, JsonlDataSource
 from .pipeline import FullSyncPipeline, create_pipeline_from_config, run_pipeline_from_config
 
 try:
@@ -41,15 +49,25 @@ __all__ = [
 	"DomainRegistry",
 	"PipelineRunConfig",
 	"ApiDataSourceConfig",
+	"BaseDataSourceConfig",
+	"DbDataSourceConfig",
 	"JsonlDataSourceConfig",
 	"build_config",
 	"build_default_config",
+	"get_datasource_config_model",
+	"get_datasource_factory",
+	"list_registered_datasource_types",
 	"load_overrides_from_file",
 	"load_named_preset_overrides",
+	"register_datasource_config_model",
+	"register_datasource_factory",
+	"register_datasource_type",
 	"ApiClient",
 	"ApiDataSource",
+	"DbDataSource",
 	"JsonlDataSource",
 	"BaseApiHandler",
+	"BaseDbHandler",
 	"BaseJsonlHandler",
 	"FullSyncPipeline",
 	"create_pipeline_from_config",

sync_state_machine/common/registry.py

@@ -48,6 +48,7 @@ class DomainRegistration:
     jsonl_handler_class: Optional[Type[Any]] = None
     api_handler_class: Optional[Type[Any]] = None
     db_handler_class: Optional[Type[Any]] = None
+    handler_classes: Dict[str, Type[Any]] = field(default_factory=dict)
     metadata: Dict[str, Any] = field(default_factory=dict)  # 额外元数据
 
 
@@ -64,6 +65,12 @@ class DomainRegistry:
     """
     
     _registry: Dict[str, DomainRegistration] = {}
+
+    _BUILTIN_HANDLER_ATTRS: Dict[str, str] = {
+        "jsonl": "jsonl_handler_class",
+        "api": "api_handler_class",
+        "db": "db_handler_class",
+    }
     
     @classmethod
     def register(
@@ -96,7 +103,7 @@ class DomainRegistry:
         if node_type in cls._registry:
             raise ValueError(f"Domain '{node_type}' already registered")
         
-        cls._registry[node_type] = DomainRegistration(
+        reg = DomainRegistration(
             node_type=node_type,
             schema=schema,
             node_class=node_class,
@@ -106,6 +113,12 @@ class DomainRegistry:
             db_handler_class=db_handler_class,
             metadata=metadata
         )
+        for datasource_type, attr_name in cls._BUILTIN_HANDLER_ATTRS.items():
+            handler_class = getattr(reg, attr_name)
+            if handler_class is not None:
+                reg.handler_classes[datasource_type] = handler_class
+
+        cls._registry[node_type] = reg
     
     @classmethod
     def get_node_class(cls, node_type: str) -> Type[Any]:
@@ -143,36 +156,40 @@ class DomainRegistry:
         return reg.strategy_class
     
     @classmethod
-    def get_jsonl_handler(cls, node_type: str) -> Optional[Type[Any]]:
-        """
-        获取 JSONL Handler 类（用于 DataSource）
-        
-        Args:
-            node_type: 节点类型
-            
-        Returns:
-            Handler 类，如果未注册则返回 None
-        """
+    def get_handler(
+        cls,
+        node_type: str,
+        datasource_type: str,
+    ) -> Optional[Type[Any]]:
+        """按 datasource 类型获取 Handler 类。"""
         reg = cls._registry.get(node_type)
         if not reg:
             raise ValueError(f"Unknown node_type: '{node_type}'")
-        return reg.jsonl_handler_class
-    
+        handler_class = reg.handler_classes.get(datasource_type)
+        if handler_class is not None:
+            return handler_class
+
+        attr_name = cls._BUILTIN_HANDLER_ATTRS.get(datasource_type)
+        if attr_name is None:
+            return None
+        return getattr(reg, attr_name)
+
     @classmethod
-    def get_api_handler(cls, node_type: str) -> Optional[Type[Any]]:
-        """获取 API Handler 类"""
+    def register_handler(cls, node_type: str, datasource_type: str, handler_class: Type[Any]) -> None:
+        """为已注册 domain 补充任意 datasource 类型的 Handler。"""
         reg = cls._registry.get(node_type)
         if not reg:
             raise ValueError(f"Unknown node_type: '{node_type}'")
-        return reg.api_handler_class
-    
+
+        reg.handler_classes[datasource_type] = handler_class
+        attr_name = cls._BUILTIN_HANDLER_ATTRS.get(datasource_type)
+        if attr_name is not None:
+            setattr(reg, attr_name, handler_class)
+
     @classmethod
-    def get_db_handler(cls, node_type: str) -> Optional[Type[Any]]:
-        """获取 DB Handler 类"""
-        reg = cls._registry.get(node_type)
-        if not reg:
-            raise ValueError(f"Unknown node_type: '{node_type}'")
-        return reg.db_handler_class
+    def register_db_handler(cls, node_type: str, db_handler_class: Type[Any]) -> None:
+        """为已注册 domain 补充 DB Handler。"""
+        cls.register_handler(node_type, "db", db_handler_class)
     
     @classmethod
     def get_schema(cls, node_type: str) -> Type[BaseModel]:
@@ -229,6 +246,11 @@ class DomainRegistry:
             print(f"   JSONL Handler:    {reg.jsonl_handler_class.__name__ if reg.jsonl_handler_class else '❌ None'}")
             print(f"   API Handler:      {reg.api_handler_class.__name__ if reg.api_handler_class else '❌ None'}")
             print(f"   DB Handler:       {reg.db_handler_class.__name__ if reg.db_handler_class else '❌ None'}")
+            extra_handler_types = sorted(
+                handler_type for handler_type in reg.handler_classes if handler_type not in cls._BUILTIN_HANDLER_ATTRS
+            )
+            if extra_handler_types:
+                print(f"   Extra Handlers:   {extra_handler_types}")
             if reg.metadata:
                 print(f"   Metadata:         {reg.metadata}")
             print()
@@ -236,7 +258,7 @@ class DomainRegistry:
         print(f"{'='*80}\n")
     
     @classmethod
-    def get_summary(cls) -> Dict[str, Dict[str, bool]]:
+    def get_summary(cls) -> Dict[str, Dict[str, Any]]:
         """
         获取注册摘要（用于自动化检查）
         
@@ -263,5 +285,6 @@ class DomainRegistry:
                 "has_jsonl_handler": reg.jsonl_handler_class is not None,
                 "has_api_handler": reg.api_handler_class is not None,
                 "has_db_handler": reg.db_handler_class is not None,
+                "handler_types": sorted(reg.handler_classes.keys()),
             }
         return summary

sync_state_machine/config/__init__.py

@@ -1,6 +1,14 @@
-from .datasource_config import ApiDataSourceConfig, JsonlDataSourceConfig, DataSourceConfig
+from .datasource_config import ApiDataSourceConfig, BaseDataSourceConfig, DbDataSourceConfig, JsonlDataSourceConfig, DataSourceConfig
 from .persist_config import PersistConfig
 from .logging_config import LoggingConfig, resolve_log_file_path
+from ..datasource.type_registry import (
+    get_datasource_config_model,
+    get_datasource_factory,
+    list_registered_datasource_types,
+    register_datasource_config_model,
+    register_datasource_factory,
+    register_datasource_type,
+)
 from .run_presets import (
     list_preset_names,
     get_preset_file_candidates,
@@ -26,11 +34,19 @@ from .builder import (
 
 __all__ = [
     "ApiDataSourceConfig",
+    "BaseDataSourceConfig",
+    "DbDataSourceConfig",
     "JsonlDataSourceConfig",
     "PipelineRunConfig",
     "DataSourceConfig",
+    "get_datasource_config_model",
+    "get_datasource_factory",
     "PersistConfig",
     "LoggingConfig",
+    "list_registered_datasource_types",
+    "register_datasource_config_model",
+    "register_datasource_factory",
+    "register_datasource_type",
     "resolve_log_file_path",
     "list_preset_names",
     "get_preset_file_candidates",

sync_state_machine/config/builder.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import json
 from pathlib import Path
 from typing import Any, Dict, Mapping, Optional, Union
 
@@ -9,6 +8,7 @@ from pydantic import BaseModel
 from ..common.registry import DomainRegistry
 from .datasource_config import DataSourceConfig
 from .pipeline_config import PipelineRunConfig
+from .run_presets import load_overrides_from_file
 from .strategy_config import ConfigPresets, StrategyRuntimeConfig
 
 
@@ -148,7 +148,7 @@ def apply_env_overrides(
     dotenv_file: Optional[Union[str, Path]] = None,
 ) -> PipelineRunConfig:
     raise NotImplementedError(
-        "Environment-based overrides are removed. Use JSON config files via overrides_file instead."
+        "Environment-based overrides are removed. Use config files via overrides_file instead."
     )
 
 
@@ -162,7 +162,7 @@ def build_config(
     dotenv_file: Optional[Union[str, Path]] = None,
 ) -> PipelineRunConfig:
     if dotenv_file is not None:
-        raise ValueError("dotenv_file is no longer supported; use overrides_file JSON instead")
+        raise ValueError("dotenv_file is no longer supported; use overrides_file instead")
 
     config = build_default_config(project_root)
 
@@ -180,7 +180,7 @@ def build_config(
 
     if overrides_file:
         override_path = Path(overrides_file)
-        file_data = json.loads(override_path.read_text(encoding="utf-8"))
+        file_data = load_overrides_from_file(override_path, project_root)
         config = apply_config_overrides(config, file_data)
 
     if overrides:

sync_state_machine/config/datasource_config.py

@@ -1,22 +1,33 @@
 from __future__ import annotations
 
-from typing import Annotated, Any, Dict, List, Literal, Union
+from typing import Annotated, Any, Dict, List, Literal
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, PlainSerializer, PlainValidator
+
+from ..datasource.type_registry import register_datasource_config_model, validate_registered_datasource_config
 
 
-class JsonlDataSourceConfig(BaseModel):
+class BaseDataSourceConfig(BaseModel):
     model_config = ConfigDict(extra="forbid", validate_assignment=True)
 
-    type: Literal["jsonl"] = "jsonl"
-    jsonl_dir: str
-    read_only: bool = False
+    type: str
     target_project_ids: List[str] = Field(default_factory=list)
     handler_configs: Dict[str, Dict[str, Any]] = Field(default_factory=dict)
 
 
-class ApiDataSourceConfig(BaseModel):
-    model_config = ConfigDict(extra="forbid", validate_assignment=True)
+class JsonlDataSourceConfig(BaseDataSourceConfig):
+
+    type: Literal["jsonl"] = "jsonl"
+    jsonl_dir: str
+    read_only: bool = False
+
+
+class DbDataSourceConfig(BaseDataSourceConfig):
+
+    type: Literal["db"] = "db"
+
+
+class ApiDataSourceConfig(BaseDataSourceConfig):
 
     type: Literal["api"] = "api"
     api_base_url: str = ""
@@ -28,7 +39,18 @@ class ApiDataSourceConfig(BaseModel):
     api_rate_limit_max_requests: int = 10
     api_rate_limit_window_seconds: float = 10.0
     target_project_ids: List[str] = Field(min_length=1)
-    handler_configs: Dict[str, Dict[str, Any]] = Field(default_factory=dict)
 
 
-DataSourceConfig = Annotated[Union[JsonlDataSourceConfig, ApiDataSourceConfig], Field(discriminator="type")]
+DataSourceConfig = Annotated[
+    BaseDataSourceConfig,
+    PlainValidator(validate_registered_datasource_config),
+    PlainSerializer(
+        lambda value: value.model_dump(mode="python") if isinstance(value, BaseModel) else value,
+        return_type=dict[str, Any],
+    ),
+]
+
+
+register_datasource_config_model("jsonl", JsonlDataSourceConfig)
+register_datasource_config_model("db", DbDataSourceConfig)
+register_datasource_config_model("api", ApiDataSourceConfig)

sync_state_machine/config/run_presets.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import importlib
 from pathlib import Path
 from typing import Any, Dict, List, Tuple
 
@@ -28,6 +29,43 @@ def _replace_placeholders(value: Any, project_root: Path) -> Any:
     return value
 
 
+def _import_from_path(import_path: str) -> Any:
+    module_path, sep, attr_path = import_path.partition(":")
+    if not module_path:
+        raise ValueError(f"Invalid import path: {import_path}")
+
+    module = importlib.import_module(module_path)
+    if not sep:
+        return module
+
+    obj: Any = module
+    for attr_name in attr_path.split("."):
+        obj = getattr(obj, attr_name)
+    return obj
+
+
+def _apply_datasource_bootstraps(payload: Dict[str, Any]) -> Dict[str, Any]:
+    extensions = payload.pop("extensions", None)
+    if extensions is None:
+        return payload
+    if not isinstance(extensions, dict):
+        raise ValueError("extensions must be an object mapping")
+
+    bootstrap_paths = extensions.get("datasource_bootstraps") or []
+    if not isinstance(bootstrap_paths, list):
+        raise ValueError("extensions.datasource_bootstraps must be a list")
+
+    for item in bootstrap_paths:
+        if not isinstance(item, str) or not item.strip():
+            raise ValueError("extensions.datasource_bootstraps entries must be non-empty strings")
+        loaded = _import_from_path(item)
+        if not callable(loaded):
+            raise TypeError(f"datasource bootstrap must resolve to a callable: {item}")
+        loaded()
+
+    return payload
+
+
 def list_preset_names() -> List[str]:
     return sorted(PRESET_FILES.keys())
 
@@ -77,7 +115,7 @@ def load_overrides_from_file(file_path: Path, project_root: Path) -> Dict[str, A
     resolved = _replace_placeholders(payload, project_root)
     if not isinstance(resolved, dict):
         raise ValueError(f"Preset file must contain object mapping: {file_path}")
-    return resolved
+    return _apply_datasource_bootstraps(resolved)
 
 
 def load_named_preset_overrides(project_root: Path, preset_name: str) -> Tuple[Dict[str, Any], Path, bool]:

sync_state_machine/config/runtime_config.py

@@ -3,7 +3,7 @@
 Prefer importing from `sync_state_machine.config` directly.
 """
 
-from .datasource_config import ApiDataSourceConfig, JsonlDataSourceConfig, DataSourceConfig
+from .datasource_config import ApiDataSourceConfig, DbDataSourceConfig, JsonlDataSourceConfig, DataSourceConfig
 from .persist_config import PersistConfig
 from .logging_config import LoggingConfig
 from .strategy_config import (
@@ -24,6 +24,7 @@ from .builder import (
 
 __all__ = [
     "ApiDataSourceConfig",
+    "DbDataSourceConfig",
     "JsonlDataSourceConfig",
     "DataSourceConfig",
     "PersistConfig",

sync_state_machine/datasource/__init__.py

@@ -6,6 +6,14 @@ Handler 负责类型转换和业务逻辑。
 """
 
 from .datasource import BaseDataSource
+from .type_registry import (
+    get_datasource_config_model,
+    get_datasource_factory,
+    list_registered_datasource_types,
+    register_datasource_config_model,
+    register_datasource_factory,
+    register_datasource_type,
+)
 from .handler import (
     NodeHandler, 
     BaseNodeHandler, 
@@ -17,15 +25,24 @@ from .task_result import TaskResult, TaskStatus
 # API DataSource
 from .api import ApiClient, ApiDataSource, BaseApiHandler
 
+# DB DataSource
+from .db import DbDataSource, BaseDbHandler
+
 # JSONL DataSource  
 from .jsonl import JsonlDataSource, BaseJsonlHandler
 
 __all__ = [
     # 基础类
     "BaseDataSource",
+    "get_datasource_config_model",
+    "get_datasource_factory",
+    "list_registered_datasource_types",
     "NodeHandler",
     "BaseNodeHandler",
     "NodeHandlerRegistry",
+    "register_datasource_config_model",
+    "register_datasource_factory",
+    "register_datasource_type",
     "TaskResult",
     "TaskStatus",
     "ValidationResult",
@@ -34,6 +51,10 @@ __all__ = [
     "ApiClient",
     "ApiDataSource",
     "BaseApiHandler",
+
+    # DB DataSource
+    "DbDataSource",
+    "BaseDbHandler",
     
     # JSONL DataSource
     "JsonlDataSource",

sync_state_machine/datasource/api/handler.py

@@ -44,6 +44,7 @@ class BaseApiHandler(NodeHandler[T]):
         self.api_client: 'ApiClient' = None  # type: ignore  # 由 DataSource 注入
         self.poll_mode: str = "async"  # async | sync
         self._collection: 'DataCollection' = None # type: ignore  # 由 DataSource 注入
+        self.handler_config: Dict[str, Any] = {}
         # 子类在 __init__ 中设置此列表以声明 create/update 中涉及的 Pydantic schema 类；
         # post-check 将只比较这些 schema 覆盖的字段，过滤后端只读的派生字段。
         self.update_schemas: List[type] = []
@@ -63,6 +64,13 @@ class BaseApiHandler(NodeHandler[T]):
     def set_collection(self, collection: 'DataCollection') -> None:
         """设置 Collection（由 DataSource 调用）"""
         self._collection = collection
+
+    def set_handler_config(self, config: Dict[str, Any]) -> None:
+        self.handler_config = dict(config)
+
+    def set_target_project_ids(self, target_project_ids: List[str]) -> None:
+        """默认忽略项目过滤列表，具体子类按需覆盖。"""
+        return
     
     @property
     def node_type(self) -> str:

sync_state_machine/datasource/db/__init__.py

@@ -0,0 +1,6 @@
+"""DB DataSource 模块。"""
+
+from .datasource import DbDataSource
+from .handler import BaseDbHandler
+
+__all__ = ["DbDataSource", "BaseDbHandler"]

sync_state_machine/datasource/db/datasource.py

@@ -0,0 +1,22 @@
+"""DbDataSource - 数据库数据源协调层。"""
+
+from __future__ import annotations
+
+from ..datasource import BaseDataSource
+
+
+class DbDataSource(BaseDataSource):
+    """基于数据库 Handler 的数据源实现。"""
+
+    async def close(self) -> None:
+        return None
+
+    def _on_node_loaded(self, handler, node) -> None:
+        node.append_log(
+            f"DB链路[load] node_type={node.node_type} data_id={node.data_id or '<empty>'}"
+        )
+
+    def _on_in_progress_result(self, handler, node, result) -> None:
+        node.append_log(
+            f"DB链路[submit] action={node.action.value} data_id={node.data_id or '<empty>'}"
+        )

sync_state_machine/datasource/db/handler.py

@@ -0,0 +1,68 @@
+"""BaseDbHandler - DB Handler 基类。"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional, Set, TypeVar, TYPE_CHECKING
+
+from pydantic import BaseModel
+
+from ...common.type_safety import get_pydantic_model_fields
+from ..handler import BaseNodeHandler
+
+if TYPE_CHECKING:
+    from ...common.sync_node import SyncNode
+    from .datasource import DbDataSource
+
+T = TypeVar("T", bound=BaseModel)
+
+
+class BaseDbHandler(BaseNodeHandler[T]):
+    """数据库 Handler 基类。"""
+
+    def __init__(
+        self,
+        datasource: "DbDataSource",
+        node_type: str,
+        node_class: type["SyncNode[T]"],
+        schema: type[T],
+    ):
+        super().__init__(node_type, node_class, schema)
+        self.datasource = datasource
+        self.update_schemas: List[type] = []
+        self._target_project_ids: Optional[Set[str]] = None
+
+    def set_target_project_ids(self, target_project_ids: List[str]) -> None:
+        if target_project_ids:
+            self._target_project_ids = {str(item) for item in target_project_ids if str(item)}
+        else:
+            self._target_project_ids = None
+
+    def get_update_fields(self, *, exclude: frozenset = frozenset({"id"})) -> List[str]:
+        fields: set[str] = set()
+        for schema in self.update_schemas:
+            for field_name in get_pydantic_model_fields(schema):
+                if field_name not in exclude:
+                    fields.add(field_name)
+        return sorted(fields)
+
+    def _create_basic_node(self, data: Dict[str, Any], depend_ids: Optional[List[str]] = None) -> "SyncNode[T]":
+        data_id = self.extract_id(data)
+        if self._collection is None:
+            raise RuntimeError("Collection not set. Call set_collection() before creating nodes.")
+
+        if data_id:
+            existing_node = self._collection.get_by_data_id(self.node_type, data_id)
+            if existing_node is not None:
+                existing_node.depend_ids = list(depend_ids or [])
+                existing_node.set_data(data.copy())
+                existing_node.set_origin_data(data.copy())
+                return existing_node
+
+        node_id = self._collection.generate_node_id()
+        return self.node_class(
+            node_id=node_id,
+            data_id=data_id,
+            data=data.copy(),
+            depend_ids=depend_ids or [],
+            origin_data=data.copy(),
+        )

sync_state_machine/datasource/handler.py

@@ -70,6 +70,10 @@ class NodeHandler(Protocol[T]):
         """
         ...
 
+    def set_handler_config(self, config: Dict[str, Any]) -> None:
+        """注入 handler 级别配置。默认可忽略。"""
+        ...
+
     def set_target_project_ids(self, target_project_ids: List[str]) -> None:
         """注入项目过滤列表。默认可忽略。"""
         ...
@@ -345,6 +349,7 @@ class BaseNodeHandler(ABC, Generic[T]):
         self._node_class = node_class
         self._schema = schema
         self._collection: Optional["DataCollection"] = None
+        self.handler_config: Dict[str, Any] = {}
     
     def set_collection(self, collection: "DataCollection") -> None:
         """
@@ -359,6 +364,10 @@ class BaseNodeHandler(ABC, Generic[T]):
         """默认忽略项目过滤能力。"""
         return
 
+    def set_handler_config(self, config: Dict[str, Any]) -> None:
+        """默认保存 handler 级别配置，供子类按需读取。"""
+        self.handler_config = dict(config)
+
     def set_api_client(self, client: "ApiClient") -> None:
         """默认忽略 API 客户端注入。"""
         return
@@ -491,6 +500,11 @@ class CompatibleNodeHandler(Generic[T]):
     def set_collection(self, collection: "DataCollection") -> None:
         self._handler.set_collection(collection)
 
+    def set_handler_config(self, config: Dict[str, Any]) -> None:
+        method = getattr(self._handler, "set_handler_config", None)
+        if callable(method):
+            method(config)
+
     def set_target_project_ids(self, target_project_ids: List[str]) -> None:
         method = getattr(self._handler, "set_target_project_ids", None)
         if callable(method):

sync_state_machine/datasource/type_registry.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from pydantic import BaseModel
+
+
+DataSourceFactory = Callable[..., Any | Awaitable[Any]]
+
+
+_CONFIG_MODELS: dict[str, type[BaseModel]] = {}
+_FACTORIES: dict[str, DataSourceFactory] = {}
+
+
+def _normalize_type_name(type_name: str) -> str:
+    normalized = str(type_name).strip()
+    if not normalized:
+        raise ValueError("datasource type name must not be empty")
+    return normalized
+
+
+def register_datasource_config_model(
+    type_name: str,
+    config_model: type[BaseModel],
+    *,
+    replace: bool = False,
+) -> None:
+    normalized = _normalize_type_name(type_name)
+    existing = _CONFIG_MODELS.get(normalized)
+    if existing is not None and existing is not config_model and not replace:
+        raise ValueError(f"datasource config model already registered for type: {normalized}")
+    _CONFIG_MODELS[normalized] = config_model
+
+
+def register_datasource_factory(
+    type_name: str,
+    factory: DataSourceFactory,
+    *,
+    replace: bool = False,
+) -> None:
+    normalized = _normalize_type_name(type_name)
+    existing = _FACTORIES.get(normalized)
+    if existing is not None and existing is not factory and not replace:
+        raise ValueError(f"datasource factory already registered for type: {normalized}")
+    _FACTORIES[normalized] = factory
+
+
+def register_datasource_type(
+    type_name: str,
+    *,
+    config_model: type[BaseModel] | None = None,
+    factory: DataSourceFactory | None = None,
+    replace: bool = False,
+) -> None:
+    if config_model is None and factory is None:
+        raise ValueError("register_datasource_type() requires config_model, factory, or both")
+
+    if config_model is not None:
+        register_datasource_config_model(type_name, config_model, replace=replace)
+    if factory is not None:
+        register_datasource_factory(type_name, factory, replace=replace)
+
+
+def get_datasource_config_model(type_name: str) -> type[BaseModel] | None:
+    return _CONFIG_MODELS.get(_normalize_type_name(type_name))
+
+
+def get_datasource_factory(type_name: str) -> DataSourceFactory | None:
+    return _FACTORIES.get(_normalize_type_name(type_name))
+
+
+def list_registered_datasource_types() -> list[str]:
+    return sorted(set(_CONFIG_MODELS) | set(_FACTORIES))
+
+
+def validate_registered_datasource_config(value: Any) -> BaseModel:
+    if isinstance(value, BaseModel):
+        payload = value.model_dump(mode="python")
+    elif isinstance(value, dict):
+        payload = dict(value)
+    else:
+        raise TypeError(f"datasource config must be a mapping or BaseModel, got {type(value).__name__}")
+
+    type_name = payload.get("type")
+    if not isinstance(type_name, str) or not type_name.strip():
+        raise ValueError("datasource config must define a non-empty string field 'type'")
+
+    config_model = get_datasource_config_model(type_name)
+    if config_model is None:
+        available = ", ".join(list_registered_datasource_types()) or "<none>"
+        raise ValueError(f"Unsupported datasource type: '{type_name}'. Registered types: {available}")
+
+    return config_model.model_validate(payload)

sync_state_machine/domain/material_detail/api_handler.py

@@ -188,7 +188,7 @@ class MaterialDetailApiHandler(BaseApiHandler):
                 )
                 # 整组成功
                 for node, _ in group_items:
-                    results.append(TaskResult.success(node_id=node.node_id))
+                    results.append(TaskResult.in_progress(node_id=node.node_id, task_id=push_id))
             except Exception as e:
                 # 整组失败
                 error_msg = str(e)

sync_state_machine/domain/project/api_handler.py

@@ -18,6 +18,7 @@ PUT (6个更新接口):
 不支持: CREATE, DELETE
 """
 
+import asyncio
 from typing import List, Dict, Any
 from ...datasource.api.handler import BaseApiHandler
 from ...datasource.task_result import TaskResult
@@ -35,6 +36,35 @@ from schemas.project.project_base import (
 from schemas.project.project import ProjectInfoResponse
 
 
+_PROJECT_EXTENSION_FIELD_NAMES = {
+    "implementation_estimate",
+    "static_total_investment",
+    "completed_investment",
+    "total_contract_quantity",
+    "completed_settlement_quantity",
+}
+
+
+def _merge_project_all_info_fields(
+    project_data: Dict[str, Any],
+    all_info: Dict[str, Any],
+) -> Dict[str, Any]:
+    """把 all_info 中项目扩展字段拍平到 project 主数据上。"""
+    merged = dict(project_data)
+
+    for key, value in all_info.items():
+        if not key.endswith("_extension") or not isinstance(value, dict):
+            continue
+        for field_name in _PROJECT_EXTENSION_FIELD_NAMES:
+            if field_name not in value:
+                continue
+            current_value = merged.get(field_name)
+            if current_value in (None, ""):
+                merged[field_name] = value.get(field_name)
+
+    return merged
+
+
 class ProjectApiHandler(BaseApiHandler):
     """项目 API Handler"""
     
@@ -136,12 +166,15 @@ class ProjectApiHandler(BaseApiHandler):
             
             # 处理所有项目
             for project_data in projects:
-                node = self._create_node(project_data.model_dump())
-                # 设置 context
-                node.context["project_id"] = project_data.id
-                
                 # 预加载并缓存 all_info，同时存储到 node.context
                 all_info = await self.get_all_info(self.api_client, project_data.id)
+                normalized_project_data = _merge_project_all_info_fields(
+                    project_data.model_dump(),
+                    all_info,
+                )
+                node = self._create_node(normalized_project_data)
+                # 设置 context
+                node.context["project_id"] = project_data.id
                 node.context["all_info"] = all_info
                 
                 nodes.append(node)
@@ -217,10 +250,12 @@ class ProjectApiHandler(BaseApiHandler):
                 
                 # 执行更新（差异已由 _get_schema_diff 打印）
                 try:
-                    push_id = self._generate_push_id()
-                    # project 需要使用 _filter_update_data 返回的完整 schema 数据
                     full_update_data = self._filter_update_data(data, original_data, schema)
-                    await api_func(self.api_client, full_update_data, push_id, biz_id)
+                    await self._submit_update_and_wait(
+                        api_func=api_func,
+                        data=full_update_data,
+                        biz_id=biz_id,
+                    )
                     node_updated = True
                 except Exception as e:
                     error = str(e)
@@ -249,13 +284,39 @@ class ProjectApiHandler(BaseApiHandler):
     async def poll_tasks(self, task_ids: List[str]) -> Dict[str, TaskResult]:
         """轮询异步任务状态"""
         return await super().poll_tasks(task_ids)
+
+    async def _submit_update_and_wait(self, *, api_func, data: Dict[str, Any], biz_id: str) -> None:
+        push_id = self._generate_push_id()
+        await api_func(self.api_client, data, push_id, biz_id)
+
+        max_attempts = 20
+        poll_interval = 0.2
+        last_result: TaskResult | None = None
+
+        for attempt in range(max_attempts):
+            polled = await super().poll_tasks([push_id])
+            result = polled.get(push_id)
+            last_result = result
+
+            if result is None:
+                raise RuntimeError(f"Project update poll missing result: push_id={push_id}")
+            if result.status == result.status.SUCCESS:
+                return
+            if result.status == result.status.FAILED:
+                raise RuntimeError(result.error or f"Project update failed: push_id={push_id}")
+            if attempt < max_attempts - 1:
+                await asyncio.sleep(poll_interval)
+
+        raise RuntimeError(
+            f"Project update poll timeout: push_id={push_id}, last_status={getattr(last_result, 'status', None)}"
+        )
     
     def _create_node(self, data: Dict[str, Any]) -> SyncNode:
         node = self._create_basic_node(data, depend_ids=[])
-        
+
         # 设置 context
         node.context["project_id"] = node.data_id
-        
+
         return node
 
 

sync_state_machine/domain/project/register.py

@@ -1,7 +1,6 @@
 """Project domain registration"""
 from ...common.registry import DomainRegistry
-from schemas.project.project_base import ProjectResponseBase
-from .sync_node import ProjectSyncNode
+from .sync_node import ProjectSyncNode, ProjectSyncSchema
 from .sync_strategy import ProjectSyncStrategy
 from .jsonl_handler import ProjectJsonlHandler
 from .api_handler import ProjectApiHandler
@@ -9,7 +8,7 @@ from .api_handler import ProjectApiHandler
 # 注册 project domain
 DomainRegistry.register(
     node_type="project",
-    schema=ProjectResponseBase,
+    schema=ProjectSyncSchema,
     node_class=ProjectSyncNode,
     strategy_class=ProjectSyncStrategy,
     jsonl_handler_class=ProjectJsonlHandler,

sync_state_machine/domain/project/sync_node.py

@@ -1,11 +1,23 @@
+from pydantic import Field
+
 from ...common.sync_node import SyncNode
 from schemas.project.project_base import ProjectResponseBase
 
 
-class ProjectSyncNode(SyncNode[ProjectResponseBase]):
+class ProjectSyncSchema(ProjectResponseBase):
+    """项目同步内部 schema，补充 all_info 扩展字段。"""
+
+    implementation_estimate: float | None = Field(None, description="执行概算")
+    static_total_investment: float | None = Field(None, description="静态总投资")
+    completed_investment: float | None = Field(None, description="已完成投资")
+    total_contract_quantity: float | None = Field(None, description="合同总量")
+    completed_settlement_quantity: float | None = Field(None, description="已结算量")
+
+
+class ProjectSyncNode(SyncNode[ProjectSyncSchema]):
     """项目同步节点"""
     node_type = "project"
-    schema = ProjectResponseBase
+    schema = ProjectSyncSchema
     
     def __init__(self, **kwargs):
         super().__init__(**kwargs)

sync_state_machine/domain/project_detail/api_handler.py

@@ -17,6 +17,7 @@ PUT:
 不支持: CREATE, DELETE
 """
 
+import asyncio
 from typing import List, Dict, Any, Optional, Tuple, Set
 from pydantic import ValidationError
 
@@ -270,6 +271,8 @@ class ProjectDetailApiHandler(BaseApiHandler):
 
                 try:
                     resolved_data_id = self._resolve_data_id_from_all_info(all_info, project_id=project_id, key=key)
+                    if not resolved_data_id:
+                        resolved_data_id = await self._resolve_data_id_with_retry(project_id=project_id, key=key)
                 except Exception as exc:
                     self._pending_push_context.pop(task_id, None)
                     results[task_id] = TaskResult.failed(error=f"Poll resolve data_id failed: {exc}", push_id=task_id)
@@ -342,6 +345,9 @@ class ProjectDetailApiHandler(BaseApiHandler):
             if isinstance(all_info, dict):
                 return all_info
 
+        if force_refresh:
+            ProjectApiHandler._all_info_cache.pop(project_id, None)
+
         all_info = await ProjectApiHandler.get_all_info(self.api_client, project_id)
         if project_node is not None:
             project_node.context["all_info"] = all_info
@@ -370,6 +376,20 @@ class ProjectDetailApiHandler(BaseApiHandler):
             return matched[0]
         return None
 
+    async def _resolve_data_id_with_retry(self, *, project_id: str, key: str) -> Optional[str]:
+        attempts = 6
+        delay = 0.2
+
+        for attempt in range(attempts):
+            all_info = await self._get_project_all_info(project_id, force_refresh=True)
+            resolved = self._resolve_data_id_from_all_info(all_info, project_id=project_id, key=key)
+            if resolved:
+                return resolved
+            if attempt < attempts - 1:
+                await asyncio.sleep(delay)
+
+        return None
+
     async def _resolve_data_id_from_project_all_info(self, project_id: str, key: str) -> Optional[str]:
         all_info = await self._get_project_all_info(project_id)
         return self._resolve_data_id_from_all_info(all_info, project_id=project_id, key=key)

sync_state_machine/pipeline/__init__.py

@@ -7,6 +7,7 @@ from .full_sync_pipeline import FullSyncPipeline
 from ..config import (
     PipelineRunConfig,
     ApiDataSourceConfig,
+    DbDataSourceConfig,
     JsonlDataSourceConfig,
     DataSourceConfig,
     PersistConfig,
@@ -35,6 +36,7 @@ __all__ = [
     "run_pipeline_from_config",
     "PipelineRunConfig",
     "ApiDataSourceConfig",
+    "DbDataSourceConfig",
     "JsonlDataSourceConfig",
     "DataSourceConfig",
     "PersistConfig",

sync_state_machine/pipeline/factory.py

@@ -13,6 +13,7 @@ Pipeline Factory - 工厂函数，负责初始化所有组件并创建 Pipeline
 from __future__ import annotations
 
 import asyncio
+import inspect
 import json
 import logging
 from pathlib import Path
@@ -23,11 +24,14 @@ from ..common.collection import DataCollection
 from ..common.persistence import PersistenceBackend
 from ..common.registry import DomainRegistry
 from ..datasource.api.datasource import ApiDataSource
+from ..datasource.db.datasource import DbDataSource
 from ..datasource.jsonl.datasource import JsonlDataSource
+from ..datasource.type_registry import get_datasource_factory, register_datasource_factory
 from ..config import (
     PipelineRunConfig,
     DataSourceConfig,
     ApiDataSourceConfig,
+    DbDataSourceConfig,
     JsonlDataSourceConfig,
     PersistConfig,
     LoggingConfig,
@@ -97,16 +101,41 @@ def _ensure_pipeline_logger(
     )
 
 
-def _api_handler_kwargs(
+def _build_handler_config(
     node_type: str,
     ds_config: DataSourceConfig,
     target_project_ids: List[str],
 ) -> Dict[str, Any]:
-    kwargs = dict(ds_config.handler_configs.get(node_type, {}))
+    config = dict(ds_config.handler_configs.get(node_type, {}))
 
     if node_type == "project" and target_project_ids:
-        kwargs.setdefault("filter_project_id", target_project_ids)
-    return kwargs
+        config.setdefault("filter_project_id", target_project_ids)
+    return config
+
+
+def _create_handler(
+    *,
+    node_type: str,
+    ds_config: DataSourceConfig,
+    datasource,
+    target_project_ids: List[str],
+):
+    datasource_type = ds_config.type
+    handler_class = DomainRegistry.get_handler(node_type, datasource_type)
+    if handler_class is None:
+        raise RuntimeError(
+            f"Missing {datasource_type.upper()} handler for node_type: {node_type}"
+        )
+
+    handler_config = _build_handler_config(node_type, ds_config, target_project_ids)
+    if datasource_type == "api":
+        handler = handler_class(**handler_config)
+    else:
+        handler = handler_class(datasource)
+
+    if hasattr(handler, "set_handler_config"):
+        handler.set_handler_config(handler_config)
+    return handler
 
 
 def _resolve_datasource_target_project_ids(config: PipelineRunConfig) -> tuple[List[str], List[str]]:
@@ -120,48 +149,85 @@ def _resolve_datasource_target_project_ids(config: PipelineRunConfig) -> tuple[L
     return local_target_ids, remote_target_ids
 
 
-async def _create_datasources(config: PipelineRunConfig):
-    local_cfg = config.local_datasource
-    remote_cfg = config.remote_datasource
+async def _create_datasource(
+    ds_config: DataSourceConfig,
+    *,
+    datasource_override=None,
+    endpoint_name: str,
+):
+    if datasource_override is not None:
+        return datasource_override
 
-    if isinstance(local_cfg, JsonlDataSourceConfig):
-        local_dir = Path(local_cfg.jsonl_dir)
-        if not local_dir.exists() or not local_dir.is_dir():
-            raise FileNotFoundError(
-                f"local_datasource.jsonl_dir does not exist or is not a directory: {local_dir}"
+    datasource_factory = get_datasource_factory(ds_config.type)
+    if datasource_factory is None:
+        raise RuntimeError(f"No datasource factory registered for type: {ds_config.type}")
+
+    datasource = datasource_factory(ds_config, endpoint_name)
+    if inspect.isawaitable(datasource):
+        datasource = await datasource
+    return datasource
+
+
+def _build_jsonl_datasource(ds_config: JsonlDataSourceConfig, endpoint_name: str):
+    dir_path = Path(ds_config.jsonl_dir)
+    if endpoint_name == "remote":
+        dir_path = _prepare_remote_jsonl_dir(dir_path)
+    elif not dir_path.exists() or not dir_path.is_dir():
+        raise FileNotFoundError(
+            f"{endpoint_name}_datasource.jsonl_dir does not exist or is not a directory: {dir_path}"
+        )
+    return JsonlDataSource(dir_path, read_only=ds_config.read_only)
+
+
+def _build_db_datasource(ds_config: DbDataSourceConfig, endpoint_name: str):
+    del ds_config, endpoint_name
+    return DbDataSource()
+
+
+async def _build_api_datasource(ds_config: ApiDataSourceConfig, endpoint_name: str):
+    del endpoint_name
+    datasource = ApiDataSource(
+        base_url=ds_config.api_base_url,
+        uid=ds_config.api_uid,
+        secret=ds_config.api_secret,
+        debug=ds_config.api_debug,
+        poll_max_retries=ds_config.poll_max_retries,
+        poll_interval=ds_config.poll_interval,
+        rate_limit_max_requests=ds_config.api_rate_limit_max_requests,
+        rate_limit_window_seconds=ds_config.api_rate_limit_window_seconds,
+    )
+    await datasource.initialize()
+    return datasource
+
+
+def _ensure_builtin_datasource_factories_registered() -> None:
+    if get_datasource_factory("jsonl") is None:
+        register_datasource_factory("jsonl", _build_jsonl_datasource)
+    if get_datasource_factory("db") is None:
+        register_datasource_factory("db", _build_db_datasource)
+    if get_datasource_factory("api") is None:
+        register_datasource_factory("api", _build_api_datasource)
+
+
+_ensure_builtin_datasource_factories_registered()
+
+
+def _register_handlers_for_endpoint(
+    *,
+    datasource,
+    ds_config: DataSourceConfig,
+    node_types: List[str],
+    target_project_ids: List[str],
+) -> None:
+    for node_type in node_types:
+        datasource.register_handler(
+            _create_handler(
+                node_type=node_type,
+                ds_config=ds_config,
+                datasource=datasource,
+                target_project_ids=target_project_ids,
             )
-        local_ds = JsonlDataSource(local_dir, read_only=local_cfg.read_only)
-    else:
-        local_ds = ApiDataSource(
-            base_url=local_cfg.api_base_url,
-            uid=local_cfg.api_uid,
-            secret=local_cfg.api_secret,
-            debug=local_cfg.api_debug,
-            poll_max_retries=local_cfg.poll_max_retries,
-            poll_interval=local_cfg.poll_interval,
-            rate_limit_max_requests=local_cfg.api_rate_limit_max_requests,
-            rate_limit_window_seconds=local_cfg.api_rate_limit_window_seconds,
         )
-        await local_ds.initialize()
-
-    if isinstance(remote_cfg, JsonlDataSourceConfig):
-        remote_dir = Path(remote_cfg.jsonl_dir)
-        remote_dir = _prepare_remote_jsonl_dir(remote_dir)
-        remote_ds = JsonlDataSource(remote_dir, read_only=remote_cfg.read_only)
-    else:
-        remote_ds = ApiDataSource(
-            base_url=remote_cfg.api_base_url,
-            uid=remote_cfg.api_uid,
-            secret=remote_cfg.api_secret,
-            debug=remote_cfg.api_debug,
-            poll_max_retries=remote_cfg.poll_max_retries,
-            poll_interval=remote_cfg.poll_interval,
-            rate_limit_max_requests=remote_cfg.api_rate_limit_max_requests,
-            rate_limit_window_seconds=remote_cfg.api_rate_limit_window_seconds,
-        )
-        await remote_ds.initialize()
-
-    return local_ds, remote_ds
 
 
 def _register_handlers(
@@ -170,44 +236,18 @@ def _register_handlers(
     remote_ds,
     all_types: List[str],
 ) -> None:
-    for node_type in all_types:
-        if isinstance(config.local_datasource, JsonlDataSourceConfig):
-            jsonl_handler_class = DomainRegistry.get_jsonl_handler(node_type)
-            if jsonl_handler_class is None:
-                raise RuntimeError(f"Missing JSONL handler for node_type: {node_type}")
-            local_ds.register_handler(jsonl_handler_class(local_ds))
-        else:
-            api_handler_class = DomainRegistry.get_api_handler(node_type)
-            if api_handler_class is None:
-                raise RuntimeError(f"Missing API handler for node_type: {node_type}")
-            local_ds.register_handler(
-                api_handler_class(
-                    **_api_handler_kwargs(
-                        node_type,
-                        config.local_datasource,
-                        config.target_project_ids,
-                    )
-                )
-            )
-
-        if isinstance(config.remote_datasource, JsonlDataSourceConfig):
-            jsonl_handler_class = DomainRegistry.get_jsonl_handler(node_type)
-            if jsonl_handler_class is None:
-                raise RuntimeError(f"Missing JSONL handler for node_type: {node_type}")
-            remote_ds.register_handler(jsonl_handler_class(remote_ds))
-        else:
-            api_handler_class = DomainRegistry.get_api_handler(node_type)
-            if api_handler_class is None:
-                raise RuntimeError(f"Missing API handler for node_type: {node_type}")
-            remote_ds.register_handler(
-                api_handler_class(
-                    **_api_handler_kwargs(
-                        node_type,
-                        config.remote_datasource,
-                        config.target_project_ids,
-                    )
-                )
-            )
+    _register_handlers_for_endpoint(
+        datasource=local_ds,
+        ds_config=config.local_datasource,
+        node_types=all_types,
+        target_project_ids=[str(pid) for pid in config.local_datasource.target_project_ids if str(pid)],
+    )
+    _register_handlers_for_endpoint(
+        datasource=remote_ds,
+        ds_config=config.remote_datasource,
+        node_types=all_types,
+        target_project_ids=[str(pid) for pid in config.remote_datasource.target_project_ids if str(pid)],
+    )
 
 
 def _resolve_strategy_map(config: PipelineRunConfig, node_types: List[str]) -> Dict[str, StrategyRuntimeConfig]:
@@ -260,7 +300,12 @@ def _build_strategies(
     return strategies
 
 
-async def create_pipeline_from_config(config: PipelineRunConfig) -> FullSyncPipeline:
+async def create_pipeline_from_config(
+    config: PipelineRunConfig,
+    *,
+    local_datasource_override=None,
+    remote_datasource_override=None,
+) -> FullSyncPipeline:
     if config.logging.initialize and not config.logging.file_path:
         config.logging.file_path = resolve_log_file_path(config.logging)
 
@@ -289,7 +334,16 @@ async def create_pipeline_from_config(config: PipelineRunConfig) -> FullSyncPipe
         persistence = PersistenceBackend(str(config.persist.db_path), backend=config.persist.backend)
         await persistence.initialize()
 
-        local_ds, remote_ds = await _create_datasources(config)
+        local_ds = await _create_datasource(
+            config.local_datasource,
+            datasource_override=local_datasource_override,
+            endpoint_name="local",
+        )
+        remote_ds = await _create_datasource(
+            config.remote_datasource,
+            datasource_override=remote_datasource_override,
+            endpoint_name="remote",
+        )
         local_cfg_with_target = config.local_datasource.model_copy(
             update={"target_project_ids": list(local_target_project_ids)},
             deep=True,
@@ -340,7 +394,7 @@ async def create_pipeline_from_config(config: PipelineRunConfig) -> FullSyncPipe
 
         return pipeline
     except Exception:
-        async def _safe_close(name: str, resource: ApiDataSource | JsonlDataSource | PersistenceBackend | None) -> None:
+        async def _safe_close(name: str, resource) -> None:
             if resource is None:
                 return
             try:
@@ -359,6 +413,8 @@ async def run_pipeline_from_config(
     *,
     title: str = "sync_state_machine pipeline",
     print_summary: bool = True,
+    local_datasource_override=None,
+    remote_datasource_override=None,
 ) -> Dict[str, Any]:
     if config.logging.initialize and not config.logging.file_path:
         config.logging.file_path = resolve_log_file_path(config.logging)
@@ -401,7 +457,11 @@ async def run_pipeline_from_config(
             print(f"🔧 Resolved Config:\n{resolved_cfg_text}")
             print("=" * 80)
 
-        pipeline = await create_pipeline_from_config(config)
+        pipeline = await create_pipeline_from_config(
+            config,
+            local_datasource_override=local_datasource_override,
+            remote_datasource_override=remote_datasource_override,
+        )
         stats = await pipeline.run()
 
         if print_summary:

tests/unit/project_detail_upsert/test_project_detail_upsert_poll.py

@@ -355,3 +355,55 @@ async def test_poll_tasks_queries_all_info_once_per_project_for_multiple_keys(mo
     assert polled[task_ids[1]].status == TaskStatus.SUCCESS
     assert polled[task_ids[1]].data_id == "remote-detail-2"
     assert client.all_info_calls == ["project-1"]
+
+
+@pytest.mark.asyncio
+async def test_poll_tasks_retries_until_project_detail_appears(monkeypatch: pytest.MonkeyPatch) -> None:
+    collection = DataCollection("remote")
+    project = _build_project_node(node_id="p-node", project_id="project-1", all_info={"project_detail": []})
+    await collection.add(project)
+
+    handler = ProjectDetailApiHandler()
+    handler.set_collection(collection)
+    handler.api_client = _MockApiClient()
+
+    call_count = {"value": 0}
+
+    async def _fake_get_all_info(cls, _client, project_id: str):
+        call_count["value"] += 1
+        if call_count["value"] >= 2:
+            return {
+                "project_detail": [
+                    {
+                        "id": "remote-detail-1",
+                        "project_id": project_id,
+                        "key": "ensure_production",
+                        "value": [{"year": 2026, "capacity": 21.0}],
+                    }
+                ]
+            }
+        return {"project_detail": []}
+
+    monkeypatch.setattr(ProjectApiHandler, "get_all_info", classmethod(_fake_get_all_info))
+
+    node = ProjectDetailSyncNode(
+        node_id="detail-node-1",
+        data_id="local-detail-1",
+        data={
+            "id": "local-detail-1",
+            "project_id": "project-1",
+            "key": "ensure_production",
+            "value": [{"year": 2026, "capacity": 21.0}],
+        },
+    )
+
+    create_result = (await handler.create_all([node]))[0]
+    assert create_result.task_id
+    handler.api_client.push_logs[create_result.task_id] = {"code": 200, "data": {"biz_id": "project-1"}}
+
+    polled = await handler.poll_tasks([create_result.task_id])
+    final = polled[create_result.task_id]
+
+    assert final.status == TaskStatus.SUCCESS
+    assert final.data_id == "remote-detail-1"
+    assert call_count["value"] >= 2

tests/unit/test_auto_test_project_profile.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+import sys
+
+import yaml
+
+
+def _load_base_module():
+    scripts_dir = Path(__file__).resolve().parents[2] / "scripts"
+    if str(scripts_dir) not in sys.path:
+        sys.path.insert(0, str(scripts_dir))
+    file_path = Path(__file__).resolve().parents[2] / "scripts" / "auto_test_domains" / "base.py"
+    spec = importlib.util.spec_from_file_location("auto_test_domain_base_for_test", file_path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Unable to load module from {file_path}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_build_temp_run_profile_uses_explicit_project_prebind(tmp_path: Path) -> None:
+    module = _load_base_module()
+
+    source_profile = tmp_path / "source.yaml"
+    source_profile.write_text(
+        yaml.safe_dump(
+            {
+                "local_datasource": {"target_project_ids": ["old-local"]},
+                "remote_datasource": {"target_project_ids": ["old-remote"]},
+                "strategies": {"project": {"config": {}}},
+            },
+            allow_unicode=True,
+            sort_keys=False,
+        ),
+        encoding="utf-8",
+    )
+
+    config = {
+        "pipeline": {
+            "run_profile": str(source_profile),
+            "command": ["python", "run.py", "--config_path", "placeholder.yaml"],
+        }
+    }
+
+    profile_path, command = module.build_temp_run_profile_for_project_auto_bind(
+        config=config,
+        report_root=tmp_path,
+        profile_name="demo",
+        local_project_id="local-project",
+        remote_project_id="remote-project",
+    )
+
+    profile = yaml.safe_load(Path(profile_path).read_text(encoding="utf-8"))
+    project_config = profile["strategies"]["project"]["config"]
+
+    assert profile["local_datasource"]["target_project_ids"] == ["local-project"]
+    assert profile["remote_datasource"]["target_project_ids"] == ["remote-project"]
+    assert project_config["auto_bind"] is False
+    assert project_config["auto_bind_fields"] == []
+    assert project_config["pre_bind_data_id"] == [
+        {"local_data_id": "local-project", "remote_data_id": "remote-project"}
+    ]
+    assert command[-1] == profile_path

tests/unit/test_datasource_target_project_ids_resolution.py

@@ -7,6 +7,7 @@ from sync_state_machine.config import (
     PipelineRunConfig,
     JsonlDataSourceConfig,
     ApiDataSourceConfig,
+    DbDataSourceConfig,
     PersistConfig,
     LoggingConfig,
 )
@@ -101,3 +102,11 @@ def test_api_datasource_rate_limit_config_fields() -> None:
     )
     assert overridden.api_rate_limit_max_requests == 5
     assert overridden.api_rate_limit_window_seconds == 2.5
+
+
+def test_db_datasource_accepts_minimal_config() -> None:
+    cfg = DbDataSourceConfig(type="db")
+
+    assert cfg.type == "db"
+    assert cfg.target_project_ids == []
+    assert cfg.handler_configs == {}

tests/unit/test_datasource_type_registry.py

@@ -0,0 +1,198 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from pydantic import BaseModel, ConfigDict
+
+from sync_state_machine.common.registry import DomainRegistry
+from sync_state_machine.common.sync_node import SyncNode
+from sync_state_machine.config import (
+    BaseDataSourceConfig,
+    DbDataSourceConfig,
+    LoggingConfig,
+    PersistConfig,
+    PipelineRunConfig,
+    build_config,
+    register_datasource_type,
+)
+from sync_state_machine.datasource import BaseDataSource
+from sync_state_machine.datasource.db import BaseDbHandler, DbDataSource
+from sync_state_machine.pipeline.factory import create_pipeline_from_config
+from sync_state_machine.sync_system.strategy import DefaultSyncStrategy
+
+
+class MemoryConfig(BaseDataSourceConfig):
+    model_config = ConfigDict(extra="forbid", validate_assignment=True)
+
+    type: str = "memory"
+    namespace: str = "default"
+
+
+class MemorySchema(BaseModel):
+    __test__ = False
+
+    id: str
+    name: str | None = None
+
+
+class MemoryNode(SyncNode[MemorySchema]):
+    __test__ = False
+
+    node_type = "memory_demo"
+    schema = MemorySchema
+
+
+class MemoryStrategy(DefaultSyncStrategy[MemorySchema]):
+    __test__ = False
+
+    schema = MemorySchema
+
+
+class MemoryDataSource(BaseDataSource):
+    def __init__(self, namespace: str):
+        super().__init__()
+        self.namespace = namespace
+
+
+class MemoryHandler(BaseDbHandler[MemorySchema]):
+    __test__ = False
+
+    def __init__(self, datasource: MemoryDataSource):
+        super().__init__(datasource, "memory_demo", MemoryNode, MemorySchema)
+
+    def extract_id(self, data: dict[str, str]) -> str:
+        return str(data.get("id", ""))
+
+    async def load(self):
+        return []
+
+    async def create_all(self, nodes):
+        return []
+
+    async def update_all(self, nodes):
+        return []
+
+
+class MemoryRemoteDbHandler(BaseDbHandler[MemorySchema]):
+    __test__ = False
+
+    def __init__(self, datasource: DbDataSource):
+        super().__init__(datasource, "memory_demo", MemoryNode, MemorySchema)
+
+    def extract_id(self, data: dict[str, str]) -> str:
+        return str(data.get("id", ""))
+
+    async def load(self):
+        return []
+
+    async def create_all(self, nodes):
+        return []
+
+    async def update_all(self, nodes):
+        return []
+
+
+def _ensure_memory_domain_registered() -> None:
+    if DomainRegistry.is_registered("memory_demo"):
+        reg = DomainRegistry.get_registration("memory_demo")
+        reg.schema = MemorySchema
+        reg.node_class = MemoryNode
+        reg.strategy_class = MemoryStrategy
+        reg.db_handler_class = MemoryRemoteDbHandler
+        reg.handler_classes["db"] = MemoryRemoteDbHandler
+        reg.handler_classes["memory"] = MemoryHandler
+        return
+
+    DomainRegistry.register(
+        node_type="memory_demo",
+        schema=MemorySchema,
+        node_class=MemoryNode,
+        strategy_class=MemoryStrategy,
+        db_handler_class=MemoryRemoteDbHandler,
+    )
+    DomainRegistry.register_handler("memory_demo", "memory", MemoryHandler)
+
+
+def _memory_factory(config: MemoryConfig, endpoint_name: str) -> MemoryDataSource:
+    return MemoryDataSource(namespace=f"{endpoint_name}:{config.namespace}")
+
+
+@pytest.mark.asyncio
+async def test_create_pipeline_from_config_supports_registered_custom_datasource_type(tmp_path: Path) -> None:
+    register_datasource_type("memory", config_model=MemoryConfig, factory=_memory_factory, replace=True)
+    _ensure_memory_domain_registered()
+
+    config = PipelineRunConfig(
+        node_types=["memory_demo"],
+        local_datasource=MemoryConfig(type="memory", namespace="alpha"),
+        remote_datasource=DbDataSourceConfig(type="db"),
+        strategies=[],
+        persist=PersistConfig(db_path=str(tmp_path / "memory-registry.db")),
+        logging=LoggingConfig(initialize=False),
+    )
+
+    pipeline = await create_pipeline_from_config(config)
+    try:
+        assert isinstance(pipeline.local_datasource, MemoryDataSource)
+        assert pipeline.local_datasource.namespace == "local:alpha"
+        assert isinstance(pipeline.local_datasource.get_handler("memory_demo"), MemoryHandler)
+        assert isinstance(pipeline.remote_datasource.get_handler("memory_demo"), MemoryRemoteDbHandler)
+    finally:
+        await pipeline.close()
+
+
+def test_build_config_supports_same_file_datasource_bootstrap(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    bootstrap_module = tmp_path / "memory_bootstrap_demo.py"
+    bootstrap_module.write_text(
+        "from pydantic import ConfigDict\n"
+        "from sync_state_machine.config import BaseDataSourceConfig, register_datasource_type\n"
+        "from sync_state_machine.datasource import BaseDataSource\n"
+        "\n"
+        "class BootstrapMemoryConfig(BaseDataSourceConfig):\n"
+        "    model_config = ConfigDict(extra=\"forbid\", validate_assignment=True)\n"
+        "    type: str = \"memory_bootstrap\"\n"
+        "    namespace: str = \"demo\"\n"
+        "\n"
+        "class BootstrapMemoryDataSource(BaseDataSource):\n"
+        "    def __init__(self, namespace: str):\n"
+        "        super().__init__()\n"
+        "        self.namespace = namespace\n"
+        "\n"
+        "def build_memory_datasource(config, endpoint_name):\n"
+        "    return BootstrapMemoryDataSource(namespace=f\"{endpoint_name}:{config.namespace}\")\n"
+        "\n"
+        "def register():\n"
+        "    register_datasource_type(\n"
+        "        \"memory_bootstrap\",\n"
+        "        config_model=BootstrapMemoryConfig,\n"
+        "        factory=build_memory_datasource,\n"
+        "        replace=True,\n"
+        "    )\n",
+        encoding="utf-8",
+    )
+    monkeypatch.syspath_prepend(str(tmp_path))
+
+    profile_path = tmp_path / "custom-memory.yaml"
+    profile_path.write_text(
+        "extensions:\n"
+        "  datasource_bootstraps:\n"
+        "    - memory_bootstrap_demo:register\n"
+        "node_types: []\n"
+        "local_datasource:\n"
+        "  type: memory_bootstrap\n"
+        "  namespace: local-demo\n"
+        "remote_datasource:\n"
+        "  type: db\n"
+        "persist:\n"
+        "  db_path: ${PROJECT_ROOT}/runtime/test-memory-bootstrap.db\n"
+        "logging:\n"
+        "  initialize: false\n",
+        encoding="utf-8",
+    )
+
+    config = build_config(project_root=tmp_path, overrides_file=profile_path)
+
+    assert config.local_datasource.type == "memory_bootstrap"
+    assert getattr(config.local_datasource, "namespace") == "local-demo"
+    assert config.remote_datasource.type == "db"

tests/unit/test_factory_db_handler_registration.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from pydantic import BaseModel
+
+from sync_state_machine.common.registry import DomainRegistry
+from sync_state_machine.common.sync_node import SyncNode
+from sync_state_machine.config import DbDataSourceConfig, JsonlDataSourceConfig, LoggingConfig, PersistConfig, PipelineRunConfig
+from sync_state_machine.datasource.db import BaseDbHandler, DbDataSource
+from sync_state_machine.datasource.jsonl import BaseJsonlHandler, JsonlDataSource
+from sync_state_machine.pipeline.factory import _register_handlers
+from sync_state_machine.sync_system.strategy import DefaultSyncStrategy
+
+
+class TestDbSchema(BaseModel):
+    __test__ = False
+    id: str
+    name: str
+
+
+class TestDbNode(SyncNode[TestDbSchema]):
+    __test__ = False
+    node_type = "test_db_handler"
+    schema = TestDbSchema
+
+
+class TestDbStrategy(DefaultSyncStrategy[TestDbSchema]):
+    __test__ = False
+    schema = TestDbSchema
+
+
+class TestDbHandler(BaseDbHandler[TestDbSchema]):
+    __test__ = False
+
+    def __init__(self, datasource: DbDataSource):
+        super().__init__(datasource, "test_db_handler", TestDbNode, TestDbSchema)
+
+    def extract_id(self, data: dict[str, str]) -> str:
+        return str(data.get("id", ""))
+
+    async def load(self):
+        return []
+
+    async def create_all(self, nodes):
+        return []
+
+    async def update_all(self, nodes):
+        return []
+
+
+class TestJsonlHandler(BaseJsonlHandler):
+    __test__ = False
+
+    def __init__(self, datasource: JsonlDataSource):
+        super().__init__(datasource, "test_db_handler", TestDbNode, TestDbSchema)
+
+
+def test_register_handlers_uses_db_handler_for_db_datasource(tmp_path: Path) -> None:
+    if not DomainRegistry.is_registered("test_db_handler"):
+        DomainRegistry.register(
+            node_type="test_db_handler",
+            schema=TestDbSchema,
+            node_class=TestDbNode,
+            strategy_class=TestDbStrategy,
+            jsonl_handler_class=TestJsonlHandler,
+        )
+    DomainRegistry.register_db_handler("test_db_handler", TestDbHandler)
+
+    config = PipelineRunConfig(
+        node_types=["test_db_handler"],
+        local_datasource=DbDataSourceConfig(type="db"),
+        remote_datasource=JsonlDataSourceConfig(
+            type="jsonl",
+            jsonl_dir=str(tmp_path),
+            target_project_ids=[],
+        ),
+        strategies=[],
+        persist=PersistConfig(db_path=str(tmp_path / "pipeline.db")),
+        logging=LoggingConfig(initialize=False),
+    )
+
+    local_ds = DbDataSource()
+    remote_ds = JsonlDataSource(tmp_path)
+
+    _register_handlers(config, local_ds, remote_ds, ["test_db_handler"])
+
+    assert isinstance(local_ds.get_handler("test_db_handler"), TestDbHandler)
+    assert isinstance(remote_ds.get_handler("test_db_handler"), TestJsonlHandler)

tests/unit/test_factory_endpoint_initialization.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from pydantic import BaseModel
+
+from sync_state_machine.common.registry import DomainRegistry
+from sync_state_machine.common.sync_node import SyncNode
+from sync_state_machine.config import DbDataSourceConfig, LoggingConfig, PersistConfig, PipelineRunConfig
+from sync_state_machine.datasource.db import BaseDbHandler, DbDataSource
+from sync_state_machine.pipeline.factory import create_pipeline_from_config
+from sync_state_machine.sync_system.strategy import DefaultSyncStrategy
+
+
+class EndpointInitSchema(BaseModel):
+    __test__ = False
+    id: str
+    name: str | None = None
+
+
+class EndpointInitNode(SyncNode[EndpointInitSchema]):
+    __test__ = False
+    node_type = "factory_endpoint_demo"
+    schema = EndpointInitSchema
+
+
+class EndpointInitStrategy(DefaultSyncStrategy[EndpointInitSchema]):
+    __test__ = False
+    schema = EndpointInitSchema
+
+
+class EndpointInitDbHandler(BaseDbHandler[EndpointInitSchema]):
+    __test__ = False
+
+    def __init__(self, datasource: DbDataSource):
+        super().__init__(datasource, "factory_endpoint_demo", EndpointInitNode, EndpointInitSchema)
+
+    def extract_id(self, data: dict[str, str]) -> str:
+        return str(data.get("id", ""))
+
+    async def load(self):
+        return []
+
+    async def create_all(self, nodes):
+        return []
+
+    async def update_all(self, nodes):
+        return []
+
+
+def _ensure_registered() -> None:
+    node_type = "factory_endpoint_demo"
+    if DomainRegistry.is_registered(node_type):
+        reg = DomainRegistry.get_registration(node_type)
+        reg.schema = EndpointInitSchema
+        reg.node_class = EndpointInitNode
+        reg.strategy_class = EndpointInitStrategy
+        reg.db_handler_class = EndpointInitDbHandler
+        return
+
+    DomainRegistry.register(
+        node_type=node_type,
+        schema=EndpointInitSchema,
+        node_class=EndpointInitNode,
+        strategy_class=EndpointInitStrategy,
+        db_handler_class=EndpointInitDbHandler,
+    )
+
+
+@pytest.mark.asyncio
+async def test_factory_initializes_same_datasource_type_with_distinct_handler_configs(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _ensure_registered()
+
+    config = PipelineRunConfig(
+        node_types=["factory_endpoint_demo"],
+        local_datasource=DbDataSourceConfig(
+            type="db",
+            handler_configs={"factory_endpoint_demo": {"side": "local", "batch_size": 10}},
+            target_project_ids=["project_local"],
+        ),
+        remote_datasource=DbDataSourceConfig(
+            type="db",
+            handler_configs={"factory_endpoint_demo": {"side": "remote", "batch_size": 20}},
+            target_project_ids=["project_remote"],
+        ),
+        strategies=[],
+        persist=PersistConfig(db_path=str(tmp_path / "factory-init.db")),
+        logging=LoggingConfig(initialize=False),
+    )
+
+    pipeline = await create_pipeline_from_config(config)
+    try:
+        local_handler = pipeline.local_datasource.get_handler("factory_endpoint_demo")
+        remote_handler = pipeline.remote_datasource.get_handler("factory_endpoint_demo")
+
+        assert isinstance(local_handler, EndpointInitDbHandler)
+        assert isinstance(remote_handler, EndpointInitDbHandler)
+        assert local_handler.handler_config == {"side": "local", "batch_size": 10}
+        assert remote_handler.handler_config == {"side": "remote", "batch_size": 20}
+    finally:
+        await pipeline.close()

tests/unit/test_handler_shared_config.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel
+
+from sync_state_machine.common.sync_node import SyncNode
+from sync_state_machine.datasource.api.handler import BaseApiHandler
+from sync_state_machine.datasource.handler import BaseNodeHandler, CompatibleNodeHandler
+from sync_state_machine.datasource.task_result import TaskResult
+
+
+class SharedConfigSchema(BaseModel):
+    __test__ = False
+    id: str
+
+
+class SharedConfigNode(SyncNode[SharedConfigSchema]):
+    __test__ = False
+    node_type = "shared_config_demo"
+    schema = SharedConfigSchema
+
+
+class SharedConfigBaseHandler(BaseNodeHandler[SharedConfigSchema]):
+    __test__ = False
+
+    def __init__(self) -> None:
+        super().__init__("shared_config_demo", SharedConfigNode, SharedConfigSchema)
+
+    async def load(self):
+        return []
+
+    async def create_all(self, nodes):
+        return []
+
+    async def update_all(self, nodes):
+        return []
+
+    def extract_id(self, data: dict[str, Any]) -> str:
+        return str(data.get("id", ""))
+
+
+class SharedConfigApiHandler(BaseApiHandler[SharedConfigSchema]):
+    __test__ = False
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._node_type = "shared_api_demo"
+        self._node_class = SharedConfigNode
+        self._schema = SharedConfigSchema
+
+    async def load(self):
+        return []
+
+    async def create_all(self, nodes):
+        return [TaskResult.success(node_id=node.node_id) for node in nodes]
+
+    async def update_all(self, nodes):
+        return [TaskResult.success(node_id=node.node_id) for node in nodes]
+
+    async def delete_all(self, nodes):
+        return [TaskResult.success(node_id=node.node_id) for node in nodes]
+
+
+def test_base_handler_supports_shared_handler_config() -> None:
+    handler = SharedConfigBaseHandler()
+
+    handler.set_handler_config({"mode": "local", "batch_size": 10})
+
+    assert handler.handler_config == {"mode": "local", "batch_size": 10}
+
+
+def test_compatible_handler_forwards_shared_handler_config() -> None:
+    handler = CompatibleNodeHandler(SharedConfigBaseHandler())
+
+    handler.set_handler_config({"mode": "remote"})
+
+    assert handler.handler_config == {"mode": "remote"}
+
+
+def test_api_handler_supports_shared_handler_config() -> None:
+    handler = SharedConfigApiHandler()
+
+    handler.set_handler_config({"filter_project_id": ["p1", "p2"]})
+
+    assert handler.handler_config == {"filter_project_id": ["p1", "p2"]}

tests/unit/test_material_detail_api_handler.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from pydantic import BaseModel
+
+from sync_state_machine.common.sync_node import SyncNode
+from sync_state_machine.domain.material_detail.api_handler import MaterialDetailApiHandler
+from sync_state_machine.datasource.task_result import TaskStatus
+
+
+class _MaterialDetailSchema(BaseModel):
+    id: str
+    parent_id: str
+    quantity: float
+    remark: str | None = None
+
+
+class _MaterialDetailNode(SyncNode[_MaterialDetailSchema]):
+    node_type = "material_detail"
+    schema = _MaterialDetailSchema
+
+
+class _MockApiClient:
+    def __init__(self) -> None:
+        self.requests: list[dict[str, Any]] = []
+
+    async def request(self, method: str, endpoint: str, **kwargs: Any) -> dict[str, Any]:
+        self.requests.append({"method": method, "endpoint": endpoint, **kwargs})
+        return {"code": 200, "message": "OK", "data": {"biz_id": kwargs.get("data", {}).get("biz_id")}}
+
+
+@pytest.mark.asyncio
+async def test_material_detail_update_returns_in_progress_with_shared_push_id() -> None:
+    handler = MaterialDetailApiHandler()
+    handler.api_client = _MockApiClient()
+
+    node = _MaterialDetailNode(
+        node_id="detail-node-1",
+        data_id="detail-1",
+        data={
+            "id": "detail-1",
+            "parent_id": "material-1",
+            "quantity": 10.0,
+            "remark": "new",
+        },
+        origin_data={
+            "id": "detail-1",
+            "parent_id": "material-1",
+            "quantity": 5.0,
+            "remark": None,
+        },
+        context={"material_id": "material-1"},
+    )
+
+    results = await handler.update_all([node])
+
+    assert len(results) == 1
+    assert results[0].status == TaskStatus.IN_PROGRESS
+    assert results[0].task_id
+    assert handler.api_client.requests[0]["endpoint"] == "/material/detail/update"

tests/unit/test_project_api_handler_normalization.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from uuid import uuid4
+
+from sync_state_machine.domain.project.api_handler import _merge_project_all_info_fields
+from sync_state_machine.domain.project.sync_node import ProjectSyncNode
+
+
+def test_merge_project_all_info_fields_flattens_extension_investment_fields() -> None:
+    project_data = {
+        "id": "project-1",
+        "name": "Demo Project",
+        "dynamic_investment": 123.0,
+    }
+    all_info = {
+        "project": {"id": "project-1"},
+        "hydropower_extension": {
+            "implementation_estimate": 1000.0,
+            "static_total_investment": 990.0,
+            "completed_investment": 120.0,
+            "total_contract_quantity": 50.0,
+        },
+    }
+
+    merged = _merge_project_all_info_fields(project_data, all_info)
+
+    assert merged["dynamic_investment"] == 123.0
+    assert merged["implementation_estimate"] == 1000.0
+    assert merged["static_total_investment"] == 990.0
+    assert merged["completed_investment"] == 120.0
+    assert merged["total_contract_quantity"] == 50.0
+
+
+def test_merge_project_all_info_fields_keeps_existing_project_values() -> None:
+    project_data = {
+        "id": "project-1",
+        "implementation_estimate": 2000.0,
+        "static_total_investment": 1800.0,
+    }
+    all_info = {
+        "hydropower_extension": {
+            "implementation_estimate": 1000.0,
+            "static_total_investment": 990.0,
+            "completed_investment": 120.0,
+        },
+    }
+
+    merged = _merge_project_all_info_fields(project_data, all_info)
+
+    assert merged["implementation_estimate"] == 2000.0
+    assert merged["static_total_investment"] == 1800.0
+    assert merged["completed_investment"] == 120.0
+
+
+def test_project_sync_node_preserves_extension_backed_fields() -> None:
+    project_id = str(uuid4())
+    data = {
+        "_id": project_id,
+        "id": project_id,
+        "project_type": "hydropower",
+        "region_name": ["中国", "测试省"],
+        "region_path": ["100000", "110000"],
+        "province": "110000",
+        "is_domestic": True,
+        "company_id": str(uuid4()),
+        "company_name": "测试公司",
+        "address": "测试地址",
+        "code": "P-001",
+        "name": "测试项目",
+        "progress_type": "A",
+        "dynamic_investment": 100.0,
+        "complete_investment": 80.0,
+        "implementation_estimate": 120.0,
+        "static_total_investment": 110.0,
+        "completed_investment": 70.0,
+    }
+
+    node = ProjectSyncNode(node_id=str(uuid4()), data_id=data["id"], data=data)
+    normalized = node.get_data()
+
+    assert normalized is not None
+    assert normalized["implementation_estimate"] == 120.0
+    assert normalized["static_total_investment"] == 110.0
+    assert normalized["completed_investment"] == 70.0

tests/unit/test_project_api_handler_updates.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from pydantic import BaseModel
+
+from sync_state_machine.common.sync_node import SyncNode
+from sync_state_machine.domain.project.api_handler import ProjectApiHandler
+from sync_state_machine.datasource.task_result import TaskStatus
+
+
+class _ProjectUpdateSchema(BaseModel):
+    id: str
+    dynamic_investment: float
+
+
+class _ProjectUpdateNode(SyncNode[_ProjectUpdateSchema]):
+    node_type = "project"
+    schema = _ProjectUpdateSchema
+
+
+class _MockApiClient:
+    def __init__(self) -> None:
+        self.requests: list[dict[str, Any]] = []
+        self._poll_count = 0
+
+    async def request(self, method: str, endpoint: str, **kwargs: Any) -> dict[str, Any]:
+        self.requests.append({"method": method, "endpoint": endpoint, **kwargs})
+        return {"code": 200, "message": "OK", "data": {"biz_id": kwargs.get("data", {}).get("biz_id")}}
+
+    async def get_push_logs(self, task_ids: list[str]) -> dict[str, dict[str, Any]]:
+        self._poll_count += 1
+        if self._poll_count == 1:
+            return {task_ids[0]: {}}
+        return {task_ids[0]: {"code": 200, "data": {"biz_id": "project-1"}}}
+
+
+@pytest.mark.asyncio
+async def test_project_update_waits_for_async_push_completion() -> None:
+    handler = ProjectApiHandler(filter_project_id=["project-1"])
+    handler.api_client = _MockApiClient()
+
+    node = _ProjectUpdateNode(
+        node_id="project-node-1",
+        data_id="project-1",
+        data={"id": "project-1", "dynamic_investment": 200.0},
+        origin_data={"id": "project-1", "dynamic_investment": 100.0},
+    )
+
+    results = await handler.update_all([node])
+
+    assert len(results) == 1
+    assert results[0].status == TaskStatus.SUCCESS
+    assert handler.api_client.requests[0]["endpoint"] == "/project/dynamic_investment"
+    assert handler.api_client._poll_count == 2

tests/unit/test_project_case_plan.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+import sys
+
+
+def _load_case_module():
+    scripts_dir = Path(__file__).resolve().parents[2] / "scripts"
+    if str(scripts_dir) not in sys.path:
+        sys.path.insert(0, str(scripts_dir))
+    file_path = scripts_dir / "auto_test_domains" / "project_case.py"
+    spec = importlib.util.spec_from_file_location("project_case_for_test", file_path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Unable to load module from {file_path}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_project_case_round_3_focuses_on_stable_readback_fields(monkeypatch, tmp_path: Path) -> None:
+    module = _load_case_module()
+    captured_rounds: dict[str, dict] = {}
+
+    def _fake_run_pipeline_round(**kwargs):
+        captured_rounds[kwargs["round_name"]] = kwargs
+        return {"name": kwargs["round_name"], "passed": True, "issues": []}
+
+    def _fake_build_domain_report(**kwargs):
+        return kwargs
+
+    monkeypatch.setattr(module, "run_pipeline_round", _fake_run_pipeline_round)
+    monkeypatch.setattr(module, "build_domain_report", _fake_build_domain_report)
+
+    module.run_case(config={"project_root": str(tmp_path)}, report_root=tmp_path)
+
+    round_3 = captured_rounds["round_3_investment_update"]
+    expected_fields = round_3["expected_bound_fields"]["remote"]["project"][module.LOCAL_PROJECT_ID]
+    changed_fields = round_3["expected_changes"]["project"][module.LOCAL_PROJECT_ID]
+
+    assert set(changed_fields) == {
+        "investment_decision_capacity",
+        "investment_decision_date",
+        "complete_investment",
+        "dynamic_investment",
+    }
+    assert set(expected_fields.keys()) == {
+        "investment_decision_capacity",
+        "investment_decision_date",
+        "complete_investment",
+        "dynamic_investment",
+    }
+    assert "PUT /project/settlement_investment" in round_3["coverage"]["api"]

tests/unit/test_project_detail_case_plan.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+import sys
+
+
+def _load_case_module():
+    scripts_dir = Path(__file__).resolve().parents[2] / "scripts"
+    if str(scripts_dir) not in sys.path:
+        sys.path.insert(0, str(scripts_dir))
+    file_path = scripts_dir / "auto_test_domains" / "project_detail_data_case.py"
+    spec = importlib.util.spec_from_file_location("project_detail_data_case_for_test", file_path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Unable to load module from {file_path}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_project_detail_case_excludes_production_from_unified_rounds(monkeypatch, tmp_path: Path) -> None:
+    module = _load_case_module()
+    recorded_rounds: list[str] = []
+
+    def _fake_run_pipeline_round(**kwargs):
+        recorded_rounds.append(kwargs["round_name"])
+        return {"name": kwargs["round_name"], "passed": True, "issues": []}
+
+    def _fake_build_domain_report(**kwargs):
+        return kwargs
+
+    monkeypatch.setattr(module, "run_pipeline_round", _fake_run_pipeline_round)
+    monkeypatch.setattr(module, "build_domain_report", _fake_build_domain_report)
+
+    report = module.run_case(config={"project_root": str(tmp_path)}, report_root=tmp_path)
+
+    assert recorded_rounds == [
+        "round_1_create",
+        "round_2_update",
+        "round_3_plan_construction_create",
+        "round_5_ensure_production_push",
+        "round_6_climb_production_push",
+        "round_7_challenge_production_push",
+    ]
+    assert "production" in report["description"]
+    assert "不混入本组统一回归" in report["description"]

tests/unit/test_registry_unified_handler_lookup.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+from sync_state_machine.common.registry import DomainRegistry
+
+
+class RegistryLookupSchema(BaseModel):
+    __test__ = False
+    id: str
+
+
+class RegistryLookupNode:
+    __test__ = False
+
+
+class RegistryLookupJsonlHandler:
+    __test__ = False
+
+
+class RegistryLookupApiHandler:
+    __test__ = False
+
+
+class RegistryLookupDbHandler:
+    __test__ = False
+
+
+def test_registry_supports_unified_handler_lookup() -> None:
+    node_type = "registry_lookup_demo"
+
+    if DomainRegistry.is_registered(node_type):
+        reg = DomainRegistry.get_registration(node_type)
+        reg.jsonl_handler_class = RegistryLookupJsonlHandler
+        reg.api_handler_class = RegistryLookupApiHandler
+        reg.db_handler_class = RegistryLookupDbHandler
+    else:
+        DomainRegistry.register(
+            node_type=node_type,
+            schema=RegistryLookupSchema,
+            node_class=RegistryLookupNode,
+            jsonl_handler_class=RegistryLookupJsonlHandler,
+            api_handler_class=RegistryLookupApiHandler,
+            db_handler_class=RegistryLookupDbHandler,
+        )
+
+    assert DomainRegistry.get_handler(node_type, "jsonl") is RegistryLookupJsonlHandler
+    assert DomainRegistry.get_handler(node_type, "api") is RegistryLookupApiHandler
+    assert DomainRegistry.get_handler(node_type, "db") is RegistryLookupDbHandler