ecm_sync_system/sync_state_machine/datasource/datasource.py

"""
DataSource V2 - 状态管理 + Handler 编排层

DataSource 职责：
1. 注册和管理 Handler
2. 按依赖顺序加载数据
3. 执行同步（调用 Handler）
4. 管理节点状态流转（PENDING → IN_PROGRESS → SUCCESS/FAILED）
5. 处理异步任务轮询
6. 统一异常处理

DataSource 不负责：
- ❌ 后端 API 调用（由 Handler 负责）
- ❌ ID 映射（由 Strategy 负责）
- ❌ 业务逻辑（由 Strategy 负责）
"""
from __future__ import annotations

import asyncio
import logging
from typing import Any, List, Dict, Optional, TYPE_CHECKING
from abc import ABC
from pathlib import Path

if TYPE_CHECKING:
    from ..common.collection import DataCollection
    from .handler import NodeHandler
    from ..common.sync_node import SyncNode
    from ..common.types import SyncAction
    from .task_result import TaskStatus

from ..common.types import SyncStatus
from ..common.type_safety import get_pydantic_model_fields
from .handler import ensure_handler_compat
from .task_result import TaskResult
from ..engine import (
    StateMachineConfig,
    StateMachineRuntime,
    e40_sync_execute,
)


logger = logging.getLogger(__name__)


