diff --git a/api/controllers/console/app/app.py b/api/controllers/console/app/app.py
index dad184c54b..d71864302c 100644
--- a/api/controllers/console/app/app.py
+++ b/api/controllers/console/app/app.py
@@ -99,6 +99,11 @@ class AppExportQuery(BaseModel):
     workflow_id: str | None = Field(default=None, description="Specific workflow ID to export")
 
 
+class AppExportBundleQuery(BaseModel):
+    include_secret: bool = Field(default=False, description="Include secrets in export")
+    workflow_id: str | None = Field(default=None, description="Specific workflow ID to export")
+
+
 class AppNamePayload(BaseModel):
     name: str = Field(..., min_length=1, description="Name to check")
 
@@ -650,6 +655,36 @@ class AppExportApi(Resource):
         return payload.model_dump(mode="json")
 
 
+@console_ns.route("/apps/<uuid:app_id>/export-bundle")
+class AppExportBundleApi(Resource):
+    @get_app_model
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def get(self, app_model):
+        from io import BytesIO
+
+        from flask import send_file
+
+        from services.app_bundle_service import AppBundleService
+
+        args = AppExportBundleQuery.model_validate(request.args.to_dict(flat=True))
+
+        result = AppBundleService.export_bundle(
+            app_model=app_model,
+            include_secret=args.include_secret,
+            workflow_id=args.workflow_id,
+        )
+
+        return send_file(
+            BytesIO(result.zip_bytes),
+            mimetype="application/zip",
+            as_attachment=True,
+            download_name=result.filename,
+        )
+
+
 @console_ns.route("/apps/<uuid:app_id>/name")
 class AppNameApi(Resource):
     @console_ns.doc("check_app_name")
diff --git a/api/controllers/console/app/app_asset.py b/api/controllers/console/app/app_asset.py
index 80d7673995..e71c4a83c2 100644
--- a/api/controllers/console/app/app_asset.py
+++ b/api/controllers/console/app/app_asset.py
@@ -243,22 +243,6 @@ class AppAssetNodeReorderResource(Resource):
             raise AppAssetNodeNotFoundError()
 
 
