初步实现多project pipeline

2026-04-07 14:45:19 +08:00
parent e3eab5ade1
commit 966f2e1270
11 changed files with 1083 additions and 169 deletions
@@ -17,11 +17,12 @@ import inspect
 import json
 import logging
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Literal, Optional

 from ..common.binding import BindingManager
 from ..common.collection import DataCollection
 from ..common.persistence import PersistenceBackend, create_persistence_backend
+from ..common.persistence_service import PersistenceService
 from ..common.registry import DomainRegistry
 from ..datasource import ensure_builtin_datasource_types_registered
 from ..datasource.type_registry import get_datasource_factory
@@ -47,10 +48,13 @@ from ..config.strategy_config import resolve_domain_option_config
 from ..sync_system.strategy import BaseSyncStrategy
 from ..logging import configure_app_logging, ensure_app_logging, get_logger
 from .full_sync_pipeline import FullSyncPipeline
+from .project_scope_runtime import ProjectScopeRuntime


 logger = get_logger(__name__)

+PipelineType = Literal["auto", "full", "multi-project"]
+

 def _project_root() -> Path:
    return Path(__file__).resolve().parents[2]
@@ -229,12 +233,32 @@ def _add_handlers(

 def _build_collections_and_bindings(
    *,
-    persistence: PersistenceBackend,
+    persistence,
    scope: str,
+    local_fallback_collection: Optional[DataCollection] = None,
+    remote_fallback_collection: Optional[DataCollection] = None,
+    binding_fallback_manager: Optional[BindingManager] = None,
 ) -> tuple[DataCollection, DataCollection, BindingManager]:
-    local_collection = DataCollection("local", scope=scope, persistence=persistence, auto_persist=False)
-    remote_collection = DataCollection("remote", scope=scope, persistence=persistence, auto_persist=False)
-    binding_manager = BindingManager(persistence, auto_persist=False, scope=scope)
+    local_collection = DataCollection(
+        "local",
+        scope=scope,
+        persistence=persistence,
+        auto_persist=False,
+        global_fallback_collection=local_fallback_collection,
+    )
+    remote_collection = DataCollection(
+        "remote",
+        scope=scope,
+        persistence=persistence,
+        auto_persist=False,
+        global_fallback_collection=remote_fallback_collection,
+    )
+    binding_manager = BindingManager(
+        persistence,
+        auto_persist=False,
+        scope=scope,
+        global_fallback_manager=binding_fallback_manager,
+    )
    return local_collection, remote_collection, binding_manager


@@ -331,22 +355,86 @@ async def create_pipeline_from_config(
    ephemeral_node_types: Optional[List[str]] = None,
    local_datasource_override=None,
    remote_datasource_override=None,
+    local_fallback_collection: Optional[DataCollection] = None,
+    remote_fallback_collection: Optional[DataCollection] = None,
+    binding_fallback_manager: Optional[BindingManager] = None,
+    persistence_service: Optional[PersistenceService] = None,
+    scope_runtime: Optional[ProjectScopeRuntime] = None,
 ) -> FullSyncPipeline:
    ensure_app_logging(config.logging)

    all_types = list(config.node_types)
    strategy_map = _resolve_strategy_map(config, all_types)

-    db_path = Path(config.persist.db_path)
-    if config.persist.wipe_on_start and db_path.exists():
-        db_path.unlink()
+    runtime = scope_runtime
+    close_resources = runtime is None

-    persistence: Optional[PersistenceBackend] = None
+    if runtime is None:
+        runtime = await create_scope_runtime_from_config(
+            config,
+            scope=scope,
+            local_datasource_override=local_datasource_override,
+            remote_datasource_override=remote_datasource_override,
+            local_fallback_collection=local_fallback_collection,
+            remote_fallback_collection=remote_fallback_collection,
+            binding_fallback_manager=binding_fallback_manager,
+            persistence_service=persistence_service,
+        )
+
+    strategies = _build_strategies(
+        config,
+        all_types,
+        runtime.local_collection,
+        runtime.remote_collection,
+        runtime.binding_manager,
+        strategy_map,
+    )
+
+    return FullSyncPipeline(
+        local_datasource=runtime.local_datasource,
+        remote_datasource=runtime.remote_datasource,
+        strategies=strategies,
+        node_types=all_types,
+        persistence=runtime.persistence_backend,
+        local_collection=runtime.local_collection,
+        remote_collection=runtime.remote_collection,
+        binding_manager=runtime.binding_manager,
+        scope=scope,
+        pipeline_name=pipeline_name,
+        bootstrap_binding_node_types=bootstrap_binding_node_types,
+        ephemeral_node_types=ephemeral_node_types,
+        enable_persistence=config.persist.enable,
+        scope_runtime=runtime,
+        close_resources=close_resources,
+    )
+
+
+async def create_scope_runtime_from_config(
+    config: PipelineRunConfig,
+    *,
+    scope: str = "global",
+    local_datasource_override=None,
+    remote_datasource_override=None,
+    local_fallback_collection: Optional[DataCollection] = None,
+    remote_fallback_collection: Optional[DataCollection] = None,
+    binding_fallback_manager: Optional[BindingManager] = None,
+    persistence_service: Optional[PersistenceService] = None,
+    global_runtime: Optional[ProjectScopeRuntime] = None,
+) -> ProjectScopeRuntime:
+    ensure_app_logging(config.logging)
+
+    db_path = Path(config.persist.db_path)
+    owns_persistence_service = persistence_service is None
+    created_backend: Optional[PersistenceBackend] = None
    local_ds = None
    remote_ds = None
    try:
-        persistence = create_persistence_backend(config.persist)
-        await persistence.initialize()
+        if persistence_service is None:
+            if config.persist.wipe_on_start and db_path.exists():
+                db_path.unlink()
+            created_backend = create_persistence_backend(config.persist)
+            persistence_service = PersistenceService(created_backend)
+            await persistence_service.initialize()

        local_ds = await _create_and_initialize_datasource(
            config.local_datasource,
@@ -360,43 +448,34 @@ async def create_pipeline_from_config(
            endpoint_name="remote",
            project_root=_project_root(),
        )
-        _add_handlers(config, local_ds, remote_ds, all_types)
+        _add_handlers(config, local_ds, remote_ds, list(config.node_types))

+        persistence_view = persistence_service.view(scope)
        local_collection, remote_collection, binding_manager = _build_collections_and_bindings(
-            persistence=persistence,
+            persistence=persistence_view,
            scope=scope,
+            local_fallback_collection=local_fallback_collection,
+            remote_fallback_collection=remote_fallback_collection,
+            binding_fallback_manager=binding_fallback_manager,
        )

-        strategies = _build_strategies(
-            config,
-            all_types,
-            local_collection,
-            remote_collection,
-            binding_manager,
-            strategy_map,
-        )
-
-        pipeline = FullSyncPipeline(
-            local_datasource=local_ds,
-            remote_datasource=remote_ds,
-            strategies=strategies,
-            node_types=all_types,
-            persistence=persistence,
+        return ProjectScopeRuntime(
+            scope=scope,
+            persistence_service=persistence_service,
+            persistence_view=persistence_view,
            local_collection=local_collection,
            remote_collection=remote_collection,
            binding_manager=binding_manager,
-            scope=scope,
-            pipeline_name=pipeline_name,
-            bootstrap_binding_node_types=bootstrap_binding_node_types,
-            ephemeral_node_types=ephemeral_node_types,
-            enable_persistence=config.persist.enable,
+            local_datasource=local_ds,
+            remote_datasource=remote_ds,
+            global_runtime=global_runtime,
+            owns_persistence_service=owns_persistence_service,
        )
-
-        return pipeline
    except Exception:
        await _safe_close_resource("remote_datasource", remote_ds)
        await _safe_close_resource("local_datasource", local_ds)
-        await _safe_close_resource("persistence", persistence)
+        if owns_persistence_service:
+            await _safe_close_resource("persistence", persistence_service or created_backend)
        raise


@@ -409,9 +488,27 @@ async def run_pipeline_from_config(
    bootstrap_binding_node_types: Optional[List[str]] = None,
    local_datasource_override=None,
    remote_datasource_override=None,
+    pipeline_type: PipelineType = "auto",
 ) -> Dict[str, Any]:
    ensure_app_logging(config.logging)

+    from .multi_project_pipeline import is_implicit_project_batch_config, run_implicit_project_batch_from_config
+
+    implicit_multi_project = is_implicit_project_batch_config(config)
+    if pipeline_type == "multi-project" and not implicit_multi_project:
+        raise ValueError("pipeline_type=multi-project requires an implicit multi-project pipeline config")
+    if pipeline_type == "full" and implicit_multi_project:
+        logger.info("pipeline_type=full specified, skipping implicit multi-project routing")
+
+    if pipeline_type != "full" and implicit_multi_project:
+        _log_resolved_config(config)
+        return await run_implicit_project_batch_from_config(
+            config,
+            title=title,
+            print_summary=print_summary,
+            max_concurrency=getattr(config, "max_concurrency", None),
+        )
+
    _log_resolved_config(config)

    pipeline: Optional[FullSyncPipeline] = None
@@ -15,7 +15,7 @@ if TYPE_CHECKING:
 from ..datasource.jsonl import JsonlDataSource
 from ..common.collection import DataCollection
 from ..common.binding import BindingManager
-from ..logging import get_logger
+from ..logging import get_scope_logger
 from ..common.persistence import PersistenceBackend
 from ..sync_system.config import UpdateDirection
 from ..sync_system.strategy import BaseSyncStrategy
@@ -25,9 +25,7 @@ from ..sync_system.strategy_ops import finalize_peer_creating_sources, run_phase
 from ..sync_system.strategy_ops.bind_ops import refresh_bind_data_id
 from ..sync_system.strategy_ops.post_check_ops import evaluate_consistency_for_node_type
 from .summary_report import print_pipeline_summary
-
-
-logger = get_logger(__name__)
+from .project_scope_runtime import ProjectScopeRuntime


 class NodeTypeLoadError(RuntimeError):
@@ -72,6 +70,8 @@ class FullSyncPipeline:
        load_order: Optional[List[str]] = None,
        sync_order: Optional[List[str]] = None,
        enable_persistence: bool = True,
+        scope_runtime: Optional[ProjectScopeRuntime] = None,
+        close_resources: bool = True,
    ):
        # 验证必需参数（由工厂函数创建）
        if strategies is None:
@@ -91,6 +91,8 @@ class FullSyncPipeline:
        self.sync_order = sync_order or self.node_types
        self.persistence = persistence
        self.enable_persistence = enable_persistence
+        self.scope_runtime = scope_runtime
+        self.close_resources = close_resources
        
        self.local_collection = local_collection
        self.remote_collection = remote_collection
@@ -113,12 +115,9 @@ class FullSyncPipeline:
            "post_check_passed": True,
        }
        self._consistency_by_type: Dict[str, Dict[str, Any]] = {}
-        self._logger = logger
+        self._logger = get_scope_logger(scope, __name__)
        self._closed = False
        self._loaded_node_types: set[str] = set()
-        self._preloaded_local_nodes: List[Any] = []
-        self._preloaded_remote_nodes: List[Any] = []
-        self._preloaded_bindings: Dict[str, Dict[str, Optional[str]]] = {}

    def _resolve_pipeline_runtime(self) -> StateMachineRuntime:
        if self.strategies:
@@ -290,12 +289,17 @@ class FullSyncPipeline:
                continue

    async def phase_bootstrap(self) -> None:
-        await self._load_persistence_state()
+        if self.scope_runtime is not None:
+            await self.scope_runtime.load(
+                node_types=self.node_types,
+                bootstrap_binding_node_types=self.bootstrap_binding_node_types,
+                enable_persistence=self.enable_persistence,
+                ephemeral_node_types=self.ephemeral_node_types,
+            )
+        else:
+            await self._load_persistence_state()
        self._logger.info("✅ Bootstrap: persistence loaded")

-        await self._apply_preloaded_state()
-        self._logger.info("✅ Bootstrap: shared preload applied")
-
        self._prepare_datasources()
        self._logger.info("✅ Bootstrap: datasource prepared")

@@ -383,6 +387,14 @@ class FullSyncPipeline:
            return  # 已经关闭过，避免重复关闭
        self._closed = True

+        if not self.close_resources:
+            return
+
+        if self.scope_runtime is not None:
+            await self.scope_runtime.unload()
+            await self.scope_runtime.close()
+            return
+
        await self._safe_close("remote_datasource", self.remote_datasource.close())
        await self._safe_close("local_datasource", self.local_datasource.close())
        await self._safe_close("persistence", self.persistence.close())
@@ -459,28 +471,6 @@ class FullSyncPipeline:
        if total_cleaned > 0:
            self._logger.info(f"🧹 Reset cleanup: {total_cleaned} bootstrap zombies cleaned")

-    async def _apply_preloaded_state(self) -> None:
-        if self._preloaded_local_nodes:
-            await self.local_collection.import_nodes(self._preloaded_local_nodes)
-        if self._preloaded_remote_nodes:
-            await self.remote_collection.import_nodes(self._preloaded_remote_nodes)
-        for node_type, bindings in self._preloaded_bindings.items():
-            self.binding_manager.import_bindings(node_type, bindings)
-
-    def preload_shared_state(
-        self,
-        *,
-        local_nodes: List[Any],
-        remote_nodes: List[Any],
-        bindings_by_type: Dict[str, Dict[str, Optional[str]]],
-    ) -> None:
-        self._preloaded_local_nodes = list(local_nodes)
-        self._preloaded_remote_nodes = list(remote_nodes)
-        self._preloaded_bindings = {
-            node_type: dict(bindings)
-            for node_type, bindings in bindings_by_type.items()
-        }
-
    def _prepare_datasources(self) -> None:
        """为后续按 node_type 加载准备 collection 绑定。"""
        self.local_datasource.set_collection(self.local_collection)
@@ -772,6 +762,18 @@ class FullSyncPipeline:

    async def phase_persistence(self) -> None:
        """持久化所有数据。"""
+        if self.scope_runtime is not None:
+            if not self.enable_persistence:
+                self._logger.info("⏭️  Persistence disabled, skipping...")
+                return
+            self._logger.info("🔍 Persisting scope runtime...")
+            await self.scope_runtime.persist(
+                enable_persistence=self.enable_persistence,
+                ephemeral_node_types=self.ephemeral_node_types,
+            )
+            self._logger.info("✅ All data persisted successfully")
+            return
+
        if not self.enable_persistence:
            self._logger.info("⏭️  Persistence disabled, skipping...")
            return
@@ -1,28 +1,34 @@
 from __future__ import annotations

+import asyncio
 import logging
-import copy
+from dataclasses import dataclass
 from pathlib import Path
-from typing import Dict, Optional, List, TypedDict, Any
+from typing import Any, Dict, Optional, List

 from ..config import (
    MultiProjectItemConfig,
    MultiProjectRunConfig,
    PipelineRunConfig,
+    StrategyRuntimeConfig,
    apply_config_overrides,
    build_config_from_file,
 )
-from ..logging import get_logger
-from .factory import create_pipeline_from_config
+from ..logging import get_scope_logger
+from .factory import create_pipeline_from_config, create_scope_runtime_from_config


-logger = get_logger(__name__)
+logger = get_scope_logger("multi-project", __name__)


-class _SharedStateSnapshot(TypedDict):
-    local_nodes: List[Any]
-    remote_nodes: List[Any]
-    bindings_by_type: Dict[str, Dict[str, Optional[str]]]
+@dataclass(frozen=True)
+class ImplicitProjectItem:
+    local_project_id: str
+    remote_project_id: str
+
+    @property
+    def scope(self) -> str:
+        return f"project_id:{self.local_project_id}"


 def _resolve_embedded_config_path(raw_path: str, *, project_root: Path, profile_path: Path) -> Path:
@@ -65,7 +71,215 @@ def _resolve_project_bootstrap_node_types(
    return list(dict.fromkeys(shared_node_types))


-class MultiProjectPipeline:
+def _normalize_project_id_items(raw_items: Any) -> list[str]:
+    items = raw_items if isinstance(raw_items, list) else []
+    normalized: list[str] = []
+    for item in items:
+        value = str(item).strip()
+        if value and value not in normalized:
+            normalized.append(value)
+    return normalized
+
+
+def _get_project_handler_filters(config: PipelineRunConfig) -> tuple[list[str], list[str]]:
+    local_filters = _normalize_project_id_items(
+        config.local_datasource.handler_configs.get("project", {}).get("data_id_filter", [])
+    )
+    remote_filters = _normalize_project_id_items(
+        config.remote_datasource.handler_configs.get("project", {}).get("data_id_filter", [])
+    )
+    return local_filters, remote_filters
+
+
+def _get_project_strategy_runtime(config: PipelineRunConfig) -> StrategyRuntimeConfig | None:
+    for item in config.strategies:
+        if item.node_type == "project":
+            return item
+    return None
+
+
+def _get_project_prebind_pairs(config: PipelineRunConfig) -> list[tuple[str, str]]:
+    runtime_cfg = _get_project_strategy_runtime(config)
+    if runtime_cfg is None:
+        return []
+    pairs: list[tuple[str, str]] = []
+    for item in runtime_cfg.config.pre_bind_data_id:
+        local_id = str(item.local_data_id).strip()
+        remote_id = str(item.remote_data_id).strip()
+        if not local_id or not remote_id:
+            continue
+        pairs.append((local_id, remote_id))
+    return pairs
+
+
+def is_implicit_project_batch_config(config: PipelineRunConfig) -> bool:
+    if "project" not in config.node_types:
+        return False
+    local_filters, remote_filters = _get_project_handler_filters(config)
+    prebind_pairs = _get_project_prebind_pairs(config)
+    return max(len(local_filters), len(remote_filters), len(prebind_pairs)) > 1
+
+
+def _extract_implicit_project_items(config: PipelineRunConfig) -> list[ImplicitProjectItem]:
+    local_filters, remote_filters = _get_project_handler_filters(config)
+    prebind_pairs = _get_project_prebind_pairs(config)
+    if not prebind_pairs:
+        raise ValueError("implicit multi-project config requires strategies.project.config.pre_bind_data_id")
+
+    pair_local_ids = [local_id for local_id, _ in prebind_pairs]
+    pair_remote_ids = [remote_id for _, remote_id in prebind_pairs]
+
+    effective_local_filters = local_filters or list(dict.fromkeys(pair_local_ids))
+    effective_remote_filters = remote_filters or list(dict.fromkeys(pair_remote_ids))
+
+    if set(effective_local_filters) != set(pair_local_ids):
+        raise ValueError(
+            "implicit multi-project config mismatch: local project data_id_filter must match project pre_bind_data_id local_data_id"
+        )
+    if set(effective_remote_filters) != set(pair_remote_ids):
+        raise ValueError(
+            "implicit multi-project config mismatch: remote project data_id_filter must match project pre_bind_data_id remote_data_id"
+        )
+
+    seen_local_ids: set[str] = set()
+    seen_remote_ids: set[str] = set()
+    items: list[ImplicitProjectItem] = []
+    for local_id, remote_id in prebind_pairs:
+        if local_id in seen_local_ids:
+            raise ValueError(f"duplicate local project id in implicit multi-project config: {local_id}")
+        if remote_id in seen_remote_ids:
+            raise ValueError(f"duplicate remote project id in implicit multi-project config: {remote_id}")
+        seen_local_ids.add(local_id)
+        seen_remote_ids.add(remote_id)
+        items.append(ImplicitProjectItem(local_project_id=local_id, remote_project_id=remote_id))
+    return items
+
+
+def _split_implicit_project_node_types(node_types: list[str]) -> tuple[list[str], list[str]]:
+    if "project" not in node_types:
+        raise ValueError("implicit multi-project config requires node_types to include 'project'")
+    project_index = node_types.index("project")
+    return node_types[:project_index], node_types[project_index:]
+
+
+def _build_implicit_project_config(
+    base_config: PipelineRunConfig,
+    *,
+    item: ImplicitProjectItem,
+    project_node_types: list[str],
+) -> PipelineRunConfig:
+    return apply_config_overrides(
+        base_config,
+        {
+            "node_types": project_node_types,
+            "persist": {"wipe_on_start": False},
+            "local_datasource": {
+                "handler_configs": {
+                    "project": {"data_id_filter": [item.local_project_id]},
+                }
+            },
+            "remote_datasource": {
+                "handler_configs": {
+                    "project": {"data_id_filter": [item.remote_project_id]},
+                }
+            },
+            "strategies": {
+                "project": {
+                    "config": {
+                        "pre_bind_data_id": [
+                            {
+                                "local_data_id": item.local_project_id,
+                                "remote_data_id": item.remote_project_id,
+                            }
+                        ]
+                    }
+                }
+            },
+        },
+    )
+
+
+async def run_implicit_project_batch_from_config(
+    config: PipelineRunConfig,
+    *,
+    title: str,
+    print_summary: bool = True,
+    max_concurrency: int | None = None,
+) -> Dict[str, Dict[str, object]]:
+    from .factory import create_pipeline_from_config, create_scope_runtime_from_config
+
+    project_items = _extract_implicit_project_items(config)
+    shared_node_types, project_node_types = _split_implicit_project_node_types(list(config.node_types))
+    results: Dict[str, Dict[str, object]] = {}
+
+    global_runtime = None
+    global_pipeline = None
+    global_config = None
+
+    if print_summary:
+        logger.info("\n%s", "=" * 80)
+        logger.info("🚀 Creating implicit multi-project pipeline from in-memory config")
+        logger.info("📦 Projects=%d", len(project_items))
+        logger.info("=" * 80)
+
+    try:
+        if shared_node_types:
+            global_config = apply_config_overrides(
+                config,
+                {
+                    "node_types": shared_node_types,
+                },
+            )
+            global_runtime = await create_scope_runtime_from_config(global_config, scope="global")
+            global_pipeline = await create_pipeline_from_config(
+                global_config,
+                scope="global",
+                pipeline_name=f"{title} [global]",
+                scope_runtime=global_runtime,
+            )
+            results["global"] = await global_pipeline.run()
+
+        semaphore = asyncio.Semaphore(max(1, max_concurrency or len(project_items)))
+
+        async def _run_project(item: ImplicitProjectItem) -> None:
+            project_config = _build_implicit_project_config(
+                config,
+                item=item,
+                project_node_types=project_node_types,
+            )
+            async with semaphore:
+                project_runtime = await create_scope_runtime_from_config(
+                    project_config,
+                    scope=item.scope,
+                    persistence_service=global_runtime.persistence_service if global_runtime is not None else None,
+                    local_fallback_collection=global_runtime.local_collection if global_runtime is not None else None,
+                    remote_fallback_collection=global_runtime.remote_collection if global_runtime is not None else None,
+                    binding_fallback_manager=global_runtime.binding_manager if global_runtime is not None else None,
+                    global_runtime=global_runtime,
+                )
+                try:
+                    project_pipeline = await create_pipeline_from_config(
+                        project_config,
+                        scope=item.scope,
+                        pipeline_name=f"{title} [{item.scope}]",
+                        bootstrap_binding_node_types=shared_node_types,
+                        ephemeral_node_types=shared_node_types,
+                        scope_runtime=project_runtime,
+                    )
+                    results[item.scope] = await project_pipeline.run()
+                finally:
+                    await project_runtime.unload()
+                    await project_runtime.close()
+
+        await asyncio.gather(*[_run_project(item) for item in project_items])
+        return results
+    finally:
+        if global_runtime is not None:
+            await global_runtime.unload()
+            await global_runtime.close()
+
+
+class ProjectBatchPipeline:
    def __init__(self, *, project_root: Path, config_path: Path, config: MultiProjectRunConfig):
        self.project_root = project_root
        self.config_path = config_path
@@ -87,27 +301,6 @@ class MultiProjectPipeline:
        cfg = apply_config_overrides(cfg, {"persist": {**shared_persist, "wipe_on_start": False}})
        return apply_config_overrides(cfg, _project_scope_overrides(item, node_types=list(cfg.node_types)))

-    async def _snapshot_shared_state(self, *, pipeline, node_types: List[str]) -> _SharedStateSnapshot:
-        local_nodes = [
-            copy.deepcopy(node)
-            for node_type in node_types
-            for node in pipeline.local_collection.filter(node_type=node_type)
-        ]
-        remote_nodes = [
-            copy.deepcopy(node)
-            for node_type in node_types
-            for node in pipeline.remote_collection.filter(node_type=node_type)
-        ]
-        bindings_by_type: Dict[str, Dict[str, Optional[str]]] = {}
-        for node_type in node_types:
-            bindings = await pipeline.binding_manager.get_all_bindings(node_type)
-            bindings_by_type[node_type] = {local_id: remote_id for local_id, remote_id in bindings}
-        return {
-            "local_nodes": local_nodes,
-            "remote_nodes": remote_nodes,
-            "bindings_by_type": bindings_by_type,
-        }
-
    async def run(self) -> Dict[str, Dict[str, object]]:
        results: Dict[str, Dict[str, object]] = {}

@@ -119,49 +312,67 @@ class MultiProjectPipeline:
            shared_node_types=shared_node_types,
        )
        logger.info("🚀 Multi-project pipeline start: projects=%d", len(self.config.projects))
-        global_pipeline = await create_pipeline_from_config(
+        global_runtime = await create_scope_runtime_from_config(
            global_config,
            scope="global",
-            pipeline_name="multi-project pipeline [global]",
        )
+        global_pipeline = None
        try:
+            global_pipeline = await create_pipeline_from_config(
+                global_config,
+                scope="global",
+                pipeline_name="multi-project pipeline [global]",
+                scope_runtime=global_runtime,
+            )
            results["global"] = await global_pipeline.run()
-            shared_state = await self._snapshot_shared_state(
-                pipeline=global_pipeline,
-                node_types=project_bootstrap_node_types,
-            )
-        finally:
-            await global_pipeline.close()
+            semaphore = asyncio.Semaphore(self.config.max_concurrency)

-        for item in self.config.projects:
-            project_config = self._prepare_project_config(
-                shared_persist=shared_persist,
-                item=item,
-                base_path=item.config_path,
-            )
-            scope = item.scope
-            title = f"multi-project pipeline [{scope}]"
-            logger.info(
-                "🚀 Project pipeline start: scope=%s local_project_id=%s remote_project_id=%s",
-                scope,
-                item.local_project_id,
-                item.remote_project_id,
-            )
-            project_pipeline = await create_pipeline_from_config(
-                project_config,
-                scope=scope,
-                pipeline_name=title,
-                bootstrap_binding_node_types=project_bootstrap_node_types,
-                ephemeral_node_types=project_bootstrap_node_types,
-            )
-            try:
-                project_pipeline.preload_shared_state(
-                    local_nodes=shared_state["local_nodes"],
-                    remote_nodes=shared_state["remote_nodes"],
-                    bindings_by_type=shared_state["bindings_by_type"],
+            async def _run_project(item: MultiProjectItemConfig) -> None:
+                project_config = self._prepare_project_config(
+                    shared_persist=shared_persist,
+                    item=item,
+                    base_path=item.config_path,
                )
-                results[scope] = await project_pipeline.run()
-            finally:
-                await project_pipeline.close()
+                scope = item.scope
+                title = f"multi-project pipeline [{scope}]"
+                logger.info(
+                    "🚀 Project pipeline start: scope=%s local_project_id=%s remote_project_id=%s",
+                    scope,
+                    item.local_project_id,
+                    item.remote_project_id,
+                )
+                async with semaphore:
+                    project_runtime = await create_scope_runtime_from_config(
+                        project_config,
+                        scope=scope,
+                        persistence_service=global_runtime.persistence_service,
+                        local_fallback_collection=global_runtime.local_collection,
+                        remote_fallback_collection=global_runtime.remote_collection,
+                        binding_fallback_manager=global_runtime.binding_manager,
+                        global_runtime=global_runtime,
+                    )
+                    project_pipeline = None
+                    try:
+                        project_pipeline = await create_pipeline_from_config(
+                            project_config,
+                            scope=scope,
+                            pipeline_name=title,
+                            bootstrap_binding_node_types=project_bootstrap_node_types,
+                            ephemeral_node_types=project_bootstrap_node_types,
+                            scope_runtime=project_runtime,
+                        )
+                        results[scope] = await project_pipeline.run()
+                    finally:
+                        await project_runtime.unload()
+                        await project_runtime.close()

-        return results
+            await asyncio.gather(*[_run_project(item) for item in self.config.projects])
+        finally:
+            await global_runtime.unload()
+            await global_runtime.close()
+
+        return results
+
+
+class MultiProjectPipeline(ProjectBatchPipeline):
+    pass
@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import asyncio
+from typing import List, Optional
+
+from ..common.binding import BindingManager
+from ..common.collection import DataCollection
+from ..common.persistence_service import PersistenceService, ScopedPersistenceView
+
+
+class ProjectScopeRuntime:
+    """Thin runtime container for one scope execution."""
+
+    def __init__(
+        self,
+        *,
+        scope: str,
+        persistence_service: PersistenceService,
+        persistence_view: ScopedPersistenceView,
+        local_collection: DataCollection,
+        remote_collection: DataCollection,
+        binding_manager: BindingManager,
+        local_datasource,
+        remote_datasource,
+        global_runtime: Optional["ProjectScopeRuntime"] = None,
+        owns_persistence_service: bool = False,
+    ) -> None:
+        self.scope = scope
+        self.persistence_service = persistence_service
+        self.persistence_view = persistence_view
+        self.local_collection = local_collection
+        self.remote_collection = remote_collection
+        self.binding_manager = binding_manager
+        self.local_datasource = local_datasource
+        self.remote_datasource = remote_datasource
+        self.global_runtime = global_runtime
+        self.owns_persistence_service = owns_persistence_service
+        self._closed = False
+
+    @property
+    def persistence_backend(self):
+        return self.persistence_service.backend
+
+    async def load(
+        self,
+        *,
+        node_types: List[str],
+        bootstrap_binding_node_types: Optional[List[str]] = None,
+        enable_persistence: bool = True,
+        ephemeral_node_types: Optional[List[str]] = None,
+    ) -> None:
+        if not enable_persistence:
+            return
+
+        excluded_node_types = list(dict.fromkeys(ephemeral_node_types or [])) or None
+        await self.local_collection.load_from_persistence_excluding(excluded_node_types)
+        await self.remote_collection.load_from_persistence_excluding(excluded_node_types)
+
+        excluded = set(ephemeral_node_types or [])
+        binding_node_types = [
+            node_type
+            for node_type in dict.fromkeys([*node_types, *(bootstrap_binding_node_types or [])])
+            if node_type not in excluded
+        ]
+        for node_type in binding_node_types:
+            await self.binding_manager.load_from_persistence(node_type)
+
+    async def persist(
+        self,
+        *,
+        enable_persistence: bool = True,
+        ephemeral_node_types: Optional[List[str]] = None,
+    ) -> None:
+        if not enable_persistence:
+            return
+
+        await self.local_collection.persist(exclude_node_types=ephemeral_node_types)
+        await self.remote_collection.persist(exclude_node_types=ephemeral_node_types)
+        await self.binding_manager.persist(exclude_node_types=ephemeral_node_types)
+
+    async def unload(self) -> None:
+        await self.local_collection.unload()
+        await self.remote_collection.unload()
+        await self.binding_manager.unload()
+
+    async def close(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+
+        await self._safe_close("remote_datasource", self.remote_datasource.close())
+        await self._safe_close("local_datasource", self.local_datasource.close())
+        if self.owns_persistence_service:
+            await self._safe_close("persistence_service", self.persistence_service.close())
+
+    async def _safe_close(self, name: str, close_coro, timeout: float = 3.0) -> None:
+        try:
+            await asyncio.wait_for(close_coro, timeout=timeout)
+        except asyncio.TimeoutError:
+            return
+        except Exception:
+            return