优化sync_system代码质量

sync_state_machine/sync_system/strategy_ops/__init__.py

@@ -1,5 +1,11 @@
 from .create_ops import add_create_action, run_create, finalize_peer_creating_sources
-from .update_ops import add_update_action, run_update
+from .update_ops import (
+    BoundNodePair,
+    add_update_action,
+    collect_bound_node_pairs,
+    prepare_directional_update,
+    run_update,
+)
 from .delete_ops import run_delete
 from .bind_ops import (
     run_bind,
@@ -10,13 +16,16 @@ from .bind_ops import (
     check_dependency_satisfied,
     collect_dependency_errors,
 )
-from .reset_ops import get_phase1_reset_defaults, run_phase2_cleanup
+from .reset_ops import run_phase2_cleanup
 
 __all__ = [
     "add_create_action",
     "run_create",
     "finalize_peer_creating_sources",
+    "BoundNodePair",
     "add_update_action",
+    "collect_bound_node_pairs",
+    "prepare_directional_update",
     "run_update",
     "run_delete",
     "run_bind",
@@ -26,6 +35,5 @@ __all__ = [
     "phase_auto_binding",
     "check_dependency_satisfied",
     "collect_dependency_errors",
-    "get_phase1_reset_defaults",
     "run_phase2_cleanup",
 ]

sync_state_machine/sync_system/strategy_ops/bind_ops.py

@@ -16,6 +16,10 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+def get_id_field_hints(strategy) -> Dict[str, str]:
+    return dict(strategy.config.depend_fields) if strategy.config.depend_fields else {}
+
+
 def _stable_pick_bind_pair(local_candidates: List["SyncNode"], remote_candidates: List["SyncNode"]):
     local_node = sorted(local_candidates, key=lambda item: ((item.data_id or ""), item.node_id))[0]
     remote_node = sorted(remote_candidates, key=lambda item: ((item.data_id or ""), item.node_id))[0]
@@ -399,7 +403,7 @@ async def phase_auto_binding(strategy, local_nodes: List["SyncNode"], remote_nod
             resolved_key_dict, resolve_error = await resolve_id_fields_in_key_dict(
                 key_dict,
                 id_resolver,
-                strategy._get_id_field_hints(),
+                get_id_field_hints(strategy),
             )
             if not resolved_key_dict:
                 if node.binding_status == BindingStatus.MISSING:

sync_state_machine/sync_system/strategy_ops/post_check_ops.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Dict, List
+from typing import Any, Callable, Dict, List
 
 from ...sync_system.resolve_ids import IDResolver
 from ...validation import SchemaDiffValidator
@@ -27,11 +27,11 @@ def _append_post_check_error(node, message: str) -> None:
 async def evaluate_consistency_for_node_type(
     *,
     node_type: str,
-    strategy,
     local_collection,
     remote_collection,
     binding_manager,
     logger,
+    normalize_compare_payload: Callable[[Dict[str, Any] | None, Dict[str, str] | None], Dict[str, Any]],
     depend_fields: Dict[str, str] | None = None,
     compare_to_remote: bool = True,
     ignore_fields: List[str] | None = None,
@@ -44,6 +44,7 @@ async def evaluate_consistency_for_node_type(
         schema_name=node_type,
         ignore_fields=ignore_fields,
     )
+    mismatches: list[str] = []
     data_id_map = await build_data_id_normalization_map(
         node_type=node_type,
         local_collection=local_collection,
@@ -93,8 +94,8 @@ async def evaluate_consistency_for_node_type(
                 if resolved_remote is not None:
                     remote_data = resolved_remote
 
-        local_payload = strategy.normalize_compare_payload(local_data, data_id_map)
-        remote_payload = strategy.normalize_compare_payload(remote_data, data_id_map)
+        local_payload = normalize_compare_payload(local_data, data_id_map)
+        remote_payload = normalize_compare_payload(remote_data, data_id_map)
         diff_map = validator.compare_payloads(
             local_payload,
             remote_payload,
@@ -129,6 +130,7 @@ async def evaluate_consistency_for_node_type(
         "fields": report["fields"],
         "schema_name": report["schema_name"],
         "ignore_fields": report["ignore_fields"],
+        "mismatches": mismatches,
     }
 
     return result

sync_state_machine/sync_system/strategy_ops/reset_ops.py

@@ -14,13 +14,6 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-def get_phase1_reset_defaults() -> Dict[str, Any]:
-    return {
-        "binding_status": BindingStatus.UNCHECKED,
-        "error": None,
-    }
-
-
 async def _resolve_local_id_for_unbind(
     *,
     node_type: str,

sync_state_machine/sync_system/strategy_ops/update_ops.py

@@ -1,17 +1,19 @@
 from __future__ import annotations
 
 import copy
+from dataclasses import dataclass
 import logging
-from typing import TYPE_CHECKING, Callable, List, Optional
+from typing import TYPE_CHECKING, Any, Iterable, List, Optional, Sequence
 
 from ...common.types import SyncAction
 from ...sync_system.config import UpdateDirection
 from ...sync_system.resolve_ids import IDResolver
 from ...engine import decision_note, e30_update_prepare
+from ...validation import SchemaDiffValidator
 from .compare_ops import (
     build_data_id_normalization_map,
-    normalized_data_for_compare,
     collect_payload_diffs,
+    normalized_data_for_compare,
 )
 
 if TYPE_CHECKING:
@@ -21,6 +23,36 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+def _build_schema_diff_validator(strategy) -> SchemaDiffValidator:
+    schema = getattr(strategy, "schema", None)
+    schema_name = getattr(schema, "__name__", None) or str(getattr(strategy, "node_type", "unknown"))
+    return SchemaDiffValidator(
+        schema_name=schema_name,
+        ignore_fields=strategy.config.compare_ignore_fields,
+        ignore_list_item_fields=strategy.config.post_check_ignore_list_item_fields,
+    )
+
+
+def _get_active_schema_diff_validator(strategy) -> SchemaDiffValidator:
+    validator = getattr(strategy, "_active_schema_diff_validator", None)
+    if validator is not None:
+        return validator
+    return _build_schema_diff_validator(strategy)
+
+
+def emit_schema_diff_report(strategy, validator: SchemaDiffValidator) -> None:
+    if not validator.has_records():
+        return
+    for line in validator.format_summary_lines():
+        logger.info(f"[{strategy.node_type}] validate {line}")
+
+
+@dataclass(frozen=True)
+class BoundNodePair:
+    local_node: "SyncNode"
+    remote_node: "SyncNode"
+
+
 def _extract_snapshot_steps(snapshot) -> List:
     if not isinstance(snapshot, dict):
         return []
@@ -41,16 +73,10 @@ def _safe_repr(value, *, max_len: int = 120) -> str:
 
 def default_needs_update(
     strategy,
-    source_node,
-    target_node,
-    resolved_data=None,
+    source_data,
+    target_data,
     data_id_map=None,
 ) -> bool:
-    if resolved_data is None:
-        raise ValueError(f"[{strategy.node_type}] resolved_data is required for differential sync.")
-
-    source_data = resolved_data
-    target_data = target_node.get_data()
     if source_data is None or target_data is None:
         return False
 
@@ -67,10 +93,10 @@ def default_needs_update(
         ignore_fields=ignore_fields,
         ignore_list_item_fields=strategy.config.post_check_ignore_list_item_fields,
     )
-    diff_map = strategy.get_schema_diff_validator().compare_payloads(
+    diff_map = _get_active_schema_diff_validator(strategy).compare_payloads(
         normalized_source,
         normalized_target,
-        sample_id=str(target_node.data_id or target_node.node_id),
+        sample_id=str(source_data.get("id") or target_data.get("id") or "unknown"),
         schema_refs=[strategy.schema.__name__],
     )
     is_different = bool(diff_map)
@@ -82,8 +108,8 @@ def default_needs_update(
     return is_different
 
 
-def resolve_needs_update_check(strategy) -> Callable:
-    return strategy._needs_update
+def default_get_node_update_payload(node: "SyncNode") -> dict[str, Any]:
+    return node.get_data() or {}
 
 
 def _collect_diff_descriptions(
@@ -125,235 +151,268 @@ def _collect_diff_descriptions(
     return descriptions
 
 
+async def collect_bound_node_pairs(strategy) -> List[BoundNodePair]:
+    local_nodes = strategy.local_collection.filter_by_state_ids(
+        node_type=strategy.node_type,
+        state_ids=["S01"],
+    )
+    if not local_nodes:
+        return []
+
+    pairs: List[BoundNodePair] = []
+    for local_node in local_nodes:
+        remote_node_id = await strategy.binding_manager.get_remote_id(strategy.node_type, local_node.node_id)
+        if not remote_node_id:
+            continue
+
+        remote_node = strategy.remote_collection.get_by_node_id(remote_node_id)
+        if remote_node is None:
+            continue
+
+        pairs.append(BoundNodePair(local_node=local_node, remote_node=remote_node))
+
+    return pairs
+
+
+async def default_update_pair(
+    strategy,
+    local_node: "SyncNode",
+    remote_node: "SyncNode",
+    data_id_map: Optional[dict[str, str]] = None,
+) -> List["SyncNode"]:
+    direction = strategy.config.update_direction
+    if direction == UpdateDirection.NONE:
+        return []
+
+    updated_node = await prepare_directional_update(
+        strategy,
+        local_node=local_node,
+        remote_node=remote_node,
+        direction=direction,
+        data_id_map=data_id_map,
+    )
+    return [updated_node] if updated_node is not None else []
+
+
+def _sync_pending_approval_snapshot(source_node: "SyncNode", target_node: "SyncNode") -> bool:
+    source_snapshot = source_node.context.get("approval_snapshot") if isinstance(source_node.context, dict) else None
+    target_snapshot = target_node.context.get("approval_snapshot") if isinstance(target_node.context, dict) else None
+    if isinstance(source_snapshot, dict):
+        steps_changed = _steps_payload_changed(source_snapshot, target_snapshot)
+        pending_snapshot = copy.deepcopy(source_snapshot)
+        pending_snapshot["steps_changed"] = steps_changed
+        target_node.context["pending_approval_snapshot"] = pending_snapshot
+        return steps_changed
+
+    target_node.context.pop("pending_approval_snapshot", None)
+    return False
+
+
+def _select_fields(
+    payload: dict[str, Any],
+    *,
+    include_fields: Optional[Sequence[str]] = None,
+    exclude_fields: Optional[Sequence[str]] = None,
+) -> dict[str, Any]:
+    if include_fields:
+        include_set = set(include_fields)
+        return {field_name: value for field_name, value in payload.items() if field_name in include_set}
+    if exclude_fields:
+        exclude_set = set(exclude_fields)
+        return {field_name: value for field_name, value in payload.items() if field_name not in exclude_set}
+    return dict(payload)
+
+
+def build_merged_update_payload(
+    *,
+    source_data: dict[str, Any],
+    target_data: dict[str, Any],
+    include_fields: Optional[Sequence[str]] = None,
+    exclude_fields: Optional[Sequence[str]] = None,
+) -> dict[str, Any]:
+    if include_fields:
+        merged_payload = dict(target_data)
+        for field_name, value in _select_fields(source_data, include_fields=include_fields).items():
+            merged_payload[field_name] = value
+        return merged_payload
+
+    if exclude_fields:
+        exclude_set = set(exclude_fields)
+        merged_payload = dict(source_data)
+        for field_name, value in target_data.items():
+            if field_name in exclude_set:
+                merged_payload[field_name] = value
+        return merged_payload
+
+    return dict(source_data)
+
+
+async def _resolve_update_source_payload(
+    strategy,
+    *,
+    source_node: "SyncNode",
+    source_is_local: bool,
+) -> tuple[Optional[dict[str, Any]], Optional[str]]:
+    source_payload = strategy.get_node_update_payload(source_node)
+    if not source_payload:
+        return None, "Node data is missing or invalid"
+
+    id_resolver = IDResolver(strategy.local_collection, strategy.remote_collection, strategy.binding_manager)
+    return await id_resolver.resolve_and_validate(
+        data=source_payload,
+        node_type=strategy.node_type,
+        depend_fields=dict(strategy.config.depend_fields) if strategy.config.depend_fields else {},
+        to_remote=source_is_local,
+    )
+
+
+async def prepare_directional_update(
+    strategy,
+    *,
+    local_node: "SyncNode",
+    remote_node: "SyncNode",
+    direction: UpdateDirection,
+    data_id_map: Optional[dict] = None,
+    include_fields: Optional[Sequence[str]] = None,
+    exclude_fields: Optional[Sequence[str]] = None,
+) -> Optional["SyncNode"]:
+    if direction == UpdateDirection.NONE:
+        return None
+
+    source_is_local = direction == UpdateDirection.PUSH
+    source_node = local_node if source_is_local else remote_node
+    target_node = remote_node if source_is_local else local_node
+
+    if target_node.action == SyncAction.CREATE:
+        return None
+
+    target_has_data_id = bool(target_node.data_id)
+    resolved_source_data = None
+    resolve_error = None
+    if target_has_data_id:
+        resolved_source_data, resolve_error = await _resolve_update_source_payload(
+            strategy,
+            source_node=source_node,
+            source_is_local=source_is_local,
+        )
+
+    update_id_resolve_ok = resolved_source_data is not None
+    target_payload = strategy.get_node_update_payload(target_node)
+    prepared_payload: dict[str, Any] | None = None
+    has_diff = False
+    diff_descriptions: List[str] = []
+
+    steps_changed = _sync_pending_approval_snapshot(source_node, target_node)
+
+    if update_id_resolve_ok and resolved_source_data is not None:
+        prepared_payload = build_merged_update_payload(
+            source_data=resolved_source_data,
+            target_data=target_payload,
+            include_fields=include_fields,
+            exclude_fields=exclude_fields,
+        )
+        has_diff = bool(
+            strategy.should_update_pair(
+                source_node,
+                target_node,
+                source_data=prepared_payload,
+                target_data=target_payload,
+                data_id_map=data_id_map,
+            )
+        )
+        if has_diff:
+            diff_descriptions = _collect_diff_descriptions(
+                prepared_payload,
+                target_payload,
+                data_id_map=data_id_map or {},
+                ignore_fields=set(strategy.config.compare_ignore_fields),
+                ignore_list_item_fields=dict(strategy.config.post_check_ignore_list_item_fields),
+            )
+
+    if steps_changed:
+        has_diff = True
+        diff_descriptions.append("steps: approval flow changed")
+
+    if target_has_data_id and update_id_resolve_ok and not has_diff:
+        return None
+
+    decision = e30_update_prepare(
+        strategy.ensure_runtime(),
+        source_node=source_node,
+        target_node=target_node,
+        update_enabled=True,
+        target_has_data_id=target_has_data_id,
+        update_id_resolve_ok=update_id_resolve_ok,
+        has_diff=has_diff,
+        id_resolve_detail=resolve_error,
+    )
+    reason = decision_note(decision)
+    if decision.to_state == "S08":
+        target_node.error = reason
+        logger.debug(f"[{strategy.node_type}] 更新准备阻断: target={target_node.node_id}, reason={reason}")
+        return None
+    if decision.to_state != "S07":
+        raise RuntimeError(f"[{strategy.node_type}] unexpected update_prepare state: {decision.to_state}")
+    if prepared_payload is None:
+        raise RuntimeError(f"[{strategy.node_type}] E30(S07) requires resolved update data")
+
+    target_node.set_data(prepared_payload)
+    target_node.error = None
+    diff_text = "; ".join(diff_descriptions) if diff_descriptions else "(未生成差异明细)"
+    target_node.append_log(f"检测到差异: 从节点 {source_node.node_id} 更新数据 | fields={diff_text}")
+    logger.info(f"[{strategy.node_type}] 更新差异字段: target={target_node.node_id}, {diff_text}")
+    logger.debug(f"[{strategy.node_type}] 更新准备完成: target={target_node.node_id}, reason={reason}")
+    return target_node
+
+
 async def add_update_action(
     strategy,
     *,
     from_collection: "DataCollection",
     to_collection: "DataCollection",
-    needs_update_check,
     source_is_local: bool,
     data_id_map: Optional[dict] = None,
-    node_filter: Optional[Callable[["SyncNode"], bool]] = None,
+    include_fields: Optional[Iterable[str]] = None,
+    exclude_fields: Optional[Iterable[str]] = None,
 ) -> List["SyncNode"]:
-    nodes_to_update = from_collection.filter_by_state_ids(
-        node_type=strategy.node_type,
-        state_ids=["S01"],
-    )
-    if not nodes_to_update:
-        return []
+    del from_collection
+    del to_collection
 
-    if node_filter is not None:
-        nodes_to_update = [n for n in nodes_to_update if node_filter(n)]
-    if not nodes_to_update:
-        return []
-
-    id_resolver = IDResolver(strategy.local_collection, strategy.remote_collection, strategy.binding_manager)
     updated_nodes: List["SyncNode"] = []
-    total_nodes = len(nodes_to_update)
-
-    for idx, src_node in enumerate(nodes_to_update, start=1):
-        if idx % 100 == 0 or idx == total_nodes:
-            logger.debug(f"[{strategy.node_type}] 更新检查 {idx}/{total_nodes}")
-
-        src_data = src_node.get_data()
-        if not src_data:
-            continue
-
-        if source_is_local:
-            target_node_id = await strategy.binding_manager.get_remote_id(strategy.node_type, src_node.node_id)
-        else:
-            target_node_id = await strategy.binding_manager.get_local_id(strategy.node_type, src_node.node_id)
-        if not target_node_id:
-            continue
-
-        target_node = to_collection.get_by_node_id(target_node_id)
-        if not target_node or target_node.action == SyncAction.CREATE:
-            continue
-
-        target_has_data_id = bool(target_node.data_id)
-        data = None
-        resolve_error = None
-        if target_has_data_id:
-            data, resolve_error = await id_resolver.resolve_and_validate(
-                data=src_data,
-                node_type=strategy.node_type,
-                depend_fields=dict(strategy.config.depend_fields) if strategy.config.depend_fields else {},
-                to_remote=source_is_local,
-            )
-
-        update_id_resolve_ok = data is not None
-        has_diff = False
-        diff_descriptions: List[str] = []
-        source_snapshot = src_node.context.get("approval_snapshot") if isinstance(src_node.context, dict) else None
-        target_snapshot = target_node.context.get("approval_snapshot") if isinstance(target_node.context, dict) else None
-        steps_changed = False
-        if isinstance(source_snapshot, dict):
-            steps_changed = _steps_payload_changed(source_snapshot, target_snapshot)
-            pending_snapshot = copy.deepcopy(source_snapshot)
-            pending_snapshot["steps_changed"] = steps_changed
-            target_node.context["pending_approval_snapshot"] = pending_snapshot
-        else:
-            target_node.context.pop("pending_approval_snapshot", None)
-
-        if update_id_resolve_ok:
-            try:
-                should_update = needs_update_check(src_node, target_node, resolved_data=data, data_id_map=data_id_map)
-            except TypeError:
-                try:
-                    should_update = needs_update_check(src_node, target_node)
-                except TypeError:
-                    should_update = needs_update_check(src_node)
-            has_diff = bool(should_update)
-            if has_diff:
-                diff_descriptions = _collect_diff_descriptions(
-                    data,
-                    target_node.get_data(),
-                    data_id_map=data_id_map or {},
-                    ignore_fields=set(strategy.config.compare_ignore_fields),
-                    ignore_list_item_fields=dict(strategy.config.post_check_ignore_list_item_fields),
-                )
-
-        if steps_changed:
-            has_diff = True
-            diff_descriptions.append("steps: approval flow changed")
-
-        if target_has_data_id and update_id_resolve_ok and not has_diff:
-            continue
-
-        decision = e30_update_prepare(
-            strategy.ensure_runtime(),
-            source_node=src_node,
-            target_node=target_node,
-            update_enabled=True,
-            target_has_data_id=target_has_data_id,
-            update_id_resolve_ok=update_id_resolve_ok,
-            has_diff=has_diff,
-            id_resolve_detail=resolve_error,
+    direction = UpdateDirection.PUSH if source_is_local else UpdateDirection.PULL
+    for pair in await collect_bound_node_pairs(strategy):
+        updated_node = await prepare_directional_update(
+            strategy,
+            local_node=pair.local_node,
+            remote_node=pair.remote_node,
+            direction=direction,
+            data_id_map=data_id_map,
+            include_fields=list(include_fields) if include_fields is not None else None,
+            exclude_fields=list(exclude_fields) if exclude_fields is not None else None,
         )
-        reason = decision_note(decision)
-        if decision.to_state == "S08":
-            target_node.error = reason
-            logger.debug(f"[{strategy.node_type}] 更新准备阻断: target={target_node.node_id}, reason={reason}")
-            continue
-        if decision.to_state != "S07":
-            raise RuntimeError(f"[{strategy.node_type}] unexpected update_prepare state: {decision.to_state}")
-        if data is None:
-            raise RuntimeError(f"[{strategy.node_type}] E30(S07) requires resolved update data")
-
-        target_node.set_data(data)
-        target_node.error = None
-        diff_text = "; ".join(diff_descriptions) if diff_descriptions else "(未生成差异明细)"
-        target_node.append_log(f"检测到差异: 从节点 {src_node.node_id} 更新数据 | fields={diff_text}")
-        logger.info(f"[{strategy.node_type}] 更新差异字段: target={target_node.node_id}, {diff_text}")
-        logger.debug(f"[{strategy.node_type}] 更新准备完成: target={target_node.node_id}, reason={reason}")
-        updated_nodes.append(target_node)
+        if updated_node is not None:
+            updated_nodes.append(updated_node)
 
     return updated_nodes
 
 
 async def run_update(strategy) -> List["SyncNode"]:
-    updated_nodes: List["SyncNode"] = []
-    strategy.reset_schema_diff_validator()
-    needs_update_check = resolve_needs_update_check(strategy)
+    updated_by_id: dict[str, "SyncNode"] = {}
+    validator = _build_schema_diff_validator(strategy)
+    strategy._active_schema_diff_validator = validator
     data_id_map = await build_data_id_normalization_map(
         node_type=strategy.node_type,
         local_collection=strategy.local_collection,
         remote_collection=strategy.remote_collection,
         binding_manager=strategy.binding_manager,
     )
+    try:
+        for pair in await collect_bound_node_pairs(strategy):
+            nodes = await strategy.update_pair(pair.local_node, pair.remote_node, data_id_map)
+            for node in nodes:
+                updated_by_id[node.node_id] = node
+    finally:
+        strategy._active_schema_diff_validator = None
 
-    field_key = strategy.config.field_direction_key
-    overrides: dict = strategy.config.field_direction_overrides
-
-    if field_key and overrides:
-        # 按方向归组：对每种出现的方向各调一次 add_update_action，传入过滤器。
-        # 覆盖表中未命中的节点沿用 update_direction 作为默认方向。
-        default_direction = strategy.config.update_direction
-        active_directions = set(overrides.values()) | {default_direction}
-
-        for direction in (UpdateDirection.PUSH, UpdateDirection.PULL):
-            if direction not in active_directions:
-                continue
-
-            captured_direction = direction  # 避免闭包捕获循环变量
-            captured_overrides = overrides
-            captured_default = default_direction
-            captured_field_key = field_key
-
-            def make_filter(d=captured_direction, fk=captured_field_key, ov=captured_overrides, dflt=captured_default):
-                def _filter(node: "SyncNode") -> bool:
-                    node_data = node.get_data() or {}
-                    key_val = node_data.get(fk)
-                    effective = ov.get(key_val, dflt)
-                    return effective == d
-                return _filter
-
-            node_filter = make_filter()
-
-            if direction == UpdateDirection.PUSH:
-                nodes = await add_update_action(
-                    strategy,
-                    from_collection=strategy.local_collection,
-                    to_collection=strategy.remote_collection,
-                    needs_update_check=needs_update_check,
-                    source_is_local=True,
-                    data_id_map=data_id_map,
-                    node_filter=node_filter,
-                )
-                updated_nodes.extend(nodes)
-                logger.info(
-                    f"[{strategy.node_type}] Updated {len(nodes)} nodes in remote collection"
-                    f" (Push, field_direction_key={field_key})"
-                )
-
-            elif direction == UpdateDirection.PULL:
-                nodes = await add_update_action(
-                    strategy,
-                    from_collection=strategy.remote_collection,
-                    to_collection=strategy.local_collection,
-                    needs_update_check=needs_update_check,
-                    source_is_local=False,
-                    data_id_map=data_id_map,
-                    node_filter=node_filter,
-                )
-                updated_nodes.extend(nodes)
-                logger.info(
-                    f"[{strategy.node_type}] Updated {len(nodes)} nodes in local collection"
-                    f" (Pull, field_direction_key={field_key})"
-                )
-
-        strategy.emit_schema_diff_report()
-        return updated_nodes
-
-    # ── 无字段级覆盖：原有逻辑不变 ──
-    if strategy.config.update_direction == UpdateDirection.PUSH:
-        nodes = await add_update_action(
-            strategy,
-            from_collection=strategy.local_collection,
-            to_collection=strategy.remote_collection,
-            needs_update_check=needs_update_check,
-            source_is_local=True,
-            data_id_map=data_id_map,
-        )
-        updated_nodes.extend(nodes)
-        logger.info(f"[{strategy.node_type}] Updated {len(nodes)} nodes in remote collection (Push)")
-
-    elif strategy.config.update_direction == UpdateDirection.PULL:
-        nodes = await add_update_action(
-            strategy,
-            from_collection=strategy.remote_collection,
-            to_collection=strategy.local_collection,
-            needs_update_check=needs_update_check,
-            source_is_local=False,
-            data_id_map=data_id_map,
-        )
-        updated_nodes.extend(nodes)
-        logger.info(f"[{strategy.node_type}] Updated {len(nodes)} nodes in local collection (Pull)")
-
-    elif strategy.config.update_direction == UpdateDirection.BIDIRECTIONAL:
-        raise NotImplementedError(
-            f"[{strategy.node_type}] update_direction=BIDIRECTIONAL requires custom implementation"
-        )
-
-    strategy.emit_schema_diff_report()
-    return updated_nodes
+    emit_schema_diff_report(strategy, validator)
+    return list(updated_by_id.values())