-@console_ns.route("/apps/<string:app_id>/assets/publish")
-class AppAssetPublishResource(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        current_user, _ = current_account_with_tenant()
-        published = AppAssetService.publish(app_model, current_user.id)
-        return {
-            "id": published.id,
-            "version": published.version,
-            "asset_tree": published.asset_tree.model_dump(),
-        }, 201
-
-
 @console_ns.route("/apps/<string:app_id>/assets/files/<string:node_id>/download-url")
 class AppAssetFileDownloadUrlResource(Resource):
     @setup_required
diff --git a/api/controllers/console/app/app_import.py b/api/controllers/console/app/app_import.py
index 22e2aeb720..eb7c3ce1c3 100644
--- a/api/controllers/console/app/app_import.py
+++ b/api/controllers/console/app/app_import.py
@@ -51,6 +51,14 @@ class AppImportPayload(BaseModel):
     app_id: str | None = None
 
 
+class AppImportBundlePayload(BaseModel):
+    name: str | None = None
+    description: str | None = None
+    icon_type: str | None = None
+    icon: str | None = None
+    icon_background: str | None = None
+
+
 console_ns.schema_model(
     AppImportPayload.__name__, AppImportPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
 )
@@ -139,3 +147,55 @@ class AppImportCheckDependenciesApi(Resource):
             result = import_service.check_dependencies(app_model=app_model)
 
         return result.model_dump(mode="json"), 200
+
+
+@console_ns.route("/apps/imports-bundle")
+class AppImportBundleApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(app_import_model)
+    @cloud_edition_billing_resource_check("apps")
+    @edit_permission_required
+    def post(self):
+        from flask import request
+
+        from core.app.entities.app_bundle_entities import BundleFormatError
+        from services.app_bundle_service import AppBundleService
+
+        current_user, _ = current_account_with_tenant()
+
+        if "file" not in request.files:
+            return {"error": "No file provided"}, 400
+
+        file = request.files["file"]
+        if not file.filename or not file.filename.endswith(".zip"):
+            return {"error": "Invalid file format, expected .zip"}, 400
+
+        zip_bytes = file.read()
+
+        form_data = request.form.to_dict()
+        args = AppImportBundlePayload.model_validate(form_data)
+
+        try:
+            result = AppBundleService.import_bundle(
+                account=current_user,
+                zip_bytes=zip_bytes,
+                name=args.name,
+                description=args.description,
+                icon_type=args.icon_type,
+                icon=args.icon,
+                icon_background=args.icon_background,
+            )
+        except BundleFormatError as e:
+            return {"error": str(e)}, 400
+
+        if result.app_id and FeatureService.get_system_features().webapp_auth.enabled:
+            EnterpriseService.WebAppAuth.update_app_access_mode(result.app_id, "private")
+
+        status = result.status
+        if status == ImportStatus.FAILED:
+            return result.model_dump(mode="json"), 400
+        elif status == ImportStatus.PENDING:
+            return result.model_dump(mode="json"), 202
+        return result.model_dump(mode="json"), 200
diff --git a/api/controllers/console/app/workflow.py b/api/controllers/console/app/workflow.py
index 085c5cec3f..29200a3d22 100644
--- a/api/controllers/console/app/workflow.py
+++ b/api/controllers/console/app/workflow.py
@@ -686,13 +686,14 @@ class PublishedWorkflowApi(Resource):
         """
         Publish workflow
         """
+        from services.app_bundle_service import AppBundleService
+
         current_user, _ = current_account_with_tenant()
 
         args = PublishWorkflowPayload.model_validate(console_ns.payload or {})
 
-        workflow_service = WorkflowService()
         with Session(db.engine) as session:
-            workflow = workflow_service.publish_workflow(
+            workflow = AppBundleService.publish(
                 session=session,
                 app_model=app_model,
                 account=current_user,
diff --git a/api/core/app/entities/app_bundle_entities.py b/api/core/app/entities/app_bundle_entities.py
new file mode 100644
index 0000000000..b81fec62ae
--- /dev/null
+++ b/api/core/app/entities/app_bundle_entities.py
@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import re
+
+from pydantic import BaseModel, Field
+
+# Constants
+BUNDLE_DSL_FILENAME_PATTERN = re.compile(r"^[^/]+\.ya?ml$")
+BUNDLE_MAX_SIZE = 50 * 1024 * 1024  # 50MB
+
+
+# Exceptions
+class BundleFormatError(Exception):
+    """Raised when bundle format is invalid."""
+
+    pass
+
+
+class ZipSecurityError(Exception):
+    """Raised when zip file contains security violations."""
+
+    pass
+
+
+# Entities
+class BundleExportResult(BaseModel):
+    zip_bytes: bytes = Field(description="ZIP file content as bytes")
+    filename: str = Field(description="Suggested filename for the ZIP")
+
+
+class SourceFileEntry(BaseModel):
+    path: str = Field(description="File path within the ZIP")
+    node_id: str = Field(description="Node ID in the asset tree")
+
+
+class ExtractedFile(BaseModel):
+    path: str = Field(description="Relative path of the extracted file")
+    content: bytes = Field(description="File content as bytes")
+
+
+class ExtractedFolder(BaseModel):
+    path: str = Field(description="Relative path of the extracted folder")
diff --git a/api/core/app_assets/__init__.py b/api/core/app_assets/__init__.py
index fa57145176..4490a8eda7 100644
--- a/api/core/app_assets/__init__.py
+++ b/api/core/app_assets/__init__.py
@@ -4,7 +4,7 @@ from .entities import (
     FileAsset,
     SkillAsset,
 )
-from .packager import AssetPackager, ZipPackager
+from .packager import AssetPackager, AssetZipPackager
 from .parser import AssetItemParser, AssetParser, FileAssetParser, SkillAssetParser
 from .paths import AssetPaths
 
@@ -15,9 +15,9 @@ __all__ = [
     "AssetPackager",
     "AssetParser",
     "AssetPaths",
+    "AssetZipPackager",
     "FileAsset",
     "FileAssetParser",
     "SkillAsset",
     "SkillAssetParser",
-    "ZipPackager",
 ]
diff --git a/api/core/app_assets/converters.py b/api/core/app_assets/converters.py
new file mode 100644
index 0000000000..c610fff542
--- /dev/null
+++ b/api/core/app_assets/converters.py
@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from core.app.entities.app_asset_entities import AppAssetFileTree, AssetNodeType
+from core.app_assets.entities import FileAsset
+from core.app_assets.paths import AssetPaths
+
+
+def tree_to_asset_items(
+    tree: AppAssetFileTree,
+    tenant_id: str,
+    app_id: str,
+) -> list[FileAsset]:
+    """
+    Convert AppAssetFileTree to list of FileAsset for packaging.
+
+    Args:
+        tree: The asset file tree to convert
+        tenant_id: Tenant ID for storage key generation
+        app_id: App ID for storage key generation
+
+    Returns:
+        List of FileAsset items ready for packaging
+    """
+    items: list[FileAsset] = []
+    for node in tree.nodes:
+        if node.node_type == AssetNodeType.FILE:
+            path = tree.get_path(node.id)
+            storage_key = AssetPaths.draft_file(tenant_id, app_id, node.id)
+            items.append(
+                FileAsset(
+                    asset_id=node.id,
+                    path=path,
+                    file_name=node.name,
+                    extension=node.extension or "",
+                    storage_key=storage_key,
+                )
+            )
+    return items
diff --git a/api/core/app_assets/packager/__init__.py b/api/core/app_assets/packager/__init__.py
index a9bc9147fa..2a52edcaa5 100644
--- a/api/core/app_assets/packager/__init__.py
+++ b/api/core/app_assets/packager/__init__.py
@@ -1,7 +1,7 @@
+from .asset_zip_packager import AssetZipPackager
 from .base import AssetPackager
-from .zip_packager import ZipPackager
 
 __all__ = [
     "AssetPackager",
-    "ZipPackager",
+    "AssetZipPackager",
 ]
diff --git a/api/core/app_assets/packager/asset_zip_packager.py b/api/core/app_assets/packager/asset_zip_packager.py
new file mode 100644
index 0000000000..b48099bf3c
--- /dev/null
+++ b/api/core/app_assets/packager/asset_zip_packager.py
@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import io
+import zipfile
+from concurrent.futures import ThreadPoolExecutor
+from threading import Lock
+from typing import TYPE_CHECKING
+
+from core.app_assets.entities import AssetItem
+
+if TYPE_CHECKING:
+    from extensions.ext_storage import Storage
+
+
+class AssetZipPackager:
+    """
+    Unified ZIP packager for assets.
+    Automatically creates directory entries from asset paths.
+    """
+
+    def __init__(self, storage: Storage, *, max_workers: int = 8) -> None:
+        self._storage = storage
+        self._max_workers = max_workers
+
+    def package(self, assets: list[AssetItem], *, prefix: str = "") -> bytes:
+        """
+        Package assets into a ZIP file.
+
+        Args:
+            assets: List of assets to package
+            prefix: Optional prefix to add to all paths in the ZIP
+
+        Returns:
+            ZIP file content as bytes
+        """
+        zip_buffer = io.BytesIO()
+
+        # Extract folder paths from asset paths
+        folder_paths = self._extract_folder_paths(assets, prefix)
+
+        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
+            # Create directory entries
+            for folder_path in sorted(folder_paths):
+                zf.writestr(zipfile.ZipInfo(folder_path + "/"), "")
+
+            # Write files in parallel
+            if assets:
+                self._write_files_parallel(zf, assets, prefix)
+
+        return zip_buffer.getvalue()
+
+    def _extract_folder_paths(self, assets: list[AssetItem], prefix: str) -> set[str]:
+        """Extract all folder paths from asset paths."""
+        folders: set[str] = set()
+        for asset in assets:
+            full_path = f"{prefix}/{asset.path}" if prefix else asset.path
+            parts = full_path.split("/")[:-1]  # Remove filename
+            folders.update("/".join(parts[:i]) for i in range(1, len(parts) + 1))
+        return folders
+
+    def _write_files_parallel(
+        self,
+        zf: zipfile.ZipFile,
+        assets: list[AssetItem],
+        prefix: str,
+    ) -> None:
+        lock = Lock()
+
+        def load_and_write(asset: AssetItem) -> None:
+            content = self._storage.load_once(asset.get_storage_key())
+            full_path = f"{prefix}/{asset.path}" if prefix else asset.path
+            with lock:
+                zf.writestr(full_path, content)
+
+        with ThreadPoolExecutor(max_workers=self._max_workers) as executor:
+            futures = [executor.submit(load_and_write, a) for a in assets]
+            for future in futures:
+                future.result()
diff --git a/api/core/app_assets/packager/zip_packager.py b/api/core/app_assets/packager/zip_packager.py
deleted file mode 100644
index 09a7d860d7..0000000000
--- a/api/core/app_assets/packager/zip_packager.py
+++ /dev/null
@@ -1,42 +0,0 @@
-import io
-import zipfile
-from concurrent.futures import Future, ThreadPoolExecutor
-from threading import Lock
-from typing import TYPE_CHECKING
-
-from core.app_assets.entities import AssetItem
-
-from .base import AssetPackager
-
-if TYPE_CHECKING:
-    from extensions.ext_storage import Storage
-
-
-class ZipPackager(AssetPackager):
-    _storage: "Storage"
-
-    def __init__(self, storage: "Storage") -> None:
-        self._storage = storage
-
-    def package(self, assets: list[AssetItem]) -> bytes:
-        zip_buffer = io.BytesIO()
-
-        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
-            lock = Lock()
-            # FOR DELVELPMENT AND TESTING ONLY, TODO: optimize
-            with ThreadPoolExecutor(max_workers=8) as executor:
-                futures: list[Future[None]] = []
-                for asset in assets:
-
-                    def _write_asset(a: AssetItem) -> None:
-                        content = self._storage.load_once(a.get_storage_key())
-                        with lock:
-                            zf.writestr(a.path, content)
-
-                    futures.append(executor.submit(_write_asset, asset))
-
-                # Wait for all futures to complete
-                for future in futures:
-                    future.result()
-
-        return zip_buffer.getvalue()
diff --git a/api/core/app_assets/paths.py b/api/core/app_assets/paths.py
index fd900729ee..2eceadb798 100644
--- a/api/core/app_assets/paths.py
+++ b/api/core/app_assets/paths.py
@@ -16,3 +16,8 @@ class AssetPaths:
     @staticmethod
     def build_skill_artifact_set(tenant_id: str, app_id: str, assets_id: str) -> str:
         return f"{AssetPaths._BASE}/{tenant_id}/{app_id}/artifacts/{assets_id}/skill_artifact_set.json"
+
+    @staticmethod
+    def build_source_zip(tenant_id: str, app_id: str, workflow_id: str) -> str:
+        """Storage key for source assets zip (editable files snapshot at publish time)."""
+        return f"{AssetPaths._BASE}/{tenant_id}/{app_id}/sources/{workflow_id}.zip"
diff --git a/api/core/app_bundle/__init__.py b/api/core/app_bundle/__init__.py
new file mode 100644
index 0000000000..5c1c22f206
--- /dev/null
+++ b/api/core/app_bundle/__init__.py
@@ -0,0 +1,5 @@
+from .source_zip_extractor import SourceZipExtractor
+
+__all__ = [
+    "SourceZipExtractor",
+]
diff --git a/api/core/app_bundle/source_zip_extractor.py b/api/core/app_bundle/source_zip_extractor.py
new file mode 100644
index 0000000000..d5bb6a4cac
--- /dev/null
+++ b/api/core/app_bundle/source_zip_extractor.py
@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+import hashlib
+import io
+import zipfile
+from collections.abc import Callable
+from typing import TYPE_CHECKING
+from uuid import uuid4
+
+from core.app.entities.app_asset_entities import AppAssetFileTree, AppAssetNode
+from core.app.entities.app_bundle_entities import ExtractedFile, ExtractedFolder, ZipSecurityError
+
+if TYPE_CHECKING:
+    from extensions.ext_storage import Storage
+
+
+class SourceZipExtractor:
+    def __init__(self, storage: Storage) -> None:
+        self._storage = storage
+
+    def extract_entries(
+        self, zip_bytes: bytes, *, expected_prefix: str
+    ) -> tuple[list[ExtractedFolder], list[ExtractedFile]]:
+        folders: list[ExtractedFolder] = []
+        files: list[ExtractedFile] = []
+
+        with zipfile.ZipFile(io.BytesIO(zip_bytes), "r") as zf:
+            for info in zf.infolist():
+                name = info.filename
+                self._validate_path(name)
+
+                if not name.startswith(expected_prefix):
+                    continue
+
+                relative_path = name[len(expected_prefix) :].lstrip("/")
+                if not relative_path:
+                    continue
+
+                if info.is_dir():
+                    folders.append(ExtractedFolder(path=relative_path.rstrip("/")))
+                else:
+                    content = zf.read(info)
+                    files.append(ExtractedFile(path=relative_path, content=content))
+
+        return folders, files
+
+    def build_tree_and_save(
+        self,
+        folders: list[ExtractedFolder],
+        files: list[ExtractedFile],
+        tenant_id: str,
+        app_id: str,
+        storage_key_fn: Callable[[str, str, str], str],
+    ) -> AppAssetFileTree:
+        tree = AppAssetFileTree()
+        path_to_node_id: dict[str, str] = {}
+
+        all_folder_paths = {f.path for f in folders}
+        for file in files:
+            self._ensure_parent_folders(file.path, all_folder_paths)
+
+        sorted_folders = sorted(all_folder_paths, key=lambda p: p.count("/"))
+        for folder_path in sorted_folders:
+            node_id = str(uuid4())
+            name = folder_path.rsplit("/", 1)[-1]
+            parent_path = folder_path.rsplit("/", 1)[0] if "/" in folder_path else None
+            parent_id = path_to_node_id.get(parent_path) if parent_path else None
+
+            node = AppAssetNode.create_folder(node_id, name, parent_id)
+            tree.add(node)
+            path_to_node_id[folder_path] = node_id
+
+        sorted_files = sorted(files, key=lambda f: f.path)
+        for file in sorted_files:
+            node_id = str(uuid4())
+            name = file.path.rsplit("/", 1)[-1]
+            parent_path = file.path.rsplit("/", 1)[0] if "/" in file.path else None
+            parent_id = path_to_node_id.get(parent_path) if parent_path else None
+
+            checksum = hashlib.sha256(file.content).hexdigest()
+            node = AppAssetNode.create_file(node_id, name, parent_id, len(file.content), checksum)
+            tree.add(node)
+
+            storage_key = storage_key_fn(tenant_id, app_id, node_id)
+            self._storage.save(storage_key, file.content)
+
+        return tree
+
+    def _validate_path(self, path: str) -> None:
+        if ".." in path:
+            raise ZipSecurityError(f"Path traversal detected: {path}")
+        if path.startswith("/"):
+            raise ZipSecurityError(f"Absolute path detected: {path}")
+        if "\\" in path:
+            raise ZipSecurityError(f"Backslash in path: {path}")
+
+    def _ensure_parent_folders(self, file_path: str, folder_set: set[str]) -> None:
+        parts = file_path.split("/")[:-1]
+        for i in range(1, len(parts) + 1):
+            parent = "/".join(parts[:i])
+            folder_set.add(parent)
diff --git a/api/core/sandbox/initializer/app_assets_initializer.py b/api/core/sandbox/initializer/app_assets_initializer.py
index ba21739014..5d73ce2bcd 100644
--- a/api/core/sandbox/initializer/app_assets_initializer.py
+++ b/api/core/sandbox/initializer/app_assets_initializer.py
@@ -37,6 +37,11 @@ class AppAssetsInitializer(AsyncSandboxInitializer):
                 ["sh", "-c", f"unzip {AppAssets.ZIP_PATH} -d {AppAssets.PATH} 2>/dev/null || [ $? -eq 1 ]"],
                 error_message="Failed to unzip assets",
             )
+            # Ensure directories have execute permission for traversal and files are readable
+            .add(
+                ["sh", "-c", f"chmod -R u+rwX,go+rX {AppAssets.PATH}"],
+                error_message="Failed to set permissions on assets",
+            )
             .execute(timeout=APP_ASSETS_DOWNLOAD_TIMEOUT, raise_on_error=True)
         )
 
diff --git a/api/core/skill/skill_compiler.py b/api/core/skill/skill_compiler.py
index 57ba9d9214..4f344f46bd 100644
--- a/api/core/skill/skill_compiler.py
+++ b/api/core/skill/skill_compiler.py
@@ -1,248 +1,390 @@
 import hashlib
-import logging
 import re
-from collections.abc import Mapping
-from datetime import UTC, datetime
-from typing import Any
+from collections.abc import Iterable, Mapping, Sequence
+from dataclasses import dataclass
+from typing import Any, Protocol, cast
 
 from core.app.entities.app_asset_entities import AppAssetFileTree
 from core.skill.entities.asset_references import AssetReferences
 from core.skill.entities.skill_bundle import SkillBundle
 from core.skill.entities.skill_bundle_entry import SkillBundleEntry, SourceInfo
 from core.skill.entities.skill_document import SkillDocument
-from core.skill.entities.skill_metadata import (
-    FileReference,
-    SkillMetadata,
-    ToolConfiguration,
-    ToolReference,
-)
+from core.skill.entities.skill_metadata import FileReference, SkillMetadata, ToolConfiguration, ToolReference
 from core.skill.entities.tool_dependencies import ToolDependencies, ToolDependency
 from core.tools.entities.tool_entities import ToolProviderType
 
-logger = logging.getLogger(__name__)
 
-TOOL_REFERENCE_PATTERN = re.compile(r"§\[tool\]\.\[([^\]]+)\]\.\[([^\]]+)\]\.\[([^\]]+)\]§")
-FILE_REFERENCE_PATTERN = re.compile(r"§\[file\]\.\[([^\]]+)\]\.\[([^\]]+)\]§")
+class PathResolver(Protocol):
+    def resolve(self, source_id: str, target_id: str) -> str: ...
+
+
+class ToolResolver(Protocol):
+    def resolve(self, tool_ref: ToolReference) -> str: ...
+
+
+@dataclass(frozen=True)
+class CompilerConfig:
+    tool_pattern: re.Pattern[str] = re.compile(r"§\[tool\]\.\[.*?\]\.\[.*?\]\.\[(.*?)\]§")
+    file_pattern: re.Pattern[str] = re.compile(r"§\[file\]\.\[.*?\]\.\[(.*?)\]§")
+
+
+class FileTreePathResolver:
+    def __init__(self, tree: AppAssetFileTree, base_path: str = ""):
+        self._tree = tree
+        self._base_path = base_path.rstrip("/")
+
+    def resolve(self, source_id: str, target_id: str) -> str:
+        source_node = self._tree.get(source_id)
+        target_node = self._tree.get(target_id)
+
+        if target_node is None:
+            return "[File not found]"
+
+        if source_node is not None:
+            return self._tree.relative_path(source_node, target_node)
+
+        full_path = self._tree.get_path(target_node.id)
+        if self._base_path:
+            return f"{self._base_path}/{full_path}"
+        return full_path
+
+
+class DefaultToolResolver:
+    def resolve(self, tool_ref: ToolReference) -> str:
+        return f"[Executable: {tool_ref.tool_name}_{tool_ref.uuid} --help command]"
 
 
 class SkillCompiler:
-    def _parse_metadata(self, content: str, raw_metadata: Mapping[str, Any]) -> SkillMetadata:
-        tools_raw: dict[str, Any] = dict(raw_metadata.get("tools", {}))
-        tools: dict[str, ToolReference] = {}
-        files: list[FileReference] = []
-
-        for match in TOOL_REFERENCE_PATTERN.finditer(content):
-            tool_id = match.group(3)
-            tool_name = match.group(2)
-            tool_provider = match.group(1)
-            tool_meta = tools_raw.get(tool_id)
-            if tool_meta is None:
-                continue
-
-            config_raw = tool_meta.get("configuration", {})
-            configuration = ToolConfiguration.model_validate(config_raw) if config_raw else None
-            tools[tool_id] = ToolReference(
-                uuid=tool_id,
-                type=ToolProviderType.value_of(tool_meta.get("type")),
-                provider=tool_provider,
-                tool_name=tool_name,
-                credential_id=tool_meta.get("credential_id"),
-                configuration=configuration,
-            )
-
-        for match in FILE_REFERENCE_PATTERN.finditer(content):
-            files.append(
-                FileReference(
-                    source=match.group(1),
-                    asset_id=match.group(2),
-                )
-            )
-
-        return SkillMetadata(tools=tools, files=files)
+    def __init__(
+        self,
+        path_resolver: PathResolver | None = None,
+        tool_resolver: ToolResolver | None = None,
+        config: CompilerConfig | None = None,
+    ):
+        self._path_resolver = path_resolver
+        self._tool_resolver = tool_resolver or DefaultToolResolver()
+        self._config = config or CompilerConfig()
 
     def compile_all(
         self,
-        documents: list[SkillDocument],
+        documents: Iterable[SkillDocument],
         file_tree: AppAssetFileTree,
         assets_id: str,
     ) -> SkillBundle:
-        bundle = SkillBundle(
-            assets_id=assets_id,
-            built_at=datetime.now(UTC),
-        )
-
-        doc_map: dict[str, SkillDocument] = {doc.skill_id: doc for doc in documents}
-        parsed_metadata: dict[str, SkillMetadata] = {}
-
-        for doc in documents:
-            metadata = self._parse_metadata(doc.content, doc.metadata)
-            parsed_metadata[doc.skill_id] = metadata
-            direct_skill_refs = self._extract_skill_refs(metadata, doc_map)
-            bundle.dependency_graph[doc.skill_id] = list(direct_skill_refs)
-            for ref_id in direct_skill_refs:
-                if ref_id not in bundle.reverse_graph:
-                    bundle.reverse_graph[ref_id] = []
-                bundle.reverse_graph[ref_id].append(doc.skill_id)
-
-        for doc in documents:
-            metadata = parsed_metadata[doc.skill_id]
-            entry = self._compile_single(doc, metadata, bundle, parsed_metadata, file_tree)
-            bundle.upsert(entry)
-
-        return bundle
+        path_resolver = self._path_resolver or FileTreePathResolver(file_tree)
+        return self._compile_batch_internal(documents, assets_id, path_resolver)
 
     def compile_one(
         self,
         bundle: SkillBundle,
         document: SkillDocument,
         file_tree: AppAssetFileTree,
-        all_documents: dict[str, SkillDocument] | None = None,
+        base_path: str = "",
     ) -> SkillBundleEntry:
-        doc_map = all_documents or {}
-        if document.skill_id not in doc_map:
-            doc_map[document.skill_id] = document
-
-        parsed_metadata: dict[str, SkillMetadata] = {}
-        for skill_id, doc in doc_map.items():
-            parsed_metadata[skill_id] = self._parse_metadata(doc.content, doc.metadata)
-
-        metadata = parsed_metadata[document.skill_id]
-        direct_skill_refs = self._extract_skill_refs(metadata, doc_map)
-        bundle.dependency_graph[document.skill_id] = list(direct_skill_refs)
-        for ref_id in direct_skill_refs:
-            if ref_id not in bundle.reverse_graph:
-                bundle.reverse_graph[ref_id] = []
-            if document.skill_id not in bundle.reverse_graph[ref_id]:
-                bundle.reverse_graph[ref_id].append(document.skill_id)
-
-        return self._compile_single(document, metadata, bundle, parsed_metadata, file_tree)
-
-    def _compile_single(
-        self,
-        document: SkillDocument,
-        metadata: SkillMetadata,
-        bundle: SkillBundle,
-        parsed_metadata: dict[str, SkillMetadata],
-        file_tree: AppAssetFileTree,
-    ) -> SkillBundleEntry:
-        all_tools, all_files = self._compute_transitive_closure(
-            document.skill_id, bundle, parsed_metadata
+        path_resolver = self._path_resolver or FileTreePathResolver(file_tree, base_path)
+        resolved_content, tool_dependencies = self._compile_template_internal(
+            document.content, document.metadata, bundle, path_resolver
         )
 
-        current_node = file_tree.get(document.skill_id)
-
-        resolved_content = self._resolve_content(
-            document.content, metadata, current_node, file_tree
-        )
-
-        content_digest = hashlib.sha256(document.content.encode("utf-8")).hexdigest()
+        metadata = self._parse_metadata(document.content, document.metadata)
+        final_files: dict[str, FileReference] = {f.asset_id: f for f in metadata.files}
 
         return SkillBundleEntry(
             skill_id=document.skill_id,
             source=SourceInfo(
                 asset_id=document.skill_id,
-                content_digest=content_digest,
+                content_digest=hashlib.sha256(document.content.encode("utf-8")).hexdigest(),
+            ),
+            tools=tool_dependencies,
+            files=AssetReferences(references=list(final_files.values())),
+            content=resolved_content,
+        )
+
+    def _compile_batch_internal(
+        self,
+        documents: Iterable[SkillDocument],
+        assets_id: str,
+        path_resolver: PathResolver,
+    ) -> SkillBundle:
+        doc_map = {doc.skill_id: doc for doc in documents}
+        graph: dict[str, set[str]] = {}
+        metadata_cache: dict[str, SkillMetadata] = {}
+
+        # Phase 1: Parse metadata and build dependency graph
+        for doc in doc_map.values():
+            metadata = self._parse_metadata(doc.content, doc.metadata)
+            metadata_cache[doc.skill_id] = metadata
+
+            deps: set[str] = set()
+            for file_ref in metadata.files:
+                if file_ref.asset_id in doc_map:
+                    deps.add(file_ref.asset_id)
+            graph[doc.skill_id] = deps
+
+        bundle = SkillBundle(assets_id=assets_id)
+        bundle.dependency_graph = {k: list(v) for k, v in graph.items()}
+
+        # Build reverse graph for propagation
+        reverse_graph: dict[str, set[str]] = {skill_id: set() for skill_id in doc_map}
+        for skill_id, deps in graph.items():
+            for dep_id in deps:
+                if dep_id in reverse_graph:
+                    reverse_graph[dep_id].add(skill_id)
+        bundle.reverse_graph = {k: list(v) for k, v in reverse_graph.items()}
+
+        # Phase 2: Compile each skill independently (content + direct dependencies only)
+        for skill_id, doc in doc_map.items():
+            metadata = metadata_cache[skill_id]
+            entry = self._compile_node_direct(doc, metadata, path_resolver)
+            bundle.upsert(entry)
+
+        # Phase 3: Propagate transitive dependencies until fixed-point
+        self._propagate_transitive_dependencies(bundle, graph)
+
+        return bundle
+
+    def _compile_node_direct(
+        self,
+        doc: SkillDocument,
+        metadata: SkillMetadata,
+        path_resolver: PathResolver,
+    ) -> SkillBundleEntry:
+        """Compile a single skill with only its direct dependencies (no transitive)."""
+        direct_tools: dict[str, ToolDependency] = {}
+        direct_refs: dict[str, ToolReference] = {}
+
+        for tool_ref in metadata.tools.values():
+            key = f"{tool_ref.provider}.{tool_ref.tool_name}"
+            if key not in direct_tools:
+                direct_tools[key] = ToolDependency(
+                    type=tool_ref.type,
+                    provider=tool_ref.provider,
+                    tool_name=tool_ref.tool_name,
+                )
+            direct_refs[tool_ref.uuid] = tool_ref
+
+        direct_files: dict[str, FileReference] = {f.asset_id: f for f in metadata.files}
+        resolved_content = self._resolve_content(doc.content, metadata, path_resolver, doc.skill_id)
+
+        return SkillBundleEntry(
+            skill_id=doc.skill_id,
+            source=SourceInfo(
+                asset_id=doc.skill_id,
+                content_digest=hashlib.sha256(doc.content.encode("utf-8")).hexdigest(),
             ),
             tools=ToolDependencies(
-                dependencies=list(all_tools.values()),
-                references=list(metadata.tools.values()),
+                dependencies=list(direct_tools.values()),
+                references=list(direct_refs.values()),
             ),
             files=AssetReferences(
-                references=list(all_files.values()),
+                references=list(direct_files.values()),
             ),
             content=resolved_content,
         )
 
-    def _extract_skill_refs(
+    def _propagate_transitive_dependencies(
         self,
-        metadata: SkillMetadata,
-        doc_map: dict[str, SkillDocument],
-    ) -> set[str]:
-        skill_refs: set[str] = set()
-        for file_ref in metadata.files:
-            if file_ref.asset_id in doc_map:
-                skill_refs.add(file_ref.asset_id)
-        return skill_refs
-
-    def _compute_transitive_closure(
-        self,
-        skill_id: str,
         bundle: SkillBundle,
-        parsed_metadata: dict[str, SkillMetadata],
-    ) -> tuple[dict[str, ToolDependency], dict[str, FileReference]]:
-        all_tools: dict[str, ToolDependency] = {}
-        all_files: dict[str, FileReference] = {}
+        graph: dict[str, set[str]],
+    ) -> None:
+        """Iteratively propagate transitive dependencies until no changes occur."""
+        changed = True
+        while changed:
+            changed = False
+            for skill_id, dep_ids in graph.items():
+                entry = bundle.get(skill_id)
+                if not entry:
+                    continue
 
-        visited: set[str] = set()
-        queue = [skill_id]
+                # Collect current tools and files
+                current_tools: dict[str, ToolDependency] = {
+                    f"{d.provider}.{d.tool_name}": d for d in entry.tools.dependencies
+                }
+                current_refs: dict[str, ToolReference] = {r.uuid: r for r in entry.tools.references}
+                current_files: dict[str, FileReference] = {f.asset_id: f for f in entry.files.references}
 
-        while queue:
-            current_id = queue.pop(0)
-            if current_id in visited:
-                continue
-            visited.add(current_id)
+                original_tool_count = len(current_tools)
+                original_ref_count = len(current_refs)
+                original_file_count = len(current_files)
 
-            metadata = parsed_metadata.get(current_id)
-            if metadata is None:
-                existing_entry = bundle.get(current_id)
-                if existing_entry:
-                    for dep in existing_entry.tools.dependencies:
-                        key = f"{dep.provider}.{dep.tool_name}"
-                        if key not in all_tools:
-                            all_tools[key] = dep
-                    for file_ref in existing_entry.files.references:
-                        if file_ref.asset_id not in all_files:
-                            all_files[file_ref.asset_id] = file_ref
-                continue
+                # Merge from dependencies
+                for dep_id in dep_ids:
+                    dep_entry = bundle.get(dep_id)
+                    if not dep_entry:
+                        continue
 
-            for tool_ref in metadata.tools.values():
-                key = f"{tool_ref.provider}.{tool_ref.tool_name}"
-                if key not in all_tools:
-                    all_tools[key] = ToolDependency(
-                        type=tool_ref.type,
-                        provider=tool_ref.provider,
-                        tool_name=tool_ref.tool_name,
+                    for tool_dep in dep_entry.tools.dependencies:
+                        key = f"{tool_dep.provider}.{tool_dep.tool_name}"
+                        if key not in current_tools:
+                            current_tools[key] = tool_dep
+
+                    for tool_ref in dep_entry.tools.references:
+                        if tool_ref.uuid not in current_refs:
+                            current_refs[tool_ref.uuid] = tool_ref
+
+                    for file_ref in dep_entry.files.references:
+                        if file_ref.asset_id not in current_files:
+                            current_files[file_ref.asset_id] = file_ref
+
+                # Check if anything changed
+                if (
+                    len(current_tools) != original_tool_count
+                    or len(current_refs) != original_ref_count
+                    or len(current_files) != original_file_count
+                ):
+                    changed = True
+                    # Update the entry with new transitive dependencies
+                    updated_entry = SkillBundleEntry(
+                        skill_id=entry.skill_id,
+                        source=entry.source,
+                        tools=ToolDependencies(
+                            dependencies=list(current_tools.values()),
+                            references=list(current_refs.values()),
+                        ),
+                        files=AssetReferences(
+                            references=list(current_files.values()),
+                        ),
+                        content=entry.content,
                     )
+                    bundle.upsert(updated_entry)
 
-            for file_ref in metadata.files:
-                if file_ref.asset_id not in all_files:
-                    all_files[file_ref.asset_id] = file_ref
-
-            for dep_id in bundle.dependency_graph.get(current_id, []):
-                if dep_id not in visited:
-                    queue.append(dep_id)
-
-        return all_tools, all_files
-
-    def _resolve_content(
+    def _compile_template_internal(
         self,
         content: str,
+        metadata_dict: Mapping[str, Any],
+        context: SkillBundle,
+        path_resolver: PathResolver,
+    ) -> tuple[str, ToolDependencies]:
+        metadata = self._parse_metadata(content, metadata_dict)
+
+        direct_deps: list[SkillBundleEntry] = []
+        for file_ref in metadata.files:
+            artifact = context.get(file_ref.asset_id)
+            if artifact:
+                direct_deps.append(artifact)
+
+        final_tools, final_refs = self._aggregate_dependencies(metadata, direct_deps)
+
+        resolved_content = self._resolve_content(content, metadata, path_resolver, current_id="<template>")
+
+        return resolved_content, ToolDependencies(
+            dependencies=list(final_tools.values()), references=list(final_refs.values())
+        )
+
+    def _compile_node(
+        self,
+        doc: SkillDocument,
         metadata: SkillMetadata,
-        current_node: Any,
-        file_tree: AppAssetFileTree,
+        direct_deps: Sequence[SkillBundleEntry],
+        path_resolver: PathResolver,
+    ) -> SkillBundleEntry:
+        final_tools, final_refs = self._aggregate_dependencies(metadata, direct_deps)
+
+        final_files: dict[str, FileReference] = {}
+        for f in metadata.files:
+            final_files[f.asset_id] = f
+
+        for dep in direct_deps:
+            for f in dep.files.references:
+                if f.asset_id not in final_files:
+                    final_files[f.asset_id] = f
+
+        resolved_content = self._resolve_content(doc.content, metadata, path_resolver, doc.skill_id)
+
+        return SkillBundleEntry(
+            skill_id=doc.skill_id,
+            source=SourceInfo(
+                asset_id=doc.skill_id,
+                content_digest=hashlib.sha256(doc.content.encode("utf-8")).hexdigest(),
+            ),
+            tools=ToolDependencies(
+                dependencies=list(final_tools.values()),
+                references=list(final_refs.values()),
+            ),
+            files=AssetReferences(
+                references=list(final_files.values()),
+            ),
+            content=resolved_content,
+        )
+
+    def _aggregate_dependencies(
+        self, metadata: SkillMetadata, direct_deps: Sequence[SkillBundleEntry]
+    ) -> tuple[dict[str, ToolDependency], dict[str, ToolReference]]:
+        all_tools: dict[str, ToolDependency] = {}
+        all_refs: dict[str, ToolReference] = {}
+
+        for tool_ref in metadata.tools.values():
+            key = f"{tool_ref.provider}.{tool_ref.tool_name}"
+            if key not in all_tools:
+                all_tools[key] = ToolDependency(
+                    type=tool_ref.type,
+                    provider=tool_ref.provider,
+                    tool_name=tool_ref.tool_name,
+                )
+            all_refs[tool_ref.uuid] = tool_ref
+
+        for dep in direct_deps:
+            for tool_dep in dep.tools.dependencies:
+                key = f"{tool_dep.provider}.{tool_dep.tool_name}"
+                if key not in all_tools:
+                    all_tools[key] = tool_dep
+
+            for tool_ref in dep.tools.references:
+                if tool_ref.uuid not in all_refs:
+                    all_refs[tool_ref.uuid] = tool_ref
+
+        return all_tools, all_refs
+
+    def _resolve_content(
+        self, content: str, metadata: SkillMetadata, path_resolver: PathResolver, current_id: str
     ) -> str:
-        if not content:
-            return content
+        def replace_file(match: re.Match[str]) -> str:
+            target_id = match.group(1)
+            try:
+                return path_resolver.resolve(current_id, target_id)
+            except Exception:
+                return match.group(0)
 
-        for match in FILE_REFERENCE_PATTERN.finditer(content):
-            file_id = match.group(2)
-            file_node = file_tree.get(file_id)
-            if file_node is None:
-                logger.warning("File not found for id=%s, skipping", file_id)
-                content = content.replace(match.group(0), "[File not found]")
-                continue
-            if current_node is not None:
-                content = content.replace(match.group(0), file_tree.relative_path(current_node, file_node))
-            else:
-                content = content.replace(match.group(0), f"[{file_node.name}]")
-
-        for match in TOOL_REFERENCE_PATTERN.finditer(content):
-            tool_id = match.group(3)
-            tool = metadata.tools.get(tool_id)
-            if tool is None:
-                logger.warning("Tool not found for id=%s, skipping", tool_id)
-                content = content.replace(match.group(0), f"[Tool not found: {tool_id}]")
-                continue
-            content = content.replace(match.group(0), f"[Bash Command: {tool.tool_name}_{tool_id}]")
+        def replace_tool(match: re.Match[str]) -> str:
+            tool_id = match.group(1)
+            tool_ref = metadata.tools.get(tool_id)
+            if not tool_ref:
+                return f"[Tool not found: {tool_id}]"
+            return self._tool_resolver.resolve(tool_ref)
 
+        content = self._config.file_pattern.sub(replace_file, content)
+        content = self._config.tool_pattern.sub(replace_tool, content)
         return content
+
+    def _parse_metadata(self, content: str, raw_metadata: Mapping[str, Any]) -> SkillMetadata:
+        tools_raw = dict(raw_metadata.get("tools", {}))
+        tools: dict[str, ToolReference] = {}
+
+        tool_iter = re.finditer(r"§\[tool\]\.\[([^\]]+)\]\.\[([^\]]+)\]\.\[([^\]]+)\]§", content)
+        for match in tool_iter:
+            provider, name, uuid = match.group(1), match.group(2), match.group(3)
+            if uuid in tools_raw:
+                meta = tools_raw[uuid]
+                if isinstance(meta, ToolReference):
+                    tools[uuid] = meta
+                elif isinstance(meta, dict):
+                    tool_type_str = cast(str | None, meta.get("type"))
+                    if tool_type_str:
+                        tools[uuid] = ToolReference(
+                            uuid=uuid,
+                            type=ToolProviderType.value_of(tool_type_str),
+                            provider=provider,
+                            tool_name=name,
+                            credential_id=cast(str | None, meta.get("credential_id")),
+                            configuration=ToolConfiguration.model_validate(meta.get("configuration", {}))
+                            if meta.get("configuration")
+                            else None,
+                        )
+
+        parsed_files: list[FileReference] = []
+        file_iter = re.finditer(r"§\[file\]\.\[([^\]]+)\]\.\[([^\]]+)\]§", content)
+        for match in file_iter:
+            source, asset_id = match.group(1), match.group(2)
+            parsed_files.append(FileReference(source=source, asset_id=asset_id))
+
+        return SkillMetadata(tools=tools, files=parsed_files)
diff --git a/api/services/app_asset_service.py b/api/services/app_asset_service.py
index e9e7db602d..0bf20b8d42 100644
--- a/api/services/app_asset_service.py
+++ b/api/services/app_asset_service.py
@@ -13,13 +13,13 @@ from core.app.entities.app_asset_entities import (
     TreePathConflictError,
 )
 from core.app_assets.builder import AssetBuildPipeline, BuildContext
-from core.app_assets.packager.zip_packager import ZipPackager
+from core.app_assets.converters import tree_to_asset_items
+from core.app_assets.packager import AssetZipPackager
 from core.app_assets.paths import AssetPaths
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from extensions.ext_storage import storage
 from extensions.storage.file_presign_storage import FilePresignStorage
-from libs.datetime_utils import naive_utc_now
 from models.app_asset import AppAssets
 from models.model import App
 
@@ -445,35 +445,39 @@ class AppAssetService:
             AppAssetService._clear_draft_download_cache(cache_keys)
 
     @staticmethod
-    def publish(app_model: App, account_id: str) -> AppAssets:
+    def publish(session: Session, app_model: App, account_id: str, workflow_id: str) -> AppAssets:
         tenant_id = app_model.tenant_id
         app_id = app_model.id
-        with Session(db.engine, expire_on_commit=False) as session:
-            assets = AppAssetService.get_or_create_assets(session, app_model, account_id)
-            tree = assets.asset_tree
 
-            publish_id = str(uuid4())
+        assets = AppAssetService.get_or_create_assets(session, app_model, account_id)
+        tree = assets.asset_tree
 
-            published = AppAssets(
-                id=publish_id,
-                tenant_id=tenant_id,
-                app_id=app_id,
-                version=str(naive_utc_now()),
-                created_by=account_id,
-            )
-            published.asset_tree = tree
-            session.add(published)
-            session.flush()
+        publish_id = str(uuid4())
 
-            ctx = BuildContext(tenant_id=tenant_id, app_id=app_id, build_id=publish_id)
-            built_assets = AssetBuildPipeline().build_all(tree, ctx)
+        published = AppAssets(
+            id=publish_id,
+            tenant_id=tenant_id,
+            app_id=app_id,
+            version=workflow_id,
+            created_by=account_id,
+        )
+        published.asset_tree = tree
+        session.add(published)
+        session.flush()
 
-            packager = ZipPackager(storage)
-            zip_bytes = packager.package(built_assets)
-            zip_key = AssetPaths.build_zip(tenant_id, app_id, publish_id)
-            storage.save(zip_key, zip_bytes)
+        ctx = BuildContext(tenant_id=tenant_id, app_id=app_id, build_id=publish_id)
+        built_assets = AssetBuildPipeline().build_all(tree, ctx)
 
-            session.commit()
+        packager = AssetZipPackager(storage)
+
+        runtime_zip_bytes = packager.package(built_assets)
+        runtime_zip_key = AssetPaths.build_zip(tenant_id, app_id, publish_id)
+        storage.save(runtime_zip_key, runtime_zip_bytes)
+
+        source_items = tree_to_asset_items(tree, tenant_id, app_id)
+        source_zip_bytes = packager.package(source_items)
+        source_zip_key = AssetPaths.build_source_zip(tenant_id, app_id, workflow_id)
+        storage.save(source_zip_key, source_zip_bytes)
 
         return published
 
@@ -483,7 +487,12 @@ class AppAssetService:
         tree = assets.asset_tree
 
         ctx = BuildContext(tenant_id=tenant_id, app_id=app_id, build_id=assets.id)
-        AssetBuildPipeline().build_all(tree, ctx)
+        built_assets = AssetBuildPipeline().build_all(tree, ctx)
+
+        packager = AssetZipPackager(storage)
+        zip_bytes = packager.package(built_assets)
+        zip_key = AssetPaths.build_zip(tenant_id, app_id, assets.id)
+        storage.save(zip_key, zip_bytes)
 
     @staticmethod
     def get_file_download_url(
@@ -503,3 +512,39 @@ class AppAssetService:
             storage_key = AssetPaths.draft_file(app_model.tenant_id, app_model.id, node_id)
             presign_storage = FilePresignStorage(storage.storage_runner)
             return presign_storage.get_download_url(storage_key, expires_in)
+
+    @staticmethod
+    def get_published_assets_by_workflow_id(tenant_id: str, app_id: str, workflow_id: str) -> AppAssets | None:
+        with Session(db.engine, expire_on_commit=False) as session:
+            return (
+                session.query(AppAssets)
+                .filter(
+                    AppAssets.tenant_id == tenant_id,
+                    AppAssets.app_id == app_id,
+                    AppAssets.version == workflow_id,
+                )
+                .first()
+            )
+
+    @staticmethod
+    def get_source_zip_bytes(tenant_id: str, app_id: str, workflow_id: str) -> bytes | None:
+        source_zip_key = AssetPaths.build_source_zip(tenant_id, app_id, workflow_id)
+        try:
+            return storage.load_once(source_zip_key)
+        except Exception:
+            logger.warning("Source zip not found: %s", source_zip_key)
+            return None
+
+    @staticmethod
+    def set_draft_assets(
+        app_model: App,
+        account_id: str,
+        new_tree: AppAssetFileTree,
+    ) -> AppAssets:
+        with Session(db.engine, expire_on_commit=False) as session:
+            assets = AppAssetService.get_or_create_assets(session, app_model, account_id)
+            assets.asset_tree = new_tree
+            assets.updated_by = account_id
+            session.commit()
+
+        return assets
diff --git a/api/services/app_bundle_service.py b/api/services/app_bundle_service.py
new file mode 100644
index 0000000000..c0956ccc9e
--- /dev/null
+++ b/api/services/app_bundle_service.py
@@ -0,0 +1,255 @@
+from __future__ import annotations
+
+import io
+import logging
+import re
+import zipfile
+
+import yaml
+from sqlalchemy.orm import Session
+
+from core.app.entities.app_bundle_entities import (
+    BUNDLE_DSL_FILENAME_PATTERN,
+    BUNDLE_MAX_SIZE,
+    BundleExportResult,
+    BundleFormatError,
+    ZipSecurityError,
+)
+from core.app_assets.converters import tree_to_asset_items
+from core.app_assets.packager import AssetZipPackager
+from core.app_assets.paths import AssetPaths
+from core.app_bundle import SourceZipExtractor
+from extensions.ext_database import db
+from extensions.ext_storage import storage
+from models import Account, App
+
+from .app_asset_service import AppAssetService
+from .app_dsl_service import AppDslService, Import
+
+logger = logging.getLogger(__name__)
+
+
+class AppBundleService:
+    @staticmethod
+    def publish(
+        session: Session,
+        app_model: App,
+        account: Account,
+        marked_name: str = "",
+        marked_comment: str = "",
+    ):
+        """
+        Publish App Bundle (workflow + assets).
+        Coordinates WorkflowService and AppAssetService publishing in a single transaction.
+        """
+        from models.workflow import Workflow
+        from services.workflow_service import WorkflowService
+
+        # 1. Publish workflow
+        workflow: Workflow = WorkflowService().publish_workflow(
+            session=session,
+            app_model=app_model,
+            account=account,
+            marked_name=marked_name,
+            marked_comment=marked_comment,
+        )
+
+        # 2. Publish assets (bound to workflow_id)
+        AppAssetService.publish(
+            session=session,
+            app_model=app_model,
+            account_id=account.id,
+            workflow_id=workflow.id,
+        )
+
+        return workflow
+
+    @staticmethod
+    def export_bundle(
+        app_model: App,
+        include_secret: bool = False,
+        workflow_id: str | None = None,
+    ) -> BundleExportResult:
+        dsl_content = AppDslService.export_dsl(
+            app_model=app_model,
+            include_secret=include_secret,
+            workflow_id=workflow_id,
+        )
+
+        safe_name = AppBundleService._sanitize_filename(app_model.name)
+        assets_prefix = safe_name
+
+        zip_buffer = io.BytesIO()
+        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
+            zf.writestr(f"{safe_name}.yml", dsl_content.encode("utf-8"))
+
+            assets_zip_bytes = AppBundleService._get_assets_zip_bytes(app_model, workflow_id)
+            if assets_zip_bytes:
+                AppBundleService._merge_assets_into_bundle(zf, assets_zip_bytes, assets_prefix)
+
+        return BundleExportResult(
+            zip_bytes=zip_buffer.getvalue(),
+            filename=f"{safe_name}.zip",
+        )
+
+    @staticmethod
+    def import_bundle(
+        account: Account,
+        zip_bytes: bytes,
+        name: str | None = None,
+        description: str | None = None,
+        icon_type: str | None = None,
+        icon: str | None = None,
+        icon_background: str | None = None,
+    ) -> Import:
+        if len(zip_bytes) > BUNDLE_MAX_SIZE:
+            raise BundleFormatError(f"Bundle size exceeds limit: {BUNDLE_MAX_SIZE} bytes")
+
+        dsl_content, assets_prefix = AppBundleService._extract_dsl_from_bundle(zip_bytes)
+
+        with Session(db.engine) as session:
+            dsl_service = AppDslService(session)
+            import_result = dsl_service.import_app(
+                account=account,
+                import_mode="yaml-content",
+                yaml_content=dsl_content,
+                name=name,
+                description=description,
+                icon_type=icon_type,
+                icon=icon,
+                icon_background=icon_background,
+                app_id=None,
+            )
+            session.commit()
+
+        if import_result.app_id and assets_prefix:
+            AppBundleService._import_assets_from_bundle(
+                zip_bytes=zip_bytes,
+                assets_prefix=assets_prefix,
+                app_id=import_result.app_id,
+                account_id=account.id,
+            )
+
+        return import_result
+
+    @staticmethod
+    def _get_assets_zip_bytes(app_model: App, workflow_id: str | None) -> bytes | None:
+        tenant_id = app_model.tenant_id
+        app_id = app_model.id
+
+        if workflow_id is None:
+            return AppBundleService._package_draft_assets(app_model)
+        else:
+            return AppAssetService.get_source_zip_bytes(tenant_id, app_id, workflow_id)
+
+    @staticmethod
+    def _package_draft_assets(app_model: App) -> bytes | None:
+        assets = AppAssetService.get_assets(
+            tenant_id=app_model.tenant_id,
+            app_id=app_model.id,
+            user_id="",
+            is_draft=True,
+        )
+        if not assets:
+            return None
+
+        tree = assets.asset_tree
+        if not tree.nodes:
+            return None
+
+        items = tree_to_asset_items(tree, app_model.tenant_id, app_model.id)
+        packager = AssetZipPackager(storage)
+        return packager.package(items)
+
+    @staticmethod
+    def _merge_assets_into_bundle(
+        bundle_zf: zipfile.ZipFile,
+        assets_zip_bytes: bytes,
+        prefix: str,
+    ) -> None:
+        with zipfile.ZipFile(io.BytesIO(assets_zip_bytes), "r") as assets_zf:
+            for info in assets_zf.infolist():
+                content = assets_zf.read(info)
+                new_path = f"{prefix}/{info.filename}"
+                if info.is_dir():
+                    bundle_zf.writestr(zipfile.ZipInfo(new_path), "")
+                else:
+                    bundle_zf.writestr(new_path, content)
+
+    @staticmethod
+    def _extract_dsl_from_bundle(zip_bytes: bytes) -> tuple[str, str | None]:
+        dsl_content: str | None = None
+        dsl_filename: str | None = None
+
+        with zipfile.ZipFile(io.BytesIO(zip_bytes), "r") as zf:
+            for info in zf.infolist():
+                if info.is_dir():
+                    continue
+                if BUNDLE_DSL_FILENAME_PATTERN.match(info.filename):
+                    if dsl_content is not None:
+                        raise BundleFormatError("Multiple DSL files found in bundle")
+                    dsl_content = zf.read(info).decode("utf-8")
+                    dsl_filename = info.filename
+
+        if dsl_content is None or dsl_filename is None:
+            raise BundleFormatError("No DSL file (*.yml or *.yaml) found in bundle root")
+
+        yaml.safe_load(dsl_content)
+
+        assets_prefix = dsl_filename.rsplit(".", 1)[0]
+        has_assets = AppBundleService._check_assets_prefix_exists(zip_bytes, assets_prefix)
+
+        return dsl_content, assets_prefix if has_assets else None
+
+    @staticmethod
+    def _check_assets_prefix_exists(zip_bytes: bytes, prefix: str) -> bool:
+        with zipfile.ZipFile(io.BytesIO(zip_bytes), "r") as zf:
+            for info in zf.infolist():
+                if info.filename.startswith(f"{prefix}/"):
+                    return True
+        return False
+
+    @staticmethod
+    def _import_assets_from_bundle(
+        zip_bytes: bytes,
+        assets_prefix: str,
+        app_id: str,
+        account_id: str,
+    ) -> None:
+        app_model = db.session.query(App).filter(App.id == app_id).first()
+        if not app_model:
+            logger.warning("App not found for asset import: %s", app_id)
+            return
+
+        extractor = SourceZipExtractor(storage)
+        try:
+            folders, files = extractor.extract_entries(
+                zip_bytes,
+                expected_prefix=f"{assets_prefix}/",
+            )
+        except ZipSecurityError as e:
+            logger.warning("Zip security error during asset import: %s", e)
+            return
+
+        if not folders and not files:
+            return
+
+        new_tree = extractor.build_tree_and_save(
+            folders=folders,
+            files=files,
+            tenant_id=app_model.tenant_id,
+            app_id=app_model.id,
+            storage_key_fn=AssetPaths.draft_file,
+        )
+
+        AppAssetService.set_draft_assets(
+            app_model=app_model,
+            account_id=account_id,
+            new_tree=new_tree,
+        )
+
+    @staticmethod
+    def _sanitize_filename(name: str) -> str:
+        safe = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", name)
+        safe = safe.strip(". ")
+        return safe[:100] if safe else "app"
diff --git a/api/services/workflow_service.py b/api/services/workflow_service.py
index 2991b5998a..b9b155ce80 100644
--- a/api/services/workflow_service.py
+++ b/api/services/workflow_service.py
@@ -40,7 +40,6 @@ from models.tools import WorkflowToolProvider
 from models.workflow import Workflow, WorkflowNodeExecutionModel, WorkflowNodeExecutionTriggeredFrom, WorkflowType
 from models.workflow_features import WorkflowFeatures
 from repositories.factory import DifyAPIRepositoryFactory
-from services.app_asset_service import AppAssetService
 from services.billing_service import BillingService
 from services.enterprise.plugin_manager_service import PluginCredentialType
 from services.errors.app import IsDraftWorkflowError, TriggerNodeLimitExceededError, WorkflowHashNotEqualError
@@ -156,8 +155,6 @@ class WorkflowService:
             .first()
         )
 
-        AppAssetService.publish(app_model=app_model, account_id=app_model.created_by)
-
         return workflow
 
     def get_all_published_workflow(
diff --git a/api/tests/unit_tests/core/skill/test_skill_compiler.py b/api/tests/unit_tests/core/skill/test_skill_compiler.py
index a382c46df9..d0c5813a94 100644
--- a/api/tests/unit_tests/core/skill/test_skill_compiler.py
+++ b/api/tests/unit_tests/core/skill/test_skill_compiler.py
@@ -1,13 +1,32 @@
 from typing import Any
 
 from core.app.entities.app_asset_entities import AppAssetFileTree, AppAssetNode
-from core.skill.entities.skill_bundle import SkillBundle
 from core.skill.entities.skill_document import SkillDocument
-from core.skill.entities.skill_metadata import FileReference, ToolConfiguration, ToolReference
+from core.skill.entities.skill_metadata import FileReference, ToolConfiguration, ToolFieldConfig, ToolReference
 from core.skill.skill_compiler import SkillCompiler
 from core.tools.entities.tool_entities import ToolProviderType
 
 
+class MockPathResolver:
+    def __init__(self, tree: AppAssetFileTree):
+        self.tree = tree
+
+    def resolve(self, source_id: str, target_id: str) -> str:
+        source_node = self.tree.get(source_id)
+        target_node = self.tree.get(target_id)
+        if not source_node or not target_node:
+            if target_node:
+                return "./" + target_node.name
+            return f"./{target_id}"
+
+        return self.tree.relative_path(source_node, target_node)
+
+
+class MockToolResolver:
+    def resolve(self, tool_ref: ToolReference) -> str:
+        return f"[Executable: {tool_ref.tool_name}_{tool_ref.uuid} --help command]"
+
+
 def create_file_tree(*nodes: AppAssetNode) -> AppAssetFileTree:
     tree = AppAssetFileTree()
     for node in nodes:
@@ -119,21 +138,19 @@ class TestSkillCompilerBasic:
 
 
 class TestSkillCompilerTransitiveDependencies:
-    def test_compile_skill_with_skill_dependency(self):
+    def test_references_are_transitive(self):
         # given
-        # skill-a references skill-b
-        # skill-b has a tool dependency
-        tool_ref = ToolReference(
-            uuid="tool-1",
-            type=ToolProviderType.BUILT_IN,
-            provider="sandbox",
-            tool_name="python",
-            credential_id=None,
-            configuration=None,
+        tool_b = ToolReference(
+            uuid="tool-b",
+            type=ToolProviderType.API,
+            provider="external",
+            tool_name="api_tool",
+            credential_id="cred-123",
+            configuration=ToolConfiguration(fields=[ToolFieldConfig(id="key", value="secret")]),
         )
         doc_a = SkillDocument(
             skill_id="skill-a",
-            content="See: §[file].[app].[skill-b]§",
+            content="A refs B: §[file].[app].[skill-b]§",
             metadata=make_metadata(
                 tools={},
                 files=[FileReference(source="app", asset_id="skill-b")],
@@ -141,15 +158,12 @@ class TestSkillCompilerTransitiveDependencies:
         )
         doc_b = SkillDocument(
             skill_id="skill-b",
-            content="Run: §[tool].[sandbox].[python].[tool-1]§",
-            metadata=make_metadata(
-                tools={"tool-1": tool_ref},
-                files=[],
-            ),
+            content="B: §[tool].[external].[api_tool].[tool-b]§",
+            metadata=make_metadata(tools={"tool-b": tool_b}, files=[]),
         )
         tree = create_file_tree(
-            AppAssetNode.create_file("skill-a", "skill-a.md"),
-            AppAssetNode.create_file("skill-b", "skill-b.md"),
+            AppAssetNode.create_file("skill-a", "a.md"),
+            AppAssetNode.create_file("skill-b", "b.md"),
         )
         compiler = SkillCompiler()
 
@@ -159,32 +173,271 @@ class TestSkillCompilerTransitiveDependencies:
         # then
         artifact_a = artifact_set.get("skill-a")
         assert artifact_a is not None
-        # skill-a should have transitive tool dependency from skill-b
-        assert len(artifact_a.tools.dependencies) == 1
-        assert artifact_a.tools.dependencies[0].tool_name == "python"
+        assert len(artifact_a.tools.references) == 1
+        ref = artifact_a.tools.references[0]
+        assert ref.uuid == "tool-b"
+        assert ref.credential_id == "cred-123"
+        assert ref.configuration is not None
+        assert ref.configuration.fields == [ToolFieldConfig(id="key", value="secret")]
 
-        # dependency graph should show skill-a depends on skill-b
-        assert "skill-b" in artifact_set.dependency_graph.get("skill-a", [])
-        # reverse graph should show skill-b is depended by skill-a
-        assert "skill-a" in artifact_set.reverse_graph.get("skill-b", [])
+        artifact_b = artifact_set.get("skill-b")
+        assert artifact_b is not None
+        assert len(artifact_b.tools.references) == 1
+        assert artifact_b.tools.references[0].uuid == "tool-b"
 
-    def test_compile_chain_dependency(self):
+
+class TestSkillCompilerCompileOne:
+    def test_compile_one_resolves_context(self):
         # given
-        # skill-a -> skill-b -> skill-c
-        # each has its own tool
-        tool_a = ToolReference(
-            uuid="tool-a", type=ToolProviderType.BUILT_IN, provider="p", tool_name="tool_a"
+        tool_ref = ToolReference(
+            uuid="tool-1",
+            type=ToolProviderType.BUILT_IN,
+            provider="sandbox",
+            tool_name="python",
         )
-        tool_b = ToolReference(
-            uuid="tool-b", type=ToolProviderType.BUILT_IN, provider="p", tool_name="tool_b"
-        )
-        tool_c = ToolReference(
-            uuid="tool-c", type=ToolProviderType.BUILT_IN, provider="p", tool_name="tool_c"
+        doc_skill = SkillDocument(
+            skill_id="skill-lib",
+            content="Library Code §[tool].[sandbox].[python].[tool-1]§",
+            metadata=make_metadata(tools={"tool-1": tool_ref}, files=[]),
         )
 
+        tree = create_file_tree(
+            AppAssetNode.create_file("skill-lib", "lib.md"),
+        )
+        compiler = SkillCompiler()
+
+        context = compiler.compile_all([doc_skill], tree, "assets-1")
+
+        # when
+        template_doc = SkillDocument(
+            skill_id="anonymous",
+            content="Use the lib: §[file].[app].[skill-lib]§",
+            metadata=make_metadata(tools={}, files=[FileReference(source="app", asset_id="skill-lib")]),
+        )
+
+        result = compiler.compile_one(context, template_doc, tree)
+
+        # then
+        assert "lib.md" in result.content
+        assert len(result.tools.dependencies) == 1
+        assert result.tools.dependencies[0].tool_name == "python"
+
+
+class TestSkillCompilerComplexGraph:
+    def test_large_complex_dependency_graph(self):
+        """
+        Generate 100 skills with complex inter-dependencies:
+        - Random file references between skills
+        - Random tool assignments
+        - Multiple circular dependencies
+        - Chain dependencies
+        - Star dependencies (hub nodes)
+        """
+        import random
+
+        random.seed(42)  # Reproducible
+
+        num_skills = 100
+        num_tools = 20
+
+        # Create tools
+        tools: dict[str, ToolReference] = {}
+        for i in range(num_tools):
+            tools[f"tool-{i}"] = ToolReference(
+                uuid=f"tool-{i}",
+                type=random.choice([ToolProviderType.BUILT_IN, ToolProviderType.API]),  # noqa: S311
+                provider=f"provider-{i % 5}",
+                tool_name=f"tool_func_{i}",
+                credential_id=f"cred-{i}" if i % 3 == 0 else None,
+            )
+
+        # Create skills with various dependency patterns
+        documents: list[SkillDocument] = []
+        nodes: list[AppAssetNode] = []
+
+        for i in range(num_skills):
+            skill_id = f"skill-{i}"
+            nodes.append(AppAssetNode.create_file(skill_id, f"{skill_id}.md"))
+
+            # Determine file references (other skills this skill references)
+            file_refs: list[FileReference] = []
+            ref_placeholders: list[str] = []
+
+            # Pattern 1: Chain dependency (skill-i references skill-i+1)
+            if i < num_skills - 1 and i % 10 != 9:
+                target = f"skill-{i + 1}"
+                file_refs.append(FileReference(source="app", asset_id=target))
+                ref_placeholders.append(f"§[file].[app].[{target}]§")
+
+            # Pattern 2: Circular dependency (every 10th skill references back)
+            if i % 10 == 9 and i >= 10:
+                target = f"skill-{i - 9}"
+                file_refs.append(FileReference(source="app", asset_id=target))
+                ref_placeholders.append(f"§[file].[app].[{target}]§")
+
+            # Pattern 3: Hub pattern (skill-0, skill-50 are hubs referenced by many)
+            if i > 0 and i % 7 == 0:
+                target = "skill-0"
+                file_refs.append(FileReference(source="app", asset_id=target))
+                ref_placeholders.append(f"§[file].[app].[{target}]§")
+
+            if i > 50 and i % 11 == 0:
+                target = "skill-50"
+                file_refs.append(FileReference(source="app", asset_id=target))
+                ref_placeholders.append(f"§[file].[app].[{target}]§")
+
+            # Pattern 4: Random references (sparse)
+            if random.random() < 0.2:  # noqa: S311
+                target_idx = random.randint(0, num_skills - 1)  # noqa: S311
+                if target_idx != i:
+                    target = f"skill-{target_idx}"
+                    if not any(f.asset_id == target for f in file_refs):
+                        file_refs.append(FileReference(source="app", asset_id=target))
+                        ref_placeholders.append(f"§[file].[app].[{target}]§")
+
+            # Assign tools to skills
+            skill_tools: dict[str, ToolReference] = {}
+            tool_placeholders: list[str] = []
+
+            # Each skill has 0-3 tools
+            num_skill_tools = random.randint(0, 3)  # noqa: S311
+            assigned_tools = random.sample(list(tools.keys()), min(num_skill_tools, len(tools)))
+
+            for tool_id in assigned_tools:
+                tool = tools[tool_id]
+                skill_tools[tool_id] = tool
+                tool_placeholders.append(f"§[tool].[{tool.provider}].[{tool.tool_name}].[{tool_id}]§")
+
+            # Build content
+            content_parts = [f"# Skill {i}\n\nThis is skill number {i}.\n"]
+            if ref_placeholders:
+                content_parts.append(f"References: {', '.join(ref_placeholders)}\n")
+            if tool_placeholders:
+                content_parts.append(f"Tools: {', '.join(tool_placeholders)}\n")
+
+            content = "\n".join(content_parts)
+
+            doc = SkillDocument(
+                skill_id=skill_id,
+                content=content,
+                metadata=make_metadata(tools=skill_tools, files=file_refs),
+            )
+            documents.append(doc)
+
+        tree = create_file_tree(*nodes)
+        compiler = SkillCompiler()
+
+        # when
+        bundle = compiler.compile_all(documents, tree, "stress-test-assets")
+
+        # then
+        assert len(bundle.entries) == num_skills
+
+        # Verify all skills compiled successfully
+        for i in range(num_skills):
+            entry = bundle.get(f"skill-{i}")
+            assert entry is not None, f"skill-{i} should exist"
+            assert entry.content, f"skill-{i} should have content"
+            # Content should have resolved references (no § markers for valid refs)
+            assert "§[file].[app].[skill-" not in entry.content or "[File not found]" in entry.content
+
+        # Verify hub nodes have many dependents in reverse graph
+        assert len(bundle.reverse_graph.get("skill-0", [])) > 5, "skill-0 should be a hub"
+
+        # Verify transitive dependencies propagate through cycles
+        # skill-9 -> skill-0 (via chain 9->8->...->1->0) and skill-9 refs skill-0 directly via cycle
+        entry_9 = bundle.get("skill-9")
+        assert entry_9 is not None
+
+        # Verify tool propagation works
+        # Find a skill with tools and check its dependents have those tools
+        for i in range(num_skills):
+            entry = bundle.get(f"skill-{i}")
+            if entry and entry.tools.dependencies:
+                # This skill has tools, check if skills that reference it also have these tools
+                dependents = bundle.reverse_graph.get(f"skill-{i}", [])
+                for dep_id in dependents[:3]:  # Check first 3 dependents
+                    dep_entry = bundle.get(dep_id)
+                    if dep_entry:
+                        # Dependent should have at least as many tool dependencies
+                        # (might have more from other paths)
+                        original_tool_names = {d.tool_name for d in entry.tools.dependencies}
+                        dep_tool_names = {d.tool_name for d in dep_entry.tools.dependencies}
+                        assert original_tool_names.issubset(dep_tool_names), (
+                            f"{dep_id} should have tools from {entry.skill_id}"
+                        )
+                break  # Only need to verify one case
+
+        print(f"\n✓ Successfully compiled {num_skills} skills")
+        print(f"  - {num_tools} tool types")
+        print(f"  - {sum(len(v) for v in bundle.dependency_graph.values())} total edges")
+        print(f"  - Hub skill-0 has {len(bundle.reverse_graph.get('skill-0', []))} dependents")
+
+
+class TestSkillCompilerCircularDependencies:
+    def test_simple_circular_dependency(self):
+        """A ↔ B: Two skills reference each other."""
+        # given
         doc_a = SkillDocument(
             skill_id="skill-a",
-            content="A refs B: §[file].[app].[skill-b]§ §[tool].[p].[tool_a].[tool-a]§",
+            content="A references B: §[file].[app].[skill-b]§",
+            metadata=make_metadata(
+                tools={},
+                files=[FileReference(source="app", asset_id="skill-b")],
+            ),
+        )
+        doc_b = SkillDocument(
+            skill_id="skill-b",
+            content="B references A: §[file].[app].[skill-a]§",
+            metadata=make_metadata(
+                tools={},
+                files=[FileReference(source="app", asset_id="skill-a")],
+            ),
+        )
+        tree = create_file_tree(
+            AppAssetNode.create_file("skill-a", "a.md"),
+            AppAssetNode.create_file("skill-b", "b.md"),
+        )
+        compiler = SkillCompiler()
+
+        # when
+        bundle = compiler.compile_all([doc_a, doc_b], tree, "assets-1")
+
+        # then
+        assert len(bundle.entries) == 2
+
+        entry_a = bundle.get("skill-a")
+        assert entry_a is not None
+        assert entry_a.content == "A references B: ./b.md"
+        # Transitive closure: A refs B, B refs A, so A has both
+        file_ids_a = {f.asset_id for f in entry_a.files.references}
+        assert file_ids_a == {"skill-a", "skill-b"}
+
+        entry_b = bundle.get("skill-b")
+        assert entry_b is not None
+        assert entry_b.content == "B references A: ./a.md"
+        # Transitive closure: B refs A, A refs B, so B has both
+        file_ids_b = {f.asset_id for f in entry_b.files.references}
+        assert file_ids_b == {"skill-a", "skill-b"}
+
+    def test_circular_dependency_with_tools(self):
+        """A ↔ B with tools: Both skills should have access to both tools."""
+        # given
+        tool_a = ToolReference(
+            uuid="tool-a",
+            type=ToolProviderType.BUILT_IN,
+            provider="sandbox",
+            tool_name="bash",
+        )
+        tool_b = ToolReference(
+            uuid="tool-b",
+            type=ToolProviderType.BUILT_IN,
+            provider="sandbox",
+            tool_name="python",
+        )
+        doc_a = SkillDocument(
+            skill_id="skill-a",
+            content="A: §[tool].[sandbox].[bash].[tool-a]§ refs §[file].[app].[skill-b]§",
             metadata=make_metadata(
                 tools={"tool-a": tool_a},
                 files=[FileReference(source="app", asset_id="skill-b")],
@@ -192,986 +445,88 @@ class TestSkillCompilerTransitiveDependencies:
         )
         doc_b = SkillDocument(
             skill_id="skill-b",
-            content="B refs C: §[file].[app].[skill-c]§ §[tool].[p].[tool_b].[tool-b]§",
+            content="B: §[tool].[sandbox].[python].[tool-b]§ refs §[file].[app].[skill-a]§",
             metadata=make_metadata(
                 tools={"tool-b": tool_b},
-                files=[FileReference(source="app", asset_id="skill-c")],
+                files=[FileReference(source="app", asset_id="skill-a")],
             ),
         )
-        doc_c = SkillDocument(
-            skill_id="skill-c",
-            content="C is leaf §[tool].[p].[tool_c].[tool-c]§",
-            metadata=make_metadata(
-                tools={"tool-c": tool_c},
-                files=[],
-            ),
-        )
-
         tree = create_file_tree(
             AppAssetNode.create_file("skill-a", "a.md"),
             AppAssetNode.create_file("skill-b", "b.md"),
-            AppAssetNode.create_file("skill-c", "c.md"),
         )
         compiler = SkillCompiler()
 
         # when
-        artifact_set = compiler.compile_all([doc_a, doc_b, doc_c], tree, "assets-1")
+        bundle = compiler.compile_all([doc_a, doc_b], tree, "assets-1")
 
         # then
-        artifact_a = artifact_set.get("skill-a")
-        assert artifact_a is not None
-        # skill-a should have all 3 tools (own + transitive)
-        tool_names = {d.tool_name for d in artifact_a.tools.dependencies}
-        assert tool_names == {"tool_a", "tool_b", "tool_c"}
+        entry_a = bundle.get("skill-a")
+        assert entry_a is not None
+        # A should have both tools (its own + B's via transitive dependency)
+        assert len(entry_a.tools.dependencies) == 2
+        tool_names_a = {dep.tool_name for dep in entry_a.tools.dependencies}
+        assert tool_names_a == {"bash", "python"}
 
-        artifact_b = artifact_set.get("skill-b")
-        assert artifact_b is not None
-        tool_names_b = {d.tool_name for d in artifact_b.tools.dependencies}
-        assert tool_names_b == {"tool_b", "tool_c"}
+        entry_b = bundle.get("skill-b")
+        assert entry_b is not None
+        # B should have both tools (its own + A's via transitive dependency)
+        assert len(entry_b.tools.dependencies) == 2
+        tool_names_b = {dep.tool_name for dep in entry_b.tools.dependencies}
+        assert tool_names_b == {"bash", "python"}
 
-        artifact_c = artifact_set.get("skill-c")
-        assert artifact_c is not None
-        tool_names_c = {d.tool_name for d in artifact_c.tools.dependencies}
-        assert tool_names_c == {"tool_c"}
-
-
-class TestSkillBundleQueries:
-    def test_recompile_group_ids(self):
+    def test_three_way_circular_dependency(self):
+        """A → B → C → A: Three skills form a cycle."""
         # given
-        # skill-a -> skill-b -> skill-c
-        doc_a = SkillDocument(
-            skill_id="skill-a",
-            content="refs B: §[file].[app].[skill-b]§",
-            metadata=make_metadata(
-                tools={},
-                files=[FileReference(source="app", asset_id="skill-b")],
-            ),
-        )
-        doc_b = SkillDocument(
-            skill_id="skill-b",
-            content="refs C: §[file].[app].[skill-c]§",
-            metadata=make_metadata(
-                tools={},
-                files=[FileReference(source="app", asset_id="skill-c")],
-            ),
-        )
-        doc_c = SkillDocument(
-            skill_id="skill-c",
-            content="leaf",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-
-        tree = create_file_tree(
-            AppAssetNode.create_file("skill-a", "a.md"),
-            AppAssetNode.create_file("skill-b", "b.md"),
-            AppAssetNode.create_file("skill-c", "c.md"),
-        )
-        compiler = SkillCompiler()
-        artifact_set = compiler.compile_all([doc_a, doc_b, doc_c], tree, "assets-1")
-
-        # when - if skill-c changes, who needs recompile?
-        affected = artifact_set.recompile_group_ids("skill-c")
-
-        # then - all upstream skills need recompile
-        assert affected == {"skill-a", "skill-b", "skill-c"}
-
-    def test_referenced_skill_ids(self):
-        # given
-        doc_a = SkillDocument(
-            skill_id="skill-a",
-            content="refs B and C: §[file].[app].[skill-b]§ §[file].[app].[skill-c]§",
-            metadata=make_metadata(
-                tools={},
-                files=[
-                    FileReference(source="app", asset_id="skill-b"),
-                    FileReference(source="app", asset_id="skill-c"),
-                ],
-            ),
-        )
-        doc_b = SkillDocument(
-            skill_id="skill-b",
-            content="B",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-        doc_c = SkillDocument(
-            skill_id="skill-c",
-            content="C",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-
-        tree = create_file_tree(
-            AppAssetNode.create_file("skill-a", "a.md"),
-            AppAssetNode.create_file("skill-b", "b.md"),
-            AppAssetNode.create_file("skill-c", "c.md"),
-        )
-        compiler = SkillCompiler()
-        artifact_set = compiler.compile_all([doc_a, doc_b, doc_c], tree, "assets-1")
-
-        # when
-        refs = artifact_set.referenced_skill_ids("skill-a")
-
-        # then
-        assert refs == {"skill-b", "skill-c"}
-
-
-class TestSkillCompilerIncrementalCompile:
-    def test_compile_one_updates_artifact_set(self):
-        # given - initial compile
-        doc_a = SkillDocument(
-            skill_id="skill-a",
-            content="original content",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-        tree = create_file_tree(
-            AppAssetNode.create_file("skill-a", "a.md"),
-        )
-        compiler = SkillCompiler()
-        artifact_set = compiler.compile_all([doc_a], tree, "assets-1")
-
-        # when - update skill-a
-        updated_doc = SkillDocument(
-            skill_id="skill-a",
-            content="updated content",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-        updated_artifact = compiler.compile_one(artifact_set, updated_doc, tree)
-        artifact_set.upsert(updated_artifact)
-
-        # then
-        artifact = artifact_set.get("skill-a")
-        assert artifact is not None
-        assert artifact.content == "updated content"
-
-
-class TestSkillCompilerEdgeCases:
-    def test_missing_file_reference_replaced_with_placeholder(self):
-        # given
-        doc = SkillDocument(
-            skill_id="skill-1",
-            content="See: §[file].[app].[non-existent]§",
-            metadata=make_metadata(
-                tools={},
-                files=[FileReference(source="app", asset_id="non-existent")],
-            ),
-        )
-        tree = create_file_tree(
-            AppAssetNode.create_file("skill-1", "skill.md"),
-        )
-        compiler = SkillCompiler()
-
-        # when
-        artifact_set = compiler.compile_all([doc], tree, "assets-1")
-
-        # then
-        artifact = artifact_set.get("skill-1")
-        assert artifact is not None
-        assert "[File not found]" in artifact.content
-
-    def test_missing_tool_reference_replaced_with_placeholder(self):
-        # given
-        doc = SkillDocument(
-            skill_id="skill-1",
-            content="Run: §[tool].[sandbox].[bash].[missing-tool]§",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-        tree = create_file_tree(
-            AppAssetNode.create_file("skill-1", "skill.md"),
-        )
-        compiler = SkillCompiler()
-
-        # when
-        artifact_set = compiler.compile_all([doc], tree, "assets-1")
-
-        # then
-        artifact = artifact_set.get("skill-1")
-        assert artifact is not None
-        assert "[Tool not found: missing-tool]" in artifact.content
-
-    def test_content_digest_changes_when_content_changes(self):
-        # given
-        tree = create_file_tree(
-            AppAssetNode.create_file("skill-1", "skill.md"),
-        )
-        compiler = SkillCompiler()
-
-        doc1 = SkillDocument(
-            skill_id="skill-1",
-            content="content version 1",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-        artifact_set1 = compiler.compile_all([doc1], tree, "assets-1")
-        artifact1 = artifact_set1.get("skill-1")
-        assert artifact1 is not None
-        digest1 = artifact1.source.content_digest
-
-        doc2 = SkillDocument(
-            skill_id="skill-1",
-            content="content version 2",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-        artifact_set2 = compiler.compile_all([doc2], tree, "assets-1")
-        artifact2 = artifact_set2.get("skill-1")
-        assert artifact2 is not None
-        digest2 = artifact2.source.content_digest
-
-        # then
-        assert digest1 != digest2
-
-
-class TestSkillCompilerComplexScenarios:
-    def test_diamond_dependency(self):
-        # given
-        #     skill-a
-        #    /       \
-        # skill-b   skill-c
-        #    \       /
-        #     skill-d (has tool)
-        tool_d = ToolReference(
-            uuid="tool-d", type=ToolProviderType.BUILT_IN, provider="p", tool_name="tool_d"
-        )
-
-        doc_a = SkillDocument(
-            skill_id="skill-a",
-            content="A refs B and C: §[file].[app].[skill-b]§ §[file].[app].[skill-c]§",
-            metadata=make_metadata(
-                tools={},
-                files=[
-                    FileReference(source="app", asset_id="skill-b"),
-                    FileReference(source="app", asset_id="skill-c"),
-                ],
-            ),
-        )
-        doc_b = SkillDocument(
-            skill_id="skill-b",
-            content="B refs D: §[file].[app].[skill-d]§",
-            metadata=make_metadata(
-                tools={},
-                files=[FileReference(source="app", asset_id="skill-d")],
-            ),
-        )
-        doc_c = SkillDocument(
-            skill_id="skill-c",
-            content="C refs D: §[file].[app].[skill-d]§",
-            metadata=make_metadata(
-                tools={},
-                files=[FileReference(source="app", asset_id="skill-d")],
-            ),
-        )
-        doc_d = SkillDocument(
-            skill_id="skill-d",
-            content="D is leaf with tool: §[tool].[p].[tool_d].[tool-d]§",
-            metadata=make_metadata(
-                tools={"tool-d": tool_d},
-                files=[],
-            ),
-        )
-
-        tree = create_file_tree(
-            AppAssetNode.create_file("skill-a", "a.md"),
-            AppAssetNode.create_file("skill-b", "b.md"),
-            AppAssetNode.create_file("skill-c", "c.md"),
-            AppAssetNode.create_file("skill-d", "d.md"),
-        )
-        compiler = SkillCompiler()
-
-        # when
-        artifact_set = compiler.compile_all([doc_a, doc_b, doc_c, doc_d], tree, "assets-1")
-
-        # then
-        # skill-a should have tool_d (via both B and C paths, but only once)
-        artifact_a = artifact_set.get("skill-a")
-        assert artifact_a is not None
-        assert len(artifact_a.tools.dependencies) == 1
-        assert artifact_a.tools.dependencies[0].tool_name == "tool_d"
-
-        # if skill-d changes, all upstream need recompile
-        affected = artifact_set.recompile_group_ids("skill-d")
-        assert affected == {"skill-a", "skill-b", "skill-c", "skill-d"}
-
-    def test_multiple_tools_from_multiple_paths(self):
-        # given
-        #     skill-a
-        #    /       \
-        # skill-b   skill-c
-        # (tool_b)  (tool_c)
-        #    \       /
-        #     skill-d
-        #     (tool_d)
-        tool_b = ToolReference(
-            uuid="tool-b", type=ToolProviderType.BUILT_IN, provider="p", tool_name="tool_b"
-        )
         tool_c = ToolReference(
-            uuid="tool-c", type=ToolProviderType.API, provider="q", tool_name="tool_c"
+            uuid="tool-c",
+            type=ToolProviderType.API,
+            provider="external",
+            tool_name="api_tool",
+            credential_id="cred-123",
         )
-        tool_d = ToolReference(
-            uuid="tool-d", type=ToolProviderType.WORKFLOW, provider="r", tool_name="tool_d"
-        )
-
         doc_a = SkillDocument(
             skill_id="skill-a",
-            content="A: §[file].[app].[skill-b]§ §[file].[app].[skill-c]§",
+            content="A refs B: §[file].[app].[skill-b]§",
             metadata=make_metadata(
                 tools={},
-                files=[
-                    FileReference(source="app", asset_id="skill-b"),
-                    FileReference(source="app", asset_id="skill-c"),
-                ],
+                files=[FileReference(source="app", asset_id="skill-b")],
             ),
         )
         doc_b = SkillDocument(
             skill_id="skill-b",
-            content="B: §[file].[app].[skill-d]§ §[tool].[p].[tool_b].[tool-b]§",
+            content="B refs C: §[file].[app].[skill-c]§",
             metadata=make_metadata(
-                tools={"tool-b": tool_b},
-                files=[FileReference(source="app", asset_id="skill-d")],
+                tools={},
+                files=[FileReference(source="app", asset_id="skill-c")],
             ),
         )
         doc_c = SkillDocument(
             skill_id="skill-c",
-            content="C: §[file].[app].[skill-d]§ §[tool].[q].[tool_c].[tool-c]§",
+            content="C refs A: §[file].[app].[skill-a]§ and uses §[tool].[external].[api_tool].[tool-c]§",
             metadata=make_metadata(
                 tools={"tool-c": tool_c},
-                files=[FileReference(source="app", asset_id="skill-d")],
+                files=[FileReference(source="app", asset_id="skill-a")],
             ),
         )
-        doc_d = SkillDocument(
-            skill_id="skill-d",
-            content="D: §[tool].[r].[tool_d].[tool-d]§",
-            metadata=make_metadata(
-                tools={"tool-d": tool_d},
-                files=[],
-            ),
-        )
-
         tree = create_file_tree(
             AppAssetNode.create_file("skill-a", "a.md"),
             AppAssetNode.create_file("skill-b", "b.md"),
             AppAssetNode.create_file("skill-c", "c.md"),
-            AppAssetNode.create_file("skill-d", "d.md"),
         )
         compiler = SkillCompiler()
 
         # when
-        artifact_set = compiler.compile_all([doc_a, doc_b, doc_c, doc_d], tree, "assets-1")
+        bundle = compiler.compile_all([doc_a, doc_b, doc_c], tree, "assets-1")
 
         # then
-        artifact_a = artifact_set.get("skill-a")
-        assert artifact_a is not None
-        tool_names = {d.tool_name for d in artifact_a.tools.dependencies}
-        assert tool_names == {"tool_b", "tool_c", "tool_d"}
-
-        # verify different tool types are preserved
-        tool_types = {d.type for d in artifact_a.tools.dependencies}
-        assert tool_types == {ToolProviderType.BUILT_IN, ToolProviderType.API, ToolProviderType.WORKFLOW}
-
-    def test_deep_nested_folder_structure_with_relative_paths(self):
-        # given
-        # /root/
-        #   main.md (refs helper and asset)
-        #   helpers/
-        #     helper.md (refs deep asset)
-        #     deep/
-        #       deep-helper.md
-        #   assets/
-        #     image.png
-        folder_root = AppAssetNode.create_folder("folder-root", "root")
-        folder_helpers = AppAssetNode.create_folder("folder-helpers", "helpers", parent_id="folder-root")
-        folder_deep = AppAssetNode.create_folder("folder-deep", "deep", parent_id="folder-helpers")
-        folder_assets = AppAssetNode.create_folder("folder-assets", "assets", parent_id="folder-root")
-
-        file_main = AppAssetNode.create_file("file-main", "main.md", parent_id="folder-root")
-        file_helper = AppAssetNode.create_file("file-helper", "helper.md", parent_id="folder-helpers")
-        file_deep = AppAssetNode.create_file("file-deep", "deep-helper.md", parent_id="folder-deep")
-        file_image = AppAssetNode.create_file("file-image", "image.png", parent_id="folder-assets")
-
-        tree = create_file_tree(
-            folder_root, folder_helpers, folder_deep, folder_assets,
-            file_main, file_helper, file_deep, file_image,
-        )
-
-        doc_main = SkillDocument(
-            skill_id="file-main",
-            content="Main refs helper: §[file].[app].[file-helper]§ and image: §[file].[app].[file-image]§",
-            metadata=make_metadata(
-                tools={},
-                files=[
-                    FileReference(source="app", asset_id="file-helper"),
-                    FileReference(source="app", asset_id="file-image"),
-                ],
-            ),
-        )
-        doc_helper = SkillDocument(
-            skill_id="file-helper",
-            content="Helper refs deep: §[file].[app].[file-deep]§",
-            metadata=make_metadata(
-                tools={},
-                files=[FileReference(source="app", asset_id="file-deep")],
-            ),
-        )
-        doc_deep = SkillDocument(
-            skill_id="file-deep",
-            content="Deep helper content",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-
-        compiler = SkillCompiler()
-
-        # when
-        artifact_set = compiler.compile_all([doc_main, doc_helper, doc_deep], tree, "assets-1")
-
-        # then
-        artifact_main = artifact_set.get("file-main")
-        assert artifact_main is not None
-        # main.md -> helpers/helper.md = ./helpers/helper.md
-        assert "./helpers/helper.md" in artifact_main.content
-        # main.md -> assets/image.png = ./assets/image.png
-        assert "./assets/image.png" in artifact_main.content
-
-        artifact_helper = artifact_set.get("file-helper")
-        assert artifact_helper is not None
-        # helpers/helper.md -> helpers/deep/deep-helper.md = ./deep/deep-helper.md
-        assert "./deep/deep-helper.md" in artifact_helper.content
-
-    def test_skill_with_many_tools_and_files(self):
-        # given - skill with 10 tools and 5 file references
-        tools = {
-            f"tool-{i}": ToolReference(
-                uuid=f"tool-{i}",
-                type=ToolProviderType.BUILT_IN,
-                provider=f"provider-{i}",
-                tool_name=f"tool_name_{i}",
-            )
-            for i in range(10)
-        }
-        files = [
-            FileReference(source="app", asset_id=f"file-{i}")
-            for i in range(5)
-        ]
-
-        tool_refs_in_content = " ".join(
-            f"§[tool].[provider-{i}].[tool_name_{i}].[tool-{i}]§" for i in range(10)
-        )
-        file_refs_in_content = " ".join(
-            f"§[file].[app].[file-{i}]§" for i in range(5)
-        )
-
-        doc = SkillDocument(
-            skill_id="skill-main",
-            content=f"Tools: {tool_refs_in_content}\nFiles: {file_refs_in_content}",
-            metadata=make_metadata(tools=tools, files=files),
-        )
-
-        nodes = [AppAssetNode.create_file("skill-main", "main.md")]
-        nodes.extend(AppAssetNode.create_file(f"file-{i}", f"file-{i}.txt") for i in range(5))
-        tree = create_file_tree(*nodes)
-
-        compiler = SkillCompiler()
-
-        # when
-        artifact_set = compiler.compile_all([doc], tree, "assets-1")
-
-        # then
-        artifact = artifact_set.get("skill-main")
-        assert artifact is not None
-        assert len(artifact.tools.dependencies) == 10
-        assert len(artifact.tools.references) == 10
-        assert len(artifact.files.references) == 5
-
-        # all tool references should be replaced
-        for i in range(10):
-            assert f"[Bash Command: tool_name_{i}_tool-{i}]" in artifact.content
-        # all file references should be replaced
-        for i in range(5):
-            assert f"./file-{i}.txt" in artifact.content
-
-    def test_incremental_compile_with_new_dependency(self):
-        # given - initial state: skill-a standalone
-        tree = create_file_tree(
-            AppAssetNode.create_file("skill-a", "a.md"),
-            AppAssetNode.create_file("skill-b", "b.md"),
-        )
-
-        doc_a_v1 = SkillDocument(
-            skill_id="skill-a",
-            content="A standalone",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-        doc_b = SkillDocument(
-            skill_id="skill-b",
-            content="B with tool: §[tool].[p].[tool_b].[tool-b]§",
-            metadata=make_metadata(
-                tools={
-                    "tool-b": ToolReference(
-                        uuid="tool-b",
-                        type=ToolProviderType.BUILT_IN,
-                        provider="p",
-                        tool_name="tool_b",
-                    )
-                },
-                files=[],
-            ),
-        )
-
-        compiler = SkillCompiler()
-        artifact_set = compiler.compile_all([doc_a_v1, doc_b], tree, "assets-1")
-
-        # skill-a has no dependencies initially
-        artifact_a_v1 = artifact_set.get("skill-a")
-        assert artifact_a_v1 is not None
-        assert len(artifact_a_v1.tools.dependencies) == 0
-
-        # when - update skill-a to reference skill-b
-        doc_a_v2 = SkillDocument(
-            skill_id="skill-a",
-            content="A now refs B: §[file].[app].[skill-b]§",
-            metadata=make_metadata(
-                tools={},
-                files=[FileReference(source="app", asset_id="skill-b")],
-            ),
-        )
-        doc_map = {"skill-a": doc_a_v2, "skill-b": doc_b}
-        artifact_a_v2 = compiler.compile_one(artifact_set, doc_a_v2, tree, doc_map)
-        artifact_set.upsert(artifact_a_v2)
-
-        # then - skill-a now has tool_b from skill-b
-        artifact_a_final = artifact_set.get("skill-a")
-        assert artifact_a_final is not None
-        assert len(artifact_a_final.tools.dependencies) == 1
-        assert artifact_a_final.tools.dependencies[0].tool_name == "tool_b"
-
-        # dependency graph updated
-        assert "skill-b" in artifact_set.dependency_graph.get("skill-a", [])
-        assert "skill-a" in artifact_set.reverse_graph.get("skill-b", [])
-
-    def test_serialization_roundtrip(self):
-        # given - complex artifact set
-        tool = ToolReference(
-            uuid="tool-1",
-            type=ToolProviderType.BUILT_IN,
-            provider="sandbox",
-            tool_name="bash",
-        )
-        doc_a = SkillDocument(
-            skill_id="skill-a",
-            content="A refs B: §[file].[app].[skill-b]§ §[tool].[sandbox].[bash].[tool-1]§",
-            metadata=make_metadata(
-                tools={"tool-1": tool},
-                files=[FileReference(source="app", asset_id="skill-b")],
-            ),
-        )
-        doc_b = SkillDocument(
-            skill_id="skill-b",
-            content="B leaf",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-
-        tree = create_file_tree(
-            AppAssetNode.create_file("skill-a", "a.md"),
-            AppAssetNode.create_file("skill-b", "b.md"),
-        )
-        compiler = SkillCompiler()
-        original = compiler.compile_all([doc_a, doc_b], tree, "assets-1")
-
-        # when - serialize and deserialize
-        json_str = original.model_dump_json()
-        restored = SkillBundle.model_validate_json(json_str)
-
-        # then - all data preserved
-        assert restored.assets_id == original.assets_id
-        assert len(restored.entries) == len(original.entries)
-        assert restored.dependency_graph == original.dependency_graph
-        assert restored.reverse_graph == original.reverse_graph
-
-        original_a = original.get("skill-a")
-        assert original_a is not None
-        artifact_a = restored.get("skill-a")
-        assert artifact_a is not None
-        assert artifact_a.content == original_a.content
-        assert len(artifact_a.tools.dependencies) == 1
-
-    def test_subset_preserves_internal_dependencies(self):
-        # given
-        # skill-a -> skill-b -> skill-c -> skill-d
-        docs = [
-            SkillDocument(
-                skill_id="skill-a",
-                content="A: §[file].[app].[skill-b]§",
-                metadata=make_metadata(
-                    tools={},
-                    files=[FileReference(source="app", asset_id="skill-b")],
-                ),
-            ),
-            SkillDocument(
-                skill_id="skill-b",
-                content="B: §[file].[app].[skill-c]§",
-                metadata=make_metadata(
-                    tools={},
-                    files=[FileReference(source="app", asset_id="skill-c")],
-                ),
-            ),
-            SkillDocument(
-                skill_id="skill-c",
-                content="C: §[file].[app].[skill-d]§",
-                metadata=make_metadata(
-                    tools={},
-                    files=[FileReference(source="app", asset_id="skill-d")],
-                ),
-            ),
-            SkillDocument(
-                skill_id="skill-d",
-                content="D",
-                metadata=make_metadata(tools={}, files=[]),
-            ),
-        ]
-        tree = create_file_tree(
-            AppAssetNode.create_file("skill-a", "a.md"),
-            AppAssetNode.create_file("skill-b", "b.md"),
-            AppAssetNode.create_file("skill-c", "c.md"),
-            AppAssetNode.create_file("skill-d", "d.md"),
-        )
-        compiler = SkillCompiler()
-        full_set = compiler.compile_all(docs, tree, "assets-1")
-
-        # when - get subset of B and C only
-        subset = full_set.subset(["skill-b", "skill-c"])
-
-        # then
-        assert len(subset.entries) == 2
-        assert subset.get("skill-b") is not None
-        assert subset.get("skill-c") is not None
-        assert subset.get("skill-a") is None
-        assert subset.get("skill-d") is None
-
-        # internal dependency preserved (B -> C)
-        assert "skill-c" in subset.dependency_graph.get("skill-b", [])
-        # external dependencies filtered out
-        assert "skill-d" not in subset.dependency_graph.get("skill-c", [])
-
-
-class TestSkillCompilerIncrementalRecompile:
-    def test_single_change_triggers_upstream_recompile(self):
-        # given
-        # skill-a -> skill-b -> skill-c (leaf)
-        # each has unique tool
-        tool_a = ToolReference(uuid="t-a", type=ToolProviderType.BUILT_IN, provider="p", tool_name="tool_a")
-        tool_b = ToolReference(uuid="t-b", type=ToolProviderType.BUILT_IN, provider="p", tool_name="tool_b")
-        tool_c = ToolReference(uuid="t-c", type=ToolProviderType.BUILT_IN, provider="p", tool_name="tool_c")
-
-        doc_a = SkillDocument(
-            skill_id="a",
-            content="A content v1 §[tool].[p].[tool_a].[t-a]§ §[file].[app].[b]§",
-            metadata=make_metadata(
-                tools={"t-a": tool_a},
-                files=[FileReference(source="app", asset_id="b")],
-            ),
-        )
-        doc_b = SkillDocument(
-            skill_id="b",
-            content="B content v1 §[tool].[p].[tool_b].[t-b]§ §[file].[app].[c]§",
-            metadata=make_metadata(
-                tools={"t-b": tool_b},
-                files=[FileReference(source="app", asset_id="c")],
-            ),
-        )
-        doc_c = SkillDocument(
-            skill_id="c",
-            content="C content v1 §[tool].[p].[tool_c].[t-c]§",
-            metadata=make_metadata(tools={"t-c": tool_c}, files=[]),
-        )
-
-        tree = create_file_tree(
-            AppAssetNode.create_file("a", "a.md"),
-            AppAssetNode.create_file("b", "b.md"),
-            AppAssetNode.create_file("c", "c.md"),
-        )
-        compiler = SkillCompiler()
-        artifact_set = compiler.compile_all([doc_a, doc_b, doc_c], tree, "assets-1")
-
-        original_a_digest = artifact_set.get("a").source.content_digest
-        original_b_digest = artifact_set.get("b").source.content_digest
-        original_c_digest = artifact_set.get("c").source.content_digest
-
-        # when - skill-c changes
-        doc_c_v2 = SkillDocument(
-            skill_id="c",
-            content="C content v2 - UPDATED §[tool].[p].[tool_c].[t-c]§",
-            metadata=make_metadata(tools={"t-c": tool_c}, files=[]),
-        )
-
-        # find affected skills using recompile_group_ids
-        affected_ids = artifact_set.recompile_group_ids("c")
-
-        # then - all upstream skills are affected
-        assert affected_ids == {"a", "b", "c"}
-
-        # simulate incremental recompile for affected skills
-        doc_map = {"a": doc_a, "b": doc_b, "c": doc_c_v2}
-        for skill_id in affected_ids:
-            updated = compiler.compile_one(artifact_set, doc_map[skill_id], tree, doc_map)
-            artifact_set.upsert(updated)
-
-        # verify c's content changed
-        assert artifact_set.get("c").source.content_digest != original_c_digest
-        assert "v2 - UPDATED" in artifact_set.get("c").content
-
-        # a and b content didn't change (only their dependencies were refreshed)
-        assert artifact_set.get("a").source.content_digest == original_a_digest
-        assert artifact_set.get("b").source.content_digest == original_b_digest
-
-    def test_branch_change_only_affects_upstream_branch(self):
-        # given
-        #       skill-root
-        #      /          \
-        # skill-left    skill-right
-        #     |              |
-        # skill-l-leaf   skill-r-leaf
-        tool = ToolReference(uuid="t", type=ToolProviderType.BUILT_IN, provider="p", tool_name="tool")
-
-        doc_root = SkillDocument(
-            skill_id="root",
-            content="root §[file].[app].[left]§ §[file].[app].[right]§",
-            metadata=make_metadata(
-                tools={},
-                files=[
-                    FileReference(source="app", asset_id="left"),
-                    FileReference(source="app", asset_id="right"),
-                ],
-            ),
-        )
-        doc_left = SkillDocument(
-            skill_id="left",
-            content="left §[file].[app].[l-leaf]§",
-            metadata=make_metadata(
-                tools={},
-                files=[FileReference(source="app", asset_id="l-leaf")],
-            ),
-        )
-        doc_right = SkillDocument(
-            skill_id="right",
-            content="right §[file].[app].[r-leaf]§",
-            metadata=make_metadata(
-                tools={},
-                files=[FileReference(source="app", asset_id="r-leaf")],
-            ),
-        )
-        doc_l_leaf = SkillDocument(
-            skill_id="l-leaf",
-            content="left leaf §[tool].[p].[tool].[t]§",
-            metadata=make_metadata(tools={"t": tool}, files=[]),
-        )
-        doc_r_leaf = SkillDocument(
-            skill_id="r-leaf",
-            content="right leaf",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-
-        tree = create_file_tree(
-            AppAssetNode.create_file("root", "root.md"),
-            AppAssetNode.create_file("left", "left.md"),
-            AppAssetNode.create_file("right", "right.md"),
-            AppAssetNode.create_file("l-leaf", "l-leaf.md"),
-            AppAssetNode.create_file("r-leaf", "r-leaf.md"),
-        )
-        compiler = SkillCompiler()
-        artifact_set = compiler.compile_all(
-            [doc_root, doc_left, doc_right, doc_l_leaf, doc_r_leaf], tree, "assets-1"
-        )
-
-        # when - l-leaf changes
-        affected_by_l_leaf = artifact_set.recompile_group_ids("l-leaf")
-
-        # then - only left branch + root affected (not right branch)
-        assert affected_by_l_leaf == {"root", "left", "l-leaf"}
-        assert "right" not in affected_by_l_leaf
-        assert "r-leaf" not in affected_by_l_leaf
-
-        # when - r-leaf changes
-        affected_by_r_leaf = artifact_set.recompile_group_ids("r-leaf")
-
-        # then - only right branch + root affected (not left branch)
-        assert affected_by_r_leaf == {"root", "right", "r-leaf"}
-        assert "left" not in affected_by_r_leaf
-        assert "l-leaf" not in affected_by_r_leaf
-
-    def test_add_new_tool_to_leaf_propagates_to_all_upstream(self):
-        # given - chain without tools initially
-        doc_a = SkillDocument(
-            skill_id="a",
-            content="A §[file].[app].[b]§",
-            metadata=make_metadata(tools={}, files=[FileReference(source="app", asset_id="b")]),
-        )
-        doc_b = SkillDocument(
-            skill_id="b",
-            content="B §[file].[app].[c]§",
-            metadata=make_metadata(tools={}, files=[FileReference(source="app", asset_id="c")]),
-        )
-        doc_c = SkillDocument(
-            skill_id="c",
-            content="C - no tools",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-
-        tree = create_file_tree(
-            AppAssetNode.create_file("a", "a.md"),
-            AppAssetNode.create_file("b", "b.md"),
-            AppAssetNode.create_file("c", "c.md"),
-        )
-        compiler = SkillCompiler()
-        artifact_set = compiler.compile_all([doc_a, doc_b, doc_c], tree, "assets-1")
-
-        # initially no tools anywhere
-        assert len(artifact_set.get("a").tools.dependencies) == 0
-        assert len(artifact_set.get("b").tools.dependencies) == 0
-        assert len(artifact_set.get("c").tools.dependencies) == 0
-
-        # when - add tool to c
-        new_tool = ToolReference(uuid="new-t", type=ToolProviderType.BUILT_IN, provider="p", tool_name="new_tool")
-        doc_c_v2 = SkillDocument(
-            skill_id="c",
-            content="C - now has tool: §[tool].[p].[new_tool].[new-t]§",
-            metadata=make_metadata(tools={"new-t": new_tool}, files=[]),
-        )
-
-        # recompile affected
-        affected = artifact_set.recompile_group_ids("c")
-        doc_map = {"a": doc_a, "b": doc_b, "c": doc_c_v2}
-        for skill_id in affected:
-            updated = compiler.compile_one(artifact_set, doc_map[skill_id], tree, doc_map)
-            artifact_set.upsert(updated)
-
-        # then - new tool propagated to all upstream
-        assert len(artifact_set.get("c").tools.dependencies) == 1
-        assert len(artifact_set.get("b").tools.dependencies) == 1
-        assert len(artifact_set.get("a").tools.dependencies) == 1
-
-        assert artifact_set.get("a").tools.dependencies[0].tool_name == "new_tool"
-        assert artifact_set.get("b").tools.dependencies[0].tool_name == "new_tool"
-        assert artifact_set.get("c").tools.dependencies[0].tool_name == "new_tool"
-
-    def test_remove_dependency_link_affects_recompile_group(self):
-        # given - a -> b -> c
-        doc_a = SkillDocument(
-            skill_id="a",
-            content="A refs B §[file].[app].[b]§",
-            metadata=make_metadata(tools={}, files=[FileReference(source="app", asset_id="b")]),
-        )
-        doc_b = SkillDocument(
-            skill_id="b",
-            content="B refs C §[file].[app].[c]§",
-            metadata=make_metadata(tools={}, files=[FileReference(source="app", asset_id="c")]),
-        )
-        doc_c = SkillDocument(
-            skill_id="c",
-            content="C leaf",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-
-        tree = create_file_tree(
-            AppAssetNode.create_file("a", "a.md"),
-            AppAssetNode.create_file("b", "b.md"),
-            AppAssetNode.create_file("c", "c.md"),
-        )
-        compiler = SkillCompiler()
-        artifact_set = compiler.compile_all([doc_a, doc_b, doc_c], tree, "assets-1")
-
-        # initially c change affects a, b, c
-        assert artifact_set.recompile_group_ids("c") == {"a", "b", "c"}
-
-        # when - b no longer refs c
-        doc_b_v2 = SkillDocument(
-            skill_id="b",
-            content="B standalone now",
-            metadata=make_metadata(tools={}, files=[]),
-        )
-        doc_map = {"a": doc_a, "b": doc_b_v2, "c": doc_c}
-
-        # recompile b (which changes its dependencies)
-        updated_b = compiler.compile_one(artifact_set, doc_b_v2, tree, doc_map)
-        artifact_set.upsert(updated_b)
-
-        # then - c change now only affects c (b no longer depends on c)
-        # note: reverse_graph still has old data until we clean it
-        # in real usage, we'd rebuild graphs or clean stale entries
-        assert "c" not in artifact_set.dependency_graph.get("b", [])
-
-    def test_complex_graph_multiple_changes(self):
-        # given - complex dependency graph
-        #
-        #     A -----> B -----> E
-        #     |        |
-        #     v        v
-        #     C -----> D
-        #
-        # A depends on B, C
-        # B depends on D, E
-        # C depends on D
-        tool_d = ToolReference(uuid="t-d", type=ToolProviderType.BUILT_IN, provider="p", tool_name="tool_d")
-        tool_e = ToolReference(uuid="t-e", type=ToolProviderType.BUILT_IN, provider="p", tool_name="tool_e")
-
-        doc_a = SkillDocument(
-            skill_id="a",
-            content="A §[file].[app].[b]§ §[file].[app].[c]§",
-            metadata=make_metadata(
-                tools={},
-                files=[
-                    FileReference(source="app", asset_id="b"),
-                    FileReference(source="app", asset_id="c"),
-                ],
-            ),
-        )
-        doc_b = SkillDocument(
-            skill_id="b",
-            content="B §[file].[app].[d]§ §[file].[app].[e]§",
-            metadata=make_metadata(
-                tools={},
-                files=[
-                    FileReference(source="app", asset_id="d"),
-                    FileReference(source="app", asset_id="e"),
-                ],
-            ),
-        )
-        doc_c = SkillDocument(
-            skill_id="c",
-            content="C §[file].[app].[d]§",
-            metadata=make_metadata(
-                tools={},
-                files=[FileReference(source="app", asset_id="d")],
-            ),
-        )
-        doc_d = SkillDocument(
-            skill_id="d",
-            content="D §[tool].[p].[tool_d].[t-d]§",
-            metadata=make_metadata(tools={"t-d": tool_d}, files=[]),
-        )
-        doc_e = SkillDocument(
-            skill_id="e",
-            content="E §[tool].[p].[tool_e].[t-e]§",
-            metadata=make_metadata(tools={"t-e": tool_e}, files=[]),
-        )
-
-        tree = create_file_tree(
-            AppAssetNode.create_file("a", "a.md"),
-            AppAssetNode.create_file("b", "b.md"),
-            AppAssetNode.create_file("c", "c.md"),
-            AppAssetNode.create_file("d", "d.md"),
-            AppAssetNode.create_file("e", "e.md"),
-        )
-        compiler = SkillCompiler()
-        artifact_set = compiler.compile_all([doc_a, doc_b, doc_c, doc_d, doc_e], tree, "assets-1")
-
-        # verify initial state
-        a_tools = {t.tool_name for t in artifact_set.get("a").tools.dependencies}
-        assert a_tools == {"tool_d", "tool_e"}
-
-        # when - d changes, who needs recompile?
-        affected_by_d = artifact_set.recompile_group_ids("d")
-        # d is depended by: b, c, and transitively a
-        assert affected_by_d == {"a", "b", "c", "d"}
-        assert "e" not in affected_by_d
-
-        # when - e changes, who needs recompile?
-        affected_by_e = artifact_set.recompile_group_ids("e")
-        # e is depended by: b, and transitively a
-        assert affected_by_e == {"a", "b", "e"}
-        assert "c" not in affected_by_e
-        assert "d" not in affected_by_e
+        assert len(bundle.entries) == 3
+
+        # All three should have access to tool-c (transitive through the cycle)
+        for skill_id in ["skill-a", "skill-b", "skill-c"]:
+            entry = bundle.get(skill_id)
+            assert entry is not None
+            assert len(entry.tools.dependencies) == 1
+            assert entry.tools.dependencies[0].tool_name == "api_tool"
+            assert len(entry.tools.references) == 1
+            assert entry.tools.references[0].uuid == "tool-c"