class BaseDataSource(ABC):
    """
    数据源基类 V2（状态管理 + 编排层）
    
    核心流程：
    1. 注册 Handler：每个节点类型对应一个 Handler
    2. 加载数据：按依赖顺序调用 Handler.load()
    3. 执行同步：根据 node.action 调用 Handler 的 create/update/delete
    4. 状态管理：根据 TaskResult 更新 node.status
    5. 异步轮询：处理 IN_PROGRESS 节点
    
    示例：
        # 1. 初始化 DataSource
        datasource = MyDataSource()
        
        # 2. 注册 Handler
        datasource.register_handler(project_handler)
        datasource.register_handler(contract_handler)
        
        # 3. 设置 Collection
        datasource.set_collection(collection)
        
        # 4. 加载数据（按顺序）
        await datasource.load_all(order=["project", "contract"])
        
        # 5. 执行同步（按顺序）
        await datasource.sync_all(order=["project", "contract"])
    """
    
    def __init__(self, poll_max_retries: int = 1, poll_interval: float = 0.5):
        self._handlers: Dict[str, "NodeHandler"] = {}
        self._collection: Optional["DataCollection"] = None
        self._target_project_ids: List[str] = []
        
        # 按节点类型统计信息
        # {"company": {"loaded": 10, "synced": 10, "success": 10, "failed": 0}, ...}
        self._stats: Dict[str, Dict[str, int]] = {}
        # 记录本次 sync 的动作统计（CREATE/UPDATE/DELETE/NONE）
        self._action_summary: Dict[str, Dict[str, Dict[str, int]]] = {}
        # 记录本次 sync 中的节点动作（用于异步完成后的统计）
        self._action_by_node_id: Dict[str, "SyncAction"] = {}
        self._poll_max_retries = max(1, poll_max_retries)
        self._poll_interval = max(0, poll_interval)
        self._sm_runtime: Optional[StateMachineRuntime] = None

    def _ensure_sm_runtime(self) -> StateMachineRuntime:
        if self._sm_runtime is None:
            cfg_path = Path(__file__).resolve().parents[1] / "config" / "node_state_machine.yaml"
            self._sm_runtime = StateMachineRuntime(StateMachineConfig.load(cfg_path))
        return self._sm_runtime

    def set_state_machine_runtime(self, runtime: StateMachineRuntime) -> None:
        """注入统一状态机 runtime（通常由 pipeline 负责）。"""
        self._sm_runtime = runtime

    def _apply_sync_execute_state(
        self,
        node: "SyncNode",
        *,
        handler_status: "TaskStatus",
        poll_timeout: bool = False,
        detail: str = "",
        action_override: Optional["SyncAction"] = None,
        result_data_id: Optional[str] = None,
    ) -> bool:
        action_for_event = action_override or node.action
        decision = e40_sync_execute(
            self._ensure_sm_runtime(),
            node=node,
            handler_result=handler_status.name,
            action=action_for_event.value,
            poll_timeout=poll_timeout,
            result_data_id=result_data_id,
        )
        if decision is None:
            if detail:
                node.append_log(
                    f"执行状态未变更: action={action_for_event.value}, result={handler_status.name}, detail={detail}"
                )
            return False
        if detail:
            node.append_log(
                f"执行状态写回: action={action_for_event.value}, result={handler_status.name}, detail={detail}"
            )
        return True

    def _compose_node_error(self, node: "SyncNode", *, source: str, detail: str) -> str:
        return (
            f"{source} | node={node.node_id} | type={node.node_type} | "
            f"action={node.action.value} | status={node.status.value} | detail={detail}"
        )

    def _on_node_loaded(self, handler: "NodeHandler", node: "SyncNode") -> None:
        """Datasource-specific hook when a loaded node is upserted into collection."""
        return

    async def _intercept_success_result(
        self,
        handler: "NodeHandler",
        node: "SyncNode",
        result: "TaskResult",
        async_tasks: Dict[str, str],
    ) -> bool:
        """Datasource-specific hook to intercept SUCCESS result handling.

        Return True if handled/consumed by subclass and no further default SUCCESS flow is needed.
        """
        return False

    def _on_in_progress_result(self, handler: "NodeHandler", node: "SyncNode", result: "TaskResult") -> None:
        """Datasource-specific hook after an IN_PROGRESS task is queued."""
        return

    def _on_poll_result(self, handler: "NodeHandler", node: "SyncNode", task_id: str, result: "TaskResult") -> None:
        """Datasource-specific hook after polling result is applied to node."""
        return

    def _on_failed_result(self, handler: "NodeHandler", node: "SyncNode", result: "TaskResult") -> None:
        """Datasource-specific hook after a FAILED result is applied to node."""
        return

    async def _handle_success_result(self, node: "SyncNode", result: "TaskResult") -> None:
        previous_data_id = node.data_id
        ok = self._apply_sync_execute_state(
            node,
            handler_status=result.status,
            detail="远程操作成功完成",
            result_data_id=result.data_id,
        )
        if not ok:
            raise RuntimeError(f"[{node.node_type}] SUCCESS result cannot map to state machine: node={node.node_id}")
        self._record_action_result(node, result.status)
        await self._apply_success_result(node, result.data_id, previous_data_id=previous_data_id)
        node.error = None

    async def _handle_failed_result(self, node: "SyncNode", result: "TaskResult") -> None:
        detail = result.error or "Unknown error"
        ok = self._apply_sync_execute_state(
            node,
            handler_status=result.status,
            detail=f"远程操作失败: {detail}",
        )
        if not ok:
            raise RuntimeError(f"[{node.node_type}] FAILED result cannot map to state machine: node={node.node_id}")
        self._record_action_result(node, result.status)
        node.error = self._compose_node_error(node, source="handler_failed", detail=detail)

    async def _handle_skipped_result(self, node: "SyncNode", result: "TaskResult") -> None:
        from .task_result import TaskStatus as _TaskStatus

        original_action = self._action_by_node_id.get(node.node_id)
        if original_action is None or original_action.value != "UPDATE":
            raise RuntimeError(
                f"[{node.node_type}] SKIPPED is only allowed for UPDATE action: node={node.node_id}, action={original_action.value if original_action is not None else None}"
            )

        ok = self._apply_sync_execute_state(
            node,
            handler_status=_TaskStatus.SKIPPED,
            detail="远程操作被跳过",
            action_override=original_action,
        )
        if not ok:
            raise RuntimeError(f"[{node.node_type}] SKIPPED result cannot map to state machine: node={node.node_id}")

        self._record_action_result(node, _TaskStatus.SKIPPED)
        if result.sync_log:
            node.append_log(f"同步跳过: {result.sync_log}")
        node.error = None

    async def _handle_in_progress_result(
        self,
        node: "SyncNode",
        result: "TaskResult",
        async_tasks: Dict[str, str],
    ) -> None:
        if not result.task_id:
            node.error = self._compose_node_error(
                node,
                source="handler_in_progress_invalid",
                detail="missing task_id",
            )
            node.append_log("异步任务返回无 task_id，已忽略")
            return

        if not result.node_id:
            node.error = self._compose_node_error(
                node,
                source="handler_in_progress_invalid",
                detail="missing node_id",
            )
            node.append_log("异步任务返回无 node_id，已忽略")
            return

        async_tasks[result.node_id] = result.task_id
        ok = self._apply_sync_execute_state(
            node,
            handler_status=result.status,
            detail=f"异步任务进行中: {result.task_id}",
        )
        if not ok:
            node.append_log(f"异步任务进行中（无状态变更）: {result.task_id}")

    def set_polling_config(self, max_retries: int, interval: float = 0.5) -> None:
        """设置异步轮询参数（默认仅轮询一次）"""
        self._poll_max_retries = max(1, max_retries)
        self._poll_interval = max(0, interval)
    
    def reset_stats(self) -> None:
        """重置统计信息"""
        self._stats = {}
        self._action_summary = {}
        self._action_by_node_id = {}
    
    def get_stats(self) -> Dict[str, Dict[str, int]]:
        """
        获取按节点类型的统计信息
        
        Returns:
            dict: {"company": {"loaded": 10, "synced": 10, "success": 10, "failed": 0}, ...}
        """
        return self._stats.copy()

    def get_action_summary(self) -> Dict[str, Dict[str, Dict[str, int]]]:
        """
        获取本次 sync 的动作统计（按节点类型）。

        Returns:
            dict: {
                "company": {
                    "create": {"total": 10, "success": 8, "failed": 2},
                    "update": {"total": 2, "success": 2, "failed": 0},
                    "delete": {"total": 0, "success": 0, "failed": 0},
                    "none": {"total": 5, "success": 0, "failed": 0},
                    "total": {"total": 17, "success": 10, "failed": 2},
                },
                ...
            }
        """
        return {k: {kk: vv.copy() for kk, vv in v.items()} for k, v in self._action_summary.items()}
    
    def _init_node_stats(self, node_type: str) -> None:
        """初始化节点类型的统计信息"""
        if node_type not in self._stats:
            self._stats[node_type] = {
                "loaded": 0,
                "synced": 0,
                "success": 0,
                "failed": 0,
            }
        if node_type not in self._action_summary:
            self._action_summary[node_type] = {
                "create": {"total": 0, "success": 0, "failed": 0, "skipped": 0},
                "update": {"total": 0, "success": 0, "failed": 0, "skipped": 0},
                "delete": {"total": 0, "success": 0, "failed": 0, "skipped": 0},
                "none": {"total": 0, "success": 0, "failed": 0, "skipped": 0},
                "total": {"total": 0, "success": 0, "failed": 0, "skipped": 0},
            }

    def _set_action_totals(self, node_type: str, create: int, update: int, delete: int, none: int) -> None:
        summary = self._action_summary[node_type]
        summary["create"]["total"] = create
        summary["update"]["total"] = update
        summary["delete"]["total"] = delete
        summary["none"]["total"] = none
        summary["total"]["total"] = create + update + delete + none

    def _record_action_result(self, node: "SyncNode", result_status: "TaskStatus") -> None:
        from ..common.types import SyncAction
        from .task_result import TaskStatus as _TaskStatus

        # 情况 A: 原始动作就是 NONE
        # 情况 B: Handler 运行时发现没差异，把 action 降级为了 NONE
        action = node.action
        
        # 获取最初锁定的动作（用于修正统计列）
        original_action = self._action_by_node_id.pop(node.node_id, SyncAction.NONE)

        summary = self._action_summary.get(node.node_type)
        if summary is None:
            return

        # 如果当前 action 是 NONE，说明这次“同步”被取消了
        if action == SyncAction.NONE:
            # 如果原始动作不是 NONE，说明发生了降级动作，我们需要把 total 从对应列挪到 none 列
            if original_action != SyncAction.NONE:
                orig_key = original_action.value.lower()
                if orig_key in summary and summary[orig_key]["total"] > 0:
                    summary[orig_key]["total"] -= 1
                    summary["none"]["total"] += 1
            return

        # 正常记录成功/失败/跳过
        key = action.value.lower()
        if result_status == _TaskStatus.SUCCESS:
            summary[key]["success"] += 1
            summary["total"]["success"] += 1
        elif result_status == _TaskStatus.FAILED:
            summary[key]["failed"] += 1
            summary["total"]["failed"] += 1
        elif result_status == _TaskStatus.SKIPPED:
            summary[key]["skipped"] += 1
            summary["total"]["skipped"] += 1
    
    def set_collection(self, collection: "DataCollection") -> None:
        """
        设置 Collection（在 load_all/sync_all 前调用）
        
        Args:
            collection: 数据集合
        """
        self._collection = collection
        collection.set_state_machine_runtime(self._ensure_sm_runtime())
        # 同时设置到所有已注册的 handler
        for handler in self._handlers.values():
            handler.set_collection(collection)
    
    # ========== Handler 管理 ==========
    
    def register_handler(self, handler: "NodeHandler") -> None:
        """
        注册 Handler
        
        Args:
            handler: 节点处理器实例
            
        示例：
            datasource.register_handler(ContractNodeHandler(api_client))
        """
        compatible_handler = ensure_handler_compat(handler)
        self._handlers[compatible_handler.node_type] = compatible_handler
        self._apply_target_project_ids_to_handler(compatible_handler)

    def set_target_project_ids(self, target_project_ids: Optional[List[str]]) -> None:
        """设置项目过滤列表，并在注册阶段下发给所有 handler。"""
        normalized = [str(pid) for pid in (target_project_ids or []) if str(pid)]
        self._target_project_ids = normalized
        for handler in self._handlers.values():
            self._apply_target_project_ids_to_handler(handler)

    def _apply_target_project_ids_to_handler(self, handler: "NodeHandler") -> None:
        handler.set_target_project_ids(self._target_project_ids)
    
    def get_handler(self, node_type: str) -> "NodeHandler":
        """获取 Handler"""
        if node_type not in self._handlers:
            raise KeyError(f"Handler for '{node_type}' not registered")
        return self._handlers[node_type]

    async def _upsert_loaded_nodes(
        self,
        handler: "NodeHandler",
        node_type: str,
        nodes: List["SyncNode"]
    ) -> None:
        """将加载的节点写入 Collection（按 data_id 优先命中已有节点）"""
        if self._collection is None:
            raise RuntimeError("Collection not set. Call set_collection() first.")

        for node in nodes:
            existing_node = None
            if node.data_id:
                existing_node = self._collection.get_by_data_id(node_type, node.data_id)
            load_source = f"{self.__class__.__name__}.load"

            if existing_node is not None:
                data_payload = node.get_data()
                existing_node.depend_ids = list(node.depend_ids or [])
                if node.context:
                    existing_node.context.update(node.context)
                existing_node.set_data(data_payload)
                existing_node.set_origin_data(data_payload)
                if existing_node.context.pop("_loaded_from_persistence", False):
                    existing_node.append_log(f"加载来源: persistence -> {load_source} 覆盖刷新")
                else:
                    existing_node.append_log(f"加载来源: {load_source} 覆盖刷新")

                await self._collection.update(existing_node)
                self._on_node_loaded(handler, existing_node)
            else:
                node.append_log(f"加载来源: {load_source} 新增节点")
                await self._collection.add(node)
                self._on_node_loaded(handler, node)
    
    # ========== 数据加载 ==========
    
    async def load_all(
        self,
        order: Optional[List[str]] = None,
        data_type: Optional[str] = None
    ) -> None:
        """
        按依赖顺序加载所有数据
        
        注意：调用前需先 set_collection()
        
        Args:
            order: 节点类型顺序（如 ["project", "contract", "construction"])
            data_type: 仅加载单一类型（与 order 互斥）
            
        工作流程：
        1. 按顺序遍历每个节点类型
        2. 调用 Handler.load(collection)
        3. Handler 可以从 collection 查询依赖节点（如 Contract 查询 Project）
        4. 使用 Handler.create_node() 创建 SyncNode
        5. 添加到 Collection
        
        示例：
            datasource.set_collection(collection)
            await datasource.load_all(order=["project", "contract"])
        """
        if self._collection is None:
            raise RuntimeError("Collection not set. Call set_collection() first.")
        
        if data_type is not None:
            order_to_load = [data_type]
        elif order is not None:
            order_to_load = order
        else:
            order_to_load = list(self._handlers.keys())

        for node_type in order_to_load:
            handler = self.get_handler(node_type)

            # 初始化统计
            self._init_node_stats(node_type)

            try:
                # Handler 加载数据并创建节点
                nodes = await handler.load()

                print(f"Loaded nodes for {node_type}: {len(nodes)}")

                # 将节点写入 Collection：优先按 data_id 命中已有节点并更新
                await self._upsert_loaded_nodes(handler, node_type, nodes)

                # 统计加载的节点数
                self._stats[node_type]["loaded"] += len(nodes)
            except Exception as exc:
                print(f"❌ [{node_type}] 加载失败: {exc}")
                continue
    
    # ========== 同步执行 ==========
    
    async def sync_all(
        self,
        order: Optional[List[str]] = None,
        data_type: Optional[str] = None,
        poll_async_tasks: bool = True
    ) -> Dict[str, Dict[str, str]]:
        """
        按依赖顺序执行同步
        
        注意：调用前需先 set_collection()
        
        Args:
            order: 节点类型顺序
            data_type: 仅同步单一类型（与 order 互斥）
            
        工作流程：
        1. 按顺序遍历每个节点类型
        2. 筛选需要同步的节点（status=PENDING, action!=NONE）
        3. 调用 Handler 的 create/update/delete
        4. 根据 TaskResult 更新节点状态
        5. 处理异步任务轮询
        
        状态流转：
        - PENDING + action=CREATE/UPDATE/DELETE → IN_PROGRESS → SUCCESS/FAILED
        - PENDING + action=NONE → 保持 PENDING
        - DEPENDENCY_ERROR / ABNORMAL / WARNING → SKIPPED
        
        示例：
            datasource.set_collection(collection)
            await datasource.sync_all(order=["project", "contract"])
        """
        if self._collection is None:
            raise RuntimeError("Collection not set. Call set_collection() first.")
        
        if data_type is not None:
            order_to_sync = [data_type]
        elif order is not None:
            order_to_sync = order
        else:
            order_to_sync = list(self._handlers.keys())

        async_tasks_by_type: Dict[str, Dict[str, str]] = {}

        for node_type in order_to_sync:
            handler = self.get_handler(node_type)
            
            # 批量同步该类型的所有节点
            async_tasks = await self._sync_nodes(handler, poll_async_tasks=poll_async_tasks)
            if async_tasks:
                async_tasks_by_type[node_type] = async_tasks

        return async_tasks_by_type
    
    async def _sync_nodes(
        self,
        handler: "NodeHandler",
        poll_async_tasks: bool = True
    ) -> Dict[str, str]:
        """
        同步一批节点（批量接口）
        
        Args:
            handler: 节点处理器
            
        流程：
        1. DataSource 筛选需要处理的节点
        2. 调用 handler.create_all(nodes)
        3. 调用 handler.update_all(nodes)
        4. 调用 handler.delete_all(nodes)
        5. 根据 List[TaskResult] 更新节点状态
        6. 收集并轮询异步任务
        """
        from ..common.types import SyncAction, SyncStatus
        from .task_result import TaskStatus as _TaskStatus
        
        if self._collection is None:
            raise RuntimeError("Collection not set. Call set_collection() first.")
        collection = self._collection
        
        # 初始化统计
        self._init_node_stats(handler.node_type)
        
        async_tasks: Dict[str, str] = {}  # {node_id: task_id}
        
        # DataSource 负责筛选需要创建的节点（排除已失败的节点）
        create_nodes = [
            n
            for n in collection.filter_by_state_ids(
                node_type=handler.node_type,
                state_ids=["S06"],
            )
            if n.action == SyncAction.CREATE
        ]
        
        # DataSource 负责筛选需要更新的节点（排除已失败的节点）
        update_nodes = [
            n
            for n in collection.filter_by_state_ids(
                node_type=handler.node_type,
                state_ids=["S07"],
            )
            if n.action == SyncAction.UPDATE
        ]
        
        # DataSource 负责筛选需要删除的节点（排除已失败的节点）
        delete_nodes = [
            n
            for n in collection.filter_by_state_ids(
                node_type=handler.node_type,
                state_ids=["S09"],
            )
            if n.action == SyncAction.DELETE
        ]

        if delete_nodes:
            raise NotImplementedError(
                f"[{handler.node_type}] delete path is not implemented yet, "
                f"but found {len(delete_nodes)} node(s) in S09/DELETE."
            )

        # 记录本次 sync 的动作统计（在 action 被重置前）
        for node in create_nodes:
            self._action_by_node_id[node.node_id] = SyncAction.CREATE
        for node in update_nodes:
            self._action_by_node_id[node.node_id] = SyncAction.UPDATE
        for node in delete_nodes:
            self._action_by_node_id[node.node_id] = SyncAction.DELETE

        none_count = len(
            self._collection.filter(
                node_type=handler.node_type,
                node_filter={"action": SyncAction.NONE}
            )
        )
        self._set_action_totals(
            handler.node_type,
            create=len(create_nodes),
            update=len(update_nodes),
            delete=len(delete_nodes),
            none=none_count,
        )

        # 批量创建（传入节点列表）
        if create_nodes:
            logger.info(f"[{handler.node_type}] create提交开始: nodes={len(create_nodes)}")
            try:
                create_results = await handler.create_all(create_nodes)
                create_success = 0
                create_in_progress = 0
                create_failed = 0
                create_skipped = 0
                for result in create_results:
                    if result.status == _TaskStatus.SUCCESS:
                        create_success += 1
                    elif result.status == _TaskStatus.IN_PROGRESS:
                        create_in_progress += 1
                    elif result.status == _TaskStatus.FAILED:
                        create_failed += 1
                    elif result.status == _TaskStatus.SKIPPED:
                        create_skipped += 1
                    await self._handle_task_result(handler, result, async_tasks)
                logger.info(
                    f"[{handler.node_type}] create提交完成: success={create_success}, "
                    f"in_progress={create_in_progress}, failed={create_failed}, skipped={create_skipped}"
                )
            except Exception as e:
                print(f"❌ [{handler.node_type}] create_all 异常: {e}")
                for node in create_nodes:
                    ok = self._apply_sync_execute_state(
                        node,
                        handler_status=_TaskStatus.FAILED,
                        detail=str(e),
                    )
                    if not ok:
                        raise RuntimeError(f"[{handler.node_type}] create_all exception cannot map to state machine")
        
        # 批量更新（传入节点列表）
        if update_nodes:
            try:
                update_results = await handler.update_all(update_nodes)
                for result in update_results:
                    await self._handle_task_result(handler, result, async_tasks)
            except Exception as e:
                print(f"❌ [{handler.node_type}] update_all 异常: {e}")
                for node in update_nodes:
                    ok = self._apply_sync_execute_state(
                        node,
                        handler_status=_TaskStatus.FAILED,
                        detail=str(e),
                    )
                    if not ok:
                        raise RuntimeError(f"[{handler.node_type}] update_all exception cannot map to state machine")
        
        # 批量删除（delete stage 当前未实现，检测到 S09 会在上方直接 fail-fast）
        
        # 轮询异步任务
        if async_tasks and poll_async_tasks:
            logger.info(
                f"[{handler.node_type}] create已提交，push_id检查即将开始: pending={len(async_tasks)}, "
                f"first_wait={self._poll_interval:.3f}s"
            )
            await self._poll_async_tasks(handler, async_tasks)
        
        # 统计同步结果
        all_nodes = self._collection.filter(node_type=handler.node_type)
        synced_count = len([n for n in all_nodes if n.action != SyncAction.NONE])
        success_count = len([n for n in all_nodes if n.status == SyncStatus.SUCCESS])
        failed_count = len([n for n in all_nodes if n.status == SyncStatus.FAILED])
        
        self._stats[handler.node_type]["synced"] = synced_count
        self._stats[handler.node_type]["success"] = success_count
        self._stats[handler.node_type]["failed"] = failed_count

        return async_tasks

    async def poll_async_tasks(self, async_tasks_by_type: Dict[str, Dict[str, str]]) -> None:
        """由 Pipeline 调度的异步任务轮询"""
        for node_type, async_tasks in async_tasks_by_type.items():
            if not async_tasks:
                continue
            handler = self.get_handler(node_type)
            await self._poll_async_tasks(handler, async_tasks)
    
    async def _handle_task_result(
        self,
        handler: "NodeHandler",
        result: "TaskResult",
        async_tasks: Dict[str, str]
    ) -> None:
        """
        处理 Handler 返回的 TaskResult
        
        Args:
            collection: Collection 上下文
            result: Handler 返回的结果（包含 node_id）
            async_tasks: 异步任务字典
        """
        from .task_result import TaskStatus as _TaskStatus
        
        if not result.node_id:
            # 没有 node_id，无法定位节点
            return
        
        if self._collection is None:
            # Collection 未设置，无法处理结果
            return
        
        # 从 self._collection 中查找节点
        node = self._collection.get(result.node_id)
        if not node:
            return

        if result.status == _TaskStatus.SUCCESS:
            if await self._intercept_success_result(handler, node, result, async_tasks):
                return
        
        if result.status == _TaskStatus.SUCCESS:
            await self._handle_success_result(node, result)
            return

        if result.status == _TaskStatus.FAILED:
            self._on_failed_result(handler, node, result)
            await self._handle_failed_result(node, result)
            return

        if result.status == _TaskStatus.SKIPPED:
            await self._handle_skipped_result(node, result)
            return

        if result.status == _TaskStatus.IN_PROGRESS:
            self._on_in_progress_result(handler, node, result)
            await self._handle_in_progress_result(node, result, async_tasks)
            return

    def _write_back_data_id_to_data(self, node, data_id: str) -> None:
        data = node.get_data()
        if data is None:
            return

        schema = node.get_schema()
        candidate_keys = ["id", "_id"]
        model_fields = get_pydantic_model_fields(schema)

        for key in candidate_keys:
            if key in data:
                data[key] = data_id
                node.set_data(data)
                return
            if key in model_fields:
                data[key] = data_id
                node.set_data(data)
                return

    async def _apply_success_result(self, node, data_id: Optional[str], previous_data_id: str) -> None:
        from ..common.types import SyncAction

        if data_id and node.action == SyncAction.CREATE:
            if node.data_id != data_id:
                raise RuntimeError(
                    f"[{node.node_type}] data_id must be updated in event apply path: expected={data_id}, actual={node.data_id}"
                )
            self._write_back_data_id_to_data(node, data_id)

            if self._collection is not None and data_id != previous_data_id:
                await self._collection.update(node)

        if node.action == SyncAction.UPDATE:
            current_data = node.get_data()
            if current_data is not None:
                node.set_origin_data(current_data)

        if node.action == SyncAction.DELETE:
            if self._collection is None:
                return
            await self._collection.delete(node.node_id)
            return

        # 成功后 action 是否重置由上层策略决定
    
    async def _poll_async_tasks(
        self,
        handler: "NodeHandler",
        async_tasks: Dict[str, str]
    ) -> None:
        """
        轮询异步任务直到全部完成
        
        Args:
            handler: 节点处理器
            collection: Collection 上下文
            async_tasks: {node_id: task_id} 映射
            
        流程：
        1. 批量调用 Handler.poll_tasks()
        2. 更新节点状态
        3. 如果还有未完成任务，等待后重试
        """
        from .task_result import TaskStatus as _TaskStatus
        
        max_retries = self._poll_max_retries
        retry_count = 0
        poll_interval = self._poll_interval  # 轮询间隔（秒）
        
        if self._collection is None:
            # Collection 未设置，无法轮询任务
            return

        if async_tasks and poll_interval > 0:
            logger.info(
                f"[{handler.node_type}] push_id首次检查前等待 {poll_interval:.3f}s, pending={len(async_tasks)}"
            )
            await asyncio.sleep(poll_interval)
        
        while async_tasks and retry_count < max_retries:
            logger.info(
                f"[{handler.node_type}] push_id检查 {retry_count + 1}/{max_retries}, pending={len(async_tasks)}"
            )
            # 批量查询
            task_ids = list(async_tasks.values())
            results = await handler.poll_tasks(task_ids)
            
            # 处理结果
            completed_nodes = []
            for node_id, task_id in list(async_tasks.items()):
                if task_id not in results:
                    continue
                
                result = results[task_id]
                node = self._collection.get(node_id)
                if not node:
                    completed_nodes.append(node_id)
                    continue
                
                if result.status == _TaskStatus.SUCCESS:
                    self._on_poll_result(handler, node, task_id, result)
                    previous_data_id = node.data_id
                    ok = self._apply_sync_execute_state(
                        node,
                        handler_status=result.status,
                        detail=f"异步任务成功完成: {task_id}",
                        result_data_id=result.data_id,
                    )
                    if not ok:
                        raise RuntimeError(f"[{node.node_type}] async SUCCESS cannot map to state machine: node={node.node_id}")
                    self._record_action_result(node, result.status)
                    await self._apply_success_result(node, result.data_id, previous_data_id=previous_data_id)
                    node.error = None
                    logger.info(
                        f"[{handler.node_type}] polling成功: node={node.node_id}, push_id={task_id}, data_id={result.data_id or node.data_id}"
                    )
                    completed_nodes.append(node_id)
                    
                elif result.status == _TaskStatus.FAILED:
                    self._on_poll_result(handler, node, task_id, result)
                    ok = self._apply_sync_execute_state(
                        node,
                        handler_status=result.status,
                        detail=f"异步任务失败: {result.error}",
                    )
                    if not ok:
                        raise RuntimeError(f"[{node.node_type}] async FAILED cannot map to state machine: node={node.node_id}")
                    self._record_action_result(node, result.status)
                    node.error = result.error
                    completed_nodes.append(node_id)
                
                # IN_PROGRESS 继续等待
            
            # 移除已完成的任务
            for node_id in completed_nodes:
                async_tasks.pop(node_id, None)
            
            # 还有未完成任务，等待后重试
            if async_tasks:
                retry_count += 1
        
        # 超时未完成的任务标记为失败
        if async_tasks:
            for node_id in async_tasks:
                node = self._collection.get(node_id)
                if node:
                    timeout_msg = f"异步任务超时 (等待 {max_retries * poll_interval} 秒后仍未完成)"
                    ok = self._apply_sync_execute_state(
                        node,
                        handler_status=_TaskStatus.FAILED,
                        poll_timeout=True,
                        detail=timeout_msg,
                    )
                    if not ok:
                        raise RuntimeError(f"[{node.node_type}] poll timeout cannot map to state machine: node={node.node_id}")
                    node.error = f"Async task timeout after {max_retries * poll_interval} seconds"
                    self._record_action_result(node, _TaskStatus.FAILED)
    
    # ========== 统计信息 ==========
    
    def get_sync_summary(self, collection: "DataCollection") -> Dict[str, Any]:
        """
        获取同步统计信息
        
        Returns:
            {
                "total": 总节点数,
                "success": 成功数,
                "failed": 失败数,
                "in_progress": 进行中,
                "skipped": 跳过数,
                "by_type": {...}  # 按类型统计
            }
        """
        nodes = list(collection._nodes.values())
        
        summary = {
            "total": len(nodes),
            "success": sum(1 for n in nodes if n.status == SyncStatus.SUCCESS),
            "failed": sum(1 for n in nodes if n.status == SyncStatus.FAILED),
            "in_progress": sum(1 for n in nodes if n.status == SyncStatus.IN_PROGRESS),
            "skipped": sum(1 for n in nodes if n.status == SyncStatus.SKIPPED),
            "by_type": {}
        }
        
        # 按类型统计
        for node_type in self._handlers.keys():
            type_nodes = collection.filter(node_type=node_type)
            summary["by_type"][node_type] = {
                "total": len(type_nodes),
                "success": sum(1 for n in type_nodes if n.status == SyncStatus.SUCCESS),
                "failed": sum(1 for n in type_nodes if n.status == SyncStatus.FAILED),
            }
        
        return summary

    async def close(self) -> None:
        """关闭数据源资源（子类可覆盖以实现特定的清理逻辑）"""
        pass  # 基类默认空实现